first pass at merging the umbel-perf-improvements pass into master

attr.go

@@ -265,6 +265,8 @@ func txUpdateAttrs(tx *bolt.Tx, id uint64, m map[string]interface{}) (map[string
 			attr[k] = uint64(v)
 		case uint:
 			attr[k] = uint64(v)
+		case float64:
+			attr[k] = uint64(v)
 		case int64:
 			attr[k] = uint64(v)
 		case string, uint64, bool:

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 {
@@ -233,6 +241,24 @@ func (p Pairs) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
 func (p Pairs) Len() int           { return len(p) }
 func (p Pairs) Less(i, j int) bool { return p[i].Count > p[j].Count }
 
+type PairHeap struct {
+	Pairs
+}
+
+func (h *Pairs) Push(x interface{}) {
+	// Push and Pop use pointer receivers because they modify the slice's length,
+	// not just its contents.
+	*h = append(*h, x.(Pair))
+}
+
+func (h *Pairs) Pop() interface{} {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[0 : n-1]
+	return x
+}
+
 // Add merges other into p and returns a new slice.
 func (p Pairs) Add(other []Pair) []Pair {
 	// Create lookup of key/counts.

cmd/pilosa/main.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
+	"net/http"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -40,6 +41,10 @@ const (
 )
 
 func main() {
+	// Limit the number of connections that a server can make to a single node
+	// in order to prevent a cluster storm.
+	http.DefaultTransport.(*http.Transport).MaxIdleConnsPerHost = 64
+
 	m := NewMain()
 	m.Server.Handler.Version = Version
 	fmt.Fprintf(m.Stderr, "Pilosa %s, build time %s\n", Version, BuildTime)

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

@@ -52,13 +52,15 @@ func (e *Executor) Execute(ctx context.Context, db string, q *pql.Query, slices
 
 	// If slices aren't specified, then include all of them.
 	if len(slices) == 0 {
-		// Round up the number of slices.
-		maxSlice := e.Index.DB(db).MaxSlice()
+		if needsSlices(q.Calls) {
+			// Round up the number of slices.
+			maxSlice := e.Index.DB(db).MaxSlice()
 
-		// Generate a slices of all slices.
-		slices = make([]uint64, maxSlice+1)
-		for i := range slices {
-			slices[i] = uint64(i)
+			// Generate a slices of all slices.
+			slices = make([]uint64, maxSlice+1)
+			for i := range slices {
+				slices[i] = uint64(i)
+			}
 		}
 	}
 
@@ -992,3 +994,21 @@ func hasOnlySetBitmapAttrs(calls []*pql.Call) bool {
 	}
 	return true
 }
+
+func needsSlices(calls pql.Calls) bool {
+	if len(calls) == 0 {
+		return false
+	}
+
+	for _, call := range calls {
+		if _, ok := call.(pql.BitmapCall); !ok {
+			return true
+		} else if _, ok := call.(*pql.Count); !ok {
+			return true
+		} else if _, ok := call.(*pql.TopN); !ok {
+			return true
+		}
+
+	}
+	return false
+}

fragment.go

@@ -4,6 +4,7 @@ import (
 	"archive/tar"
 	"bufio"
 	"bytes"
+	"container/heap"
 	"context"
 	"crypto/sha1"
 	"encoding/binary"
@@ -28,7 +29,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 +51,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 +107,7 @@ type Fragment struct {
 	BitmapAttrStore *AttrStore
 
 	stats StatsClient
+	turbo BitmapCacher
 }
 
 // NewFragment returns a new instance of Fragment.
@@ -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
 }
@@ -350,6 +381,11 @@ func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error)
 		return false, err
 	}
 
+	// Don't update the cache if nothing changed.
+	if !changed {
+		return changed, nil
+	}
+
 	// Invalidate block checksum.
 	delete(f.checksums, int(bitmapID/HashBlockSize))
 
@@ -359,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
 	}
 
@@ -389,6 +425,11 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) {
 		return false, err
 	}
 
+	// Don't update the cache if nothing changed.
+	if !changed {
+		return changed, nil
+	}
+
 	// Invalidate block checksum.
 	delete(f.checksums, int(bitmapID/HashBlockSize))
 
@@ -398,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
 	}
 
@@ -453,7 +494,8 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 	}
 
 	// Iterate over rankings and add to results until we have enough.
-	results := make([]Pair, 0, opt.N)
+	//results := make(PairHeap, 0, opt.N)
+	results := &PairHeap{}
 	for _, pair := range pairs {
 		bitmapID, n := pair.ID, pair.Count
 
@@ -477,7 +519,7 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 		}
 
 		// The initial n pairs should simply be added to the results.
