Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
46 changes: 34 additions & 12 deletions cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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.
Expand Down Expand Up @@ -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
Expand All @@ -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
}
Expand All @@ -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)
Expand Down Expand Up @@ -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 {
Expand Down
18 changes: 4 additions & 14 deletions loader.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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)

Expand All @@ -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 {
Expand All @@ -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,
Expand Down Expand Up @@ -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:
Expand Down
13 changes: 4 additions & 9 deletions loader_batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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 {
Expand All @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
30 changes: 23 additions & 7 deletions shard.go
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down Expand Up @@ -153,17 +152,17 @@ 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
}

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)
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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()
Expand Down
2 changes: 0 additions & 2 deletions stats.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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)
Expand Down
7 changes: 3 additions & 4 deletions view.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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()
Expand Down
Loading
Loading