Yield the releasing goroutine's P to a blocked acquirer

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)

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

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