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,22 @@ 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
+	// cscActiveFn reports whether client-side caching is CURRENTLY active on the
+	// underlying client (nil when the client type exposes none). Consulted per
+	// solo dispatch — not captured as a bool — because CSC can disable itself
+	// mid-life (RESP3 fallback, processor damping), after which cacheable solos
+	// should return to the pipeline pool instead of the main pool. Gates the
+	// cacheable-solo routing: only an active-CSC client routes through Process
+	// (which honors the cache).
+	cscActiveFn func() 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 +809,17 @@ 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()
+	}
+	// CSC probe (in-package assertion; *ClusterClient does not expose it) — see
+	// the cscActiveFn field doc.
+	if cc, ok := pipeliner.(interface{ autopipelineCSCActive() bool }); ok {
+		ap.cscActiveFn = 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,16 @@ 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 solo on a CSC client goes through Process so the cache
+				// is honored (processPipeline bypasses processCached). Gated on
+				// the LIVE CSC state: without active CSC, Process would just run on
+				// the MAIN pool, ignoring the pipeline pool the straggler gate probed.
+				if ap.cscActiveFn != nil && ap.cscActiveFn() && isCacheable(cmd) {
+					return ap.pipeliner.process(ctx, cmd)
+				}
+				// One-command pipeline on the PIPELINE pool (falls back to the main
+				// pool when none exists).
+				return ap.pipeliner.processPipeline(ctx, []Cmder{cmd})
 			})
 			solo.SetErr(err)
 			ap.observeBatchExec(time.Since(execStart))

autopipeline_internal_test.go

@@ -1468,3 +1468,56 @@ 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
+// the live gate (cscActiveFn) must report false there and true for a CSC
+// client — and false again after CSC disables itself mid-life.
+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.cscActiveFn != nil && ap.cscActiveFn() {
+		t.Fatal("non-CSC client: gate reports active, want inactive — 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.cscActiveFn == nil || !apc.cscActiveFn() {
+		t.Fatal("CSC client: gate reports inactive, want active — cacheable solos must honor the cache")
+	}
+	// Mid-life disable (RESP3 fallback, processor damping): the LIVE gate must
+	// flip so cacheable solos return to the pipeline pool.
+	cscC.disableCSCServing(ctx, "test: mid-life disable")
+	if apc.cscActiveFn() {
+		t.Fatal("gate still reports active after CSC disabled itself mid-life")
+	}
+}

internal/pool/has_free_capacity_test.go

@@ -0,0 +1,150 @@
+package pool_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9/internal/pool"
+)
+
+// TestHasFreeCapacityHonorsMaxActiveConns: with MaxActiveConns < PoolSize and
+// no idle conn, a Get would fail ErrPoolExhausted even though Len() < Size() —
+// HasFreeCapacity must report false.
+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: with MaxActiveConns unset the gate
+// reduces to the prior idle-or-under-size heuristic (fresh pool free, full pool
+// with no idle not-free, idle conn free again). Conservative by design: it does
+// not track non-pooled overflow Gets.
+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: a not-usable idle conn (mid handoff
+// or re-auth) must not count as capacity.
+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: a handoff-marked idle conn is still
+// StateIdle/usable but an OnGet hook diverts it, so it must not count as
+// 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")
+	}
+}