diff --git a/burst_performance_test.go b/burst_performance_test.go new file mode 100644 index 0000000..c4e824e --- /dev/null +++ b/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) + } + }) + } + } +} diff --git a/logger.go b/logger.go index 215f063..75e3484 100644 --- a/logger.go +++ b/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 } diff --git a/options.go b/options.go index 6132f55..65f8ae4 100644 --- a/options.go +++ b/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. diff --git a/performance_options_test.go b/performance_options_test.go new file mode 100644 index 0000000..053d5b1 --- /dev/null +++ b/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()) +} diff --git a/queue.go b/queue.go index 1c2ae74..0c84739 100644 --- a/queue.go +++ b/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 diff --git a/ring.go b/ring.go index 10619a1..c35e533 100644 --- a/ring.go +++ b/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