Repository: blevesearch/bleve (go)

Issue:
perf: scorch/zap ReadUvarint()
On scorch search performance, @sreekanth-cb has been looking into pprof profiling reports, so capturing some of recent discussion and more thoughts on this...

From the pprof's, looks like encoding/binary.ReadUvarint() and its invocation of bytes.Reader.ReadByte() show up a bunch as top contenders of the "hot inner loop(s)" of "where does the time go?".

Especially, those functions are invoked from zap's PostingIterator.readFreqNormHasLocs() method.

At first, I thought those funcs were taking up time because they were a likely point of first contact when accessing different parts of the mmap()'ed zap file.  And, I was theorizing that page-faults are "invisible" to golang's profiler -- so those func's weren't really actually slow -- it's just that I figured that that's where disk reads was happening.

But, @sreekanth-cb was able to run some tests on large RAM machines where major page faults weren't appearing at all, but ReadUvarint() and ReadByte() were still at the top of the pprof reports.

On the one hand, those funcs from the golang standard library are seemingly highly optimized.

On the other hand, there's a couple of checks and features there that are meant to handle general cases as a standard library, where in contrast scorch/zap might have some narrower, optimizable needs.  Some ideas...

- there's safety error checking for io.EOF with the bytes.Reader.ReadByte().  In contrast, scorch/zap ought to have "perfectly sized" memory slices (knock on wood) where, theoretically, zap's freqNormReader never ought to run into an io.EOF.

- there's some minor rune related code in bytes.Reader (that's apparently for io.RuneScanner) that might be worth avoiding for scorch/zap.

- a simplified bytes.Reader.ReadByte() might be explicitly in-line'able, which may mean that we can avoid a method call in the hot-inner-loop of ReadUvarint().

- zap's PostingIterator.currChunkNext() invokes readFreqNormHasLocs() but ignores the freq/norm parts of the result, so perhaps specialized methods might be faster.  e.g., something like SkipUvarint()?


Pull request:
fixes #1166 - perf optimizations for ReadUvarint()
This commit introduces a specialized MemUvarintReader which is focused
on the "hot inner loop" of scorch/zap's reading of uvarint's -- such
as the postingsList reading freqNormHasLocs and location information.

The various changes include...

- the separate bytes.Reader vs binary.ReadUvarint() codepaths are
  unified, and ReadByte() is now inlined.

- error return codes are removed, since the memory slices are
  "perfectly sized" in scorch/zap, which meant there's never an error.

- SkipUvarint() allows entries to be skipped more efficiently (by
  postingsList.currChunkNext()).

- prevRune maintenance is removed.

- postingsList.readLocation() is always provided a non-nil output
  location.

In microbenchmarks...
```
  $ go test -benchmem -bench=. ./index/scorch/segment/
  BenchmarkUvarint-8          100000000 16.4  ns/op 0 B/op 0 allocs/op
  BenchmarkMemUvarintReader-8 200000000  7.57 ns/op 0 B/op 0 allocs/op
```

In bleve-query microbenchmarks on a scorch index with 200K en-wiki
docs, high-frequency term ("http") searches that included locations
had throughput of ~21.5 q/sec before this change, and after this
change went to ~28.49 q/sec.

Patch:
diff --git a/index/scorch/introducer.go b/index/scorch/introducer.go
index ac627796..2d04bd38 100644
--- a/index/scorch/introducer.go
+++ b/index/scorch/introducer.go
@@ -376,7 +376,6 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 				fileSegments++
 			}
 		}
-
 	}
 
 	// before the newMerge introduction, need to clean the newly
@@ -393,6 +392,7 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 			}
 		}
 	}
+
 	// In case where all the docs in the newly merged segment getting
 	// deleted by the time we reach here, can skip the introduction.
 	if nextMerge.new != nil &&
@@ -424,6 +424,7 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 	newSnapshot.AddRef() // 1 ref for the nextMerge.notify response
 
 	newSnapshot.updateSize()
