diff --git a/CHANGELOG.md b/CHANGELOG.md index d0d202c..2abc4c3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +# Unreleased + +* Yield the releasing goroutine's P to a blocked acquirer after Release, Destroy and Hijack, so a released resource does not sit unused until the releaser blocks (Mikhail Lukianchenko) + # 2.2.2 (September 10, 2024) * Add empty acquire time to stats (Maxim Ivanov) diff --git a/pool.go b/pool.go index 4b8c361..9885be3 100644 --- a/pool.go +++ b/pool.go @@ -4,6 +4,7 @@ import ( "context" "errors" "math/bits" + "runtime" "sync" "sync/atomic" "time" @@ -142,6 +143,8 @@ type Pool[T any] struct { emptyAcquireCount int64 emptyAcquireWaitTime time.Duration canceledAcquireCount atomic.Int64 + // waiting is the number of goroutines blocked in acquireSem.Acquire. + waiting atomic.Int64 resetCount int @@ -358,7 +361,9 @@ func (p *Pool[T]) acquire(ctx context.Context) (*Resource[T], error) { var waitedForLock bool if !p.acquireSem.TryAcquire(1) { waitedForLock = true + p.waiting.Add(1) err := p.acquireSem.Acquire(ctx, 1) + p.waiting.Add(-1) if err != nil { p.canceledAcquireCount.Add(1) return nil, err @@ -671,6 +676,7 @@ func (p *Pool[T]) Reset() { // releaseAcquiredResource returns res to the the pool. func (p *Pool[T]) releaseAcquiredResource(res *Resource[T], lastUsedNano int64) { + defer p.yieldToWaiter() p.mux.Lock() defer p.mux.Unlock() defer p.acquireSem.Release(1) @@ -690,6 +696,7 @@ func (p *Pool[T]) releaseAcquiredResource(res *Resource[T], lastUsedNano int64) func (p *Pool[T]) destroyAcquiredResource(res *Resource[T]) { p.destructResourceValue(res.value) + defer p.yieldToWaiter() p.mux.Lock() defer p.mux.Unlock() defer p.acquireSem.Release(1) @@ -698,6 +705,7 @@ func (p *Pool[T]) destroyAcquiredResource(res *Resource[T]) { } func (p *Pool[T]) hijackAcquiredResource(res *Resource[T]) { + defer p.yieldToWaiter() p.mux.Lock() defer p.mux.Unlock() defer p.acquireSem.Release(1) @@ -707,6 +715,17 @@ func (p *Pool[T]) hijackAcquiredResource(res *Resource[T]) { p.destructWG.Done() // not responsible for destructing hijacked resources } +// yieldToWaiter hands this goroutine's P to the acquirer a release woke. The +// runtime queues a goroutine woken by a channel close as runnext of the +// waker's P, where it runs only once the waker blocks or another P steals it, +// so without the yield a released resource sits unused until then. Must be +// called with mux unlocked. +func (p *Pool[T]) yieldToWaiter() { + if p.waiting.Load() > 0 { + runtime.Gosched() + } +} + func (p *Pool[T]) destructResourceValue(value T) { p.destructor(value) p.destructWG.Done() diff --git a/pool_test.go b/pool_test.go index a313568..580ae6e 100644 --- a/pool_test.go +++ b/pool_test.go @@ -1510,3 +1510,44 @@ func BenchmarkAcquire_MultipleCancelledWithCPULoad(b *testing.B) { r.NoError(err) } } + +// BenchmarkAcquire_ReleaseThenWork saturates a pool of 4 with ~100 goroutines +// that hold a resource for 1 ms and keep the CPU for 1 ms after releasing it. +func BenchmarkAcquire_ReleaseThenWork(b *testing.B) { + const ( + size = 4 + queue = 100 + hold = time.Millisecond + work = time.Millisecond + ) + + r := require.New(b) + ctx := context.Background() + pool, err := puddle.NewPool(&puddle.Config[int32]{ + MaxSize: size, + Constructor: func(context.Context) (int32, error) { return 0, nil }, + Destructor: func(int32) {}, + }) + r.NoError(err) + defer pool.Close() + for i := 0; i < size; i++ { + r.NoError(pool.CreateResource(ctx)) + } + + procs := runtime.GOMAXPROCS(0) + b.SetParallelism((size + queue + procs - 1) / procs) + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + res, err := pool.Acquire(ctx) + if err != nil { + b.Error(err) + return + } + time.Sleep(hold) + res.Release() + for t := time.Now(); time.Since(t) < work; { + } + } + }) +}