diff --git a/benchmark/adaptive_batch_benchmark_test.go b/benchmark/adaptive_batch_benchmark_test.go new file mode 100644 index 0000000..17e92e3 --- /dev/null +++ b/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") + }) + } +} diff --git a/task_buffer.go b/task_buffer.go index 8b4ba02..3f05728 100644 --- a/task_buffer.go +++ b/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) { diff --git a/worker.go b/worker.go index 74f8156..370a00d 100644 --- a/worker.go +++ b/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