perf(queue): reduce retry and burst allocations

burst_performance_test.go

@@ -0,0 +1,91 @@
+package queue
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"unsafe"
+
+	"github.com/golang-queue/queue/core"
+	"github.com/golang-queue/queue/job"
+)
+
+// The comparison harness supplies the same retained-capacity option to each
+// variant; on the baseline it is intentionally a no-op (the old policy).
+func performanceRingOptions(retained bool) []Option {
+	if retained {
+		return []Option{WithQueueMinCapacity(1024)}
+	}
+	return nil
+}
+
+// One operation is a full 1024-task burst and drain. Retained bytes count only
+// backing-array slots after drain, not RSS or task payloads.
+func BenchmarkRingBurstDrain(b *testing.B) {
+	for _, retained := range []bool{false, true} {
+		b.Run(fmt.Sprintf("retained=%t", retained), func(b *testing.B) {
+			w := NewRing(performanceRingOptions(retained)...)
+			m := job.NewTask(func(context.Context) error { return nil })
+			b.ReportAllocs()
+			b.ResetTimer()
+			for range b.N {
+				for range 1024 {
+					if err := w.Queue(&m); err != nil {
+						b.Fatal(err)
+					}
+				}
+				for range 1024 {
+					if _, err := w.Request(); err != nil {
+						b.Fatal(err)
+					}
+				}
+			}
+			b.StopTimer()
+			b.ReportMetric(
+				float64(len(w.taskQueue))*float64(unsafe.Sizeof(core.TaskMessage(nil))),
+				"retained-B",
+			)
+		})
+	}
+}
+
+// One operation is one publicly submitted, completed task. Fixed batches bound
+// the backlog; producer goroutine creation and final Release are included.
+func BenchmarkQueueBurstProducers(b *testing.B) {
+	for _, producers := range []int{1, 4} {
+		for _, retained := range []bool{false, true} {
+			b.Run(fmt.Sprintf("producers=%d/retained=%t", producers, retained), func(b *testing.B) {
+				opts := append(performanceRingOptions(retained), WithLogger(NewEmptyLogger()))
+				q := NewPool(4, opts...)
+				var completed sync.WaitGroup
+				task := func(context.Context) error { completed.Done(); return nil }
+				b.ReportAllocs()
+				b.ResetTimer()
+				for submitted := 0; submitted < b.N; {
+					n := min(1024, b.N-submitted)
+					completed.Add(n)
+					var senders sync.WaitGroup
+					for p := range producers {
+						senders.Go(func() {
+							for i := p; i < n; i += producers {
+								if err := q.QueueTask(task); err != nil {
+									b.Error(err)
+									completed.Done()
+								}
+							}
+						})
+					}
+					senders.Wait()
+					completed.Wait()
+					submitted += n
+				}
+				q.Release()
+				b.StopTimer()
+				if got := q.SuccessTasks(); got != uint64(b.N) {
+					b.Fatalf("got %d completed, want %d", got, b.N)
+				}
+			})
+		}
+	}
+}

logger.go

@@ -35,6 +35,21 @@ type Logger interface {
 	Fatal(args ...any)
 }
 
+// InfoLogger optionally lets a Logger disable informational messages before the
+// queue constructs variadic arguments. Loggers without it remain enabled.
+// InfoEnabled must be safe to call concurrently; it is checked on every retry.
+type InfoLogger interface {
+	Logger
+	InfoEnabled() bool
+}
+
+func infoEnabled(l Logger) bool {
+	if l, ok := l.(InfoLogger); ok {
+		return l.InfoEnabled()
+	}
+	return true
+}
+
 // NewLogger creates a standard logger that writes to stderr with timestamps.
 // This is the default logger used by queues unless overridden with WithLogger.
 //
@@ -127,3 +142,6 @@ func (l emptyLogger) Fatalf(format string, args ...any) {}
 func (l emptyLogger) Info(args ...any)                  {}
 func (l emptyLogger) Error(args ...any)                 {}
 func (l emptyLogger) Fatal(args ...any)                 {}
+
+// InfoEnabled disables informational argument construction for the no-op logger.
+func (l emptyLogger) InfoEnabled() bool { return false }

options.go

@@ -61,6 +61,17 @@ func WithQueueSize(num int) Option {
 	})
 }
 
+// WithQueueMinCapacity sets the Ring buffer's initial and minimum retained slots.
+// Values below 2 use 2 (the default). A positive WithQueueSize limit caps this
+// setting, except that the buffer always has at least 2 slots. The pending-task
+// limit is unchanged. Retaining slots reduces burst allocations at the cost of
+// idle memory; dequeued task references are still cleared. Other backends ignore it.
+func WithQueueMinCapacity(num int) Option {
+	return OptionFunc(func(q *Options) {
+		q.queueMinCapacity = num
+	})
+}
+
 // WithLogger sets a custom logger for queue events and errors.
 // By default, the queue uses a standard logger that writes to stderr.
 // Use NewEmptyLogger() to disable logging entirely.
