diff --git a/brand.go b/brand.go index 33cf34d54..c3c39ebe8 100644 --- a/brand.go +++ b/brand.go @@ -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 - } } diff --git a/commands.go b/commands.go deleted file mode 100644 index 73c723d3e..000000000 --- a/commands.go +++ /dev/null @@ -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) -} diff --git a/fragment.go b/fragment.go new file mode 100644 index 000000000..a8c2898a3 --- /dev/null +++ b/fragment.go @@ -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 +} diff --git a/fragment_container.go b/fragment_container.go index c09645593..6ad4d7f9f 100644 --- a/fragment_container.go +++ b/fragment_container.go @@ -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() - } - } - } -} diff --git a/general.go b/general.go index e7f2f3eef..5c96d9672 100644 --- a/general.go +++ b/general.go @@ -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