perf: adapt task buffer batch size to backlog

benchmark/adaptive_batch_benchmark_test.go

@@ -0,0 +1,63 @@
+package benchmark_test
+
+import (
+	"testing"
+
+	agilepool "github.com/Yiming1997/agilePool/v2"
+)
+
+// prepareOverflowBacklog blocks the only worker, then fills the one-slot
+// handoff channel plus exactly backlog tasks in the overflow buffer. Releasing
+// the returned channel starts a reproducible overflow-drain scenario.
+func prepareOverflowBacklog(tb testing.TB, backlog int) (*agilepool.Pool, chan struct{}) {
+	tb.Helper()
+
+	pool := agilepool.NewPool(agilepool.NewConfig(
+		agilepool.WithWorkerNumCapacity(1),
+		agilepool.WithTaskQueueSize(1),
+	))
+	started := make(chan struct{})
+	release := make(chan struct{})
+	pool.Submit(agilepool.TaskFunc(func() error {
+		close(started)
+		<-release
+		return nil
+	}))
+	<-started
+
+	noop := agilepool.TaskFunc(func() error { return nil })
+	for i := 0; i <= backlog; i++ {
+		pool.Submit(noop)
+	}
+
+	return pool, release
+}
+
+// BenchmarkAgilePoolAdaptiveBatchDrain compares end-to-end overflow draining
+// at the four adaptive batching thresholds. It measures the actual public pool
+// lifecycle, so results include queueing, worker scheduling, and task drain.
+func BenchmarkAgilePoolAdaptiveBatchDrain(b *testing.B) {
+	tests := []struct {
+		name    string
+		backlog int
+	}{
+		{name: "backlog_1_batch_1", backlog: 1},
+		{name: "backlog_9_batch_8", backlog: 9},
+		{name: "backlog_65_batch_32", backlog: 65},
+		{name: "backlog_513_batch_64", backlog: 513},
+	}
+
+	for _, tt := range tests {
+		b.Run(tt.name, func(b *testing.B) {
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				pool, release := prepareOverflowBacklog(b, tt.backlog)
+				close(release)
+				pool.Wait()
+				pool.Close()
+			}
+			b.ReportMetric(float64(tt.backlog), "buffer-tasks/op")
+		})
+	}
+}

task_buffer.go

@@ -10,6 +10,13 @@ const (
 	taskBufferFull
 )
 
+const (
+	// maxWorkerTaskBatchSize bounds how long one worker can drain the overflow
+	// queue before returning to the handoff channel. Keeping it bounded avoids
+	// turning high backlog throughput into unbounded latency for newer tasks.
+	maxWorkerTaskBatchSize = 64
+)
+
 // taskChunk is a fixed-size node in the linked-list task buffer.
 // Each chunk holds up to taskChunkSize Task values. Using small,
 // fixed-size nodes avoids the doubling overhead of a single slice
@@ -96,6 +103,54 @@ func (b *chunkedTaskBuffer) PopBatch(dst []Task) int {
 	return n
 }
 
+// PopAdaptiveBatch drains a backlog-dependent number of tasks into dst.
+// The batch size is selected while taskMu is held, so workers do not take a
+// separate length snapshot lock before draining. Shallow queues use tiny
+// batches for fairness; deeper queues amortize the shared lock acquisition.
+func (b *chunkedTaskBuffer) PopAdaptiveBatch(dst []Task) int {
+	b.taskMu.Lock()
+	defer b.taskMu.Unlock()
+
+	limit := adaptiveTaskBatchSize(b.chunkLen, len(dst))
+	n := 0
+	for n < limit {
+		t, ok := b.popHead()
+		if !ok {
+			break
+		}
+		dst[n] = t
+		n++
+	}
+
+	return n
+}
+
+// adaptiveTaskBatchSize balances queue fairness and lock amortization.
+// maxSize is normally the worker's stack batch capacity; zero means there is
+// no destination space and therefore no work should be removed.
+func adaptiveTaskBatchSize(backlog int64, maxSize int) int {
+	if backlog <= 0 || maxSize <= 0 {
+		return 0
+	}
+
+	var size int
+	switch {
+	case backlog <= 8:
+		size = 1
+	case backlog <= 64:
+		size = 8
+	case backlog <= 512:
+		size = 32
+	default:
+		size = maxWorkerTaskBatchSize
+	}
+
+	if size > maxSize {
+		return maxSize
+	}
+	return size
+}
+
 // pushTail appends a task to the tail of the chunked buffer.
 // Must be called with taskMu held.
 func (b *chunkedTaskBuffer) pushTail(task Task) {

worker.go

@@ -34,6 +34,9 @@ func (w *worker) run(task Task) {
 	// Submits could then concurrently spawn two goroutines on the same
 	// *worker via Pop and workerPool.Get respectively, causing a data race
 	// on w.lastActiveAt and phantom duplicates in idleWorks.
+	// Keep one reusable stack batch per worker. Processed entries are cleared
+	// below so a parked worker does not retain completed task payloads.
+	var batch [maxWorkerTaskBatchSize]Task
 
 loop:
 	for {
@@ -58,16 +61,14 @@ loop:
 			w.runTask(task)
 
 		default:
-			// Try the chunked buffer before the second channel check.
-			// Grab a batch of up to 8 tasks per lock acquisition to
-			// amortise the mutex overhead across multiple tasks and
-			// reduce contention with the submission path.
-			const batchSize = 8
-			var batch [batchSize]Task
-			n := w.pool.taskBuf.PopBatch(batch[:])
+			// Try the chunked buffer before the second channel check. The buffer
+			// chooses a larger batch only when backlog is deep enough to justify
+			// fewer lock acquisitions; the fixed upper bound preserves fairness.
+			n := w.pool.taskBuf.PopAdaptiveBatch(batch[:])
 			for i := 0; i < n; i++ {
 				w.lastActiveAt = time.Now()
 				w.runTask(batch[i])
+				batch[i] = nil // Release task payloads before this worker parks.
 			}
 			if n > 0 {
 				continue