@@ -155,14 +166,15 @@ func WithRetryInterval(d time.Duration) Option {
 // Options holds the configuration parameters for a Queue.
 // Use the With* functions to configure these options when creating a queue.
 type Options struct {
-	workerCount   int64                                         // Number of concurrent worker goroutines
-	logger        Logger                                        // Logger for queue events
-	queueSize     int                                           // Maximum queue capacity, 0 means unlimited
-	worker        core.Worker                                   // Worker implementation for queue backend
-	fn            func(context.Context, core.TaskMessage) error // Task handler function
-	afterFn       func()                                        // Callback executed after each job
-	metric        Metric                                        // Metrics collector
-	retryInterval time.Duration                                 // Polling interval when queue is empty
+	queueMinCapacity int                                           // Initial and minimum retained Ring slots; normalized by NewRing
+	workerCount      int64                                         // Number of concurrent worker goroutines
+	logger           Logger                                        // Logger for queue events
+	queueSize        int                                           // Maximum queue capacity, 0 means unlimited
+	worker           core.Worker                                   // Worker implementation for queue backend
+	fn               func(context.Context, core.TaskMessage) error // Task handler function
+	afterFn          func()                                        // Callback executed after each job
+	metric           Metric                                        // Metrics collector
+	retryInterval    time.Duration                                 // Polling interval when queue is empty
 }
 
 // NewOptions creates an Options struct with default values and applies any provided options.

performance_options_test.go

@@ -0,0 +1,133 @@
+package queue
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/golang-queue/queue/job"
+
+	"github.com/stretchr/testify/require"
+)
+
+type legacyRetryLogger struct {
+	Logger
+	messages []string
+}
+
+func (l *legacyRetryLogger) Infof(format string, args ...any) {
+	l.messages = append(l.messages, fmt.Sprintf(format, args...))
+}
+
+type filteredRetryLogger struct {
+	*legacyRetryLogger
+	enabled bool
+}
+
+func (l *filteredRetryLogger) InfoEnabled() bool { return l.enabled }
+
+func TestRetryInfoEnabled(t *testing.T) {
+	for _, mode := range []string{"legacy", "enabled", "disabled", "dynamic"} {
+		t.Run(mode, func(t *testing.T) {
+			capture := &legacyRetryLogger{Logger: NewEmptyLogger()}
+			filtered := &filteredRetryLogger{legacyRetryLogger: capture, enabled: mode == "enabled"}
+			var logger Logger = filtered
+			if mode == "legacy" {
+				logger = capture
+			}
+			q, err := NewQueue(WithWorker(NewRing()), WithLogger(logger))
+			require.NoError(t, err)
+			defer q.Release()
+			attempts := 0
+			m := job.NewTask(func(context.Context) error {
+				attempts++
+				if mode == "dynamic" {
+					filtered.enabled = attempts == 2
+				}
+				if attempts <= 3 {
+					return errors.New("retry")
+				}
+				return nil
+			}, job.AllowOption{RetryCount: job.Int64(3), RetryDelay: job.Time(time.Nanosecond)})
+			require.NoError(t, q.handle(&m))
+			require.Equal(t, 4, attempts)
+			want := 3
+			if mode == "disabled" {
+				want = 0
+			}
+			if mode == "dynamic" {
+				want = 1
+			}
+			require.Len(t, capture.messages, want)
+			if mode == "legacy" || mode == "enabled" {
+				require.Equal(t, "retry remaining times: 2, delay time: 1ns", capture.messages[0])
+			}
+		})
+	}
+	require.False(t, infoEnabled(NewEmptyLogger()))
+}
+
+func TestRingMinCapacity(t *testing.T) {
+	for _, tc := range []struct {
+		name               string
+		floor, limit, want int
+	}{
+		{"default", 0, 0, 2},
+		{"negative", -1, 0, 2},
+		{"retained", 64, 0, 64},
+		{"non-power-of-two", 7, 0, 7},
+		{"capped", 64, 10, 10},
+		{"single", 64, 1, 2},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			w := NewRing(WithQueueMinCapacity(tc.floor), WithQueueSize(tc.limit))
+			require.Len(t, w.taskQueue, tc.want)
+			n := 257
+			if tc.limit > 0 {
+				n = tc.limit
+			}
+			for range 3 {
+				expected := make([]*job.Message, n)
+				for i := range n {
+					expected[i] = &job.Message{Body: []byte{byte(i)}}
+					require.NoError(t, w.Queue(expected[i]))
+				}
+				if tc.limit > 0 {
+					require.ErrorIs(t, w.Queue(new(job.Message)), ErrMaxCapacity)
+				}
+				// Drain partly and refill to force wraparound before the final drain.
+				for i := range n / 2 {
+					got, err := w.Request()
+					require.NoError(t, err)
+					require.Same(t, expected[i], got)
+				}
+				for i := range n / 2 {
+					require.NoError(t, w.Queue(expected[i]))
+				}
+				order := append(append([]*job.Message{}, expected[n/2:]...), expected[:n/2]...)
+				for _, want := range order {
+					got, err := w.Request()
+					require.NoError(t, err)
+					require.Same(t, want, got)
+				}
+				require.Len(t, w.taskQueue, tc.want)
+				for _, slot := range w.taskQueue {
+					require.Nil(t, slot)
+				}
+				_, err := w.Request()
+				require.ErrorIs(t, err, ErrNoTaskInQueue)
+			}
+			require.NoError(t, w.Shutdown())
+			require.ErrorIs(t, w.Queue(new(job.Message)), ErrQueueShutdown)
+		})
+	}
+	// Option ordering must not affect the cap.
+	require.Len(t, NewRing(WithQueueSize(10), WithQueueMinCapacity(64)).taskQueue, 10)
+	q := NewPool(1, WithQueueMinCapacity(64), WithLogger(NewEmptyLogger()))
+	require.Equal(t, 64, q.worker.(*Ring).minCapacity)
+	require.NoError(t, q.QueueTask(func(context.Context) error { return nil }))
+	q.Release()
+	require.EqualValues(t, 1, q.SuccessTasks())
+}

