Merge remote-tracking branch 'upstream/umbel-perf-improvements' into umbel-perf-improvements

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
 
@@ -43,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)
@@ -86,6 +91,8 @@ func (c *LRUCache) Top() []BitmapPair {
 }
 
 func (c *LRUCache) onEvicted(key lru.Key, _ interface{}) { delete(c.counts, key.(uint64)) }
+func (c *LRUCache) Refresh() {
+}
 
 // Ensure LRUCache implements Cache.
 var _ Cache = &LRUCache{}
@@ -117,12 +124,11 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) {
 		return
 	}
 
-	// Add to cache.
 	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)
@@ -131,6 +137,15 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) {
 	}
 }
 
+// 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] }
 
@@ -147,16 +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()
-	}
-}
-
 // 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 {

db.go

@@ -426,8 +426,9 @@ func (p dbSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() }
 
 // DBInfo represents schema information for a database.
 type DBInfo struct {
-	Name   string       `json:"name"`
-	Frames []*FrameInfo `json:"frames"`
+	Name     string       `json:"name"`
+	Frames   []*FrameInfo `json:"frames"`
+	MaxSlice uint64
 }
 
 type dbInfoSlice []*DBInfo

executor.go

@@ -281,7 +281,7 @@ func (e *Executor) executeBitmapSlice(ctx context.Context, db string, c *pql.Bit
 
 // executeIntersectSlice executes a intersect() call for a local slice.
 func (e *Executor) executeIntersectSlice(ctx context.Context, db string, c *pql.Intersect, slice uint64) (*Bitmap, error) {
-	var other *Bitmap
+	other := &Bitmap{}
 	for i, input := range c.Inputs {
 		bm, err := e.executeBitmapCallSlice(ctx, db, input, slice)
 		if err != nil {
@@ -331,7 +331,7 @@ func (e *Executor) executeRangeSlice(ctx context.Context, db string, c *pql.Rang
 
 // executeUnionSlice executes a union() call for a local slice.
 func (e *Executor) executeUnionSlice(ctx context.Context, db string, c *pql.Union, slice uint64) (*Bitmap, error) {
-	var other *Bitmap
+	other := &Bitmap{}
 	for i, input := range c.Inputs {
 		bm, err := e.executeBitmapCallSlice(ctx, db, input, slice)
 		if err != nil {

fragment.go

@@ -28,7 +28,8 @@ import (
 
 const (
 	// SliceWidth is the number of profile IDs in a slice.
-	SliceWidth = 1048576
+	//SliceWidth = 1048576
+	SliceWidth = 262144
 
 	// SnapshotExt is the file extension used for an in-process snapshot.
 	SnapshotExt = ".snapshotting"
@@ -49,9 +50,27 @@ const (
 
 const (
 	// DefaultFragmentMaxOpN is the default value for Fragment.MaxOpN.
-	DefaultFragmentMaxOpN = 1000
+	//TODO CHANGING FOR TEST TO 10x
+	DefaultFragmentMaxOpN = 2000
 )
 
+type BitmapCacher interface {
+	Fetch(id uint64) (*Bitmap, bool)
+	Add(id uint64, b *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) Add(id uint64, p *Bitmap) {
+	s.cache[id] = p
+}
+
 // Fragment represents the intersection of a frame and slice in a database.
 type Fragment struct {
 	mu sync.Mutex
@@ -87,6 +106,7 @@ type Fragment struct {
 	BitmapAttrStore *AttrStore
 
 	stats StatsClient
+	turbo BitmapCacher
 }
 
 // NewFragment returns a new instance of Fragment.
@@ -153,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()
 
@@ -203,6 +224,7 @@ func (f *Fragment) openStorage() error {
 
 	// Attach the file to the bitmap to act as a write-ahead log.
 	f.storage.OpWriter = f.file
+	f.turbo = &Simple{make(map[uint64]*Bitmap)}
 
 	return nil
 
@@ -239,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
 }
@@ -305,10 +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 {
+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)
@@ -323,8 +351,11 @@ func (f *Fragment) bitmap(bitmapID uint64) *Bitmap {
 	}
 	bm.InvalidateCount()
 
-	// Update cache.
-	f.cache.Add(bitmapID, bm.Count())
+	if updateCache {
+		// Update cache.
+		f.cache.Add(bitmapID, bm.Count())
+		f.turbo.Add(bitmapID, bm)
+	}
 
 	return bm
 }
@@ -364,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
 	}
 
@@ -408,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
 	}
 
@@ -543,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()
@@ -570,6 +610,8 @@ func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair {
 			Count: f.Bitmap(bitmapID).Count(),
 		}
 	}
+	sort.Sort(BitmapPairs(pairs))
+	//debugDumpPairs(pairs)
 	return pairs
 }
 
@@ -879,7 +921,7 @@ 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.Invalidate()
@@ -918,10 +960,15 @@ func (f *Fragment) Snapshot() error {
 	defer f.mu.Unlock()
 	return f.snapshot()
 }
+func track(start time.Time, name string, logger *log.Logger) {
+	elapsed := time.Since(start)
+	logger.Printf("%s took %s", name, elapsed)
+}
 
 func (f *Fragment) snapshot() error {
 	logger := f.logger()
 	logger.Printf("fragment: snapshotting %s/%s/%d", f.db, f.frame, f.slice)
+	defer track(time.Now(), fmt.Sprintf("fragment: snapshot complete %s/%s/%d", f.db, f.frame, f.slice), logger)
 
 	// Create a temporary file to snapshot to.
 	snapshotPath := f.path + SnapshotExt

roaring/roaring.go

@@ -444,17 +444,30 @@ func (b *Bitmap) removeEmptyContainers() {
 		i++
 	}
 }
+func (b *Bitmap) countEmptyContainers() int {
+	result := 0
+	for i := 0; i < len(b.containers); {
+		c := b.containers[i]
+
+		if c.n == 0 {
+			result++
+		}
+		i++
+	}
+	return result
+}
 
 // WriteTo writes b to w.
 func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 	// Remove empty containers before persisting.
-	b.removeEmptyContainers()
+	//b.removeEmptyContainers()
+	containerCount := len(b.keys) - b.countEmptyContainers()
 
 	// Build header before writing individual container blocks.
-	buf := make([]byte, headerSize+(len(b.keys)*(4+8+4)))
+	buf := make([]byte, headerSize+(containerCount*(4+8+4)))
 	binary.LittleEndian.PutUint32(buf[0:], cookie)
-	binary.LittleEndian.PutUint32(buf[4:], uint32(len(b.keys)))
-
+	binary.LittleEndian.PutUint32(buf[4:], uint32(containerCount))
+	empty := 0
 	// Encode keys and cardinality.
 	for i, key := range b.keys {
 		c := b.containers[i]
@@ -463,15 +476,24 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 		// TODO: instead of commenting this out, we need to make it a configuration option
 		//count := c.count()
 		//assert(c.count() == c.n, "cannot write container count, mismatch: count=%d, n=%d", count, c.n)
-
-		binary.LittleEndian.PutUint64(buf[headerSize+i*12:], uint64(key))
-		binary.LittleEndian.PutUint32(buf[headerSize+i*12+8:], uint32(c.n-1))
+		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 {
+			empty++
+		}
 	}
 
 	// Write the offset for each container block.
 	offset := uint32(len(buf))
+	empty = 0
 	for i, c := range b.containers {
-		binary.LittleEndian.PutUint32(buf[headerSize+(len(b.keys)*12)+(i*4):], uint32(offset))
+
+		if c.n > 0 {
+			binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset))
+		} else {
+			empty++
+		}
 		offset += uint32(c.size())
 	}
 
@@ -484,10 +506,12 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 
 	// Write each container block.
 	for _, c := range b.containers {
-		nn, err := c.WriteTo(w)
-		n += nn
-		if err != nil {
-			return n, err
+		if c.n > 0 {
+			nn, err := c.WriteTo(w)
+			n += nn
+			if err != nil {
+				return n, err
+			}
 		}
 	}