+
 	s.rootLock.Lock()
 	// swap in new index snapshot
 	newSnapshot.epoch = s.nextSnapshotEpoch
@@ -501,6 +502,7 @@ func (s *Scorch) revertToSnapshot(revertTo *snapshotReversion) error {
 	}
 
 	newSnapshot.updateSize()
+
 	// swap in new snapshot
 	rootPrev := s.root
 	s.root = newSnapshot
diff --git a/index/scorch/persister.go b/index/scorch/persister.go
index e15fa2ab..349ccdc0 100644
--- a/index/scorch/persister.go
+++ b/index/scorch/persister.go
@@ -111,6 +111,7 @@ OUTER:
 		if ew != nil && ew.epoch > lastMergedEpoch {
 			lastMergedEpoch = ew.epoch
 		}
+
 		lastMergedEpoch, persistWatchers = s.pausePersisterForMergerCatchUp(lastPersistedEpoch,
 			lastMergedEpoch, persistWatchers, po)
 
@@ -178,6 +179,7 @@ OUTER:
 			s.fireEvent(EventKindPersisterProgress, time.Since(startTime))
 
 			if changed {
+				s.removeOldData()
 				atomic.AddUint64(&s.stats.TotPersistLoopProgress, 1)
 				continue OUTER
 			}
@@ -659,13 +661,13 @@ func (s *Scorch) LoadSnapshot(epoch uint64) (rv *IndexSnapshot, err error) {
 }
 
 func (s *Scorch) loadSnapshot(snapshot *bolt.Bucket) (*IndexSnapshot, error) {
-
 	rv := &IndexSnapshot{
 		parent:   s,
 		internal: make(map[string][]byte),
 		refs:     1,
 		creator:  "loadSnapshot",
 	}
+
 	var running uint64
 	c := snapshot.Cursor()
 	for k, _ := c.First(); k != nil; k, _ = c.Next() {
@@ -701,6 +703,7 @@ func (s *Scorch) loadSnapshot(snapshot *bolt.Bucket) (*IndexSnapshot, error) {
 			running += segmentSnapshot.segment.Count()
 		}
 	}
+
 	return rv, nil
 }
 
diff --git a/index/scorch/segment/int.go b/index/scorch/segment/int.go
index a4836ebf..55299d8f 100644
--- a/index/scorch/segment/int.go
+++ b/index/scorch/segment/int.go
@@ -19,7 +19,10 @@
 
 package segment
 
-import "fmt"
+import (
+	"errors"
+	"fmt"
+)
 
 const (
 	MaxVarintSize = 9
@@ -92,3 +95,82 @@ func DecodeUvarintAscending(b []byte) ([]byte, uint64, error) {
 	}
 	return b[length:], v, nil
 }