queue.go

@@ -278,7 +278,9 @@ func (q *Queue) handle(m *job.Message) error {
 
 			select {
 			case <-time.After(delay): // Wait before retrying
-				q.logger.Infof("retry remaining times: %d, delay time: %s", m.RetryCount, delay)
+				if infoEnabled(q.logger) {
+					q.logger.Infof("retry remaining times: %d, delay time: %s", m.RetryCount, delay)
+				}
 			case <-ctx.Done(): // Timeout reached
 				err = ctx.Err()
 				break loop

ring.go

@@ -21,6 +21,7 @@ var _ core.Worker = (*Ring)(nil)
 //   - When head == tail, the queue is empty
 //   - Both pointers wrap around using modulo arithmetic
 type Ring struct {
+	minCapacity int // Minimum retained slots, configured at construction
 	sync.Mutex
 	taskQueue []core.TaskMessage                            // Circular buffer storing queued tasks
 	runFunc   func(context.Context, core.TaskMessage) error // Function to execute when processing tasks
@@ -131,8 +132,11 @@ func (s *Ring) Request() (core.TaskMessage, error) {
 
 	// 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 <= len(s.taskQueue)/4 {
-		s.resize(n)
+	if n := len(
+		s.taskQueue,
+	) / 2; len(s.taskQueue) > s.minCapacity &&
+		s.count <= len(s.taskQueue)/4 {
+		s.resize(max(n, s.minCapacity))
 	}
 
 	return data, nil
@@ -179,7 +183,7 @@ func (s *Ring) resize(n int) {
 }
 
 // NewRing creates a new Ring instance with the provided options.
-// It initializes the task queue with a default size of 2, sets the capacity
+// It initializes the task queue with at least 2 slots (or WithQueueMinCapacity), sets the capacity
 // based on the provided options, and configures the logger and run function.
 // The function returns a pointer to the newly created Ring instance.
 //
@@ -192,12 +196,17 @@ func (s *Ring) resize(n int) {
 //	*Ring - A pointer to the newly created Ring instance.
 func NewRing(opts ...Option) *Ring {
 	o := NewOptions(opts...)
+	minCapacity := max(2, o.queueMinCapacity)
+	if o.queueSize > 0 {
+		minCapacity = max(2, min(minCapacity, o.queueSize))
+	}
 	w := &Ring{
-		taskQueue: make([]core.TaskMessage, 2),
-		capacity:  o.queueSize,
-		exit:      make(chan struct{}),
-		logger:    o.logger,
-		runFunc:   o.fn,
+		minCapacity: minCapacity,
+		taskQueue:   make([]core.TaskMessage, minCapacity),
+		capacity:    o.queueSize,
+		exit:        make(chan struct{}),
+		logger:      o.logger,
+		runFunc:     o.fn,
 	}
 
 	return w