feat: make lossy buffers dynamically striped

cache.go

@@ -49,7 +49,7 @@ func init() {
 }
 
 type evictionPolicy[K comparable, V any] interface {
-	Read(nodes []node.Node[K, V])
+	Read(n node.Node[K, V])
 	Add(n node.Node[K, V], nowNanos int64)
 	Delete(n node.Node[K, V])
 	MaxAvailableWeight() uint64
@@ -73,15 +73,14 @@ type Cache[K comparable, V any] struct {
 	stats          statsRecorder
 	logger         Logger
 	clock          *clock.Clock
-	stripedBuffer  []*lossy.Buffer[K, V]
+	stripedBuffer  *lossy.Striped[K, V]
 	writeBuffer    *queue.Growable[task[K, V]]
 	evictionMutex  sync.Mutex
 	closeOnce      sync.Once
 	doneClear      chan struct{}
 	doneClose      chan struct{}
 	weigher        func(key K, value V) uint32
 	onDeletion     func(e DeletionEvent[K, V])
-	mask           uint32
 	ttl            time.Duration
 	withExpiration bool
 	withEviction   bool
@@ -99,12 +98,9 @@ func newCache[K comparable, V any](b *Builder[K, V]) *Cache[K, V] {
 	maximum := b.getMaximum()
 	withEviction := maximum != nil
 
-	var stripedBuffer []*lossy.Buffer[K, V]
+	var stripedBuffer *lossy.Striped[K, V]
 	if withEviction {
-		stripedBuffer = make([]*lossy.Buffer[K, V], 0, maxStripedBufferSize)
-		for i := 0; i < maxStripedBufferSize; i++ {
-			stripedBuffer = append(stripedBuffer, lossy.New[K, V](nodeManager))
-		}
+		stripedBuffer = lossy.NewStriped(maxStripedBufferSize, nodeManager)
 	}
 
 	var hashmap *hashtable.Map[K, V]
@@ -122,10 +118,8 @@ func newCache[K comparable, V any](b *Builder[K, V]) *Cache[K, V] {
 		stripedBuffer: stripedBuffer,
 		doneClear:     make(chan struct{}),
 		doneClose:     make(chan struct{}, 1),
-		//nolint:gosec // there will never be an overflow
-		mask:       uint32(maxStripedBufferSize - 1),
-		weigher:    b.getWeigher(),
-		onDeletion: b.onDeletion,
+		weigher:       b.getWeigher(),
+		onDeletion:    b.onDeletion,
 	}
 
 	cache.withEviction = withEviction
@@ -166,10 +160,6 @@ func newCache[K comparable, V any](b *Builder[K, V]) *Cache[K, V] {
 	return cache
 }
 
-func (c *Cache[K, V]) getReadBufferIdx() int {
-	return int(xruntime.Fastrand() & c.mask)
-}
-
 func (c *Cache[K, V]) getExpiration(duration time.Duration) int64 {
 	return c.clock.Offset() + duration.Nanoseconds()
 }
@@ -234,14 +224,10 @@ func (c *Cache[K, V]) afterGet(got node.Node[K, V]) {
 		return
 	}
 
-	idx := c.getReadBufferIdx()
-	pb := c.stripedBuffer[idx].Add(got)
-	if pb != nil {
-		c.evictionMutex.Lock()
-		c.policy.Read(pb.Returned)
+	result := c.stripedBuffer.Add(got)
+	if result == lossy.Full && c.evictionMutex.TryLock() {
+		c.stripedBuffer.DrainTo(c.policy.Read)
 		c.evictionMutex.Unlock()
-
-		c.stripedBuffer[idx].Free()
 	}
 }
 
@@ -521,9 +507,7 @@ func (c *Cache[K, V]) clear(t task[K, V]) {
 	}
 
 	if c.withEviction {
-		for i := 0; i < len(c.stripedBuffer); i++ {
-			c.stripedBuffer[i].Clear()
-		}
+		c.stripedBuffer.Clear()
 	}
 
 	c.writeBuffer.Push(t)

cache_test.go

@@ -105,7 +105,7 @@ func TestCache_PinnedWeight(t *testing.T) {
 			}
 			return 1
 		}).
-		WithTTL(3 * time.Second).
+		WithTTL(2 * time.Second).
 		OnDeletion(func(e DeletionEvent[int, int]) {
 			mutex.Lock()
 			m[e.Cause]++
@@ -127,9 +127,6 @@ func TestCache_PinnedWeight(t *testing.T) {
 	}
 	for i := size; i < 2*size; i++ {
 		c.Set(i, i)
-	}
-	time.Sleep(time.Second)
-	for i := size; i < 2*size; i++ {
 		if !c.Has(i) {
 			t.Fatalf("the key must exist: %d", i)
 		}
@@ -139,7 +136,7 @@ func TestCache_PinnedWeight(t *testing.T) {
 		t.Fatalf("the key must exist: %d", pinned)
 	}
 
-	time.Sleep(3 * time.Second)
+	time.Sleep(4 * time.Second)
 
 	if c.Has(pinned) {
 		t.Fatalf("the key must not exist: %d", pinned)

internal/eviction/disabled.go

@@ -12,7 +12,7 @@ func NewDisabled[K comparable, V any]() Disabled[K, V] {
 	return Disabled[K, V]{}
 }
 
-func (d Disabled[K, V]) Read(nodes []node.Node[K, V]) {
+func (d Disabled[K, V]) Read(nodes node.Node[K, V]) {
 	panic("not implemented")
 }
 

internal/eviction/s3fifo/policy.go

@@ -48,10 +48,8 @@ func NewPolicy[K comparable, V any](maxWeight uint64, evictNode func(node.Node[K
 }
 
 // Read updates the eviction policy based on node accesses.
-func (p *Policy[K, V]) Read(nodes []node.Node[K, V]) {
-	for _, n := range nodes {
-		n.IncrementFrequency()
-	}
+func (p *Policy[K, V]) Read(n node.Node[K, V]) {
+	n.IncrementFrequency()
 }
 
 // Add adds node to the eviction policy.

internal/eviction/s3fifo/policy_test.go

@@ -26,6 +26,12 @@ func newNode(k int) node.Node[int, int] {
 	return n
 }
 
+func read[K comparable, V any](p *Policy[K, V], nodes []node.Node[K, V]) {
+	for _, n := range nodes {
+		p.Read(n)
+	}
+}
+
 func TestPolicy_ReadAndWrite(t *testing.T) {
 	n := newNode(2)
 	p := NewPolicy[int, int](10, func(n node.Node[int, int]) {
@@ -58,9 +64,9 @@ func TestPolicy_OneHitWonders(t *testing.T) {
 		p.Add(n, 1)
 	}
 
-	p.Read(oneHitWonders)
+	read(p, oneHitWonders)
 	for i := 0; i < 3; i++ {
-		p.Read(popular)
+		read(p, popular)
 	}
 
 	newNodes := make([]node.Node[int, int], 0, 11)
@@ -117,7 +123,8 @@ func TestPolicy_Update(t *testing.T) {
 	p.Delete(n)
 	p.Add(n1, 1)
 
-	p.Read([]node.Node[int, int]{n1, n1})
+	p.Read(n1)
+	p.Read(n1)
 
 	n2 := m.Create(2, 1, 0, 92)
 	collect = true