Skip to content

cache: move batching to demux so that ringBuffer stores single item #20502

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
serathius merged 1 commit into from
Aug 19, 2025
Merged

cache: move batching to demux so that ringBuffer stores single item #20502

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
27 changes: 19 additions & 8 deletions cache/demux.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ type demux struct {
// activeWatchers & laggingWatchers hold the first revision the watcher still needs (nextRev).
activeWatchers map[*watcher]int64
laggingWatchers map[*watcher]int64
history ringBuffer[*clientv3.Event]
history ringBuffer[[]*clientv3.Event]
resyncInterval time.Duration
}

Expand All @@ -45,7 +45,7 @@ func newDemux(historyWindowSize int, resyncInterval time.Duration) *demux {
return &demux{
activeWatchers: make(map[*watcher]int64),
laggingWatchers: make(map[*watcher]int64),
history: *newRingBuffer(historyWindowSize, func(ev *clientv3.Event) int64 { return ev.Kv.ModRevision }),
history: *newRingBuffer(historyWindowSize, func(batch []*clientv3.Event) int64 { return batch[0].Kv.ModRevision }),
resyncInterval: resyncInterval,
}
}
Expand Down Expand Up @@ -110,7 +110,19 @@ func (d *demux) Broadcast(events []*clientv3.Event) {
d.mu.Lock()
defer d.mu.Unlock()

d.history.Append(events)
batchStart := 0
for end := 1; end < len(events); end++ {
if events[end].Kv.ModRevision != events[batchStart].Kv.ModRevision {
if end > batchStart {
d.history.Append(events[batchStart:end])
}
batchStart = end
}
}
if batchStart < len(events) {
d.history.Append(events[batchStart:])
}

firstRev := events[0].Kv.ModRevision
lastRev := events[len(events)-1].Kv.ModRevision
for w, nextRev := range d.activeWatchers {
Expand All @@ -119,19 +131,18 @@ func (d *demux) Broadcast(events []*clientv3.Event) {
delete(d.activeWatchers, w)
continue
}
start := len(events)
sendStart := len(events)
for i, ev := range events {
if ev.Kv.ModRevision >= nextRev {
start = i
sendStart = i
break
}
}

if start == len(events) {
if sendStart == len(events) {
continue
}

if !w.enqueueEvent(events[start:]) { // overflow → lagging
if !w.enqueueEvent(events[sendStart:]) { // overflow → lagging
d.laggingWatchers[w] = nextRev
delete(d.activeWatchers, w)
} else {
Expand Down
55 changes: 8 additions & 47 deletions cache/ringbuffer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,6 @@

package cache

import "fmt"

type ringBuffer[T any] struct {
buffer []entry[T]
// head is the index immediately after the last non-empty entry in the buffer (i.e., the next write position).
Expand All @@ -25,13 +23,13 @@ type ringBuffer[T any] struct {

type entry[T any] struct {
revision int64
items []T
item T
}

type (
KeyPredicate = func([]byte) bool
RevisionOf[T any] func(T) int64
IterFunc[T any] func(rev int64, items []T) bool
IterFunc[T any] func(rev int64, item T) bool
)

func newRingBuffer[T any](capacity int, revisionOf RevisionOf[T]) *ringBuffer[T] {
Expand All @@ -42,29 +40,8 @@ func newRingBuffer[T any](capacity int, revisionOf RevisionOf[T]) *ringBuffer[T]
}
}

func (r *ringBuffer[T]) Append(items []T) {
start := 0
for end := 1; end < len(items); end++ {
if r.revisionOf(items[end]) != r.revisionOf(items[start]) {
r.append(entry[T]{
revision: r.revisionOf(items[start]),
items: items[start:end],
})
start = end
}
}
if start < len(items) {
r.append(entry[T]{
revision: r.revisionOf(items[start]),
items: items[start:],
})
}
}

func (r *ringBuffer[T]) append(entry entry[T]) {
if len(entry.items) == 0 {
return
}
func (r *ringBuffer[T]) Append(item T) {
entry := entry[T]{revision: r.revisionOf(item), item: item}
if r.size == len(r.buffer) {
r.tail = (r.tail + 1) % len(r.buffer)
} else {
Expand All @@ -84,15 +61,11 @@ func (r *ringBuffer[T]) AscendGreaterOrEqual(pivot int64, iter IterFunc[T]) {
for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
entry := r.buffer[i]

if entry.items == nil {
panic(fmt.Sprintf("ringBuffer.AscendGreaterOrEqual: unexpected nil at %d", i))
}

if entry.revision < pivot {
continue
}

if !iter(entry.revision, entry.items) {
if !iter(entry.revision, entry.item) {
return
}
}
Expand All @@ -107,15 +80,11 @@ func (r *ringBuffer[T]) AscendLessThan(pivot int64, iter IterFunc[T]) {
for n, i := 0, r.tail; n < r.size; n, i = n+1, (i+1)%len(r.buffer) {
entry := r.buffer[i]

if entry.items == nil {
panic(fmt.Sprintf("ringBuffer.AscendLessThan: unexpected nil at %d", i))
}

if entry.revision >= pivot {
return
}

if !iter(entry.revision, entry.items) {
if !iter(entry.revision, entry.item) {
return
}
}
Expand All @@ -130,15 +99,11 @@ func (r *ringBuffer[T]) DescendGreaterThan(pivot int64, iter IterFunc[T]) {
for n, i := 0, r.moduloIndex(r.head-1); n < r.size; n, i = n+1, r.moduloIndex(i-1) {
entry := r.buffer[i]

if entry.items == nil {
panic(fmt.Sprintf("ringBuffer.DescendGreaterThan: unexpected nil at %d", i))
}

if entry.revision <= pivot {
return
}

if !iter(entry.revision, entry.items) {
if !iter(entry.revision, entry.item) {
return
}
}
Expand All @@ -153,15 +118,11 @@ func (r *ringBuffer[T]) DescendLessOrEqual(pivot int64, iter IterFunc[T]) {
for n, i := 0, r.moduloIndex(r.head-1); n < r.size; n, i = n+1, r.moduloIndex(i-1) {
entry := r.buffer[i]

if entry.items == nil {
panic(fmt.Sprintf("ringBuffer.DescendLessOrEqual: unexpected nil at %d", i))
}

if entry.revision > pivot {
continue
}

if !iter(entry.revision, entry.items) {
if !iter(entry.revision, entry.item) {
return
}
}
Expand Down
25 changes: 15 additions & 10 deletions cache/ringbuffer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ func TestPeekLatestAndOldest(t *testing.T) {
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
rb := newRingBuffer(tt.capacity, func(ev *clientv3.Event) int64 { return ev.Kv.ModRevision })
rb := newRingBuffer(tt.capacity, func(batch []*clientv3.Event) int64 { return batch[0].Kv.ModRevision })
for _, r := range tt.revs {
batch, err := makeEventBatch(r, "k", 1)
if err != nil {
Expand Down Expand Up @@ -168,16 +168,21 @@ func TestIterationMethods(t *testing.T) {
}

func TestIterationWithBatching(t *testing.T) {
rb := newRingBuffer(6, func(ev *clientv3.Event) int64 { return ev.Kv.ModRevision })
events := []*clientv3.Event{
rb := newRingBuffer(6, func(batch []*clientv3.Event) int64 { return batch[0].Kv.ModRevision })
batchA := []*clientv3.Event{
{Kv: &mvccpb.KeyValue{Key: []byte("key-a"), ModRevision: 5}},
}
batchB := []*clientv3.Event{
{Kv: &mvccpb.KeyValue{Key: []byte("key-b-1"), ModRevision: 10}},
{Kv: &mvccpb.KeyValue{Key: []byte("key-b-2"), ModRevision: 10}},
{Kv: &mvccpb.KeyValue{Key: []byte("key-b-3"), ModRevision: 10}},
}
batchC := []*clientv3.Event{
{Kv: &mvccpb.KeyValue{Key: []byte("key-c"), ModRevision: 12}},
}

rb.Append(events)
rb.Append(batchA)
rb.Append(batchB)
rb.Append(batchC)

tests := []struct {
name string
Expand Down Expand Up @@ -394,7 +399,7 @@ func TestAtomicOrdered(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()

rb := newRingBuffer(tt.capacity, func(ev *clientv3.Event) int64 { return ev.Kv.ModRevision })
rb := newRingBuffer(tt.capacity, func(batch []*clientv3.Event) int64 { return batch[0].Kv.ModRevision })
for _, in := range tt.inputs {
batch, err := makeEventBatch(in.rev, in.key, in.size)
if err != nil {
Expand Down Expand Up @@ -445,7 +450,7 @@ func TestRebaseHistory(t *testing.T) {
tt := tt
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
rb := newRingBuffer(4, func(ev *clientv3.Event) int64 { return ev.Kv.ModRevision })
rb := newRingBuffer(4, func(batch []*clientv3.Event) int64 { return batch[0].Kv.ModRevision })
for _, r := range tt.revs {
batch, err := makeEventBatch(r, "k", 1)
if err != nil {
Expand Down Expand Up @@ -479,8 +484,8 @@ func TestRebaseHistory(t *testing.T) {
}
}

func setupRingBuffer(t *testing.T, capacity int, revs []int64) *ringBuffer[*clientv3.Event] {
rb := newRingBuffer(capacity, func(ev *clientv3.Event) int64 { return ev.Kv.ModRevision })
func setupRingBuffer(t *testing.T, capacity int, revs []int64) *ringBuffer[[]*clientv3.Event] {
rb := newRingBuffer(capacity, func(batch []*clientv3.Event) int64 { return batch[0].Kv.ModRevision })
for _, r := range revs {
batch, err := makeEventBatch(r, "key", 1)
if err != nil {
Expand All @@ -491,7 +496,7 @@ func setupRingBuffer(t *testing.T, capacity int, revs []int64) *ringBuffer[*clie
return rb
}

func collectRevisions(rb *ringBuffer[*clientv3.Event], method iterMethod, pivot int64) []int64 {
func collectRevisions(rb *ringBuffer[[]*clientv3.Event], method iterMethod, pivot int64) []int64 {
revs := []int64{}
rb.iterate(method, pivot, func(rev int64, events []*clientv3.Event) bool {
revs = append(revs, rev)
Expand Down