diff --git a/CHANGELOG.md b/CHANGELOG.md index a63b53e..a789998 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ While the major version is 0, the public API may change in any release. ## [Unreleased] +### Changed + +- `Len` and `Bytes` read only the shard sizes instead of taking a full `Stats` snapshot. + ## [0.8.0] - 2026-10-09 ### Added diff --git a/cache.go b/cache.go index b848e04..15ee6db 100644 --- a/cache.go +++ b/cache.go @@ -331,25 +331,25 @@ func (c *Cache[K, V]) Set(key K, value V) error { // SetTTL caches value under key for ttl, overriding Options.TTL. A ttl of zero // means the entry never expires on its own. func (c *Cache[K, V]) SetTTL(key K, value V, ttl time.Duration) error { - return c.core.setValue(key, value, c.core.valueCost(value), ttl, c.core.refreshAfter, refreshCache) + return c.core.setValue(key, value, c.core.valueCost(value), ttl, c.core.refreshAfter, refreshCache, nil) } // SetWithCost is Set with the value's cost given instead of taken from // Options.Cost, for a value whose size is known only where it was produced. A // negative cost counts as zero. func (c *Cache[K, V]) SetWithCost(key K, value V, cost int64) error { - return c.core.setValue(key, value, max(cost, 0), c.core.ttl, c.core.refreshAfter, refreshCache) + return c.core.setValue(key, value, max(cost, 0), c.core.ttl, c.core.refreshAfter, refreshCache, nil) } // SetNegative records that the upstream reports no such key, for the configured // NegativeTTL. A successful write supersedes outstanding loads. func (c *Cache[K, V]) SetNegative(key K) error { - return c.core.setNegative(key, c.core.negativeTTL) + return c.core.setNegative(key, c.core.negativeTTL, nil) } // SetNegativeTTL is SetNegative with an explicit lifetime. func (c *Cache[K, V]) SetNegativeTTL(key K, ttl time.Duration) error { - return c.core.setNegative(key, ttl) + return c.core.setNegative(key, ttl, nil) } // Delete removes key and reports whether it was present. It also invalidates @@ -371,10 +371,26 @@ func (c *Cache[K, V]) Clear() { } // Len reports how many entries are held, including expired ones not yet swept. -func (c *Cache[K, V]) Len() int { return c.Stats().Entries } +func (c *Cache[K, V]) Len() int { + n := 0 + for _, s := range c.core.shards { + entries, _ := s.stats() + n += entries + } + + return n +} // Bytes reports the summed cost of the entries held. -func (c *Cache[K, V]) Bytes() int64 { return c.Stats().Bytes } +func (c *Cache[K, V]) Bytes() int64 { + var n int64 + for _, s := range c.core.shards { + _, bytes := s.stats() + n += bytes + } + + return n +} // Stats returns a snapshot of the counters. It walks every shard, so poll it on // a metrics interval rather than per request. @@ -428,7 +444,7 @@ func (c *core[K, V]) valueCost(v V) int64 { // read by owner may refresh it; zero, or anything not shorter than ttl, means // never. func (c *core[K, V]) setValue( - key K, value V, cost int64, ttl, refreshAfter time.Duration, owner refresher, tokens ...*loadToken, + key K, value V, cost int64, ttl, refreshAfter time.Duration, owner refresher, token *loadToken, ) error { e := &entry[K, V]{key: key, value: value, cost: cost, byView: owner == refreshView} var refreshAt int64 @@ -438,10 +454,10 @@ func (c *core[K, V]) setValue( e.refreshAt.Store(refreshAt) } - return c.store(e, tokens...) + return c.store(e, token) } -func (c *core[K, V]) setNegative(key K, ttl time.Duration, tokens ...*loadToken) error { +func (c *core[K, V]) setNegative(key K, ttl time.Duration, token *loadToken) error { if ttl <= 0 { return ErrNegativeDisabled } @@ -457,13 +473,13 @@ func (c *core[K, V]) setNegative(key K, ttl time.Duration, tokens ...*loadToken) cost: cost, expiresAt: c.expiryAt(ttl), negative: true, - }, tokens...) + }, token) } -func (c *core[K, V]) store(e *entry[K, V], tokens ...*loadToken) error { +func (c *core[K, V]) store(e *entry[K, V], token *loadToken) error { s := c.shardFor(e.key) - replaced, victims, err := s.set(e, tokens...) + replaced, victims, err := s.set(e, token) if err != nil { if c.countStats { s.counters.rejections.Add(1) @@ -552,6 +568,12 @@ func (c *core[K, V]) shardFor(key K) *shard[K, V] { return c.shards[c.shardIndex(key)] } +func (c *core[K, V]) flightFor(key K) (*flightGroup[K, V], *shard[K, V]) { + i := c.shardIndex(key) + + return c.flights[i], c.shards[i] +} + // now is the time expiry is judged against: the wall clock, or what the clock // goroutine last read when ClockGranularity asked for one. func (c *core[K, V]) now() int64 { diff --git a/loader.go b/loader.go index 18b8e73..37a6a3e 100644 --- a/loader.go +++ b/loader.go @@ -172,8 +172,7 @@ func (g *flightGroup[K, V]) drop(key K, cl *call[V]) { } func (c *core[K, V]) load(ctx context.Context, key K, loader loadFunc[K, V]) (V, error) { - idx := c.shardIndex(key) - g, s := c.flights[idx], c.shards[idx] + g, s := c.flightFor(key) g.mu.Lock() if cl := g.running(key); cl != nil { @@ -184,13 +183,8 @@ func (c *core[K, V]) load(ctx context.Context, key K, loader loadFunc[K, V]) (V, return g.wait(ctx, key, cl) } - token := s.beginLoad(key) - - // A load that finished since the caller's lookup has left the map but - // published first, so looking again under the lock makes "one load per key" - // exact. Not counted: the caller's lookup already was. - if v, st, _, _ := s.get(key, c.now(), refreshNone); st != StatusMiss { - s.endLoad(key, token) + token, v, st := s.beginLoadUnlessCached(key, c.now(), nil) + if token == nil { g.mu.Unlock() c.countCoalesced(s) @@ -213,8 +207,7 @@ func (c *core[K, V]) load(ctx context.Context, key K, loader loadFunc[K, V]) (V, // refresh reloads key in the background ahead of its expiry, as an ordinary // flight that callers may join, unless a load of the key is already running. func (c *core[K, V]) refresh(ctx context.Context, key K, loader loadFunc[K, V]) { - idx := c.shardIndex(key) - g, s := c.flights[idx], c.shards[idx] + g, s := c.flightFor(key) g.mu.Lock() if g.running(key) != nil { @@ -236,8 +229,6 @@ func (c *core[K, V]) countCoalesced(s *shard[K, V]) { } } -// run performs the load and publishes the result; refresh says it is counted as -// a refresh. func (c *core[K, V]) run( ctx context.Context, g *flightGroup[K, V], s *shard[K, V], key K, cl *call[V], loader loadFunc[K, V], refresh bool, @@ -280,7 +271,6 @@ func (g *flightGroup[K, V]) run(key K, cl *call[V], load func() (V, error), reco cl.val, cl.err = load() } -// wait blocks until the load finishes or ctx is done, whichever comes first. func (g *flightGroup[K, V]) wait(ctx context.Context, key K, cl *call[V]) (V, error) { select { case <-cl.done: diff --git a/loader_batch.go b/loader_batch.go index 1accdda..30e8853 100644 --- a/loader_batch.go +++ b/loader_batch.go @@ -118,8 +118,7 @@ func (b *batchLoad) join() context.CancelFunc { // call's cancel is set before anyone else can see the call. func (c *core[K, V]) acquire(ctx context.Context, keys []K, found map[K]V) (waits, fresh []flight[K, V], batch *batchLoad) { for _, key := range keys { - idx := c.shardIndex(key) - g, s := c.flights[idx], c.shards[idx] + g, s := c.flightFor(key) g.mu.Lock() if cl := g.running(key); cl != nil { @@ -131,10 +130,8 @@ func (c *core[K, V]) acquire(ctx context.Context, keys []K, found map[K]V) (wait continue } - // The same recheck as load's. - token := s.beginLoad(key) - if v, st, _, _ := s.get(key, c.now(), refreshNone); st != StatusMiss { - s.endLoad(key, token) + token, v, st := s.beginLoadUnlessCached(key, c.now(), nil) + if token == nil { g.mu.Unlock() c.countCoalesced(s) if st == StatusHit { @@ -144,7 +141,6 @@ func (c *core[K, V]) acquire(ctx context.Context, keys []K, found map[K]V) (wait continue } - // Values, but not cancellation, as in startAndUnlock. cl := &call[V]{done: make(chan struct{}), waiters: 1, token: token} var loadCtx context.Context if c.batchLoader != nil { @@ -182,8 +178,7 @@ func (c *core[K, V]) refreshMany(ctx context.Context, keys []K) { batch := &batchLoad{} batch.ctx, batch.cancel = context.WithCancel(context.WithoutCancel(ctx)) for _, key := range keys { - idx := c.shardIndex(key) - g, s := c.flights[idx], c.shards[idx] + g, s := c.flightFor(key) g.mu.Lock() if g.running(key) != nil { diff --git a/shard.go b/shard.go index f2ce59a..e446dc9 100644 --- a/shard.go +++ b/shard.go @@ -51,7 +51,6 @@ func (e *entry[K, V]) expired(now int64) bool { return e.expiresAt != 0 && now >= e.expiresAt } -// shard is an independently locked slice of the cache. type shard[K comparable, V any] struct { mu sync.RWMutex items map[K]*entry[K, V] @@ -153,9 +152,9 @@ func (s *shard[K, V]) get(key K, now int64, by refresher) (_ V, _ Status, expire return e.value, StatusHit, false, refresh } -// set stores e, returning the entry it replaced and those evicted to make room. -// A value that cannot fit on its own is refused with ErrTooLarge. -func (s *shard[K, V]) set(e *entry[K, V], tokens ...*loadToken) (replaced *entry[K, V], victims []*entry[K, V], err error) { +// set stores e, unless token was invalidated since its load began: then it +// stores nothing and reports no error. +func (s *shard[K, V]) set(e *entry[K, V], token *loadToken) (replaced *entry[K, V], victims []*entry[K, V], err error) { if s.maxBytes > 0 && e.cost > s.maxBytes { return nil, nil, ErrTooLarge } @@ -163,7 +162,7 @@ func (s *shard[K, V]) set(e *entry[K, V], tokens ...*loadToken) (replaced *entry s.mu.Lock() defer s.mu.Unlock() - if len(tokens) > 0 && !tokens[0].valid.Load() { + if token != nil && !token.valid.Load() { return nil, nil, nil } s.invalidateLoads(e.key) @@ -194,7 +193,6 @@ func (s *shard[K, V]) delete(key K) (*entry[K, V], bool) { return e, true } -// sweep drops every entry that has expired by now. func (s *shard[K, V]) sweep(now int64) []*entry[K, V] { s.mu.Lock() defer s.mu.Unlock() @@ -265,7 +263,7 @@ func (s *shard[K, V]) over() bool { (s.maxEntries > 0 && len(s.items) > s.maxEntries) } -// remove unlinks e from both the map and the LRU list. Callers hold the lock. +// remove is called with s.mu held, as are the list helpers below. func (s *shard[K, V]) remove(e *entry[K, V]) { delete(s.items, e.key) s.unlink(e) @@ -324,6 +322,24 @@ func (s *shard[K, V]) beginLoad(key K) *loadToken { return token } +// beginLoadUnlessCached is beginLoad unless key has gained an answer since the +// caller's lookup, which it returns with a nil token instead. The caller holds +// the key's flight group lock: a load that finished in between has left the +// flight map but published first, and looking again makes "one load per key" +// exact. accept, if set, says which hits count as an answer. The lookup is not +// counted: the caller's already was. +func (s *shard[K, V]) beginLoadUnlessCached(key K, now int64, accept func(V) bool) (*loadToken, V, Status) { + token := s.beginLoad(key) + v, st, _, _ := s.get(key, now, refreshNone) + if st == StatusNegative || (st == StatusHit && (accept == nil || accept(v))) { + s.endLoad(key, token) + + return nil, v, st + } + + return token, v, StatusMiss +} + func (s *shard[K, V]) endLoad(key K, token *loadToken) { s.mu.Lock() defer s.mu.Unlock() diff --git a/stats.go b/stats.go index 36e3ae2..ad918c1 100644 --- a/stats.go +++ b/stats.go @@ -73,7 +73,6 @@ func since(cur, prev int64) int64 { return cur - prev } -// addTo folds one shard's counters into a snapshot. func (c *counters) addTo(s *Stats) { s.Hits += c.hits.Load() s.Misses += c.misses.Load() @@ -92,7 +91,6 @@ func (c *counters) addTo(s *Stats) { s.RefreshErrors += c.refreshErrors.Load() } -// countLoad records one completed load, or refresh when it was one. func (c *counters) countLoad(o loadOutcome, refresh bool) { if refresh { c.refreshes.Add(1) diff --git a/view.go b/view.go index b3314be..68c9843 100644 --- a/view.go +++ b/view.go @@ -207,7 +207,7 @@ func (v *View[T]) SetTTL(key string, value T, ttl time.Duration) error { return ErrDisabled } - return v.cache.core.setValue(v.prefix+key, value, v.valueCost(value), ttl, v.refreshAfter, refreshView) + return v.cache.core.setValue(v.prefix+key, value, v.valueCost(value), ttl, v.refreshAfter, refreshView, nil) } // SetWithCost is Cache.SetWithCost for this view, with the view's TTL. @@ -216,7 +216,7 @@ func (v *View[T]) SetWithCost(key string, value T, cost int64) error { return ErrDisabled } - return v.cache.core.setValue(v.prefix+key, value, max(cost, 0), v.ttl, v.refreshAfter, refreshView) + return v.cache.core.setValue(v.prefix+key, value, max(cost, 0), v.ttl, v.refreshAfter, refreshView, nil) } // SetNegative records that the upstream reports no such key, for the view's @@ -231,7 +231,7 @@ func (v *View[T]) SetNegativeTTL(key string, ttl time.Duration) error { return ErrDisabled } - return v.cache.core.setNegative(v.prefix+key, ttl) + return v.cache.core.setNegative(v.prefix+key, ttl, nil) } // Delete is Cache.Delete for this view's key, invalidating its loads in any view @@ -264,7 +264,6 @@ func (c *Cache[K, V]) ViewStats() map[string]ViewStats { return stats } -// viewCounters returns the counters of a view name, creating them on first use. func (c *core[K, V]) viewCounters(name string) *counters { c.viewsMu.Lock() defer c.viewsMu.Unlock() diff --git a/view_loader.go b/view_loader.go index 175ff14..21a4366 100644 --- a/view_loader.go +++ b/view_loader.go @@ -94,8 +94,7 @@ func (v *View[T]) loadUncached(ctx context.Context, key string, loader loadFunc[ func (v *View[T]) load(ctx context.Context, key string, loader loadFunc[string, T]) (T, error) { c := v.cache.core fullKey := v.prefix + key - idx := c.shardIndex(fullKey) - g, s := v.flights[idx], c.shards[idx] + g, s := v.flightFor(fullKey) coalesced := func() { v.count(&v.stats.coalesced) v.count(&s.counters.coalesced) @@ -109,12 +108,8 @@ func (v *View[T]) load(ctx context.Context, key string, loader loadFunc[string, return g.wait(ctx, key, cl) } - token := s.beginLoad(fullKey) - // load's recheck; a value of another type still needs loading. - raw, st, _, _ := s.get(fullKey, c.now(), refreshNone) - val, typed := raw.(T) - if st == StatusNegative || (st == StatusHit && typed) { - s.endLoad(fullKey, token) + token, raw, st := s.beginLoadUnlessCached(fullKey, c.now(), isType[T]) + if token == nil { g.mu.Unlock() coalesced() if st == StatusNegative { @@ -123,7 +118,7 @@ func (v *View[T]) load(ctx context.Context, key string, loader loadFunc[string, return zero, ErrNotFound } - return val, nil + return raw.(T), nil } cl := g.startAndUnlock(ctx, key, token, func(ctx context.Context, cl *call[T]) { v.run(ctx, g, s, key, cl, loader, false) @@ -134,10 +129,8 @@ func (v *View[T]) load(ctx context.Context, key string, loader loadFunc[string, // refresh is Cache's refresh for a view's key, with the view's loader. func (v *View[T]) refresh(ctx context.Context, key string, loader loadFunc[string, T]) { - c := v.cache.core fullKey := v.prefix + key - idx := c.shardIndex(fullKey) - g, s := v.flights[idx], c.shards[idx] + g, s := v.flightFor(fullKey) g.mu.Lock() if g.running(key) != nil { @@ -178,3 +171,15 @@ func (v *View[T]) run( } }) } + +func (v *View[T]) flightFor(fullKey string) (*flightGroup[string, T], *shard[string, any]) { + i := v.cache.core.shardIndex(fullKey) + + return v.flights[i], v.cache.core.shards[i] +} + +func isType[T any](v any) bool { + _, ok := v.(T) + + return ok +}