From 59d05041a4d2c8195d96883220b5cf5a1e2ac28b Mon Sep 17 00:00:00 2001 From: Bo-Yi Wu Date: Sat, 31 Oct 2026 10:58:21 +0800 Subject: [PATCH] perf(ring): prevent resize oscillation Shrink only at quarter utilization and add FIFO regression coverage and capacity-boundary benchmarks. --- ring.go | 8 ++++---- ring_performance_test.go | 36 +++++++++++++++++++++++++++++++++ ring_resize_test.go | 43 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 4 deletions(-) create mode 100644 ring_performance_test.go create mode 100644 ring_resize_test.go diff --git a/ring.go b/ring.go index 3c51a8e..10619a1 100644 --- a/ring.go +++ b/ring.go @@ -13,7 +13,7 @@ var _ core.Worker = (*Ring)(nil) // Ring is an in-memory worker implementation using a dynamic circular buffer. // It implements the core.Worker interface and provides automatic resizing: // - Doubles capacity when full -// - Halves capacity when less than 25% utilized +// - Halves capacity when at most 25% utilized // // The ring buffer uses two pointers (head and tail) to track the queue boundaries: // - head: points to the next task to dequeue @@ -95,7 +95,7 @@ func (s *Ring) Queue(task core.TaskMessage) error { } // Request dequeues and returns the next task from the ring buffer. -// The buffer shrinks automatically (halves in size) when less than 25% full. +// The buffer shrinks automatically (halves in size) when at most 25% full. // Returns: // - (task, nil) if a task is successfully dequeued // - (nil, ErrNoTaskInQueue) if the queue is currently empty @@ -129,9 +129,9 @@ func (s *Ring) Request() (core.TaskMessage, error) { s.head = (s.head + 1) % len(s.taskQueue) // Advance head with wraparound s.count-- - // Shrink buffer if less than 25% full (to save memory) + // Shrink buffer if at most 25% full (to save memory) // Minimum buffer size is 2 to maintain circular buffer structure - if n := len(s.taskQueue) / 2; n >= 2 && s.count <= n { + if n := len(s.taskQueue) / 2; n >= 2 && s.count <= len(s.taskQueue)/4 { s.resize(n) } diff --git a/ring_performance_test.go b/ring_performance_test.go new file mode 100644 index 0000000..0196286 --- /dev/null +++ b/ring_performance_test.go @@ -0,0 +1,36 @@ +package queue + +import ( + "context" + "fmt" + "testing" + + "github.com/golang-queue/queue/job" +) + +// BenchmarkRingBoundary holds the backlog near a power-of-two capacity, where +// growing at full and shrinking at half can otherwise copy the entire backlog +// twice per operation. One operation is one successful enqueue/dequeue pair. +func BenchmarkRingBoundary(b *testing.B) { + for _, backlog := range []int{64, 1024} { + b.Run(fmt.Sprintf("backlog=%d", backlog), func(b *testing.B) { + w := NewRing() + m := job.NewTask(func(context.Context) error { return nil }) + for range backlog { + if err := w.Queue(&m); err != nil { + b.Fatal(err) + } + } + b.ReportAllocs() + b.ResetTimer() + for range b.N { + if err := w.Queue(&m); err != nil { + b.Fatal(err) + } + if _, err := w.Request(); err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/ring_resize_test.go b/ring_resize_test.go new file mode 100644 index 0000000..5f16596 --- /dev/null +++ b/ring_resize_test.go @@ -0,0 +1,43 @@ +package queue + +import ( + "testing" + + "github.com/golang-queue/queue/job" + + "github.com/stretchr/testify/require" +) + +func TestRingShrinkHysteresisAndFIFO(t *testing.T) { + w := NewRing() + var expected []*job.Message + for range 16 { + m := new(job.Message) + expected = append(expected, m) + require.NoError(t, w.Queue(m)) + } + for i := range 8 { + m, err := w.Request() + require.NoError(t, err) + require.Same(t, expected[i], m) + } + require.Len(t, w.taskQueue, 16, "half-full buffers must not shrink") + for i := 8; i < 12; i++ { + m, err := w.Request() + require.NoError(t, err) + require.Same(t, expected[i], m) + } + require.Len(t, w.taskQueue, 8, "quarter-full buffers should halve") + // Refill, wrap, grow, and drain to verify FIFO survives both resize paths. + for range 20 { + m := new(job.Message) + expected = append(expected, m) + require.NoError(t, w.Queue(m)) + } + for _, want := range expected[12:] { + m, err := w.Request() + require.NoError(t, err) + require.Same(t, want, m) + } + require.Len(t, w.taskQueue, 2) +}