From 43ede4d9c8efe07e05acbc557d1fa61d5abebb5c Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Tue, 21 Feb 2017 09:28:01 -0600 Subject: [PATCH] WIP increase caching and topn adjustments --- cache.go | 42 ++++++++++++++-------------- fragment.go | 69 ++++++++++++++++++++++++++++------------------ roaring/roaring.go | 13 +++++---- 3 files changed, 71 insertions(+), 53 deletions(-) diff --git a/cache.go b/cache.go index fb5f4defd..c6fc48ec6 100644 --- a/cache.go +++ b/cache.go @@ -14,6 +14,7 @@ import ( // Cache represents a cache for bitmap counts. type Cache interface { Add(bitmapID uint64, n uint64) + BulkAdd(bitmapID uint64, n uint64) Get(bitmapID uint64) uint64 Len() int @@ -22,7 +23,6 @@ type Cache interface { // Updates the cache, if necessary. Invalidate() - Refresh() // Returns an ordered list of the top ranked bitmaps. Top() []BitmapPair @@ -44,6 +44,10 @@ func NewLRUCache(maxEntries int) *LRUCache { return c } +func (c *LRUCache) BulkAdd(bitmapID, n uint64) { + c.Add(bitmapID, n) +} + // Add adds a bitmap to the cache. func (c *LRUCache) Add(bitmapID, n uint64) { c.cache.Add(bitmapID, n) @@ -87,7 +91,7 @@ func (c *LRUCache) Top() []BitmapPair { } func (c *LRUCache) onEvicted(key lru.Key, _ interface{}) { delete(c.counts, key.(uint64)) } -func (c *LRUCache) Refresh() { +func (c *LRUCache) Refresh() { } // Ensure LRUCache implements Cache. @@ -122,18 +126,26 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) { c.entries[bitmapID] = n - + c.Invalidate() // If size is larger than the threshold then trim it. if len(c.entries) > c.ThresholdLength { - c.update() - for id, n := range c.entries { - if n <= c.ThresholdValue { - delete(c.entries, id) - } + for id, n := range c.entries { + if n <= c.ThresholdValue { + delete(c.entries, id) } + } } } +// Add adds a bitmap to the cache unsorted you should Invalidate after completion +func (c *RankCache) BulkAdd(bitmapID uint64, n uint64) { + if n < c.ThresholdValue { + return + } + + c.entries[bitmapID] = n +} + // Get returns a bitmap with a given id. func (c *RankCache) Get(bitmapID uint64) uint64 { return c.entries[bitmapID] } @@ -150,19 +162,9 @@ func (c *RankCache) BitmapIDs() []uint64 { return a } -// Invalidate reorders the entries, if necessary. -func (c *RankCache) Invalidate() { - // Update if there aren't many items or it hasn't been updated recently. - if len(c.rankings) < 50 || (c.updateN > 0 && time.Since(c.updateTime) > 5*time.Minute) { - c.update() - } -} -func (c *RankCache) Refresh() { - c.update() -} - // update reorders the entries by rank. -func (c *RankCache) update() { +func (c *RankCache) Invalidate() { + //fmt.Println("RankCache Update") // Convert cache to a sorted list. rankings := make([]BitmapPair, 0, len(c.entries)) for id, n := range c.entries { diff --git a/fragment.go b/fragment.go index 51cb1dc61..f16bf6ddb 100644 --- a/fragment.go +++ b/fragment.go @@ -53,21 +53,22 @@ const ( //TODO CHANGING FOR TEST TO 10x DefaultFragmentMaxOpN = 2000 ) + type BitmapCacher interface { - Fetch(id uint64)(*Bitmap,bool) + Fetch(id uint64) (*Bitmap, bool) Add(id uint64, b *Bitmap) } +type Simple struct { + cache map[uint64]*Bitmap +} -type Simple struct{ - cache map[uint64]*Bitmap +func (s *Simple) Fetch(id uint64) (*Bitmap, bool) { + m, ok := s.cache[id] + return m, ok } -func (s *Simple)Fetch(id uint64)(*Bitmap,bool){ - m,ok:=s.cache[id] - return m,ok -} -func (s *Simple)Add(id uint64,p*Bitmap){ - s.cache[id]=p +func (s *Simple) Add(id uint64, p *Bitmap) { + s.cache[id] = p } // Fragment represents the intersection of a frame and slice in a database. @@ -105,8 +106,7 @@ type Fragment struct { BitmapAttrStore *AttrStore stats StatsClient - turbo BitmapCacher - + turbo BitmapCacher } // NewFragment returns a new instance of Fragment. @@ -173,6 +173,7 @@ func (f *Fragment) Open() error { // openStorage opens the storage bitmap. func (f *Fragment) openStorage() error { + //f.logger().Printf("Open Storage %s/%s/%d", f.db, f.frame, f.slice) // Create a roaring bitmap to serve as storage for the slice. f.storage = roaring.NewBitmap() @@ -260,9 +261,11 @@ func (f *Fragment) openCache() error { // Read in all bitmaps by ID. // This will cause them to be added to the cache. for _, bitmapID := range pb.BitmapIDs { - n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) - f.cache.Add(bitmapID, n) + //n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) + n := f.bitmap(bitmapID, false).Count() + f.cache.BulkAdd(bitmapID, n) } + f.cache.Invalidate() return nil } @@ -326,15 +329,14 @@ func (f *Fragment) logger() *log.Logger { return log.New(f.LogOutput, "", log.Ls func (f *Fragment) Bitmap(bitmapID uint64) *Bitmap { f.mu.Lock() defer f.mu.Unlock() - return f.bitmap(bitmapID) + return f.bitmap(bitmapID, false) } - -func (f *Fragment) bitmap(bitmapID uint64) *Bitmap { - r,ok:=f.turbo.Fetch(bitmapID) - if ok && r != nil{ - return r - } +func (f *Fragment) bitmap(bitmapID uint64, updateCache bool) *Bitmap { + r, ok := f.turbo.Fetch(bitmapID) + if ok && r != nil { + return r + } // Only use a subset of the containers. // NOTE: The start & end ranges must be divisible by data := f.storage.OffsetRange(f.slice*SliceWidth, bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) @@ -349,9 +351,11 @@ func (f *Fragment) bitmap(bitmapID uint64) *Bitmap { } bm.InvalidateCount() - // Update cache. - f.cache.Add(bitmapID, bm.Count()) - f.turbo.Add(bitmapID, bm) + if updateCache { + // Update cache. + f.cache.Add(bitmapID, bm.Count()) + f.turbo.Add(bitmapID, bm) + } return bm } @@ -391,7 +395,7 @@ func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error) } // Update the cache. - if f.bitmap(bitmapID).SetBit(profileID) { + if f.bitmap(bitmapID, true).SetBit(profileID) { changed = true } @@ -435,7 +439,7 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { } // Update the cache. - if f.bitmap(bitmapID).ClearBit(profileID) { + if f.bitmap(bitmapID, true).ClearBit(profileID) { return true, nil } @@ -570,7 +574,16 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { return results, nil } +func debugDumpPairs(pairs []BitmapPair) { + fmt.Println("=====Start") + for i, pair := range pairs { + fmt.Println(i, pair.ID, pair.Count) + } + fmt.Println("=====Stop") +} + func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair { + //fmt.Println("DEBUG topBitmapPairs") // If no specific bitmaps are requested, retrieve top bitmaps. if len(bitmapIDs) == 0 { f.mu.Lock() @@ -597,6 +610,8 @@ func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair { Count: f.Bitmap(bitmapID).Count(), } } + sort.Sort(BitmapPairs(pairs)) + //debugDumpPairs(pairs) return pairs } @@ -906,10 +921,10 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { // Update cache counts for all bitmaps. for bitmapID := range set { - f.cache.Add(bitmapID, f.bitmap(bitmapID).Count()) + f.cache.BulkAdd(bitmapID, f.bitmap(bitmapID, false).Count()) } - f.cache.Refresh() + f.cache.Invalidate() return nil }(); err != nil { _ = f.closeStorage() diff --git a/roaring/roaring.go b/roaring/roaring.go index d714577da..58b4d1e19 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -467,7 +467,7 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) { buf := make([]byte, headerSize+(containerCount*(4+8+4))) binary.LittleEndian.PutUint32(buf[0:], cookie) binary.LittleEndian.PutUint32(buf[4:], uint32(containerCount)) - empty:=0 + empty := 0 // Encode keys and cardinality. for i, key := range b.keys { c := b.containers[i] @@ -479,20 +479,20 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) { if c.n > 0 { binary.LittleEndian.PutUint64(buf[headerSize+(i-empty)*12:], uint64(key)) binary.LittleEndian.PutUint32(buf[headerSize+(i-empty)*12+8:], uint32(c.n-1)) - }else{ + } else { empty++ } } // Write the offset for each container block. offset := uint32(len(buf)) - empty=0 + empty = 0 for i, c := range b.containers { if c.n > 0 { - binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset)) - }else{ - empty++ + binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset)) + } else { + empty++ } offset += uint32(c.size()) } @@ -1392,6 +1392,7 @@ func unionArrayBitmap(a, b *container) *container { break } else if i >= len(a.array) { output.add(vb) + continue } else if eof { output.add(a.array[i]) i++