Perf/sharded handoff queue

benchmark/ants_compare_test.go

@@ -0,0 +1,109 @@
+package benchmark_test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	agilepool "github.com/Yiming1997/agilePool/v2"
+	"github.com/panjf2000/ants/v2"
+)
+
+// cmpWorkerCap mirrors the worker capacity used by the agilePool benchmarks so
+// ants and agilePool run under the same concurrency budget.
+const cmpWorkerCap = 20000
+
+func newAgilePoolForCmp() *agilepool.Pool {
+	return agilepool.NewPool(agilepool.NewConfig(
+		agilepool.WithCleanPeriod(500*time.Millisecond),
+		agilepool.WithTaskQueueSize(100000),
+		agilepool.WithWorkerNumCapacity(cmpWorkerCap),
+		agilepool.WithIdleContainerType(agilepool.LinkedListType),
+	))
+}
+
+// ---------------------------------------------------------------------------
+// Metric 1 — hand-off throughput: no-op tasks.
+//
+// The task body does no work, so the measured cost is entirely the pool's
+// task hand-off (submit -> worker pickup). ns/op therefore reads directly as
+// the per-task hand-off cost; lower is better.
+// ---------------------------------------------------------------------------
+
+func BenchmarkCmpChurnAnts(b *testing.B) {
+	pool, err := ants.NewPool(cmpWorkerCap)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer pool.Release()
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			_ = pool.Submit(func() {})
+		}
+	})
+}
+
+func BenchmarkCmpChurnAgilePool(b *testing.B) {
+	pool := newAgilePoolForCmp()
+	defer pool.Close()
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			pool.Submit(agilepool.TaskFunc(func() error { return nil }))
+		}
+	})
+	pool.Wait()
+}
+
+// ---------------------------------------------------------------------------
+// Metric 2 — CPU usage under an execution-bound load: 10ms sleep tasks (the
+// same shape as the agilePool SequentialLinkedList benchmark). Here the task
+// duration is the limiter, so the meaningful metric is CPU burned per
+// completed task, read from the CPU profile of each benchmark.
+// ---------------------------------------------------------------------------
+
+func BenchmarkCmpSleepAnts(b *testing.B) {
+	pool, err := ants.NewPool(cmpWorkerCap)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer pool.Release()
+
+	var wg sync.WaitGroup
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			wg.Add(1)
+			_ = pool.Submit(func() {
+				defer wg.Done()
+				time.Sleep(10 * time.Millisecond)
+			})
+		}
+	})
+	wg.Wait()
+}
+
+func BenchmarkCmpSleepAgilePool(b *testing.B) {
+	pool := newAgilePoolForCmp()
+	defer pool.Close()
+
+	// Use the same WaitGroup-based completion tracking as the ants variant so
+	// neither side gets a harness-specific advantage (both closures capture
+	// &wg and therefore heap-allocate identically).
+	var wg sync.WaitGroup
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			wg.Add(1)
+			pool.Submit(agilepool.TaskFunc(func() error {
+				defer wg.Done()
+				time.Sleep(10 * time.Millisecond)
+				return nil
+			}))
+		}
+	})
+	wg.Wait()
+}

benchmark/scaling_matrix_test.go

@@ -0,0 +1,65 @@
+package benchmark_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	agilepool "github.com/Yiming1997/agilePool/v2"
+)
+
+// Implementation-agnostic scaling matrix. It deliberately avoids any
+// version-specific config so the exact same file compiles and runs against
+// both the pre-sharding implementation and the current one, letting the two
+// be compared by swapping the checkout only.
+var scalingCaps = []int{128, 1024, 8192, 20000}
+
+func newScalingPool(capacity int) *agilepool.Pool {
+	return agilepool.NewPool(agilepool.NewConfig(
+		agilepool.WithCleanPeriod(500*time.Millisecond),
+		agilepool.WithTaskQueueSize(100000),
+		agilepool.WithWorkerNumCapacity(int64(capacity)),
+		agilepool.WithIdleContainerType(agilepool.LinkedListType),
+	))
+}
+
+// BenchmarkScalingNoop is hand-off bound: no-op tasks, so throughput is the
+// pool's submit -> worker pickup rate.
+func BenchmarkScalingNoop(b *testing.B) {
+	for _, capacity := range scalingCaps {
+		b.Run(fmt.Sprintf("cap=%d", capacity), func(b *testing.B) {
+			pool := newScalingPool(capacity)
+			defer pool.Close()
+			b.ReportAllocs()
+			b.ResetTimer()
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					pool.Submit(agilepool.TaskFunc(func() error { return nil }))
+				}
+			})
+		})
+	}
+}
+
+// BenchmarkScalingSleep10ms is execution bound: each task sleeps 10ms, so the
+// limiter is capacity/10ms and the channel hand-off overhead shows up only as
+// extra CPU and scheduling jitter, not throughput.
+func BenchmarkScalingSleep10ms(b *testing.B) {
+	for _, capacity := range scalingCaps {
+		b.Run(fmt.Sprintf("cap=%d", capacity), func(b *testing.B) {
+			pool := newScalingPool(capacity)
+			defer pool.Close()
+			b.ReportAllocs()
+			b.ResetTimer()
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					pool.Submit(agilepool.TaskFunc(func() error {
+						time.Sleep(10 * time.Millisecond)
+						return nil
+					}))
+				}
+			})
+			pool.Wait()
+		})
+	}
+}

config.go

@@ -34,7 +34,8 @@ const (
 
 type Config struct {
 	cleanPeriod        time.Duration
-	taskQueueSize      int64 // capacity of the internal handoff channel
+	taskQueueSize      int64 // total capacity of the internal handoff queues
+	taskQueueShards    int   // number of handoff-queue shards; 0 = auto
 	workerNumCapacity  int64
 	workMode           WorkMode
 	idleContainerType  IdleContainerType
@@ -100,6 +101,18 @@ func WithTaskQueueSize(size int64) ConfigOption {
 	}
 }
 
+// WithTaskQueueShards sets how many shards the internal handoff queue is split
+// into. Sharding spreads channel-lock contention across multiple channel locks,
+// which matters when the pool runs a very large number of workers that would
+// otherwise serialise on a single channel. The value is rounded up to a power
+// of two and each shard gets at least one slot. n <= 0 selects a default: a
+// single shard for small queues and GOMAXPROCS (capped) for large ones.
+func WithTaskQueueShards(n int) ConfigOption {
+	return func(c *Config) {
+		c.taskQueueShards = n
+	}
+}
+
 func WithWorkerNumCapacity(capacity int64) ConfigOption {
 	return func(c *Config) {
 		if capacity > 0 {

go.mod

@@ -2,10 +2,14 @@ module github.com/Yiming1997/agilePool/v2
 
 go 1.23.4
 
-require github.com/stretchr/testify v1.11.1
+require (
+	github.com/panjf2000/ants/v2 v2.12.1
+	github.com/stretchr/testify v1.11.1
+)
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	golang.org/x/sync v0.11.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,9 +1,13 @@
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/panjf2000/ants/v2 v2.12.1 h1:BWvU2wHpyXWxhhNXsGB6JXLCNbshyLd1QxvoAmZnu10=
+github.com/panjf2000/ants/v2 v2.12.1/go.mod h1:tSQuaNQ6r6NRhPt+IZVUevvDyFMTs+eS4ztZc52uJTY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=