scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 16:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

replace command queue with mutex

Browse source 
This commit simplifies Fragment by replacing the request/response channels
with a `sync.Mutex` to restrict access. This also simplifies the
`FragmentContainer` code which can largely be removed now since most of
the methods are simply wrappers around `GetFragment()` and `Fragment`.
This commit is contained in:
Ben Johnson 2015-09-10 17:26:32 -06:00
parent 7048aa519b
commit 26bc85d565
5 changed files with 459 additions and 797 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
brand.go
View file

@ -14,11 +14,7 @@ import (
var FragmentBase string
var globalLock *sync.Mutex
func init() {
globalLock = new(sync.Mutex)
}
var globalLock sync.Mutex
type Pair struct {
Key, Count uint64
@ -199,10 +195,15 @@ func (b *Brand) Stats() interface{} {
"skip": b.skip}
return stats
}
func (b *Brand) Store(bitmap_id uint64, bm *Bitmap, filter uint64) {
b.storage.Store(bitmap_id, b.db, b.frame, b.slice, filter, bm)
func (b *Brand) Store(bitmap_id uint64, bm *Bitmap, filter uint64) error {
if err := b.storage.Store(bitmap_id, b.db, b.frame, b.slice, filter, bm); err != nil {
return err
}
b.cache_it(bm, bitmap_id, filter)
return nil
}
func (b *Brand) checkRank() {
if len(b.rankings) < 50 {
b.Rank()
@ -393,7 +394,7 @@ func (b *Brand) Persist() error {
return encoder.Encode(results)
}
func (b *Brand) Load(requestChan chan Command, f *Fragment) {
func (b *Brand) Load(f *Fragment) {
log.Warn("Brand Load")
time.Sleep(time.Duration(rand.Intn(32)) * time.Second) //trying to avoid mass cassandra hit
r, err := openFile(b.getFileName())
@ -406,15 +407,14 @@ func (b *Brand) Load(requestChan chan Command, f *Fragment) {
if err := dec.Decode(&keys); err != nil {
return
}
globalLock.Lock()
defer globalLock.Unlock()
// probaly need to get a etcd lock too someday
for _, k := range keys {
request := NewLoadRequest(k)
requestChan <- request
request.Response()
b.Get(k)
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) //trying to avoid mass cassandra hit
}
}

372
commands.go
View file

@ -1,372 +0,0 @@
package pilosa
import (
"bytes"
"compress/gzip"
"encoding/base64"
"io/ioutil"
"time"
log "github.com/cihub/seelog"
)
type Result struct {
answer Calculation
exec_time time.Duration
}
type Responder struct {
result chan Result
queryType string
}
func NewResponder(queryType string) *Responder {
return &Responder{
result: make(chan Result),
queryType: queryType,
}
}
func (r *Responder) QueryType() string { return r.queryType }
func (r *Responder) Response() Result { return <-r.result }
func (r *Responder) ResponseChannel() chan Result { return r.result }
type Calculation interface{}
type Command interface {
Execute(*Fragment) Calculation
QueryType() string
Response() Result
ResponseChannel() chan Result
}
type CmdGet struct {
*Responder
bitmap_id uint64
}
func NewGet(bitmap_id uint64) *CmdGet {
return &CmdGet{NewResponder("Get"), bitmap_id}
}
func (self *CmdGet) Execute(f *Fragment) Calculation {
return f.NewHandle(self.bitmap_id)
}
type CmdCount struct {
*Responder
bitmap BitmapHandle
}
func NewCount(bitmap_handle BitmapHandle) *CmdCount {
return &CmdCount{NewResponder("Count"), bitmap_handle}
}
func (self *CmdCount) Execute(f *Fragment) Calculation {
bm, ok := f.getBitmap(self.bitmap)
if ok == false {
return uint64(0)
}
return bm.BitCount()
}
type CmdUnion struct {
*Responder
bitmap_ids []BitmapHandle
}
func NewUnion(bitmaps []BitmapHandle) *CmdUnion {
result := &CmdUnion{NewResponder("Union"), bitmaps}
return result
}
func (self *CmdUnion) Execute(f *Fragment) Calculation {
return f.union(self.bitmap_ids)
}
type CmdDifference struct {
*Responder
bitmap_ids []BitmapHandle
}
func NewDifference(bitmaps []BitmapHandle) *CmdDifference {
result := &CmdDifference{NewResponder("Difference"), bitmaps}
return result
}
func (self *CmdDifference) Execute(f *Fragment) Calculation {
return f.difference(self.bitmap_ids)
}
type CmdIntersect struct {
*Responder
bitmaps []BitmapHandle
}
func NewIntersect(bh []BitmapHandle) *CmdIntersect {
return &CmdIntersect{NewResponder("Intersect"), bh}
}
func (self *CmdIntersect) Execute(f *Fragment) Calculation {
return f.intersect(self.bitmaps)
}
type BitArgs struct {
Bitmap_id uint64
Bit_pos uint64
}
type CmdSetBit struct {
*Responder
bitmap_id uint64
bit_pos uint64
filter uint64
}
func NewSetBit(bitmap_id uint64, bit_pos uint64, filter uint64) *CmdSetBit {
result := &CmdSetBit{NewResponder("SetBit"), bitmap_id, bit_pos, filter}
return result
}
func (self *CmdSetBit) Execute(f *Fragment) Calculation {
return f.impl.SetBit(self.bitmap_id, self.bit_pos, self.filter)
}
type CmdGetBytes struct {
*Responder
bitmap BitmapHandle
}
func NewGetBytes(bh BitmapHandle) *CmdGetBytes {
return &CmdGetBytes{NewResponder("GetBytes"), bh}
}
func (self *CmdGetBytes) Execute(f *Fragment) Calculation {
bm, ok := f.getBitmap(self.bitmap)
//*Compress it
if !ok {
bm = NewBitmap()
log.Warn("cache miss")
}
var b bytes.Buffer
w := gzip.NewWriter(&b)
w.Write(bm.ToBytes())
w.Flush()
w.Close()
return b.Bytes()
}
type CmdFromBytes struct {
*Responder
compressed_bytes []byte
}
func NewFromBytes(bytes []byte) *CmdFromBytes {
return &CmdFromBytes{NewResponder("FromBytes"), bytes}
}
func (self *CmdFromBytes) Execute(f *Fragment) Calculation {
reader, _ := gzip.NewReader(bytes.NewReader(self.compressed_bytes))
b, _ := ioutil.ReadAll(reader)
result := NewBitmap()
result.FromBytes(b)
return f.AllocHandle(result)
}
type CmdEmpty struct {
*Responder
}
func NewEmpty() *CmdEmpty {
return &CmdEmpty{NewResponder("Empty")}
}
func (self *CmdEmpty) Execute(f *Fragment) Calculation {
result := NewBitmap()
return f.AllocHandle(result)
}
type CmdGetList struct {
*Responder
bitmap_ids []uint64
}
func NewGetList(bitmap_ids []uint64) *CmdGetList {
return &CmdGetList{NewResponder("GetList"), bitmap_ids}
}
func (self *CmdGetList) Execute(f *Fragment) Calculation {
ret := make([]BitmapHandle, len(self.bitmap_ids))
for i, v := range self.bitmap_ids {
ret[i] = f.NewHandle(v)
}
return ret
}
type CmdTopN struct {
*Responder
bitmap BitmapHandle
n int
categories []uint64
}
func NewTopN(b BitmapHandle, n int, categories []uint64) *CmdTopN {
return &CmdTopN{NewResponder("TopN"), b, n, categories}
}
func (self *CmdTopN) Execute(f *Fragment) Calculation {
return f.TopN(self.bitmap, self.n, self.categories)
}
type CmdClear struct {
*Responder
}
func NewClear() *CmdClear {
return &CmdClear{NewResponder("Clear")}
}
func (self *CmdClear) Execute(f *Fragment) Calculation {
return f.impl.Clear()
}
type CmdLoader struct {
*Responder
bitmap_id uint64
compressed_bitmap string
filter uint64
}
func NewLoader(bitmap_id uint64, compressed_bitmap string, filter uint64) *CmdLoader {
return &CmdLoader{NewResponder("Loader"), bitmap_id, compressed_bitmap, filter}
}
func (self *CmdLoader) Execute(f *Fragment) Calculation {
buf, err := base64.StdEncoding.DecodeString(self.compressed_bitmap)
if err != nil {
log.Warn(err)
return "ok"
}
reader, _ := gzip.NewReader(bytes.NewReader(buf))
data, _ := ioutil.ReadAll(reader)
bm := NewBitmap()
bm.FromBytes(data)
f.impl.Store(self.bitmap_id, bm, self.filter)
return "ok"
}
type CmdStats struct {
*Responder
}
func NewStats() *CmdStats {
return &CmdStats{NewResponder("Stats")}
}
func (self *CmdStats) Execute(f *Fragment) Calculation {
return f.impl.Stats()
}
type CmdLoadRequest struct {
*Responder
bitmap_id uint64
}
func NewLoadRequest(bitmap_id uint64) *CmdLoadRequest {
result := &CmdLoadRequest{NewResponder("LoadRequest"), bitmap_id}
return result
}
func (self *CmdLoadRequest) Execute(f *Fragment) Calculation {
f.impl.Get(self.bitmap_id)
return 0
}
type CmdRange struct {
*Responder
bitmap_id uint64
start_time time.Time
end_time time.Time
}
func NewRange(bitmap_id uint64, start, end time.Time) *CmdRange {
return &CmdRange{NewResponder("Range"), bitmap_id, start, end}
}
func (self *CmdRange) Execute(f *Fragment) Calculation {
return f.build_time_range_bitmap(self.bitmap_id, self.start_time, self.end_time)
}
type CmdMask struct {
*Responder
start, end uint64
}
func NewMask(start, end uint64) *CmdMask {
return &CmdMask{NewResponder("Mask"), start, end}
}
func (self *CmdMask) Execute(f *Fragment) Calculation {
result := NewBitmap()
for i := self.start; i < self.end; i++ {
result.SetBit(i)
}
return f.AllocHandle(result)
}
type CmdTopFill struct {
*Responder
args FillArgs
}
func NewTopFill(a FillArgs) *CmdTopFill {
return &CmdTopFill{NewResponder("TopFill"), a}
}
func (self *CmdTopFill) Execute(f *Fragment) Calculation {
result := make([]Pair, 0)
for _, v := range self.args.Bitmaps {
if f.exists(v) {
a := f.NewHandle(v)
if self.args.Handle == 0 {
// return just the count
bm, ok := f.getBitmap(a)
if ok && bm.Count() > 0 {
result = append(result, Pair{v, bm.Count()})
}
} else {
res := f.intersect([]BitmapHandle{self.args.Handle, a})
bm, ok := f.getBitmap(res)
if ok {
bc := bm.BitCount()
if bc > 0 {
result = append(result, Pair{v, bc})
}
}
}
}
}
return result
}
type CmdTopNAll struct {
*Responder
n int
categories []uint64
}
func NewTopNAll(n int, categories []uint64) *CmdTopNAll {
return &CmdTopNAll{NewResponder("TopNAll"), n, categories}
}
func (self *CmdTopNAll) Execute(f *Fragment) Calculation {
return f.TopNAll(self.n, self.categories)
}
type CmdClearBit struct {
*Responder
bitmap_id uint64
bit_pos uint64
}
func NewClearBit(bitmap_id uint64, bit_pos uint64) *CmdClearBit {
result := &CmdClearBit{NewResponder("ClearBit"), bitmap_id, bit_pos}
return result
}
func (self *CmdClearBit) Execute(f *Fragment) Calculation {
return f.impl.ClearBit(self.bitmap_id, self.bit_pos)
}

213
fragment.go Normal file
View file

@ -0,0 +1,213 @@
package pilosa
import (
"strings"
"sync"
"time"
log "github.com/cihub/seelog"
"github.com/golang/groupcache/lru"
)
type Fragment struct {
mu sync.Mutex
id SUUID
slice int
impl Pilosa
cache *lru.Cache
// Provide an autoincrementing index for bitmap handles.
seq uint64
// Stats for how many messages have been processed.
stats FragmentStats
}
func NewFragment(id SUUID, db string, slice int, frame string) *Fragment {
storage := NewStorage(Backend, StorageOptions{
DB: db,
Slice: slice,
Frame: frame,
FragmentID: id,
LevelDBPath: LevelDBPath,
})
var impl Pilosa
if strings.HasSuffix(frame, ".n") {
impl = NewBrand(db, frame, slice, storage, 50000, 45000, 100)
} else {
impl = NewGeneral(db, frame, slice, storage)
}
return &Fragment{
id: id,
cache: lru.New(50000),
impl: impl,
slice: slice,
}
}
func (f *Fragment) Bitmap(bh BitmapHandle) (*Bitmap, bool) {
f.mu.Lock()
defer f.mu.Unlock()
return f.bitmap(bh)
}
func (f *Fragment) bitmap(bh BitmapHandle) (*Bitmap, bool) {
bm, ok := f.cache.Get(bh)
if ok && bm != nil {
return bm.(*Bitmap), ok
}
return NewBitmap(), false //cache fail
}
func (f *Fragment) exists(bitmapID uint64) bool {
f.mu.Lock()
defer f.mu.Unlock()
return f.impl.Exists(bitmapID)
}
func (f *Fragment) TopNAll(n int, categories []uint64) []Pair {
f.mu.Lock()
defer f.mu.Unlock()
return f.impl.TopNAll(n, categories)
}
func (f *Fragment) TopN(bitmap BitmapHandle, n int, categories []uint64) []Pair {
f.mu.Lock()
defer f.mu.Unlock()
bm, ok := f.cache.Get(bitmap)
if ok {
return f.impl.TopN(bm.(*Bitmap), n, categories)
}
return nil
}
func (f *Fragment) NewHandle(bitmapID uint64) BitmapHandle {
f.mu.Lock()
defer f.mu.Unlock()
return f.allocHandle(f.impl.Get(bitmapID))
}
func (f *Fragment) AllocHandle(bm *Bitmap) BitmapHandle {
f.mu.Lock()
defer f.mu.Unlock()
return f.allocHandle(bm)
}
func (f *Fragment) allocHandle(bm *Bitmap) BitmapHandle {
handle := f.nextHandle()
f.cache.Add(handle, bm)
return handle
}
func (f *Fragment) nextHandle() BitmapHandle {
millis := uint64(time.Now().UTC().UnixNano())
id := millis << (64 - 41)
id |= uint64(f.slice) << (64 - 41 - 13)
id |= f.seq % 1024
f.seq += 1
return BitmapHandle(id)
}
func (f *Fragment) Union(bitmaps []BitmapHandle) BitmapHandle {
f.mu.Lock()
defer f.mu.Unlock()
result := NewBitmap()
for i, id := range bitmaps {
bm, _ := f.bitmap(id)
if i == 0 {
result = bm
} else {
result = result.Union(bm)
}
}
return f.allocHandle(result)
}
func (f *Fragment) build_time_range_bitmap(bitmapID uint64, start, end time.Time) BitmapHandle {
f.mu.Lock()
defer f.mu.Unlock()
result := NewBitmap()
for i, bid := range GetRange(start, end, bitmapID) {
bm := f.impl.Get(bid)
if i == 0 {
result = bm
} else {
result = result.Union(bm)
}
}
return f.AllocHandle(result)
}
func (f *Fragment) Intersect(bitmaps []BitmapHandle) BitmapHandle {
f.mu.Lock()
defer f.mu.Unlock()
var result *Bitmap
for i, id := range bitmaps {
bm, _ := f.bitmap(id)
if i == 0 {
result = bm.Clone()
} else {
result = result.Intersection(bm)
}
}
return f.allocHandle(result)
}
func (f *Fragment) Difference(bitmaps []BitmapHandle) BitmapHandle {
f.mu.Lock()
defer f.mu.Unlock()
result := NewBitmap()
for i, id := range bitmaps {
bm, _ := f.bitmap(id)
if i == 0 {
result = bm
} else {
result = result.Difference(bm)
}
}
return f.allocHandle(result)
}
func (f *Fragment) Persist() {
f.mu.Lock()
defer f.mu.Unlock()
err := f.impl.Persist()
if err != nil {
log.Warn("Error saving:", err)
}
}
func (f *Fragment) Load() {
f.mu.Lock()
defer f.mu.Unlock()
f.impl.Load(f)
}
// FragmentStats represents in-memory stats for a single fragment.
type FragmentStats struct {
// Messages processed by the fragment
ProcessN uint64
ProcessTime time.Duration
}
type Pilosa interface {
Get(id uint64) *Bitmap
SetBit(id uint64, bit_pos uint64, filter uint64) bool
ClearBit(id uint64, bit_pos uint64) bool
TopN(b *Bitmap, n int, categories []uint64) []Pair
TopNAll(n int, categories []uint64) []Pair
Clear() bool
Store(bitmapID uint64, bm *Bitmap, filter uint64) error
Stats() interface{}
Persist() error
Load(fragment *Fragment)
Exists(id uint64) bool
}

632
fragment_container.go
View file

@ -1,17 +1,19 @@
package pilosa
import (
"bytes"
"compress/gzip"
"encoding/base64"
"encoding/gob"
"errors"
"fmt"
"strings"
"io/ioutil"
"sync"
"syscall"
"time"
log "github.com/cihub/seelog"
"github.com/golang/groupcache/lru"
"github.com/umbel/pilosa/statsd"
// "github.com/umbel/pilosa/statsd"
)
// DefaultBackend is the default data storage layer.
@ -21,179 +23,160 @@ var Backend = DefaultBackend
var LevelDBPath string
var (
ErrFragmentNotFound = errors.New("fragment not found")
)
func init() {
gob.Register(BitmapHandle(0))
gob.Register([]Pair{})
}
type FragmentContainer struct {
mu sync.Mutex
fragments map[SUUID]*Fragment
mutex *sync.Mutex
}
func NewFragmentContainer() *FragmentContainer {
f := new(FragmentContainer)
f.fragments = make(map[SUUID]*Fragment)
f.mutex = &sync.Mutex{}
return f
return &FragmentContainer{
fragments: make(map[SUUID]*Fragment),
}
}
type BitmapHandle uint64
type FillArgs struct {
Frag_id SUUID
Handle BitmapHandle
Bitmaps []uint64
FragmentID SUUID
Handle BitmapHandle
Bitmaps []uint64
}
func init() {
var vh BitmapHandle
gob.Register(vh)
var lp []Pair
gob.Register(lp)
func (fc *FragmentContainer) Fragment(fragmentID SUUID) (*Fragment, bool) {
fc.mu.Lock()
f, ok := fc.fragments[fragmentID]
fc.mu.Unlock()
return f, ok
}
func (self *FragmentContainer) Shutdown() {
log.Warn("Container Shutdown Started")
var wg sync.WaitGroup
wg.Add(len(self.fragments))
for _, v := range self.fragments {
v.exit <- &wg
func (fc *FragmentContainer) Get(fragmentID SUUID, bitmapID uint64) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
wg.Wait()
log.Warn("Container Shutdown Complete")
return f.NewHandle(bitmapID), nil
}
func (self *FragmentContainer) LoadBitmap(frag_id SUUID, bitmap_id uint64, compressed_bitmap string, filter uint64) {
log.Trace("LoadBitmap")
if fragment, found := self.GetFragment(frag_id); found {
request := NewLoader(bitmap_id, compressed_bitmap, filter)
fragment.requestChan <- request
request.Response()
return
func (fc *FragmentContainer) LoadBitmap(fragmentID SUUID, bitmapID uint64, data string, filter uint64) error {
f, ok := fc.Fragment(fragmentID)
if !ok {
return ErrFragmentNotFound
}
// Decode from base64 encoding.
buf, err := base64.StdEncoding.DecodeString(data)
if err != nil {
return err
}
// Decompress data.
reader, err := gzip.NewReader(bytes.NewReader(buf))
if err != nil {
return err
}
b, err := ioutil.ReadAll(reader)
if err != nil {
return err
}
// Build bitmap from data.
bm := NewBitmap()
bm.FromBytes(b)
// Write bitmap to the underlying store.
return f.impl.Store(bitmapID, bm, filter)
}
func (self *FragmentContainer) GetFragment(frag_id SUUID) (*Fragment, bool) {
log.Trace("index.GetFragment")
self.mutex.Lock()
c, v := self.fragments[frag_id]
self.mutex.Unlock()
return c, v
func (fc *FragmentContainer) Stats(fragmentID SUUID) interface{} {
f, ok := fc.Fragment(fragmentID)
if !ok {
return nil
}
return f.impl.Stats()
}
func (self *FragmentContainer) Stats(frag_id SUUID) interface{} {
if fragment, found := self.GetFragment(frag_id); found {
request := NewStats()
fragment.requestChan <- request
return request.Response().answer
func (fc *FragmentContainer) Empty(fragmentID SUUID) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return nil
}
func (self *FragmentContainer) Empty(frag_id SUUID) (BitmapHandle, error) {
log.Trace("index.Empty")
if fragment, found := self.GetFragment(frag_id); found {
request := NewEmpty()
fragment.requestChan <- request
return request.Response().answer.(BitmapHandle), nil
}
return 0, errors.New("Invalid Bitmap Handle Empty")
return f.AllocHandle(NewBitmap()), nil
}
func (self *FragmentContainer) Intersect(frag_id SUUID, bh []BitmapHandle) (BitmapHandle, error) {
log.Trace("index.Intersect")
if fragment, found := self.GetFragment(frag_id); found {
request := NewIntersect(bh)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_Intersect", result.exec_time.Nanoseconds())
return result.answer.(BitmapHandle), nil
func (fc *FragmentContainer) Intersect(fragmentID SUUID, bh []BitmapHandle) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return 0, errors.New("Invalid Bitmap Handle Intersect")
}
func (self *FragmentContainer) Union(frag_id SUUID, bh []BitmapHandle) (BitmapHandle, error) {
log.Trace("index.Union")
if fragment, found := self.GetFragment(frag_id); found {
request := NewUnion(bh)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_Union", result.exec_time.Nanoseconds())
return result.answer.(BitmapHandle), nil
}
return 0, errors.New("Invalid Bitmap Handle Union")
return f.Intersect(bh), nil
}
func (self *FragmentContainer) Difference(frag_id SUUID, bh []BitmapHandle) (BitmapHandle, error) {
log.Trace("index.Difference")
if fragment, found := self.GetFragment(frag_id); found {
request := NewDifference(bh)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_Difference", result.exec_time.Nanoseconds())
return result.answer.(BitmapHandle), nil
func (fc *FragmentContainer) Union(fragmentID SUUID, bh []BitmapHandle) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return 0, errors.New("Invalid Bitmap Handle Diff")
return f.Union(bh), nil
}
func (self *FragmentContainer) Get(frag_id SUUID, bitmap_id uint64) (BitmapHandle, error) {
log.Trace("index.Get")
if fragment, found := self.GetFragment(frag_id); found {
request := NewGet(bitmap_id)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_Get", result.exec_time.Nanoseconds())
return result.answer.(BitmapHandle), nil
func (fc *FragmentContainer) Difference(fragmentID SUUID, bh []BitmapHandle) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return 0, errors.New("Invalid Bitmap Handle Get")
return f.Difference(bh), nil
}
func (self *FragmentContainer) Mask(frag_id SUUID, start, end uint64) (BitmapHandle, error) {
if fragment, found := self.GetFragment(frag_id); found {
request := NewMask(start, end)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_Mask", result.exec_time.Nanoseconds())
return result.answer.(BitmapHandle), nil
func (fc *FragmentContainer) Mask(fragmentID SUUID, start, end uint64) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return 0, errors.New("Invalid Bitmap Handle")
bm := NewBitmap()
for i := start; i < end; i++ {
bm.SetBit(i)
}
return f.AllocHandle(bm), nil
}
func (self *FragmentContainer) Range(frag_id SUUID, bitmap_id uint64, start, end time.Time) (BitmapHandle, error) {
if fragment, found := self.GetFragment(frag_id); found {
request := NewRange(bitmap_id, start, end)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_Range", result.exec_time.Nanoseconds())
return result.answer.(BitmapHandle), nil
func (fc *FragmentContainer) Range(fragmentID SUUID, bitmapID uint64, start, end time.Time) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return 0, errors.New("Invalid Bitmap Handle")
return f.build_time_range_bitmap(bitmapID, start, end), nil
}
func (self *FragmentContainer) TopN(frag_id SUUID, bh BitmapHandle, n int, categories []uint64) ([]Pair, error) {
log.Trace("index.TopN")
if fragment, found := self.GetFragment(frag_id); found {
request := NewTopN(bh, n, categories)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_TopN", result.exec_time.Nanoseconds())
return result.answer.([]Pair), nil
func (fc *FragmentContainer) TopN(fragmentID SUUID, bh BitmapHandle, n int, categories []uint64) ([]Pair, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return nil, ErrFragmentNotFound
}
return nil, errors.New(fmt.Sprintf("Fragment not found:%s", frag_id.String()))
return f.TopN(bh, n, categories), nil
}
func (self *FragmentContainer) TopNAll(frag_id SUUID, n int, categories []uint64) ([]Pair, error) {
log.Trace("index.TopNAll")
if fragment, found := self.GetFragment(frag_id); found {
request := NewTopNAll(n, categories)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_TopNAll", result.exec_time.Nanoseconds())
return result.answer.([]Pair), nil
func (fc *FragmentContainer) TopNAll(fragmentID SUUID, n int, categories []uint64) ([]Pair, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return nil, ErrFragmentNotFound
}
return nil, errors.New(fmt.Sprintf("Fragment not found:%s", frag_id.String()))
return f.TopNAll(n, categories), nil
}
func (self *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) {
log.Trace("index.TopFillBatch")
//should probaly make this concurrent but then all hell breaks loose
func (fc *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) {
results := make(map[uint64]uint64)
for _, v := range args {
items, _ := self.TopFillFragment(v)
items, _ := fc.TopFillFragment(v)
if len(args) == 1 {
return items, nil
}
@ -210,100 +193,143 @@ func (self *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) {
return ret_val, nil
}
func (self *FragmentContainer) TopFillFragment(arg FillArgs) ([]Pair, error) {
log.Trace("index.TopFillFragment")
if fragment, found := self.GetFragment(arg.Frag_id); found {
request := NewTopFill(arg)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_TopFillFragment", result.exec_time.Nanoseconds())
return result.answer.([]Pair), nil
func (fc *FragmentContainer) TopFillFragment(args FillArgs) ([]Pair, error) {
f, ok := fc.Fragment(args.FragmentID)
if !ok {
return nil, ErrFragmentNotFound
}
return nil, errors.New("Invalid Bitmap Handle")
result := make([]Pair, 0)
for _, v := range args.Bitmaps {
if !f.exists(v) {
continue
}
a := f.NewHandle(v)
if args.Handle == 0 {
// Return just the count
if bm, ok := f.Bitmap(a); ok && bm.Count() > 0 {
result = append(result, Pair{v, bm.Count()})
}
continue
}
res := f.Intersect([]BitmapHandle{args.Handle, a})
if bm, ok := f.Bitmap(res); ok {
bc := bm.BitCount()
if bc > 0 {
result = append(result, Pair{v, bc})
}
}
}
return result, nil
}
func (self *FragmentContainer) GetList(frag_id SUUID, bitmap_id []uint64) ([]BitmapHandle, error) {
log.Trace("index.GetList")
if fragment, found := self.GetFragment(frag_id); found {
request := NewGetList(bitmap_id)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_GetList", result.exec_time.Nanoseconds())
return result.answer.([]BitmapHandle), nil
func (fc *FragmentContainer) GetList(fragmentID SUUID, bitmapIDs []uint64) ([]BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return nil, ErrFragmentNotFound
}
return nil, errors.New("Invalid Bitmap Handle GetList")
a := make([]BitmapHandle, len(bitmapIDs))
for i, v := range bitmapIDs {
a[i] = f.NewHandle(v)
}
return a, nil
}
func (self *FragmentContainer) Count(frag_id SUUID, bitmap BitmapHandle) (uint64, error) {
log.Trace("index.Count")
if fragment, found := self.GetFragment(frag_id); found {
request := NewCount(bitmap)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_Count", result.exec_time.Nanoseconds())
return result.answer.(uint64), nil
func (fc *FragmentContainer) Count(fragmentID SUUID, bh BitmapHandle) (uint64, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return 0, errors.New("Invalid Bitmap Handle Count")
bm, ok := f.Bitmap(bh)
if ok == false {
return 0, nil
}
return bm.BitCount(), nil
}
func (self *FragmentContainer) GetBytes(frag_id SUUID, bh BitmapHandle) ([]byte, error) {
log.Trace("index.GetBytes")
if fragment, found := self.GetFragment(frag_id); found {
request := NewGetBytes(bh)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_GetBytes", result.exec_time.Nanoseconds())
return result.answer.([]byte), nil
func (fc *FragmentContainer) GetBytes(fragmentID SUUID, bh BitmapHandle) ([]byte, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return nil, ErrFragmentNotFound
}
return nil, errors.New("Invalid Bitmap Handle GetBytes")
bm, ok := f.Bitmap(bh)
if !ok {
bm = NewBitmap()
log.Warn("cache miss")
}
var b bytes.Buffer
w := gzip.NewWriter(&b)
if _, err := w.Write(bm.ToBytes()); err != nil {
return nil, err
}
if err := w.Flush(); err != nil {
return nil, err
}
if err := w.Close(); err != nil {
return nil, err
}
return b.Bytes(), nil
}
func (self *FragmentContainer) FromBytes(frag_id SUUID, bytes []byte) (BitmapHandle, error) {
log.Trace("index.FromBytes")
if fragment, found := self.GetFragment(frag_id); found {
request := NewFromBytes(bytes)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_FromBytes", result.exec_time.Nanoseconds())
return result.answer.(BitmapHandle), nil
func (fc *FragmentContainer) FromBytes(fragmentID SUUID, data []byte) (BitmapHandle, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return 0, ErrFragmentNotFound
}
return 0, errors.New("Invalid Bitmap Handle FromBytes")
r, _ := gzip.NewReader(bytes.NewReader(data))
b, _ := ioutil.ReadAll(r)
bm := NewBitmap()
bm.FromBytes(b)
return f.AllocHandle(bm), nil
}
func (self *FragmentContainer) SetBit(frag_id SUUID, bitmap_id uint64, pos uint64, category uint64) (bool, error) {
log.Trace("SetBit", frag_id, bitmap_id, pos, category)
if fragment, found := self.GetFragment(frag_id); found {
request := NewSetBit(bitmap_id, pos, category)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_SetBit", result.exec_time.Nanoseconds())
statsd.SendInc("fragmant_container_SetBit")
return result.answer.(bool), nil
func (fc *FragmentContainer) SetBit(fragmentID SUUID, bitmapID uint64, pos uint64, category uint64) (bool, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return false, ErrFragmentNotFound
}
return false, errors.New("Invalid Bitmap Handle SetBit")
return f.impl.SetBit(bitmapID, pos, category), nil
}
func (self *FragmentContainer) ClearBit(frag_id SUUID, bitmap_id uint64, pos uint64) (bool, error) {
log.Trace("ClearBit", frag_id, bitmap_id, pos)
if fragment, found := self.GetFragment(frag_id); found {
request := NewClearBit(bitmap_id, pos)
fragment.requestChan <- request
result := request.Response()
statsd.SendTimer("fragmant_container_ClearBit", result.exec_time.Nanoseconds())
statsd.SendInc("fragmant_container_ClearBit")
return result.answer.(bool), nil
func (fc *FragmentContainer) ClearBit(fragmentID SUUID, bitmapID uint64, pos uint64) (bool, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return false, ErrFragmentNotFound
}
return false, errors.New("Invalid Bitmap Handle ClearBit")
return f.impl.ClearBit(bitmapID, pos), nil
}
func (self *FragmentContainer) Clear(frag_id SUUID) (bool, error) {
if fragment, found := self.GetFragment(frag_id); found {
request := NewClear()
fragment.requestChan <- request
return request.Response().answer.(bool), nil
func (fc *FragmentContainer) Clear(fragmentID SUUID) (bool, error) {
f, ok := fc.Fragment(fragmentID)
if !ok {
return false, ErrFragmentNotFound
}
return false, errors.New("Invalid Fragment ID")
return f.impl.Clear(), nil
}
func (fc *FragmentContainer) AddFragment(db string, frame string, slice int, id SUUID) {
fc.mu.Lock()
defer fc.mu.Unlock()
_, ok := fc.fragments[id]
if ok {
return
}
log.Warn("ADD FRAGMENT", frame, db, slice, id.String())
f := NewFragment(id, db, slice, frame)
fc.fragments[id] = f
// go f.Load(loader)
}
func dumpHandlesToLog() {
var limit syscall.Rlimit
if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &limit); err != nil {
@ -312,209 +338,3 @@ func dumpHandlesToLog() {
mesg := fmt.Sprintf("%v file descriptors out of a maximum of %v available\n", limit.Cur, limit.Max)
log.Warn(mesg)
}
func (self *FragmentContainer) AddFragment(db string, frame string, slice int, id SUUID) {
self.mutex.Lock()
defer self.mutex.Unlock()
_, ok := self.fragments[id]
if !ok {
// dumpHandlesToLog()
log.Warn("ADD FRAGMENT", frame, db, slice, id.String())
f := NewFragment(id, db, slice, frame)
loader := make(chan Command)
self.fragments[id] = f
go f.ServeFragment(loader)
go f.Load(loader)
}
}
type Pilosa interface {
Get(id uint64) *Bitmap
SetBit(id uint64, bit_pos uint64, filter uint64) bool
ClearBit(id uint64, bit_pos uint64) bool
TopN(b *Bitmap, n int, categories []uint64) []Pair
TopNAll(n int, categories []uint64) []Pair
Clear() bool
Store(bitmap_id uint64, bm *Bitmap, filter uint64)
Stats() interface{}
Persist() error
Load(requestChan chan Command, fragment *Fragment)
Exists(id uint64) bool
}
type Fragment struct {
requestChan chan Command
fragment_id SUUID
impl Pilosa
counter uint64
slice int
cache *lru.Cache
mesg_count uint64
mesg_time time.Duration
exit chan *sync.WaitGroup
queue_size int
}
func NewFragment(frag_id SUUID, db string, slice int, frame string) *Fragment {
var impl Pilosa
log.Warn(fmt.Sprintf("XXXXXXXXXXXXXXXXXXXXXXXXXXX(%s)", frame))
storage := NewStorage(Backend, StorageOptions{
DB: db,
Slice: slice,
Frame: frame,
FragmentID: frag_id,
LevelDBPath: LevelDBPath,
})
if strings.HasSuffix(frame, ".n") {
impl = NewBrand(db, frame, slice, storage, 50000, 45000, 100)
} else {
impl = NewGeneral(db, frame, slice, storage)
}
f := new(Fragment)
f.requestChan = make(chan Command, 64)
f.fragment_id = frag_id
f.cache = lru.New(50000)
f.impl = impl //NewGeneral(db, slice, NewMemoryStorage())
f.slice = slice
f.exit = make(chan *sync.WaitGroup)
f.queue_size = 0
return f
}
func (self *Fragment) getBitmap(bitmap BitmapHandle) (*Bitmap, bool) {
bm, ok := self.cache.Get(bitmap)
if ok && bm != nil {
return bm.(*Bitmap), ok
}
return NewBitmap(), false //cache fail but return ting em
}
func (self *Fragment) exists(bitmap_id uint64) bool {
return self.impl.Exists(bitmap_id)
}
func (self *Fragment) TopNAll(n int, categories []uint64) []Pair {
return self.impl.TopNAll(n, categories)
}
func (self *Fragment) TopN(bitmap BitmapHandle, n int, categories []uint64) []Pair {
bm, ok := self.cache.Get(bitmap)
if ok {
return self.impl.TopN(bm.(*Bitmap), n, categories)
}
return nil
}
func (self *Fragment) NewHandle(bitmap_id uint64) BitmapHandle {
bm := self.impl.Get(bitmap_id)
return self.AllocHandle(bm)
//given a bitmap_id return a newly allocated handle
}
func (self *Fragment) AllocHandle(bm *Bitmap) BitmapHandle {
handle := self.nextHandle()
self.cache.Add(handle, bm)
return handle
}
func (self *Fragment) nextHandle() BitmapHandle {
millis := uint64(time.Now().UTC().UnixNano())
id := millis << (64 - 41)
id |= uint64(self.slice) << (64 - 41 - 13)
id |= self.counter % 1024
self.counter += 1
return BitmapHandle(id)
}
func (self *Fragment) union(bitmaps []BitmapHandle) BitmapHandle {
result := NewBitmap()
for i, id := range bitmaps {
bm, _ := self.getBitmap(id)
if i == 0 {
result = bm
} else {
result = result.Union(bm)
}
}
return self.AllocHandle(result)
}
func (self *Fragment) build_time_range_bitmap(bitmap_id uint64, start, end time.Time) BitmapHandle {
result := NewBitmap()
for i, bid := range GetRange(start, end, bitmap_id) {
bm := self.impl.Get(bid)
if i == 0 {
result = bm
} else {
result = result.Union(bm)
}
}
return self.AllocHandle(result)
}
func (self *Fragment) intersect(bitmaps []BitmapHandle) BitmapHandle {
var result *Bitmap
for i, id := range bitmaps {
bm, _ := self.getBitmap(id)
if i == 0 {
result = bm.Clone()
} else {
result = result.Intersection(bm)
}
}
return self.AllocHandle(result)
}
func (self *Fragment) difference(bitmaps []BitmapHandle) BitmapHandle {
result := NewBitmap()
for i, id := range bitmaps {
bm, _ := self.getBitmap(id)
if i == 0 {
result = bm
} else {
result = result.Difference(bm)
}
}
return self.AllocHandle(result)
}
func (self *Fragment) Persist() {
err := self.impl.Persist()
if err != nil {
log.Warn("Error saving:", err)
}
}
func (self *Fragment) Load(loadChan chan Command) {
self.impl.Load(loadChan, self)
}
func (self *Fragment) processCommand(req Command) {
self.mesg_count++
start := time.Now()
answer := req.Execute(self)
delta := time.Since(start)
self.mesg_count += 1
self.mesg_time += delta
req.ResponseChannel() <- Result{answer, delta}
}
func (self *Fragment) ServeFragment(loadChan chan Command) {
for {
select {
case req := <-self.requestChan:
self.processCommand(req)
default:
select {
case req := <-self.requestChan:
self.processCommand(req)
case req := <-loadChan:
self.processCommand(req)
case wg := <-self.exit:
log.Warn("Fragment Shutdown")
self.Persist()
wg.Done()
}
}
}
}

15
general.go
View file

@ -63,11 +63,15 @@ func (self *General) TopN(b *Bitmap, n int, categories []uint64) []Pair {
return empty
}
func (self *General) Store(bitmap_id uint64, bm *Bitmap, filter uint64) {
self.storage.Store(bitmap_id, self.db, self.frame, self.slice, filter, bm)
func (self *General) Store(bitmap_id uint64, bm *Bitmap, filter uint64) error {
if err := self.storage.Store(bitmap_id, self.db, self.frame, self.slice, filter, bm); err != nil {
return err
}
self.bitmap_cache.Add(bitmap_id, bm)
self.keys[bitmap_id] = 0
return nil
}
func (self *General) OnEvicted(key lru.Key, value interface{}) {
delete(self.keys, key.(uint64))
}
@ -108,7 +112,7 @@ func (self *General) Persist() error {
return encoder.Encode(results)
}
func (self *General) Load(requestChan chan Command, f *Fragment) {
func (self *General) Load(f *Fragment) {
log.Warn("General Load")
r, err := openFile(self.getFileName())
if err != nil {
@ -123,14 +127,11 @@ func (self *General) Load(requestChan chan Command, f *Fragment) {
return
}
for _, k := range keys {
request := NewLoadRequest(k)
requestChan <- request
request.Response()
self.Get(k)
}
}
func (self *General) TopNAll(n int, categories []uint64) []Pair {
results := make([]Pair, 0, 0)
count := 0