reduce bucket tail latency with 2 map rotation

fastcache.go

@@ -228,6 +228,13 @@ type bucket struct {
 	// m maps hash(k) to idx of (k, v) pair in chunks.
 	m map[uint64]uint64
 
+	// mPrev contains entries written during the previous gen.
+	mPrev map[uint64]uint64
+
+	// mPrevEntriesMask is a mask when a hash is present in both m and mPrev.
+	// Used to calculate EntriesCount without scanning.
+	mPrevEntriesMask int
+
 	// gen is the generation of chunks.
 	gen uint64
 
@@ -259,6 +266,8 @@ func (b *bucket) Reset() {
 		chunks[i] = nil
 	}
 	b.m = make(map[uint64]uint64)
+	b.mPrev = nil
+	b.mPrevEntriesMask = 0
 	b.idx = 0
 	b.gen = 1
 	atomic.StoreUint64(&b.getCalls, 0)
@@ -281,20 +290,42 @@ func (b *bucket) cleanLocked() {
 			newItems++
 		}
 	}
-	if newItems < len(bm) {
-		// Re-create b.m with valid items, which weren't expired yet instead of deleting expired items from b.m.
-		// This should reduce memory fragmentation and the number Go objects behind b.m.
-		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5379
-		bmNew := make(map[uint64]uint64, newItems)
-		for k, v := range bm {
-			gen := v >> bucketSizeBits
-			idx := v & ((1 << bucketSizeBits) - 1)
-			if (gen+1 == bGen || gen == maxGen && bGen == 1) && idx >= bIdx || gen == bGen && idx < bIdx {
-				bmNew[k] = v
-			}
+	bmPrev := b.mPrev
+	if bmPrev == nil && newItems == len(bm) {
+		return
+	}
+	for k, v := range bmPrev {
+		if _, ok := bm[k]; ok {
+			continue
+		}
+		gen := v >> bucketSizeBits
+		idx := v & ((1 << bucketSizeBits) - 1)
+		if (gen+1 == bGen || gen == maxGen && bGen == 1) && idx >= bIdx || gen == bGen && idx < bIdx {
+			newItems++
 		}
-		b.m = bmNew
 	}
+
+	bmNew := make(map[uint64]uint64, newItems)
+	for k, v := range bmPrev {
+		if _, ok := bm[k]; ok {
+			continue
+		}
+		gen := v >> bucketSizeBits
+		idx := v & ((1 << bucketSizeBits) - 1)
+		if (gen+1 == bGen || gen == maxGen && bGen == 1) && idx >= bIdx || gen == bGen && idx < bIdx {
+			bmNew[k] = v
+		}
+	}
+	for k, v := range bm {
+		gen := v >> bucketSizeBits
+		idx := v & ((1 << bucketSizeBits) - 1)
+		if (gen+1 == bGen || gen == maxGen && bGen == 1) && idx >= bIdx || gen == bGen && idx < bIdx {
+			bmNew[k] = v
+		}
+	}
+	b.m = bmNew
+	b.mPrev = nil
+	b.mPrevEntriesMask = 0
 }
 
 func (b *bucket) UpdateStats(s *Stats) {
@@ -305,7 +336,8 @@ func (b *bucket) UpdateStats(s *Stats) {
 	s.Corruptions += atomic.LoadUint64(&b.corruptions)
 
 	b.mu.RLock()
-	s.EntriesCount += uint64(len(b.m))
+	entriesCount := len(b.m) + len(b.mPrev) - b.mPrevEntriesMask
+	s.EntriesCount += uint64(entriesCount)
 	bytesSize := uint64(0)
 	for _, chunk := range b.chunks {
 		bytesSize += uint64(cap(chunk))
@@ -336,7 +368,6 @@ func (b *bucket) Set(k, v []byte, h uint64) {
 
 	b.mu.Lock()
 	chunks := b.chunks
-	needClean := false
 	idx := b.idx
 	idxNew := idx + kvLen
 	chunkIdx := idx / chunkSize
@@ -350,7 +381,9 @@ func (b *bucket) Set(k, v []byte, h uint64) {
 			if b.gen&((1<<genSizeBits)-1) == 0 {
 				b.gen++
 			}
-			needClean = true
+			b.mPrev = b.m
+			b.m = make(map[uint64]uint64)
+			b.mPrevEntriesMask = 0
 		} else {
 			idx = chunkIdxNew * chunkSize
 			idxNew = idx + kvLen
@@ -367,11 +400,14 @@ func (b *bucket) Set(k, v []byte, h uint64) {
 	chunk = append(chunk, k...)
 	chunk = append(chunk, v...)
 	chunks[chunkIdx] = chunk
+	if _, ok := b.mPrev[h]; ok {
+		if _, ok := b.m[h]; !ok {
+			// first time this hash is going to be in b.m
+			b.mPrevEntriesMask++
+		}
+	}
 	b.m[h] = idx | (b.gen << bucketSizeBits)
 	b.idx = idxNew
-	if needClean {
-		b.cleanLocked()
-	}
 	b.mu.Unlock()
 }
 
@@ -380,9 +416,12 @@ func (b *bucket) Get(dst, k []byte, h uint64, returnDst bool) ([]byte, bool) {
 	atomic.AddUint64(&b.getCalls, 1)
 	found := false
 	chunks := b.chunks
-	v := b.m[h]
+	v, ok := b.m[h]
+	if !ok {
+		v, ok = b.mPrev[h]
+	}
 	bGen := b.gen & ((1 << genSizeBits) - 1)
-	if v > 0 {
+	if ok && v > 0 {
 		gen := v >> bucketSizeBits
 		idx := v & ((1 << bucketSizeBits) - 1)
 		if gen == bGen && idx < b.idx || gen+1 == bGen && idx >= b.idx || gen == maxGen && bGen == 1 && idx >= b.idx {
@@ -429,6 +468,12 @@ end:
 
 func (b *bucket) Del(h uint64) {
 	b.mu.Lock()
+	_, cur := b.m[h]
+	_, prev := b.mPrev[h]
+	if cur && prev {
+		b.mPrevEntriesMask--
+	}
 	delete(b.m, h)
+	delete(b.mPrev, h)
 	b.mu.Unlock()
 }

fastcache_bench_test.go

@@ -0,0 +1,82 @@
+package fastcache
+
+import (
+	"testing"
+	"time"
+)
+
+// BenchmarkBucketSetAtNextGen compares writes that create a next gen with
+// writes that remain in the current gen. A next gen is created when the key
+// cannot fit into the last chunk in this bucket.
+//
+//	go test -run '^$' -bench '^BenchmarkBucketSetAtNextGen$' -benchtime=5000000x -count=5 -benchmem
+func BenchmarkBucketSetAtNextGen(b *testing.B) {
+	var shard bucket
+	// each chunk is 64 * 1024 bytes, so 1024 * 1024 bytes has 16 chunks
+	shard.Init(1024 * 1024)
+	defer shard.Reset()
+
+	key := []byte("12345678")
+	val := []byte(nil)
+	// 4+len(key) because kvLen := uint64(len(kvLenBuf) + len(k) + len(v))
+	// in the fastcache implementation.
+	// - len(kvLenBuf) == 4,
+	// - len(k) == len(Key)
+	// - len(v) == len(val) == 0
+	kvLen := uint64(4 + len(key))
+	// do some warmups
+	var h uint64 = 1
+	for shard.gen < 3 {
+		shard.Set(key, val, h)
+		h++
+	}
+
+	var nextGenTotal time.Duration
+	var nextGenMax time.Duration
+	var nextGenCount uint64
+	var currentGenTotal time.Duration
+	var currentGenMax time.Duration
+	var currentGenCount uint64
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		start := time.Now()
+		if nextGen(&shard, kvLen) {
+			shard.Set(key, val, h)
+			d := time.Since(start)
+			nextGenTotal += d
+			if d > nextGenMax {
+				nextGenMax = d
+			}
+			nextGenCount++
+		} else {
+			shard.Set(key, val, h)
+			d := time.Since(start)
+			currentGenTotal += d
+			if d > currentGenMax {
+				currentGenMax = d
+			}
+			currentGenCount++
+		}
+		h++
+	}
+	b.StopTimer()
+
+	if nextGenCount > 0 {
+		b.ReportMetric(float64(nextGenTotal.Nanoseconds())/float64(nextGenCount), "next-gen/ns/avg")
+		b.ReportMetric(float64(nextGenMax.Nanoseconds()), "next-gen/ns/max")
+		b.ReportMetric(float64(nextGenCount), "next-gen/count")
+	}
+	if currentGenCount > 0 {
+		b.ReportMetric(float64(currentGenTotal.Nanoseconds())/float64(currentGenCount), "cur-gen/ns/avg")
+		b.ReportMetric(float64(currentGenMax.Nanoseconds()), "cur-gen/ns/max")
+		b.ReportMetric(float64(currentGenCount), "cur-gen/count")
+	}
+}
+
+func nextGen(b *bucket, kvLen uint64) bool {
+	idxNew := b.idx + kvLen
+	chunkIdx := b.idx / chunkSize
+	chunkIdxNew := idxNew / chunkSize
+	return chunkIdxNew > chunkIdx && chunkIdxNew >= uint64(len(b.chunks))
+}

file.go

@@ -282,11 +282,18 @@ func loadBuckets(buckets []bucket, dataPath string, maxChunks uint64) error {
 }
 
 func (b *bucket) Save(w io.Writer) error {
-	b.mu.Lock()
-	b.cleanLocked()
-	b.mu.Unlock()
+	for {
+		b.mu.Lock()
+		b.cleanLocked()
+		b.mu.Unlock()
 
-	b.mu.RLock()
+		b.mu.RLock()
+		if b.mPrev != nil {
+			b.mu.RUnlock()
+			continue
+		}
+		break // b.mu is readlocked
+	}
 	defer b.mu.RUnlock()
 
 	// Store b.idx, b.gen and b.m to w.
@@ -407,6 +414,8 @@ func (b *bucket) Load(r io.Reader, maxChunks uint64) error {
 	}
 	b.chunks = chunks
 	b.m = m
+	b.mPrev = nil
+	b.mPrevEntriesMask = 0
 	b.idx = bIdx
 	b.gen = bGen
 	b.mu.Unlock()