perf(queue): reduce dispatch and resize overhead

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:

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)
+	}
+}

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
 	}
 }

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)
 	}