Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
8 changes: 4 additions & 4 deletions ring.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
}

Expand Down
36 changes: 36 additions & 0 deletions ring_performance_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
})
}
}
43 changes: 43 additions & 0 deletions ring_resize_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
Loading