From 0de4661195db1cae9e7e7a47d2f1dc2234805fe1 Mon Sep 17 00:00:00 2001 From: Siyuan Liu Date: Mon, 10 Aug 2026 12:19:14 -0700 Subject: [PATCH] reduce bucket tail latency with 2 map rotation --- fastcache.go | 85 +++++++++++++++++++++++++++++++---------- fastcache_bench_test.go | 82 +++++++++++++++++++++++++++++++++++++++ file.go | 17 +++++++-- 3 files changed, 160 insertions(+), 24 deletions(-) create mode 100644 fastcache_bench_test.go diff --git a/fastcache.go b/fastcache.go index 9358a2b..3fbdde2 100644 --- a/fastcache.go +++ b/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< 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() } diff --git a/fastcache_bench_test.go b/fastcache_bench_test.go new file mode 100644 index 0000000..e371db8 --- /dev/null +++ b/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)) +} diff --git a/file.go b/file.go index 0d35c50..bb3029e 100644 --- a/file.go +++ b/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()