package platform import ( "container/list" "context" "errors" "sync" "time" ) var ErrReadCacheBusy = errors.New("热点读取正在回源,请稍后重试") var errReadCacheLoadAborted = errors.New("缓存回源异常中断") // 仅用于可短暂延迟的读投影。限制条目、正文预算和同时回源数;失败短暂冷却, // 不返回过期成功值,也不允许大量不同 key 将 singleflight 本身变成无界队列。 type BoundedReadCache[K comparable, V any] struct { mu sync.Mutex entries map[K]*readCacheEntry[K, V] lru list.List bytes, loads int maxEntries, maxBytes, maxLoads int ttl time.Duration } type readCacheEntry[K comparable, V any] struct { key K ready chan struct{} value V err error expires time.Time bytes int position *list.Element } func NewBoundedReadCache[K comparable, V any](entries, bytes, loads int, ttl time.Duration) *BoundedReadCache[K, V] { return &BoundedReadCache[K, V]{entries: make(map[K]*readCacheEntry[K, V]), maxEntries: entries, maxBytes: bytes, maxLoads: loads, ttl: ttl} } func (c *BoundedReadCache[K, V]) Get(ctx context.Context, key K, load func(context.Context) (V, int, error)) (V, error) { var zero V if err := ctx.Err(); err != nil { return zero, err } c.mu.Lock() if e := c.entries[key]; e != nil { if ready := e.ready; ready != nil { c.mu.Unlock() select { case <-ctx.Done(): return zero, ctx.Err() case <-ready: return e.value, e.err } } if time.Now().Before(e.expires) { c.lru.MoveToFront(e.position) value, err := e.value, e.err c.mu.Unlock() return value, err } c.remove(e) } if c.loads >= c.maxLoads { c.mu.Unlock() return zero, ErrReadCacheBusy } for len(c.entries) >= c.maxEntries { if !c.evict() { c.mu.Unlock() return zero, ErrReadCacheBusy } } e := &readCacheEntry[K, V]{key: key, ready: make(chan struct{})} c.entries[key] = e c.loads++ c.mu.Unlock() // 让原始 panic 继续传播给调用方,同时释放等待者和容量,避免永久卡住热点 key。 loaded := false defer func() { if loaded { return } c.mu.Lock() defer c.mu.Unlock() c.loads-- e.err = errReadCacheLoadAborted if c.entries[key] == e { delete(c.entries, key) } close(e.ready) e.ready = nil }() value, bytes, err := load(ctx) loaded = true c.mu.Lock() defer c.mu.Unlock() c.loads-- e.value, e.err = value, err ready := e.ready e.ready = nil if c.entries[key] == e { ttl := c.ttl if err != nil { ttl, bytes = time.Second, 256 } bytes = max(bytes, 256) if bytes > c.maxBytes || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { delete(c.entries, key) } else { for c.bytes+bytes > c.maxBytes { if !c.evict() { break } } e.bytes, e.expires = bytes, time.Now().Add(ttl) e.position = c.lru.PushFront(e) c.bytes += bytes } } close(ready) return value, err } func (c *BoundedReadCache[K, V]) evict() bool { if oldest := c.lru.Back(); oldest != nil { c.remove(oldest.Value.(*readCacheEntry[K, V])) return true } return false } func (c *BoundedReadCache[K, V]) remove(e *readCacheEntry[K, V]) { delete(c.entries, e.key) if e.position != nil { c.lru.Remove(e.position) c.bytes -= e.bytes } } func (c *BoundedReadCache[K, V]) Clear() { c.ClearAndCount() } func (c *BoundedReadCache[K, V]) ClearAndCount() int { c.mu.Lock() defer c.mu.Unlock() count := len(c.entries) c.entries = make(map[K]*readCacheEntry[K, V]) c.lru.Init() c.bytes = 0 return count }