+
+// ------------------------------------------------------------
+
+type MemUvarintReader struct {
+	C int // index of next byte to read from S
+	S []byte
+}
+
+func NewMemUvarintReader(s []byte) *MemUvarintReader {
+	return &MemUvarintReader{S: s}
+}
+
+// Len returns the number of unread bytes.
+func (r *MemUvarintReader) Len() int {
+	n := len(r.S) - r.C
+	if n < 0 {
+		return 0
+	}
+	return n
+}
+
+var ErrMemUvarintReaderOverflow = errors.New("MemUvarintReader overflow")
+
+// ReadUvarint reads an encoded uint64.  The original code this was
+// based on is at encoding/binary/ReadUvarint().
+func (r *MemUvarintReader) ReadUvarint() (uint64, error) {
+	var x uint64
+	var s uint
+	var C = r.C
+	var S = r.S
+
+	for {
+		b := S[C]
+		C++
+
+		if b < 0x80 {
+			r.C = C
+
+			// why 63?  The original code had an 'i += 1' loop var and
+			// checked for i > 9 || i == 9 ...; but, we no longer
+			// check for the i var, but instead check here for s,
+			// which is incremented by 7.  So, 7*9 == 63.
+			//
+			// why the "extra" >= check?  The normal case is that s <
+			// 63, so we check this single >= guard first so that we
+			// hit the normal, nil-error return pathway sooner.
+			if s >= 63 && (s > 63 || s == 63 && b > 1) {
+				return 0, ErrMemUvarintReaderOverflow
+			}
+
+			return x | uint64(b)<<s, nil
+		}
+
+		x |= uint64(b&0x7f) << s
+		s += 7
+	}
+}
+
+// SkipUvarint skips ahead one encoded uint64.
+func (r *MemUvarintReader) SkipUvarint() {
+	for {
+		b := r.S[r.C]
+		r.C++
+
+		if b < 0x80 {
+			return
+		}
+	}
+}
+
+// SkipBytes skips a count number of bytes.
+func (r *MemUvarintReader) SkipBytes(count int) {
+	r.C = r.C + count
+}
+
+func (r *MemUvarintReader) Reset(s []byte) {
+	r.C = 0
+	r.S = s
+}
diff --git a/index/scorch/segment/zap/posting.go b/index/scorch/segment/zap/posting.go
index 93b51ae7..0656c8e9 100644
--- a/index/scorch/segment/zap/posting.go
+++ b/index/scorch/segment/zap/posting.go
@@ -15,10 +15,8 @@
 package zap
 
 import (
-	"bytes"
 	"encoding/binary"
 	"fmt"
-	"io"
 	"math"
 	"reflect"
 
@@ -334,8 +332,8 @@ type PostingsIterator struct {
 	currChunkFreqNorm []byte
 	currChunkLoc      []byte
 
-	freqNormReader *bytes.Reader
-	locReader      *bytes.Reader
+	freqNormReader *segment.MemUvarintReader
+	locReader      *segment.MemUvarintReader
 
 	freqChunkOffsets []uint64
 	freqChunkStart   uint64
@@ -386,7 +384,7 @@ func (i *PostingsIterator) loadChunk(chunk int) error {
 		end += e
 		i.currChunkFreqNorm = i.postings.sb.mem[start:end]
 		if i.freqNormReader == nil {
-			i.freqNormReader = bytes.NewReader(i.currChunkFreqNorm)
+			i.freqNormReader = segment.NewMemUvarintReader(i.currChunkFreqNorm)
 		} else {
 			i.freqNormReader.Reset(i.currChunkFreqNorm)
 		}
@@ -404,7 +402,7 @@ func (i *PostingsIterator) loadChunk(chunk int) error {
 		end += e
 		i.currChunkLoc = i.postings.sb.mem[start:end]
 		if i.locReader == nil {
-			i.locReader = bytes.NewReader(i.currChunkLoc)
+			i.locReader = segment.NewMemUvarintReader(i.currChunkLoc)
 		} else {
 			i.locReader.Reset(i.currChunkLoc)
 		}
@@ -419,18 +417,34 @@ func (i *PostingsIterator) readFreqNormHasLocs() (uint64, uint64, bool, error) {
 		return 1, i.normBits1Hit, false, nil
 	}
 
-	freqHasLocs, err := binary.ReadUvarint(i.freqNormReader)
+	freqHasLocs, err := i.freqNormReader.ReadUvarint()
 	if err != nil {
 		return 0, 0, false, fmt.Errorf("error reading frequency: %v", err)
 	}
+
 	freq, hasLocs := decodeFreqHasLocs(freqHasLocs)
 
-	normBits, err := binary.ReadUvarint(i.freqNormReader)
+	normBits, err := i.freqNormReader.ReadUvarint()
 	if err != nil {
 		return 0, 0, false, fmt.Errorf("error reading norm: %v", err)
 	}
 
-	return freq, normBits, hasLocs, err
+	return freq, normBits, hasLocs, nil
+}
+
+func (i *PostingsIterator) skipFreqNormReadHasLocs() (bool, error) {
+	if i.normBits1Hit != 0 {
+		return false, nil
+	}
+
+	freqHasLocs, err := i.freqNormReader.ReadUvarint()
+	if err != nil {
+		return false, fmt.Errorf("error reading freqHasLocs: %v", err)
+	}
+
+	i.freqNormReader.SkipUvarint() // Skip normBits.
+
+	return freqHasLocs&0x01 != 0, nil // See decodeFreqHasLocs() / hasLocs.
 }
 
 func encodeFreqHasLocs(freq uint64, hasLocs bool) uint64 {
@@ -448,58 +462,53 @@ func decodeFreqHasLocs(freqHasLocs uint64) (uint64, bool) {
 }
 
 // readLocation processes all the integers on the stream representing a single
-// location.  if you care about it, pass in a non-nil location struct, and we
-// will fill it.  if you don't care about it, pass in nil and we safely consume
-// the contents.
+// location.
 func (i *PostingsIterator) readLocation(l *Location) error {
 	// read off field
-	fieldID, err := binary.ReadUvarint(i.locReader)
+	fieldID, err := i.locReader.ReadUvarint()
 	if err != nil {
 		return fmt.Errorf("error reading location field: %v", err)
 	}
 	// read off pos
-	pos, err := binary.ReadUvarint(i.locReader)
+	pos, err := i.locReader.ReadUvarint()
 	if err != nil {
 		return fmt.Errorf("error reading location pos: %v", err)
 	}
 	// read off start
-	start, err := binary.ReadUvarint(i.locReader)
+	start, err := i.locReader.ReadUvarint()
 	if err != nil {
 		return fmt.Errorf("error reading location start: %v", err)
 	}
 	// read off end
-	end, err := binary.ReadUvarint(i.locReader)
+	end, err := i.locReader.ReadUvarint()
 	if err != nil {
 		return fmt.Errorf("error reading location end: %v", err)
 	}
 	// read off num array pos
-	numArrayPos, err := binary.ReadUvarint(i.locReader)
+	numArrayPos, err := i.locReader.ReadUvarint()
 	if err != nil {
 		return fmt.Errorf("error reading location num array pos: %v", err)
 	}
 
-	// group these together for less branching
-	if l != nil {
-		l.field = i.postings.sb.fieldsInv[fieldID]
-		l.pos = pos
-		l.start = start
-		l.end = end
-		if cap(l.ap) < int(numArrayPos) {
-			l.ap = make([]uint64, int(numArrayPos))
-		} else {
-			l.ap = l.ap[:int(numArrayPos)]
-		}
+	l.field = i.postings.sb.fieldsInv[fieldID]
+	l.pos = pos
+	l.start = start
+	l.end = end
+
+	if cap(l.ap) < int(numArrayPos) {
+		l.ap = make([]uint64, int(numArrayPos))
+	} else {
+		l.ap = l.ap[:int(numArrayPos)]
 	}
 
 	// read off array positions
 	for k := 0; k < int(numArrayPos); k++ {
-		ap, err := binary.ReadUvarint(i.locReader)
+		ap, err := i.locReader.ReadUvarint()
 		if err != nil {
 			return fmt.Errorf("error reading array position: %v", err)
 		}
-		if l != nil {
-			l.ap[k] = ap
-		}
+
+		l.ap[k] = ap
 	}
 
 	return nil
@@ -556,7 +565,7 @@ func (i *PostingsIterator) nextAtOrAfter(atOrAfter uint64) (segment.Posting, err
 		}
 		rv.locs = i.nextSegmentLocs[:0]
 
-		numLocsBytes, err := binary.ReadUvarint(i.locReader)
+		numLocsBytes, err := i.locReader.ReadUvarint()
 		if err != nil {
 			return nil, fmt.Errorf("error reading location numLocsBytes: %v", err)
 		}
@@ -612,17 +621,14 @@ func (i *PostingsIterator) nextBytes() (
 	if hasLocs {
 		startLoc := len(i.currChunkLoc) - i.locReader.Len()
 
-		numLocsBytes, err := binary.ReadUvarint(i.locReader)
+		numLocsBytes, err := i.locReader.ReadUvarint()
 		if err != nil {
 			return 0, 0, 0, nil, nil,
 				fmt.Errorf("error reading location nextBytes numLocs: %v", err)
 		}
 
 		// skip over all the location bytes
-		_, err = i.locReader.Seek(int64(numLocsBytes), io.SeekCurrent)
-		if err != nil {
-			return 0, 0, 0, nil, nil, err
-		}
+		i.locReader.SkipBytes(int(numLocsBytes))
 
 		endLoc := len(i.currChunkLoc) - i.locReader.Len()
 		bytesLoc = i.currChunkLoc[startLoc:endLoc]
@@ -763,22 +769,19 @@ func (i *PostingsIterator) currChunkNext(nChunk uint32) error {
 	}
 
 	// read off freq/offsets even though we don't care about them
-	_, _, hasLocs, err := i.readFreqNormHasLocs()
+	hasLocs, err := i.skipFreqNormReadHasLocs()
 	if err != nil {
 		return err
 	}
 
 	if i.includeLocs && hasLocs {
-		numLocsBytes, err := binary.ReadUvarint(i.locReader)
+		numLocsBytes, err := i.locReader.ReadUvarint()
 		if err != nil {
 			return fmt.Errorf("error reading location numLocsBytes: %v", err)
 		}
 
 		// skip over all the location bytes
-		_, err = i.locReader.Seek(int64(numLocsBytes), io.SeekCurrent)
-		if err != nil {
-			return err
-		}
+		i.locReader.SkipBytes(int(numLocsBytes))
 	}
 
 	return nil


Test patch:
diff --git a/index/scorch/segment/int_test.go b/index/scorch/segment/int_test.go
index 3d2ab6fd..449f4a07 100644
--- a/index/scorch/segment/int_test.go
+++ b/index/scorch/segment/int_test.go
@@ -21,6 +21,7 @@ package segment
 
 import (
 	"bytes"
+	"encoding/binary"
 	"math"
 	"testing"
 )
@@ -94,3 +95,73 @@ func testCustomEncodeUint64(
 		}
 	}
 }
+
+func BenchmarkUvarint(b *testing.B) {
+	n, buf := generateCommonUvarints(64, 512)
+
+	reader := bytes.NewReader(buf)
+	seen := 0
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i = i + 1 {
+		if seen >= n {
+			reader.Reset(buf)
+			seen = 0
+		}
+
+		_, _ = binary.ReadUvarint(reader)
+		seen = seen + 1
+	}
+}
+
+func BenchmarkMemUvarintReader(b *testing.B) {
+	n, buf := generateCommonUvarints(64, 512)
+
+	reader := &MemUvarintReader{S: buf}
+	seen := 0
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i = i + 1 {
+		if seen >= n {
+			reader.Reset(buf)
+			seen = 0
+		}
+
+		_, _ = reader.ReadUvarint()
+		seen = seen + 1
+	}
+}
+
+// generate some common, encoded uvarint's that we might see as
+// freq-norm's or locations.
+func generateCommonUvarints(maxFreq, maxFieldLen int) (n int, rv []byte) {
+	buf := make([]byte, binary.MaxVarintLen64)
+
+	var out bytes.Buffer
+
+	encode := func(val uint64) {
+		bufLen := binary.PutUvarint(buf, val)
+		out.Write(buf[:bufLen])
+		n = n + 1
+	}
+
+	for i := 1; i < maxFreq; i = i * 2 { // Common freqHasLoc's.
+		freqHasLocs := uint64(i << 1)
+		encode(freqHasLocs)
+		encode(freqHasLocs | 0x01) // 0'th LSB encodes whether there are locations.
+	}
+
+	encodeNorm := func(fieldLen int) {
+		norm := float32(1.0 / math.Sqrt(float64(fieldLen)))
+		normUint64 := uint64(math.Float32bits(float32(norm)))
+		encode(normUint64)
+	}
+
+	for i := 1; i < maxFieldLen; i = i * 2 { // Common norm's.
+		encodeNorm(i)
+	}
+
+	return n, out.Bytes()
+}