package cacheqm
import (
"context"
"fmt"
"sort"
"sync"
"time"
"github.com/google/trillian/quota"
"k8s.io/klog/v2"
)
const (
DefaultMinBatchSize = 100
DefaultMaxCacheEntries = 1000
)
var now = time.Now
type manager struct {
quota.Manager
minBatchSize, maxEntries int
mu sync.Mutex
cache map[quota.Spec]*bucket
evictWg sync.WaitGroup
}
type bucket struct {
tokens int
lastModified time.Time
}
func NewCachedManager(qm quota.Manager, minBatchSize, maxEntries int) (quota.Manager, error) {
switch {
case minBatchSize <= 0:
return nil, fmt.Errorf("invalid minBatchSize: %v", minBatchSize)
case maxEntries <= 0:
return nil, fmt.Errorf("invalid maxEntries: %v", minBatchSize)
}
return &manager{
Manager: qm,
minBatchSize: minBatchSize,
maxEntries: maxEntries,
cache: make(map[quota.Spec]*bucket),
}, nil
}
func (m *manager) GetTokens(ctx context.Context, numTokens int, specs []quota.Spec) error {
m.mu.Lock()
defer m.mu.Unlock()
specsToRefill := []quota.Spec{}
for _, spec := range specs {
bucket, ok := m.cache[spec]
if !ok || bucket.tokens < numTokens {
specsToRefill = append(specsToRefill, spec)
}
}
if len(specsToRefill) != 0 {
defer func() {
m.evictWg.Add(1)
go func() {
m.evict(ctx)
m.evictWg.Done()
}()
}()
tokens := numTokens + m.minBatchSize
if err := m.Manager.GetTokens(ctx, tokens, specsToRefill); err != nil {
return err
}
for _, spec := range specsToRefill {
b, ok := m.cache[spec]
if !ok {
b = &bucket{}
m.cache[spec] = b
}
b.tokens += tokens
}
}
lastModified := now()
for _, spec := range specs {
bucket, ok := m.cache[spec]
if !ok || bucket.tokens < 0 || bucket.tokens < numTokens {
klog.Errorf("Bucket invariants failed for spec %+v: ok = %v, bucket = %+v", spec, ok, bucket)
return nil
}
bucket.tokens -= numTokens
bucket.lastModified = lastModified
}
return nil
}
func (m *manager) evict(ctx context.Context) {
m.mu.Lock()
if len(m.cache) <= m.maxEntries {
m.mu.Unlock()
return
}
var buckets bucketsByTime = make([]specBucket, 0, len(m.cache))
for spec, b := range m.cache {
if spec.Group != quota.Global {
buckets = append(buckets, specBucket{bucket: b, spec: spec})
}
}
sort.Sort(buckets)
wg := sync.WaitGroup{}
evicts := len(m.cache) - m.maxEntries
for i := 0; i < evicts; i++ {
b := buckets[i]
klog.V(1).Infof("Too many tokens cached, returning least recently used (%v tokens for %+v)", b.tokens, b.spec)
delete(m.cache, b.spec)
wg.Add(1)
go func() {
if err := m.Manager.PutTokens(ctx, b.tokens, []quota.Spec{b.spec}); err != nil {
klog.Warningf("Error replenishing tokens from evicted bucket (spec = %+v, bucket = %+v): %v", b.spec, b.bucket, err)
}
wg.Done()
}()
}
m.mu.Unlock()
wg.Wait()
}
func (m *manager) wait() {
m.evictWg.Wait()
}
type specBucket struct {
*bucket
spec quota.Spec
}
type bucketsByTime []specBucket
func (b bucketsByTime) Len() int {
return len(b)
}
func (b bucketsByTime) Less(i, j int) bool {
return b[i].lastModified.Before(b[j].lastModified)
}
func (b bucketsByTime) Swap(i, j int) {
b[i], b[j] = b[j], b[i]
}