perf(autopipeline): better pool utilization on stranglers - *Client

.github/workflows/govulncheck.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v7
         with:
-          go-version: "1.26.x"
+          go-version: "stable"
           cache: true
 
       - name: Install govulncheck

autopipeline.go

@@ -15,6 +15,7 @@ import (
 	"golang.org/x/sys/cpu"
 
 	"github.com/redis/go-redis/v9/internal"
+	"github.com/redis/go-redis/v9/internal/pool"
 )
 
 // AutoPipelineOptions configures the autopipelining behavior.
@@ -545,7 +546,21 @@ type AutoPipeliner struct {
 	cmdable // Embed cmdable to get all Redis command methods
 
 	pipeliner cmdableClient
-	config    *AutoPipelineOptions
+	// pipelinePool is the connection pool that backs autopipelined batch
+	// dispatch (distinct from the client's main pool). Captured once at
+	// construction via an in-package assertion; nil when the underlying client
+	// does not expose one (e.g. *ClusterClient). The straggler-hold reads it to
+	// tell whether flushing a tiny batch now would contend for a scarce pooled
+	// connection — see awaitExpectedArrivals / pipelineHasFreeConn.
+	pipelinePool pool.Pooler
+	// cscEnabled records whether the underlying client has client-side caching
+	// active. Captured once at construction (CSC attaches at client creation,
+	// before the autopipeliner is lazily built). Only a CSC client benefits from
+	// routing a cacheable solo straggler through the single-command Process path
+	// (which honors the cache); a non-CSC client must instead dispatch it on the
+	// pipeline pool the straggler gate probed, not the main pool (#3962).
+	cscEnabled bool
+	config     *AutoPipelineOptions
 	// blocking selects how the typed command surface (Set, Get, ...) behaves:
 	// when true the command call itself blocks until the command has executed
 	// (drop-in, synchronous shape); when false the call returns immediately and
@@ -793,6 +808,18 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 		ctx:       ctx,
 		cancel:    cancel,
 	}
+	// Capture the pipeline pool (in-package, promoted to *Client). nil for a
+	// client that has none (e.g. *ClusterClient) — the straggler-hold then
+	// keeps its conservative long hold rather than guess at pool pressure.
+	if pp, ok := pipeliner.(interface{ getPipelinePool() pool.Pooler }); ok {
+		ap.pipelinePool = pp.getPipelinePool()
+	}
+	// Capture whether client-side caching is active (in-package, promoted to
+	// *Client; a *ClusterClient does not expose it). Used to route a cacheable
+	// solo straggler: through Process only when the cache can actually serve it.
+	if cc, ok := pipeliner.(interface{ autopipelineCSCActive() bool }); ok {
+		ap.cscEnabled = cc.autopipelineCSCActive()
+	}
 
 	// Route the typed command surface. Blocking: the command call blocks until
 	// executed (synchronous drop-in shape). Deferred: the call returns at once
