From 79971d22e2bb1d81f93f95c66dce0e7f7403bfb0 Mon Sep 17 00:00:00 2001 From: Omar Polo Date: Sun, 12 Jul 2026 14:20:32 +0000 Subject: [PATCH 1/2] caching/fifo: add a FIFO strategy with the rework in parallel of caching/lru to become a *real* LRU cache, introduce a FIFO as well so existing consumers can keep the same behavior. The APIs exposed by this package and caching/lru are the same. --- caching/fifo/fifo.go | 140 ++++++++++++++++++++++++++ caching/fifo/fifo_test.go | 203 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 343 insertions(+) create mode 100644 caching/fifo/fifo.go create mode 100644 caching/fifo/fifo_test.go diff --git a/caching/fifo/fifo.go b/caching/fifo/fifo.go new file mode 100644 index 00000000..21d505ec --- /dev/null +++ b/caching/fifo/fifo.go @@ -0,0 +1,140 @@ +package fifo + +import ( + "sync" + "sync/atomic" +) + +// Cache implements a FIFO caching strategy. +type Cache[K comparable, V any] struct { + mtx sync.RWMutex + + size int + + items map[K]V + + buffer []K + head int + tail int + + onevict func(K, V) error + + hits atomic.Uint64 + misses atomic.Uint64 +} + +// New constructs a new FIFO cache of the given size. The optional +// callback onevict is called when an item is evicted: if it retuns an +// error, the operation is aborted. +func New[K comparable, V any](size int, onevict func(K, V) error) *Cache[K, V] { + // since we're working in modulo fashion, bump the size by + // one, so that we actually can hold `size' items. Also, + // silently bump the size if it's non-sensical. + size = max(size, 1) + 1 + + return &Cache[K, V]{ + size: size, + items: make(map[K]V, size), + buffer: make([]K, size), + tail: -1, + onevict: onevict, + } +} + +func (c *Cache[K, V]) put(key K, val V) error { + // special case: on duplicate, we override the value and leave + // it in its place. + if _, ok := c.items[key]; ok { + c.items[key] = val + return nil + } + + nexthead := (c.head + 1) % c.size + + if nexthead == c.tail { + oldk := c.buffer[c.tail] + if c.onevict != nil { + if err := c.onevict(oldk, c.items[oldk]); err != nil { + return err + } + } + delete(c.items, oldk) + } + c.buffer[c.head] = key + c.items[key] = val + + c.head = nexthead + if c.tail == -1 || c.tail == c.head { + c.tail = (c.tail + 1) % c.size + } + + return nil +} + +// Put inserts an item in the cache. If the cache is full, the oldest +// element is evicted. Put can fail only if the onevict callback +// returns an error. +func (c *Cache[K, V]) Put(key K, val V) error { + c.mtx.Lock() + err := c.put(key, val) + c.mtx.Unlock() + return err +} + +// Get retrieves the given key from the cache and returns its value +// and a boolean indicating whether it was found. +func (c *Cache[K, V]) Get(key K) (V, bool) { + c.mtx.RLock() + val, ok := c.items[key] + c.mtx.RUnlock() + + if ok { + c.hits.Add(1) + } else { + c.misses.Add(1) + } + + return val, ok +} + +// Close empties the cache, and calls onevict on all pending items. +// It is safe to re-use a cache after Close was called assuming the +// size and onevict are okay, as it really is a reset operation. If +// onevict fails, Close keeps evicting all the other items, and +// returns the last error occurred. +func (c *Cache[K, V]) Close() error { + c.mtx.Lock() + defer c.mtx.Unlock() + + var err error + if c.onevict != nil { + for key, val := range c.items { + if e := c.onevict(key, val); e != nil { + err = e + } + } + } + + // reset + c.tail = -1 + c.head = 0 + clear(c.items) + clear(c.buffer) + + return err +} + +// Stats return the number of cache hit, misses and the size of the +// cache. It does *not* reset them. +func (c *Cache[K, V]) Stats() (hit, miss, size uint64) { + c.mtx.RLock() + head := uint64(c.head) + tail := c.tail + c.mtx.RUnlock() + + if tail != -1 { + size = (head - uint64(tail) + uint64(c.size)) % uint64(c.size) + } + + return c.hits.Load(), c.misses.Load(), size +} diff --git a/caching/fifo/fifo_test.go b/caching/fifo/fifo_test.go new file mode 100644 index 00000000..8f661aad --- /dev/null +++ b/caching/fifo/fifo_test.go @@ -0,0 +1,203 @@ +package fifo + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNew(t *testing.T) { + cache := New[int, string](10, nil) + require.NotNil(t, cache) + require.Equal(t, 11, cache.size) // works in modulo, so size is bumped + require.NotNil(t, cache.buffer) + require.NotNil(t, cache.items) +} + +func TestPutAndGet(t *testing.T) { + cache := New[int, string](2, nil) + + err := cache.Put(1, "one") + require.NoError(t, err) + + // validate head and tail + require.Equal(t, 1, cache.head) + require.Equal(t, 0, cache.tail) + + val, ok := cache.Get(1) + require.True(t, ok) + require.Equal(t, "one", val) + + // non-existent + val, ok = cache.Get(2) + require.False(t, ok) + require.Empty(t, val) +} + +func TestPutOverrides(t *testing.T) { + cache := New[int, string](2, nil) + + var err error + + err = cache.Put(1, "one") + require.NoError(t, err) + + err = cache.Put(1, "ONE") + require.NoError(t, err) + + val, ok := cache.Get(1) + require.True(t, ok) + require.Equal(t, "ONE", val) +} + +func TestPutEvicts(t *testing.T) { + cache := New[int, string](2, nil) + + var err error + + // Fill the cache + err = cache.Put(1, "one") + require.NoError(t, err) + err = cache.Put(2, "two") + require.NoError(t, err) + + // validate head and tail + require.Equal(t, 2, cache.head) + require.Equal(t, 0, cache.tail) + + // inserting 3 should work + err = cache.Put(3, "three") + require.NoError(t, err) + + // one should be evicted + val, ok := cache.Get(1) + require.False(t, ok) + require.Empty(t, val) + + // but two and three should still be there + val, ok = cache.Get(2) + require.True(t, ok) + require.Equal(t, "two", val) + + val, ok = cache.Get(3) + require.True(t, ok) + require.Equal(t, "three", val) + + // (re)validate head and tail + require.Equal(t, 0, cache.head) + require.Equal(t, 1, cache.tail) + +} + +func TestPutCallsOnEvict(t *testing.T) { + evicted := make(map[int]string) + onevict := func(key int, val string) error { + evicted[key] = val + return nil + } + + cache := New(2, onevict) + + var err error + + // Fill the cache + err = cache.Put(1, "one") + require.NoError(t, err) + err = cache.Put(2, "two") + require.NoError(t, err) + + // inserting 3 should evict one + err = cache.Put(3, "three") + require.NoError(t, err) + + // check that the eviction happened, and only one was evicted. + require.Equal(t, "one", evicted[1]) + require.Equal(t, 1, len(evicted)) +} + +func TestPutFailsOnEvictFailure(t *testing.T) { + expectedErr := errors.New("eviction error") + + cache := New(2, func(key int, val string) error { return expectedErr }) + + var err error + + // Fill the cache + err = cache.Put(1, "one") + require.NoError(t, err) + err = cache.Put(2, "two") + require.NoError(t, err) + + // This should fail since it hits onevict + err = cache.Put(3, "three") + require.Error(t, err) + require.ErrorIs(t, err, expectedErr) +} + +func TestCloseFailsOnEvictFailure(t *testing.T) { + expectedErr := errors.New("eviction error") + + cache := New(2, func(key int, val string) error { return expectedErr }) + + var err error + + // Fill the cache + err = cache.Put(1, "one") + require.NoError(t, err) + err = cache.Put(2, "two") + require.NoError(t, err) + + // This should fail since it hits onevict + err = cache.Close() + require.Error(t, err) + require.ErrorIs(t, err, expectedErr) +} + +func TestStats(t *testing.T) { + cache := New[int, string](2, nil) + + hits, misses, size := cache.Stats() + require.Zero(t, hits) + require.Zero(t, misses) + require.Zero(t, size) + + var ( + err error + val string + ok bool + ) + + // Add some items + err = cache.Put(1, "one") + require.NoError(t, err) + + // test updated stats + hits, misses, size = cache.Stats() + require.Zero(t, hits) + require.Zero(t, misses) + require.Equal(t, uint64(1), size) + + err = cache.Put(2, "two") + require.NoError(t, err) + + hits, misses, size = cache.Stats() + require.Zero(t, hits) + require.Zero(t, misses) + require.Equal(t, uint64(2), size) + + // test some gets + val, ok = cache.Get(1) // hit + require.True(t, ok) + require.Equal(t, "one", val) + + val, ok = cache.Get(3) // miss + require.False(t, ok) + require.Zero(t, val) + + // test updated stats + hits, misses, size = cache.Stats() + require.Equal(t, uint64(1), hits) + require.Equal(t, uint64(1), misses) + require.Equal(t, uint64(2), size) +} From 2e5327ab29c68f69539a592bd7ebc1e37233e4ac Mon Sep 17 00:00:00 2001 From: Omar Polo Date: Sun, 12 Jul 2026 14:51:08 +0000 Subject: [PATCH 2/2] caching/lru: add some parallel tests --- caching/fifo/fifo_test.go | 78 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/caching/fifo/fifo_test.go b/caching/fifo/fifo_test.go index 8f661aad..0098ce6e 100644 --- a/caching/fifo/fifo_test.go +++ b/caching/fifo/fifo_test.go @@ -2,6 +2,7 @@ package fifo import ( "errors" + "sync" "testing" "github.com/stretchr/testify/require" @@ -201,3 +202,80 @@ func TestStats(t *testing.T) { require.Equal(t, uint64(1), misses) require.Equal(t, uint64(2), size) } + +func TestDisjointKeySpaceConcurrentPutGet(t *testing.T) { + const ( + capacity = 64 + workers = 16 + opsPerWorker = 2000 + ) + + cache := New[int, int](capacity, nil) + defer cache.Close() + + var wg sync.WaitGroup + + for w := range workers { + wg.Go(func() { + for i := range opsPerWorker { + key := w*opsPerWorker + i + err := cache.Put(key, key*2) + require.NoError(t, err, "Put(%d)", key) + + val, ok := cache.Get(key) + // cannot assert ok being true because + // the key could have been evicted by + // a concurrenc goroutine. + if ok { + require.Equal(t, key*2, val, "Get(%d)", key) + } + + _, _, size := cache.Stats() + require.LessOrEqual(t, size, uint64(capacity)) + } + }) + } + + wg.Wait() + + hits, misses, size := cache.Stats() + require.LessOrEqual(t, size, uint64(capacity)) + require.Greater(t, hits+misses, uint64(0)) +} + +func TestConcurrentSharedKeySpacePutGet(t *testing.T) { + const ( + capacity = 8 + keyspace = 4 // smaller than capacity: contention on the same slots via the update path + workers = 16 + iterations = 2000 + ) + + cache := New[int, int](capacity, nil) + defer cache.Close() + + var wg sync.WaitGroup + + for w := range workers { + wg.Go(func() { + for i := range iterations { + key := i % keyspace + err := cache.Put(key, w) + require.NoError(t, err, "Put(%d)", key) + + // don't attempt to Get(key) here + // since it could either be evicted or + // overwritten by another goroutine + // concurrently. + + _, _, size := cache.Stats() + require.LessOrEqual(t, size, uint64(capacity)) + } + }) + } + + wg.Wait() + + _, _, size := cache.Stats() + require.LessOrEqual(t, size, uint64(capacity)) +}