From f4fc11c374948f639835682c6855ac5ac47eb2ed Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 18:36:47 +0700 Subject: [PATCH] execution/cache: track GenericCache entry count in an atomic, not freelru's all-shard Len Every insert into a cache below its growth ceiling ran the grow check through freelru's ShardedLRU.Len, which RLocks all shards in turn (up to GOMAXPROCS*16). That serialised writers on shards they never touch, and ShardedLRU.Get takes the shard's write lock, so readers convoyed behind it too. Pair each generation of the LRU with an atomic entry count maintained by the insert path and the OnEvict callback, and read that instead. BenchmarkGenericCacheParallelPutGrow, M4 Max, -cpu 10: 466.3n -> 163.7n (-64.9%, p=0.002 n=6). The gap widens with core count. Claude-Session: https://claude.ai/code/session_01WFkAYPPhqPe1NXg41Nph78 --- execution/cache/cache_test.go | 49 ++++++++++++ execution/cache/generic_cache.go | 75 ++++++++++++------- .../cache/generic_cache_concurrency_test.go | 46 ++++++++++++ 3 files changed, 143 insertions(+), 27 deletions(-) diff --git a/execution/cache/cache_test.go b/execution/cache/cache_test.go index f0d8a7e02cd..580ad509e09 100644 --- a/execution/cache/cache_test.go +++ b/execution/cache/cache_test.go @@ -1771,3 +1771,52 @@ func BenchmarkPublishVsViewBindLock(b *testing.B) { }) } } + +// The O(1) entry counter that replaced freelru's all-shard Len on the grow +// check must not drift from the LRU's real length on any mutation path. +func TestGenericCache_LenTracksLRU(t *testing.T) { + c := closeOnCleanup(t, NewGenericCacheWithAvg[[]byte](8*datasize.MB, 256, func(v []byte) int { return len(v) }, ModeEvictLRU)) + key := func(i int) []byte { + k := make([]byte, 8) + binary.BigEndian.PutUint64(k, uint64(i)) + return k + } + check := func(phase string) { + t.Helper() + require.Equal(t, c.data.Load().lru.Len(), c.Len(), "entry counter drifted after %s", phase) + } + + for i := range 500 { + c.Put(key(i), []byte("v"), 10) + } + check("inserts") + + for i := range 200 { + c.Put(key(i), []byte("updated"), 20) + } + check("updates") + + for i := range 100 { + c.Delete(key(i)) + } + check("deletes") + + // Floor 15 leaves the txNum-10 entries live and strands the txNum-20 ones, + // which their next read drops. + c.Unwind(15) + for i := 100; i < 500; i++ { + c.Get(key(i)) + } + check("stale drops") + + before := c.data.Load() + for i := 500; i < 4000; i++ { + c.Put(key(i), []byte("v"), 30) + } + require.NotEqual(t, before, c.data.Load(), "grow did not happen") + check("grow") + + c.Clear() + require.Equal(t, 0, c.Len()) + check("clear") +} diff --git a/execution/cache/generic_cache.go b/execution/cache/generic_cache.go index 926c664c30e..25dc4397e41 100644 --- a/execution/cache/generic_cache.go +++ b/execution/cache/generic_cache.go @@ -58,6 +58,26 @@ type entry[T any] struct { epoch uint32 // unwind generation the entry was written in } +// lruGen is one generation of the sharded LRU plus an O(1) live-entry count. +// freelru's own Len RLocks every shard, so the grow check — which every insert +// runs while the cache is below its ceiling — cannot use it without serialising +// writers that would otherwise touch disjoint shards. +type lruGen[T any] struct { + lru *freelru.ShardedLRU[uint64, entry[T]] + n atomic.Int64 +} + +func (g *lruGen[T]) len() int { return int(g.n.Load()) } + +// add inserts a key the LRU does not already hold — every call site removes an +// existing one first — so the count rises by one; a capacity eviction inside +// freelru fires OnEvict, which takes it back down. +func (g *lruGen[T]) add(h uint64, e entry[T]) (evicted bool) { + evicted = g.lru.Add(h, e) + g.n.Add(1) + return evicted +} + // GenericCache is a sharded, LRU-evicting bounded cache for key-value // data. Eviction mode is fixed at construction (see policy.go). type GenericCache[T any] struct { @@ -65,7 +85,7 @@ type GenericCache[T any] struct { // held — on a jump-grow (fully copied generation) and on Clear (fresh // empty one) — so no write lands in a retired generation and no reader // sees a partial copy (see maybeGrow, Clear). - data atomic.Pointer[freelru.ShardedLRU[uint64, entry[T]]] + data atomic.Pointer[lruGen[T]] capacityB datasize.ByteSize mode Mode @@ -198,15 +218,17 @@ func newGenericCacheEntries[T any](capacityBytes datasize.ByteSize, capacityEntr // entry. The callback must not feed the evictions metric — it also fires for // intentional Removes — so capacity evictions are counted from Add's evicted // return at the call sites. -func (c *GenericCache[T]) newShards(capacity, shards uint32) *freelru.ShardedLRU[uint64, entry[T]] { +func (c *GenericCache[T]) newShards(capacity, shards uint32) *lruGen[T] { lru, err := freelru.NewShardedWithSize[uint64, entry[T]](shards, capacity, capacity+capacity/4, u64identity) if err != nil { panic(err) } + g := &lruGen[T]{lru: lru} lru.SetOnEvict(func(_ uint64, e entry[T]) { c.currentSize.Add(-int64(e.size)) + g.n.Add(-1) }) - return lru + return g } // maybeGrow jump-resizes the LRU one step larger when it is full, the ceiling @@ -226,7 +248,7 @@ func (c *GenericCache[T]) maybeGrow() { old := c.data.Load() curCap := c.curCap.Load() - if curCap >= c.maxCap || old.Len() < int(curCap) { + if curCap >= c.maxCap || old.len() < int(curCap) { return } newCap := min(curCap*genericCacheGrowFactor, c.maxCap) @@ -251,9 +273,9 @@ func (c *GenericCache[T]) maybeGrow() { c.putStripes[i].Lock() } copied, evicted := 0, 0 - for _, k := range old.Keys() { - if v, ok := old.Get(k); ok { - if next.Add(k, v) { + for _, k := range old.lru.Keys() { + if v, ok := old.lru.Get(k); ok { + if next.add(k, v) { evicted++ } copied++ @@ -320,8 +342,8 @@ func (c *GenericCache[T]) GetWithTxNum(key []byte) (T, uint64, bool) { // snapshot can only cause a safe miss because dropStale rechecks the current // generation before removing it. coh := c.coh.Snapshot() - lru := c.data.Load() - e, ok := lru.Get(h) + gen := c.data.Load() + e, ok := gen.lru.Get(h) if !ok || !bytes.Equal(e.key, key) { c.misses.Add(1) var zero T @@ -369,8 +391,7 @@ func (c *GenericCache[T]) put(key []byte, value T, txNum uint64, overwrite bool) // putStriped performs the write under the key's stripe and reports whether the // insert landed in a full LRU with ceiling headroom, i.e. the caller should -// grow. Detection stays on the insert path — Len locks every shard, too costly -// per warm update. +// grow. func (c *GenericCache[T]) putStriped(key []byte, value T, txNum uint64, overwrite bool) bool { h := maphash.Hash(key) valBytes := c.sizeFunc(value) @@ -384,8 +405,8 @@ func (c *GenericCache[T]) putStriped(key []byte, value T, txNum uint64, overwrit // generation swap and coherence reset, so the stamp cannot belong to a // different generation from the one where the entry lands. ep := c.coh.Epoch() - lru := c.data.Load() - existing, hasExisting := lru.Get(h) + gen := c.data.Load() + existing, hasExisting := gen.lru.Get(h) // Existing key — update by remove-then-add (see newShards for why a size // delta would be wrong). Reuse the stored key buffer to avoid an extra @@ -399,8 +420,8 @@ func (c *GenericCache[T]) putStriped(key []byte, value T, txNum uint64, overwrit // another stripe over-admits past the budget. Over-stating is safe — at // worst a new key is dropped, which is within "drop new keys when full". c.currentSize.Add(int64(newSize)) - lru.Remove(h) - if lru.Add(h, entry[T]{key: existing.key, val: value, size: newSize, txNum: txNum, epoch: ep}) { + gen.lru.Remove(h) + if gen.add(h, entry[T]{key: existing.key, val: value, size: newSize, txNum: txNum, epoch: ep}) { c.evictions.Add(1) } return false @@ -409,7 +430,7 @@ func (c *GenericCache[T]) putStriped(key []byte, value T, txNum uint64, overwrit if c.mode == ModeNoOp { // Refuse once full by either bound — freelru would otherwise evict at the // entry-count cap, which ModeNoOp ("drop new keys when full") must not do. - if c.currentSize.Load()+int64(newSize) > int64(c.capacityB) || lru.Len() >= int(c.maxCap) { + if c.currentSize.Load()+int64(newSize) > int64(c.capacityB) || gen.len() >= int(c.maxCap) { c.dropped.Add(1) return false } @@ -419,7 +440,7 @@ func (c *GenericCache[T]) putStriped(key []byte, value T, txNum uint64, overwrit // The insert lands before the grow (which must run outside the stripe), so // it and any racers until the swap evict at the pre-grow cap — a transient // bounded by the grow window. - needGrow := c.mode != ModeNoOp && curCap < c.maxCap && lru.Len() >= int(curCap) + needGrow := c.mode != ModeNoOp && curCap < c.maxCap && gen.len() >= int(curCap) // In ModeEvictLRU the byte budget is enforced through the entry-count cap, // not a separate currentSize check: capacityEntries is derived from @@ -439,10 +460,10 @@ func (c *GenericCache[T]) putStriped(key []byte, value T, txNum uint64, overwrit // is reserved before the removal (see the update path above). c.currentSize.Add(int64(newSize)) if hasExisting { - lru.Remove(h) + gen.lru.Remove(h) } keyCopy := bytes.Clone(key) - if lru.Add(h, entry[T]{key: keyCopy, val: value, size: newSize, txNum: txNum, epoch: ep}) { + if gen.add(h, entry[T]{key: keyCopy, val: value, size: newSize, txNum: txNum, epoch: ep}) { c.evictions.Add(1) } c.inserts.Add(1) @@ -457,9 +478,9 @@ func (c *GenericCache[T]) Delete(key []byte) { mu := &c.putStripes[h&(putStripeCount-1)] mu.Lock() defer mu.Unlock() - lru := c.data.Load() - if existing, ok := lru.Get(h); ok && bytes.Equal(existing.key, key) { - lru.Remove(h) + gen := c.data.Load() + if existing, ok := gen.lru.Get(h); ok && bytes.Equal(existing.key, key) { + gen.lru.Remove(h) } } @@ -470,9 +491,9 @@ func (c *GenericCache[T]) dropStale(h uint64, key []byte) { mu := &c.putStripes[h&(putStripeCount-1)] mu.Lock() defer mu.Unlock() - lru := c.data.Load() - if e, ok := lru.Get(h); ok && bytes.Equal(e.key, key) && c.coh.IsStale(e.txNum, e.epoch) { - lru.Remove(h) + gen := c.data.Load() + if e, ok := gen.lru.Get(h); ok && bytes.Equal(e.key, key) && c.coh.IsStale(e.txNum, e.epoch) { + gen.lru.Remove(h) } } @@ -533,7 +554,7 @@ func (c *GenericCache[T]) Unwind(unwindToTxNum uint64) { // Len returns the number of entries in the cache. func (c *GenericCache[T]) Len() int { - return c.data.Load().Len() + return c.data.Load().len() } // SizeBytes returns the current size of the cache in bytes. @@ -566,7 +587,7 @@ func (c *GenericCache[T]) PrintStatsAndReset(name string) { "hits", hits, "misses", misses, "hit_rate", hitRate, "inserts", inserts, "evictions", evictions, "dropped", dropped, "stale_evicted", staleEvicted, "epoch", c.coh.Epoch(), - "entries", c.data.Load().Len(), "size_mb", sizeBytes/(1024*1024), + "entries", c.data.Load().len(), "size_mb", sizeBytes/(1024*1024), "capacity_mb", int64(c.capacityB/datasize.MB), "usage_pct", usagePct, ) } diff --git a/execution/cache/generic_cache_concurrency_test.go b/execution/cache/generic_cache_concurrency_test.go index a4bdc74a1ea..82dd4217763 100644 --- a/execution/cache/generic_cache_concurrency_test.go +++ b/execution/cache/generic_cache_concurrency_test.go @@ -417,3 +417,49 @@ func TestGenericCache_StatsResetAtomicWithDelete_NoPhantomEvictions(t *testing.T total += c.evictions.Swap(0) require.Zero(t, total, "intentional removals surfaced in the evictions metric") } + +// The entry counter is maintained by the insert path (+1) and freelru's +// OnEvict (-1) instead of by locking every shard, so it must still agree with +// the LRU's real length after concurrent inserts, updates, deletes and +// capacity evictions have raced across several grow steps. +func TestGenericCache_LenCounterUnderConcurrency(t *testing.T) { + c := closeOnCleanup(t, NewGenericCacheWithAvg[[]byte](8*datasize.MB, 256, func(v []byte) int { return len(v) }, ModeEvictLRU)) + + const workers = 8 + const perWorker = 4000 + var wg sync.WaitGroup + for w := range workers { + wg.Go(func() { + k := make([]byte, 8) + for i := range perWorker { + binary.BigEndian.PutUint64(k, uint64(w*perWorker+i)) + c.Put(k, []byte("v"), uint64(i)) + binary.BigEndian.PutUint64(k, uint64(w*perWorker+i/2)) + c.Put(k, []byte("updated"), uint64(i)) + if i%16 == 0 { + binary.BigEndian.PutUint64(k, uint64(w*perWorker+i/4)) + c.Delete(k) + } + } + }) + } + wg.Wait() + + require.Equal(t, c.data.Load().lru.Len(), c.Len(), "entry counter drifted from the LRU") +} + +// Parallel inserts into a cache that is still below its ceiling — the window +// where every insert runs the grow check. +func BenchmarkGenericCacheParallelPutGrow(b *testing.B) { + c := NewGenericCacheWithAvg[[]byte](256*datasize.MB, 96, func(v []byte) int { return len(v) }, ModeEvictLRU) + defer c.Close() + var seq atomic.Uint64 + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + k := make([]byte, 8) + for pb.Next() { + binary.BigEndian.PutUint64(k, seq.Add(1)) + c.Put(k, []byte("0123456789abcdef"), 1) + } + }) +}