-		if opt.N == 0 || len(results) < opt.N {
+		if opt.N == 0 || results.Len() < opt.N {
 			// Calculate count and append.
 			count := n
 			if opt.Src != nil {
@@ -486,23 +528,26 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 			if count == 0 {
 				continue
 			}
-			results = append(results, Pair{Key: bitmapID, Count: count})
+			//results = append(results, Pair{Key: bitmapID, Count: count})
+			heap.Push(results, Pair{Key: bitmapID, Count: count})
 
 			// If we reach the requested number of pairs and we are not computing
 			// intersections then simply exit. If we are intersecting then sort
 			// and then only keep pairs that are higher than the lowest count.
-			if opt.N > 0 && len(results) == opt.N {
+			if opt.N > 0 && results.Len() == opt.N {
 				if opt.Src == nil {
 					break
 				}
-				sort.Sort(Pairs(results))
+				//	sort.Sort(Pairs(results))
 			}
 			continue
 		}
 
 		// Retrieve the lowest count we have.
 		// If it's too low then don't try finding anymore pairs.
-		threshold := results[len(results)-1].Count
+		//threshold := results[len(results)-1].Count
+
+		threshold := results.Pairs[0].Count
 		if threshold < MinThreshold {
 			break
 		}
@@ -515,22 +560,22 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 
 		// Calculate the intersecting bit count and skip if it's below our
 		// last bitmap in our current result set.
+
 		count := opt.Src.IntersectionCount(f.Bitmap(bitmapID))
 		if count < threshold {
 			continue
 		}
 
-		// Swap out the last pair for this new count.
-		results[len(results)-1] = Pair{Key: bitmapID, Count: count}
-
-		// If it's count is also higher than the second to last item then resort.
-		if len(results) >= 2 && count > results[len(results)-2].Count {
-			sort.Sort(Pairs(results))
-		}
+		heap.Push(results, Pair{Key: bitmapID, Count: count})
 	}
-
-	sort.Sort(Pairs(results))
-	return results, nil
+	r := make(Pairs, results.Len(), results.Len())
+	x := results.Len()
+	i := 1
+	for results.Len() > 0 {
+		r[x-i] = heap.Pop(results).(Pair)
+		i++
+	}
+	return r, nil
 }
 
 func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair {
@@ -560,6 +605,8 @@ func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair {
 			Count: f.Bitmap(bitmapID).Count(),
 		}
 	}
+	sort.Sort(BitmapPairs(pairs))
+	//debugDumpPairs(pairs)
 	return pairs
 }
 
@@ -838,7 +885,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 	// Process every bit.
 	// If an error occurs then reopen the storage.
 	lastID := uint64(0)
-	bmCounter := 0
 	if err := func() error {
 		set := make(map[uint64]struct{})
 		for i := range bitmapIDs {
@@ -851,7 +897,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 			}
 
 			// Write to storage.
-			changed, err := f.storage.Add(pos)
+			_, err = f.storage.Add(pos)
 			if err != nil {
 				return err
 			}
@@ -863,9 +909,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 				lastID = bitmapID
 				set[bitmapID] = struct{}{}
 			}
-			if changed {
-				bmCounter += 1
-			}
 
 			// Invalidate block checksum.
 			delete(f.checksums, int(bitmapID/HashBlockSize))
@@ -873,7 +916,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()
@@ -912,10 +955,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

handler.go

@@ -201,7 +201,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		http.NotFound(w, r)
 	}
 
-	h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), time.Since(t).Seconds())
+	dif := time.Since(t).Seconds()
+	if dif > 90 {
+		h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), dif)
+	}
 }
 
 // handleGetSchema handles GET /schema requests.

roaring/roaring.go

@@ -444,33 +444,56 @@ 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]
 
 		// Verify container count before writing.
-		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))
+		// 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)
+		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())
 	}
 
@@ -483,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
+			}
 		}
 	}
 
@@ -532,9 +557,10 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
 		c := b.containers[i]
 		if c.n <= ArrayMaxSize {
 			c.array = (*[0xFFFFFFF]uint32)(unsafe.Pointer(&data[offset]))[:c.n]
-			for _, v := range c.array {
-				assert(lowbits(uint64(v)) == v, "array value out of range: %d", v)
-			}
+			// TODO: instead of commenting this out, we need to make it a configuration option
+			//for _, v := range c.array {
+			//    assert(lowbits(uint64(v)) == v, "array value out of range: %d", v)
+			//}
 			opsOffset = int(offset) + len(c.array)*4
 		} else {
 			c.bitmap = (*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN]
@@ -542,8 +568,9 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
 		}
 
 		// Verify container count on load.
-		count := c.count()
-		assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n)
+		// TODO: instead of commenting this out, we need to make it a configuration option
+		//count := c.count()
+		//assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n)
 	}
 
 	// Read ops log until the end of the file.
@@ -1074,9 +1101,10 @@ func (c *container) arrayWriteTo(w io.Writer) (n int64, err error) {
 	}
 
 	// Verify all elements are valid.
-	for _, v := range c.array {
-		assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v)
-	}
+	// TODO: instead of commenting this out, we need to make it a configuration option
+	//for _, v := range c.array {
+	//  	assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v)
+	//}
 
 	nn, err := w.Write((*[0xFFFFFFF]byte)(unsafe.Pointer(&c.array[0]))[:4*c.n])
 	return int64(nn), err