diff --git a/options.go b/options.go index c45161f..6132f55 100644 --- a/options.go +++ b/options.go @@ -123,7 +123,8 @@ func WithFn(fn func(context.Context, core.TaskMessage) error) Option { // WithAfterFn sets a callback function that will be executed after each job completes. // This callback runs regardless of whether the job succeeded or failed. -// It executes after metrics are updated but before the worker picks up the next task. +// It executes after metrics are updated without occupying a worker slot. +// Callbacks may overlap; Release waits for them to finish. // Useful for cleanup, logging, or triggering post-processing workflows. // // Example: diff --git a/performance_test.go b/performance_test.go new file mode 100644 index 0000000..23a113a --- /dev/null +++ b/performance_test.go @@ -0,0 +1,84 @@ +package queue + +import ( + "context" + "fmt" + "sync" + "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) + } + } + }) + } +} + +// BenchmarkQueueEndToEnd measures public submission through handler completion +// and final Release. Batches bound the backlog independently of b.N. +func BenchmarkQueueEndToEnd(b *testing.B) { + for _, workers := range []int64{1, 4, 12} { + b.Run(fmt.Sprintf("workers=%d", workers), func(b *testing.B) { + benchmarkQueueCompletion(b, workers, false) + }) + } +} + +// BenchmarkQueueAfterFnEndToEnd also waits for every configured callback. +func BenchmarkQueueAfterFnEndToEnd(b *testing.B) { + benchmarkQueueCompletion(b, 4, true) +} + +func benchmarkQueueCompletion(b *testing.B, workers int64, withCallback bool) { + var completed sync.WaitGroup + opts := []Option{WithLogger(NewEmptyLogger())} + task := func(context.Context) error { + completed.Done() + return nil + } + if withCallback { + opts = append(opts, WithAfterFn(completed.Done)) + task = func(context.Context) error { return nil } + } + q := NewPool(workers, opts...) + b.ReportAllocs() + b.ResetTimer() + for submitted := 0; submitted < b.N; { + n := min(1024, b.N-submitted) + completed.Add(n) + for range n { + if err := q.QueueTask(task); err != nil { + b.Fatal(err) + } + } + completed.Wait() + submitted += n + } + q.Release() + b.StopTimer() + if got := q.SuccessTasks(); got != uint64(b.N) { + b.Fatalf("completed %d tasks, want %d", got, b.N) + } +} diff --git a/queue.go b/queue.go index 6115337..f61b857 100644 --- a/queue.go +++ b/queue.go @@ -35,6 +35,7 @@ type ( ready chan struct{} // Channel to signal worker readiness notify chan struct{} // Channel to notify workers of new jobs worker core.Worker // The worker implementation that processes jobs + startOnce sync.Once // Ensures only one dispatcher requests tasks stopOnce sync.Once // Ensures shutdown is only performed once stopFlag atomic.Int32 // Atomic flag indicating if shutdown has started afterFn func() // Optional callback after each job execution @@ -71,7 +72,7 @@ func NewQueue(opts ...Option) (*Queue, error) { return q, nil } -// Start launches all worker goroutines and begins processing jobs. +// Start launches the dispatcher and workers once and begins processing jobs. // If workerCount is zero, Start is a no-op. func (q *Queue) Start() { q.Lock() @@ -80,8 +81,8 @@ func (q *Queue) Start() { if count == 0 { return } - q.routineGroup.Run(func() { - q.start() + q.startOnce.Do(func() { + q.routineGroup.Run(q.start) }) } @@ -198,7 +199,10 @@ func (q *Queue) work(task core.TaskMessage) { q.metric.IncFailureTask() } if q.afterFn != nil { - q.afterFn() + // Completion callbacks have never occupied a worker slot. Keep them + // independent so a callback can wait for another queued task, while + // still making Release wait for all callbacks to finish. + q.routineGroup.Run(q.afterFn) } }() @@ -313,86 +317,99 @@ func (q *Queue) UpdateWorkerCount(num int64) { q.schedule() } -// schedule checks if more workers can be started based on the current busy count. -// If so, it signals readiness to start a new worker. +// schedule wakes the dispatcher after a completion or worker-count update. +// Notifications may coalesce; the dispatcher always rechecks the current limit. func (q *Queue) schedule() { - q.Lock() - defer q.Unlock() - if q.BusyWorkers() >= q.workerCount { - return - } - select { case q.ready <- struct{}{}: default: } } -/* -start launches the main worker loop, which manages job scheduling and execution. +// consume reuses a worker goroutine until it is retired or dispatch has stopped. +// Retiring a busy worker does not interrupt its current task. +func (q *Queue) consume(tasks <-chan core.TaskMessage, retire <-chan struct{}) { + for { + // Prefer retirement before accepting another task. + select { + case <-retire: + return + default: + } + select { + case <-retire: + return + case task, ok := <-tasks: + if !ok { + return + } + q.work(task) + } + } +} -- It uses a ticker to periodically retry job requests if the queue is empty. -- For each available worker slot, it requests a new task from the worker. -- If a task is available, it is sent to the tasks channel and processed by a new goroutine. -- The loop exits when the quit channel is closed. -*/ +// start serializes backend requests and dispatches tasks to reusable workers. +// Only the dispatcher owns the worker set; completed tasks and count updates wake +// it to recheck capacity. The handler goroutine in handle still enforces timeout. func (q *Queue) start() { - tasks := make(chan core.TaskMessage, 1) + tasks := make(chan core.TaskMessage) + defer close(tasks) ticker := time.NewTicker(q.retryInterval) defer ticker.Stop() + var workers []chan struct{} for { - // Ensure the number of busy workers does not exceed the configured worker count. - q.schedule() + q.Lock() + count := q.workerCount + q.Unlock() + + // A nonpositive dynamic count pauses dispatch without interrupting work. + count = max(count, 0) + for int64(len(workers)) > count { + last := len(workers) - 1 + close(workers[last]) + workers[last] = nil + workers = workers[:last] + } + for int64(len(workers)) < count { + retire := make(chan struct{}) + workers = append(workers, retire) + q.routineGroup.Run(func() { q.consume(tasks, retire) }) + } + if q.BusyWorkers() >= count { + select { + case <-q.ready: + continue + case <-q.quit: + return + } + } select { - case <-q.ready: // Wait for a worker slot to become available - case <-q.quit: // Shutdown signal received + case <-q.quit: return + default: } - // Request a task from the worker in a background goroutine. - q.routineGroup.Run(func() { - for { - t, err := q.worker.Request() - if t == nil || err != nil { - if err != nil { - select { - case <-q.quit: - if !errors.Is(err, ErrNoTaskInQueue) { - close(tasks) - return - } - case <-ticker.C: - case <-q.notify: - } - } - } - if t != nil { - tasks <- t - return - } - - select { - case <-q.quit: - if !errors.Is(err, ErrNoTaskInQueue) { - close(tasks) - return - } - default: - } + // Request was already serialized by the old dispatcher waiting for each + // result. Calling it here removes a goroutine and channel hop per task. + task, err := q.worker.Request() + if task == nil || err != nil { + select { + case <-q.quit: + return + case <-ticker.C: + case <-q.notify: + case <-q.ready: + } + if task == nil { + continue } - }) - - task, ok := <-tasks - if !ok { - return } - // Start processing the new task in a separate goroutine. + // Once dequeued, the task must be handed off even during shutdown. + // The workers remain alive until this dispatcher closes tasks. q.metric.IncBusyWorker() - q.routineGroup.Run(func() { - q.work(task) - }) + tasks <- task } } 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/scheduling_test.go b/scheduling_test.go new file mode 100644 index 0000000..a6e4a21 --- /dev/null +++ b/scheduling_test.go @@ -0,0 +1,294 @@ +package queue + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/golang-queue/queue/core" + "github.com/golang-queue/queue/job" + + "github.com/stretchr/testify/require" +) + +func receiveWithin[T any](t *testing.T, ch <-chan T) T { + t.Helper() + select { + case value := <-ch: + return value + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for queue progress") + var zero T + return zero + } +} + +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) +} + +func TestQueueResizePauseAndDrain(t *testing.T) { + const total = 16 + started := make(chan int, total) + completed := make(chan struct{}, total) + release := make(chan struct{}) + var releaseOnce sync.Once + q := NewPool(2, WithLogger(NewEmptyLogger()), WithAfterFn(func() { completed <- struct{}{} })) + t.Cleanup(func() { + releaseOnce.Do(func() { close(release) }) + q.UpdateWorkerCount(2) + q.Release() + }) + for id := range total { + require.NoError(t, q.QueueTask(func(context.Context) error { + started <- id + <-release + return nil + })) + } + seen := make(map[int]bool) + receiveStarted := func(n int) { + for range n { + id := receiveWithin(t, started) + require.False(t, seen[id], "task executed twice") + seen[id] = true + } + } + receiveStarted(2) + q.UpdateWorkerCount(4) + receiveStarted(2) + require.EqualValues(t, 4, q.BusyWorkers()) + q.UpdateWorkerCount(1) + for range 4 { + select { + case release <- struct{}{}: + case <-time.After(5 * time.Second): + t.Fatal("workers did not finish during scale-down") + } + } + for range 4 { + receiveWithin(t, completed) + } + receiveStarted(1) + require.EqualValues(t, 1, q.BusyWorkers()) + q.UpdateWorkerCount(0) + release <- struct{}{} + receiveWithin(t, completed) + select { + case <-started: + t.Fatal("task started while paused") + case <-time.After(20 * time.Millisecond): + } + q.UpdateWorkerCount(2) + receiveStarted(2) + require.EqualValues(t, 2, q.BusyWorkers()) + releaseOnce.Do(func() { close(release) }) + q.Release() + receiveStarted(total - len(seen)) + require.Len(t, seen, total) + require.EqualValues(t, total, q.SuccessTasks()) + require.EqualValues(t, 0, q.BusyWorkers()) +} + +func TestReusableWorkerOutcomes(t *testing.T) { + completed := make(chan struct{}, 5) + lateRelease := make(chan struct{}) + lateDone := make(chan struct{}) + q := NewPool(1, WithLogger(NewEmptyLogger()), WithAfterFn(func() { completed <- struct{}{} })) + t.Cleanup(func() { + close(lateRelease) + q.Release() + }) + wantErr := errors.New("task failed") + require.NoError(t, q.QueueTask(func(context.Context) error { return wantErr })) + require.NoError(t, q.QueueTask(func(context.Context) error { panic("task panic") })) + var attempts atomic.Int32 + require.NoError(t, q.QueueTask(func(context.Context) error { + if attempts.Add(1) == 1 { + return wantErr + } + return nil + }, job.AllowOption{RetryCount: job.Int64(1), RetryDelay: job.Time(time.Millisecond)})) + require.NoError(t, q.QueueTask(func(context.Context) error { + defer close(lateDone) + <-lateRelease // Deliberately ignore cancellation to check timeout isolation. + return nil + }, job.AllowOption{Timeout: job.Time(10 * time.Millisecond)})) + require.NoError(t, q.QueueTask(func(context.Context) error { return nil })) + for range 5 { + receiveWithin(t, completed) + } + require.EqualValues(t, 2, attempts.Load()) + require.EqualValues(t, 2, q.SuccessTasks()) + require.EqualValues(t, 3, q.FailureTasks()) + require.EqualValues(t, 5, q.CompletedTasks()) + select { + case <-lateDone: + t.Fatal("timeout test handler finished before being released") + default: + } + // Release still completes even though the timed-out handler has not returned. + q.Release() +} + +func TestAfterFnCanWaitForNextTask(t *testing.T) { + var q *Queue + var callbacks atomic.Int32 + callbackDone := make(chan error, 1) + q = NewPool(1, WithLogger(NewEmptyLogger()), WithAfterFn(func() { + if callbacks.Add(1) != 1 { + return + } + nextTask := make(chan struct{}) + if err := q.QueueTask(func(context.Context) error { + close(nextTask) + return nil + }); err != nil { + callbackDone <- err + return + } + select { + case <-nextTask: + callbackDone <- nil + case <-time.After(time.Second): + callbackDone <- errors.New("afterFn blocked the next task") + } + })) + t.Cleanup(q.Release) + require.NoError(t, q.QueueTask(func(context.Context) error { return nil })) + require.NoError(t, receiveWithin(t, callbackDone)) + q.Release() + require.EqualValues(t, 2, q.SuccessTasks()) + require.EqualValues(t, 2, callbacks.Load()) +} + +func TestReleaseWaitsForAfterFn(t *testing.T) { + entered := make(chan struct{}) + unblock := make(chan struct{}) + var once sync.Once + q := NewPool(1, WithLogger(NewEmptyLogger()), WithAfterFn(func() { + close(entered) + <-unblock + })) + t.Cleanup(func() { + once.Do(func() { close(unblock) }) + q.Release() + }) + require.NoError(t, q.QueueTask(func(context.Context) error { return nil })) + receiveWithin(t, entered) + released := make(chan struct{}) + go func() { + q.Release() + close(released) + }() + select { + case <-released: + t.Fatal("Release returned before afterFn finished") + case <-time.After(20 * time.Millisecond): + } + once.Do(func() { close(unblock) }) + receiveWithin(t, released) +} + +// blockingWorker models backends whose Request blocks until work or shutdown. +// Counting active calls detects accidental parallel Request calls. +type blockingWorker struct { + requests chan core.TaskMessage + closed chan struct{} + entered chan struct{} + once sync.Once + active atomic.Int32 + parallel atomic.Bool +} + +func (w *blockingWorker) Queue(m core.TaskMessage) error { + w.requests <- m + return nil +} + +func (w *blockingWorker) Request() (core.TaskMessage, error) { + if w.active.Add(1) != 1 { + w.parallel.Store(true) + } + defer w.active.Add(-1) + select { + case w.entered <- struct{}{}: + default: + } + select { + case m := <-w.requests: + return m, nil + case <-w.closed: + return nil, ErrQueueHasBeenClosed + } +} + +func (w *blockingWorker) Run(context.Context, core.TaskMessage) error { return nil } + +func (w *blockingWorker) Shutdown() error { + w.once.Do(func() { close(w.closed) }) + return nil +} + +func TestBlockingBackendAndRepeatedStart(t *testing.T) { + w := &blockingWorker{ + requests: make(chan core.TaskMessage, 64), + closed: make(chan struct{}), + entered: make(chan struct{}, 128), + } + completed := make(chan struct{}, 64) + q, err := NewQueue(WithWorker(w), WithWorkerCount(4), WithLogger(NewEmptyLogger()), + WithAfterFn(func() { completed <- struct{}{} })) + require.NoError(t, err) + t.Cleanup(q.Release) + q.Start() + receiveWithin(t, w.entered) + q.Start() + for range 64 { + require.NoError(t, q.QueueTask(func(context.Context) error { return nil })) + } + for range 64 { + receiveWithin(t, completed) + } + released := make(chan struct{}) + go func() { + q.Release() + close(released) + }() + receiveWithin(t, released) + require.False(t, w.parallel.Load(), "backend Request must remain serialized") + require.EqualValues(t, 64, q.SuccessTasks()) +}