@@ -1820,6 +1847,23 @@ const (
 // near-empty flush; once nothing is in flight, any size flushes immediately.
 const coalesceMinFlush = 8
 
+// stragglerHoldGaps bounds the straggler-hold WHEN the pipeline pool has a free
+// connection (see awaitExpectedArrivals): at most this many silence gaps (each
+// clamp(execEWMA/8, 200µs, 2ms), so the bound tracks the round trip) pass before
+// queued stragglers flush. The old behavior re-armed until
+// autoPipelinePermitBackstop — effectively until the in-flight batch's reply
+// landed, ~1 RTT — so a straggler that enqueued behind an in-flight batch waited
+// a full round trip BEFORE its own, i.e. every such op paid ~2x RTT (measured
+// with a phase trace on a deterministic 50ms link: uncached p95 pinned at 2x RTT
+// / 107ms at low-to-mid concurrency, straggler-hold avg == 1 RTT; the bound
+// collapsed that to ~1 RTT / 62ms). The bound is applied ONLY when a pooled
+// connection is idle or dial-able — when the pool is saturated the long hold is
+// kept, because flushing tiny batches into a full pool thrashes it and cuts
+// throughput (measured ~7x at a squeezed pool). The 30s
+// autoPipelinePermitBackstop remains the absolute safety ceiling on the flush
+// path itself (a wedged connection).
+const stragglerHoldGaps = 3
+
 // observeBatchExec folds one batch execution duration into execEWMA.
 func (ap *AutoPipeliner) observeBatchExec(d time.Duration) {
 	sample := int64(d)
@@ -1847,6 +1891,29 @@ func (ap *AutoPipeliner) silenceGap() time.Duration {
 	return g
 }
 
+// pipelineHasFreeConn reports whether the pipeline pool can serve another batch
+// without blocking: an idle connection is ready, or the pool has not yet dialed
+// to capacity (the pipeline pool runs MinIdleConns=0, so it dials on demand up
+// to Size). When the pool is unknown (nil — e.g. a cluster client) it returns
+// false, so the straggler-hold keeps its conservative long hold. Called only on
+// a gap fire, not per command.
+//
+// Prefer the pool's own HasFreeCapacity probe (all *ConnPool implement it): the
+// plain IdleLen()/Len()<Size() heuristic below ignores MaxActiveConns, so a pool
+// with MaxActiveConns < PoolSize and no idle conn would report free even though
+// the flush's Get would hit ErrPoolExhausted (codex #3962). The heuristic stays
+// as a fallback for any Pooler that does not implement the probe.
+func (ap *AutoPipeliner) pipelineHasFreeConn() bool {
+	p := ap.pipelinePool
+	if p == nil {
+		return false
+	}
+	if hc, ok := p.(interface{ HasFreeCapacity() bool }); ok {
+		return hc.HasFreeCapacity()
+	}
+	return p.IdleLen() > 0 || p.Len() < p.Size()
+}
+
 // awaitExpectedArrivals holds the flusher while related work is in motion, so
 // commands flush as deep pipelines instead of fragmenting into small batches
 // (each fragment costs a pipeline connection for a full round trip). Two
@@ -1905,15 +1972,33 @@ func (s *apShard) awaitExpectedArrivals(batchSize int) {
 				// flushing a near-empty pipeline burns a connection for a full
 				// round trip (measured at high WAN concurrency: straggler
 				// flushes of 1-3 commands starved the connection pool and
-				// doubled p50). Hold them — the next completed batch's wave
-				// sweeps them along, and the wave path below flushes promptly.
-				// The hold is bounded like the permit wait: with read timeouts
-				// disabled a wedged batch could pin inFlight forever, and the
-				// held stragglers must not hang with it.
+				// doubled p50). How long to hold depends on whether the pipeline
+				// pool has a connection to spare:
+				//
+				//   - a connection is free -> bound the hold at stragglerHoldGaps
+				//     silence gaps (a few ms). Flushing then costs an otherwise-
+				//     idle connection and saves the straggler ~1 RTT. Waiting a
+				//     whole round trip here (the old behavior) is what pinned
+				//     low-concurrency stragglers at 2x RTT.
+				//   - the pool is saturated -> keep the original long hold. Tiny
+				//     flushes into a full pool thrash it: they cannot coalesce
+				//     into the deep pipelines the scarce connections need, and
+				//     throughput collapses (measured at a squeezed pool: an
+				//     unconditional few-ms bound cut throughput ~7x). Holding
+				//     lets the next completed batch's wave sweep the stragglers
+				//     along.
+				//
+				// A wedged in-flight batch cannot hang the held stragglers past
+				// the bound (the free-conn case caps at a few ms; the flush path
+				// keeps its own autoPipelinePermitBackstop safety ceiling).
 				if holdStart.IsZero() {
 					holdStart = time.Now()
 				}
-				if time.Since(holdStart) < autoPipelinePermitBackstop {
+				stragCap := autoPipelinePermitBackstop
+				if ap.pipelineHasFreeConn() {
+					stragCap = stragglerHoldGaps * gap
+				}
+				if time.Since(holdStart) < stragCap {
 					lastSeenExpected = ap.expectedArrivals.Load()
 					fallback.Reset(gap)
 					continue
@@ -2328,7 +2413,22 @@ func (s *apShard) flushBatchSlice() {
 			// deadlock-free via the dispGid guard stamped above.
 			// A successful short-circuit stays successful (see dispatchCmds).
 			err := ap.pipeliner.withProcessHook(context.Background(), solo, func(ctx context.Context, cmd Cmder) error {
-				return ap.pipeliner.process(ctx, cmd)
+				// A cacheable command goes through the single-command Process path so
+				// client-side caching is honored (processCommand -> processCached) —
+				// processPipeline bypasses that branch and would silently drop CSC
+				// hits/fills for one-command flushes (codex #3962). But ONLY when the
+				// client actually has CSC: on a non-CSC client Process just runs on the
+				// MAIN pool, so a cacheable solo would ignore the dedicated pipeline pool
+				// the straggler gate probed and could contend on a saturated main pool
+				// while pipeline capacity sits free. So gate on cscEnabled.
+				if ap.cscEnabled && isCacheable(cmd) {
+					return ap.pipeliner.process(ctx, cmd)
+				}
+				// Otherwise dispatch as a one-command pipeline on the PIPELINE pool, the
+				// same pool the straggler-hold gate probes (processPipeline falls back to
+				// the main pool when none exists). The solo goroutine dispatch (the 2xRTT
+				// phase-lock fix) is unchanged.
+				return ap.pipeliner.processPipeline(ctx, []Cmder{cmd})
 			})
 			solo.SetErr(err)
 			ap.observeBatchExec(time.Since(execStart))

autopipeline_internal_test.go

@@ -1468,3 +1468,49 @@ func TestAsyncProcessReportsSubmitRejection(t *testing.T) {
 		t.Fatalf("Process after Close = %v, want ErrClosed", err)
 	}
 }
+
+// TestAutopipelineSoloRoutingGate pins #3962: a cacheable solo straggler is
+// routed through the cache-honoring Process path (main pool) ONLY when the
+// client has client-side caching active. A non-CSC client (the default) must
+// keep cacheable solos on the pipeline pool the straggler gate probed, so
+// ap.cscEnabled — the gate — must be false there and true for a CSC client.
+func TestAutopipelineSoloRoutingGate(t *testing.T) {
+	ctx := context.Background()
+
+	// Non-CSC client: dial-free and deterministic.
+	plain := NewClient(&Options{Addr: "localhost:1", Protocol: 3})
+	defer plain.Close()
+	ap, err := plain.AsyncAutoPipeline()
+	if err != nil {
+		t.Fatalf("AsyncAutoPipeline: %v", err)
+	}
+	defer ap.Close()
+	if ap.cscEnabled {
+		t.Fatal("non-CSC client: ap.cscEnabled = true, want false — a cacheable solo would wrongly use the main pool instead of the pipeline pool")
+	}
+
+	// CSC client: needs a live RESP3 server for CLIENT TRACKING to attach.
+	if err := NewClient(&Options{Addr: internalTestRedisAddr(), Protocol: 3}).Ping(ctx).Err(); err != nil {
+		t.Skipf("no redis for the CSC half: %v", err)
+	}
+	cscC := NewClient(&Options{
+		Addr:                  internalTestRedisAddr(),
+		Protocol:              3,
+		ClientSideCacheConfig: &ClientSideCacheConfig{MaxEntries: 128},
+	})
+	defer cscC.Close()
+	if err := cscC.Ping(ctx).Err(); err != nil {
+		t.Skipf("no redis: %v", err)
+	}
+	if !cscC.autopipelineCSCActive() {
+		t.Skip("client-side caching did not attach (server lacks CLIENT TRACKING?)")
+	}
+	apc, err := cscC.AsyncAutoPipeline()
+	if err != nil {
+		t.Fatalf("AsyncAutoPipeline (csc): %v", err)
+	}
+	defer apc.Close()
+	if !apc.cscEnabled {
+		t.Fatal("CSC client: ap.cscEnabled = false, want true — cacheable solos must honor the cache")
+	}
+}

internal/pool/has_free_capacity_test.go

@@ -0,0 +1,161 @@
+package pool_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9/internal/pool"
+)
+
+// TestHasFreeCapacityHonorsMaxActiveConns pins the branch the plain
+// IdleLen()>0 || Len()<Size() heuristic missed: with MaxActiveConns < PoolSize
+// and no idle connection, the pool cannot serve another Get (newConn returns
+// ErrPoolExhausted once poolSize >= MaxActiveConns), yet Len() < Size() still
+// holds. HasFreeCapacity must report false there — that is what makes the
+// autopipeline straggler-hold gate stop shortening the hold into a pool the
+// flush's Get would exhaust (#3962).
+func TestHasFreeCapacityHonorsMaxActiveConns(t *testing.T) {
+	connPool := pool.NewConnPool(&pool.Options{
+		Dialer:             dummyDialer,
+		PoolSize:           4, // Len()<Size() alone would report "free"...
+		MaxActiveConns:     1, // ...but only one active connection is allowed.
+		MaxConcurrentDials: 1,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	})
+	t.Cleanup(func() { _ = connPool.Close() })
+	ctx := context.Background()
+
+	// Fresh pool: nothing dialed yet, below PoolSize and MaxActiveConns — the
+	// first Get can dial, so there is free capacity.
+	if !connPool.HasFreeCapacity() {
+		t.Fatal("fresh pool: HasFreeCapacity() = false, want true (first Get can dial)")
+	}
+
+	first, err := connPool.Get(ctx)
+	if err != nil {
+		t.Fatalf("first Get: %v", err)
+	}
+
+	// One active connection (== MaxActiveConns) and none idle: the next Get would
+	// return ErrPoolExhausted even though Len()(=1) < Size()(=4). The old heuristic
+	// would wrongly report free; HasFreeCapacity must report false.
+	if connPool.HasFreeCapacity() {
+		t.Fatal("at MaxActiveConns with no idle conn: HasFreeCapacity() = true, want false")
+	}
+
+	// Return it: an idle connection is now ready, so there is capacity again.
+	connPool.Put(ctx, first)
+	if !connPool.HasFreeCapacity() {
+		t.Fatal("with an idle conn available: HasFreeCapacity() = false, want true")
+	}
+}
+
+// TestHasFreeCapacityWithoutMaxActiveConns confirms that with MaxActiveConns
+// unset HasFreeCapacity reduces EXACTLY to the prior `IdleLen()>0 || Len()<Size()`
+// heuristic: a fresh pool is free, a pool grown to PoolSize with no idle conn is
+// reported not-free, and an idle conn makes it free again. This is a conservative
+// gate, not an admission check: a second Get on the PoolSize-full pool would in
+// fact still succeed with a non-pooled connection (PoolSize does not hard-block
+// dials) — HasFreeCapacity deliberately does not track that, it just keeps the
+// straggler-hold conservative.
+func TestHasFreeCapacityWithoutMaxActiveConns(t *testing.T) {
+	connPool := pool.NewConnPool(&pool.Options{
+		Dialer:             dummyDialer,
+		PoolSize:           1,
+		MaxConcurrentDials: 1,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	})
+	t.Cleanup(func() { _ = connPool.Close() })
+	ctx := context.Background()
+
+	if !connPool.HasFreeCapacity() {
+		t.Fatal("fresh pool: HasFreeCapacity() = false, want true")
+	}
+
+	cn, err := connPool.Get(ctx)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	// Grown to PoolSize (1), no idle conn and no free turn (the one turn is held):
+	// the gate reports false.
+	if connPool.HasFreeCapacity() {
+		t.Fatal("at PoolSize with no idle conn / no free turn: HasFreeCapacity() = true, want false")
+	}
+	connPool.Put(ctx, cn)
+	if !connPool.HasFreeCapacity() {
+		t.Fatal("with a usable idle conn available: HasFreeCapacity() = false, want true")
+	}
+}
+
+// TestHasFreeCapacityExcludesUnusableIdle pins the refinement that an idle
+// connection which is not usable (mid handoff / re-auth) does NOT count as
+// capacity — the old IdleLen()>0 term would have wrongly reported free.
+func TestHasFreeCapacityExcludesUnusableIdle(t *testing.T) {
+	connPool := pool.NewConnPool(&pool.Options{
+		Dialer:             dummyDialer,
+		PoolSize:           1,
+		MaxConcurrentDials: 1,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	})
+	t.Cleanup(func() { _ = connPool.Close() })
+	ctx := context.Background()
+
+	cn, err := connPool.Get(ctx)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	connPool.Put(ctx, cn) // one usable idle conn
+	if !connPool.HasFreeCapacity() {
+		t.Fatal("usable idle conn: HasFreeCapacity() = false, want true")
+	}
+
+	// Mark the idle conn unusable (as a handoff / re-auth would). It is still in
+	// idleConns, so IdleLen()>0, but it cannot serve a Get; at PoolSize there is
+	// also nothing to dial, so capacity must read false.
+	cn.SetUsable(false)
+	if connPool.HasFreeCapacity() {
+		t.Fatal("idle conn is UNUSABLE and pool is at PoolSize: HasFreeCapacity() = true, want false")
+	}
+}
+
+// TestHasFreeCapacityExcludesHandoffIdle pins the OnGet-reject refinement (#3962):
+// an idle connection marked ShouldHandoff is still StateIdle/usable, but an OnGet
+// hook (maintnotifications) diverts it to handoff instead of serving it, so
+// HasFreeCapacity must not count it as capacity. With the pool at PoolSize (no
+// dial possible), a lone handoff-marked idle conn means no free capacity.
+func TestHasFreeCapacityExcludesHandoffIdle(t *testing.T) {
+	connPool := pool.NewConnPool(&pool.Options{
+		Dialer:             dummyDialer,
+		PoolSize:           1,
+		MaxActiveConns:     1,
+		MaxConcurrentDials: 1,
+		PoolTimeout:        time.Second,
+		DialTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	})
+	t.Cleanup(func() { _ = connPool.Close() })
+	ctx := context.Background()
+
+	cn, err := connPool.Get(ctx)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	if err := cn.MarkForHandoff("new-endpoint:6379", 1); err != nil {
+		t.Fatalf("MarkForHandoff: %v", err)
+	}
+	if !cn.IsUsable() {
+		t.Fatal("precondition: a handoff-marked conn should still be IsUsable (StateIdle)")
+	}
+	connPool.Put(ctx, cn)
+
+	if connPool.HasFreeCapacity() {
+		t.Fatal("HasFreeCapacity() = true with only a handoff-marked idle conn — an OnGet hook would divert it, so it must not count")
+	}
+}