feat(autopipeline): add FullDuplexPipes option (#4107)

NumShards meant two different things: the number of half-duplex queue
shards (raising it usually hurts and needs Unordered on the async face),
and, with FullDuplex, the number of held full-duplex connections
(raising it is the only way past one connection's ceiling).

FullDuplexPipes now names the full-duplex pipe count. NumShards > 1 is
still honored under FullDuplex when FullDuplexPipes is 0, so 9.23.0
configs keep working; Validate rejects the two set to different values.
Unlike NumShards, FullDuplexPipes has no effect when full duplex cannot
engage, so it never turns into unordered half-duplex shards. Cluster
full duplex keeps one pipe per master.

.claude/specs/full-duplex.md

@@ -55,7 +55,8 @@ aggregating correctly; only single-node commands route through a child.
 
 - **Child config**: derived once from the parent's `AutoPipelineOptions`,
   forcing the ordered single-shard FD combo (`FullDuplex/!Unordered/
-  MaxConcurrentBatches<=1/NumShards<=1`) the engine requires, and stripping the
+  MaxConcurrentBatches<=1/NumShards<=1/FullDuplexPipes<=1`) the engine requires
+  (one pipe per master: a parent's `FullDuplexPipes` must not multiply per node), and stripping the
   cluster-only `contentSharded` bit so it cannot leak onto a node child.
 - **Redirects**: each child's `fdEngine.reprocess` is wired to
   `clusterReprocess`, which re-runs a MOVED/ASK (or otherwise retryable) reply

autopipeline.go

@@ -99,9 +99,9 @@ type AutoPipelineOptions struct {
 	//
 	// Honored on the ordered (Unordered:false, MaxConcurrentBatches<=1) face of a
 	// standalone *Client that has a pipeline pool — BOTH the deferred
-	// (AsyncAutoPipeline) and the blocking (AutoPipeline) face. NumShards > 1
-	// runs that many full-duplex engines, each on its own held connection; see
-	// NumShards for the ordering that mode keeps. A SINGLE
+	// (AsyncAutoPipeline) and the blocking (AutoPipeline) face. FullDuplexPipes > 1
+	// runs that many full-duplex pipes, each on its own held connection; see
+	// FullDuplexPipes for the ordering that mode keeps. A SINGLE
 	// blocking caller gains nothing: it has one command in flight, so there is
 	// nothing to overlap, and it still pays the held connection and goroutine
 	// overhead; the win needs MANY concurrent blocking callers, whose commands then
@@ -211,10 +211,10 @@ type AutoPipelineOptions struct {
 	// set; 0 means the default (65536, covering ~50ms links at ~1.3M ops/s) and a
 	// negative value is rejected by Validate.
 	//
-	// The window is PER ENGINE. With NumShards > 1 each full-duplex engine gets
-	// its own window and submit queue of this size, so the client-wide bound is
-	// NumShards × FullDuplexWindow in flight, plus as many queued. Size it for
-	// one wire; to cap the total, divide by NumShards.
+	// The window is PER PIPE. With FullDuplexPipes > 1 each pipe gets its own
+	// window and submit queue of this size, so the client-wide bound is
+	// FullDuplexPipes × FullDuplexWindow in flight, plus as many queued. Size it
+	// for one pipe; to cap the total, divide by FullDuplexPipes.
 	FullDuplexWindow int
 
 	// FullDuplexIdleTimeout is how long the held full-duplex connection may sit
@@ -230,6 +230,36 @@ type AutoPipelineOptions struct {
 	// value is rejected by Validate.
 	FullDuplexMaxHold time.Duration
 
+	// FullDuplexPipes is the number of full-duplex pipes on a standalone *Client.
+	// Each pipe is one held pipeline-pool connection with its own writer, reader,
+	// window (see FullDuplexWindow) and submit queue. 0 means 1. Only used when
+	// FullDuplex is set; a negative value is rejected by Validate.
+	//
+	// One pipe has a throughput ceiling; more pipes are the way past it, and they
+	// pay off only with many concurrent callers: measured with 10-command
+	// pipelines, 8 pipes were 30% slower than 1 at 8 callers, broke even at about
+	// 128, and 2.2x faster at 1024. Raise it only when one pipe is saturated.
+	//
+	// Commands are routed to a pipe by a hash of their first key, so Unordered is
+	// not required, but order holds only between commands that share a first key.
+	// The key is hashed in its string form (fmt for non-string arguments, as Ring
+	// does), so pass a key as the same Go type everywhere: true and "1" are the
+	// same Redis key but can land on different pipes. Commands that touch other
+	// keys after the first one (COPY, RENAME, MSET, multi-key DEL, EVAL/FCALL with
+	// several keys) and keyless commands (including FLUSHDB and SWAPDB) are not
+	// ordered against the rest; await the first future before submitting a
+	// dependent one, or keep FullDuplexPipes at 1, which keeps the whole submit
+	// order. A Pipeline() rides a pipe only when all its keyed commands hash to
+	// one pipe; otherwise it runs as an ordinary pipeline on a pooled connection,
+	// not ordered against unawaited commands on the pipes.
+	//
+	// The pipeline pool must hold at least FullDuplexPipes connections for every
+	// full-duplex autopipeliner in use (construction checks this one); raise
+	// Options.PipelinePoolSize if needed. If full duplex cannot engage (no
+	// pipeline pool), FullDuplexPipes has no effect. On a ClusterClient full
+	// duplex runs one pipe per master and ignores FullDuplexPipes.
+	FullDuplexPipes int
+
 	// contentSharded is set internally by cluster wiring when commands are
 	// routed to shards by content (slot), so same-key commands always share a
 	// shard and per-key order holds even with several shards. It exempts that
@@ -268,28 +298,12 @@ type AutoPipelineOptions struct {
 	// concurrently, NumShards > 1 on the deferred (async) face requires
 	// Unordered: true (construction fails otherwise).
 	//
-	// With FullDuplex active on a standalone *Client, NumShards means something
-	// else: the number of full-duplex engines, each holding one pipeline-pool
-	// connection with its own window (see FullDuplexWindow). More engines pay
-	// off only with many concurrent callers: measured with 10-command
-	// pipelines, 8 engines were 30% slower than 1 at 8 callers, broke even at about
-	// 128, and 2.2x faster at 1024. Commands are routed
-	// to an engine by a hash of their first key, so Unordered is not required,
-	// but order holds only between commands that share a first key. The key is
-	// hashed in its string form (fmt for non-string arguments, as Ring does), so
-	// pass a key as the same Go type everywhere: true and "1" are the same Redis
-	// key but can land on different engines. Commands
-	// that touch other keys after the first one (COPY, RENAME, MSET, multi-key
-	// DEL, EVAL/FCALL with several keys) and keyless commands (including
-	// FLUSHDB and SWAPDB) are not ordered against the rest; await the first
-	// future before submitting a dependent one, or keep NumShards at 1, which
-	// keeps the whole submit order. A Pipeline() rides the FD wire only when all
-	// its keyed commands hash to one engine; otherwise it runs as an ordinary
-	// pipeline on a pooled connection, not ordered against unawaited commands on
-	// the engines. The pipeline pool must hold at least
-	// NumShards connections for every full-duplex autopipeliner in use
-	// (construction checks this one). If full duplex cannot engage (no
-	// pipeline pool), the Unordered rule above applies.
+	// Full duplex: use FullDuplexPipes to choose the number of pipes. For
+	// compatibility with 9.23.0, NumShards > 1 with FullDuplex still selects
+	// that many pipes when FullDuplexPipes is 0; Validate rejects the two set to
+	// different values. If full duplex cannot engage (no pipeline pool),
+	// NumShards keeps its half-duplex meaning and the Unordered rule above
+	// applies.
 	NumShards int
 
 	// MaxFlushDelay is the maximum delay after flushing before checking for more commands.
@@ -516,6 +530,17 @@ func (cfg *AutoPipelineOptions) Validate() error {
 		if cfg.FullDuplexMaxHold < 0 {
 			return fmt.Errorf("redis: AutoPipelineOptions.FullDuplexMaxHold=%s must be >= 0 (0 = default)", cfg.FullDuplexMaxHold)
 		}
+		if cfg.FullDuplexPipes < 0 {
+			return fmt.Errorf("redis: AutoPipelineOptions.FullDuplexPipes=%d must be >= 0 (0 = one pipe)", cfg.FullDuplexPipes)
+		}
+		// NumShards>1 is the 9.23.0 spelling of the pipe count under full duplex
+		// (see fdPipeCount). Two different counts are ambiguous; reject them.
+		if cfg.FullDuplexPipes > 1 && cfg.NumShards > 1 && cfg.FullDuplexPipes != cfg.NumShards &&
+			!cfg.contentSharded {
+			return fmt.Errorf("redis: AutoPipelineOptions.FullDuplexPipes=%d and NumShards=%d disagree; "+
+				"set only FullDuplexPipes (NumShards is the older spelling of the full-duplex pipe count)",
+				cfg.FullDuplexPipes, cfg.NumShards)
+		}
 	}
 	return nil
 }
@@ -1358,7 +1383,9 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 			clusterFDOn, clusterFDCC = true, cc
 			nShards = 1
 			// Report the actual shard count (1), not the cluster default, from Config().
+			// Cluster full duplex runs one pipe per master (FullDuplexPipes is ignored).
 			config.NumShards = 1
+			config.FullDuplexPipes = 1
 			// Resolve the FD tuning defaults on the PARENT config now, mirroring
 			// newFDEngine (which only writes them onto each child's config). Without
 			// this, Config() on a default-constructed cluster-FD autopipeliner would
@@ -1408,15 +1435,24 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 		// NumShards 8 pass here against a pool of 10, and the surplus spills.
 		// A client-wide engine count is a follow-up; until then size
 		// PipelinePoolSize for every full-duplex autopipeliner in use.
-		if n := fdShardCount(config); n > 1 {
+		n := fdPipeCount(config)
+		if n > 1 {
+			field := "FullDuplexPipes"
+			if config.FullDuplexPipes <= 0 {
+				field = "NumShards"
+			}
 			if pp := fdClient.getPipelinePool(); pp != nil && n > pp.Size() {
 				return nil, fmt.Errorf(
-					"redis: AutoPipelineOptions.NumShards=%d needs a pipeline pool of at "+
-						"least %d connections (each full-duplex engine holds one), but "+
+					"redis: AutoPipelineOptions.%s=%d needs a pipeline pool of at "+
+						"least %d connections (each full-duplex pipe holds one), but "+
 						"PipelinePoolSize gives %d; raise Options.PipelinePoolSize or lower "+
-						"NumShards", n, n, pp.Size())
+						"%s", field, n, n, pp.Size(), field,
+				)
 			}
 		}
+		// Report the effective pipe count from Config(), whichever field asked
+		// for it.
+		config.FullDuplexPipes = n
 		// One flusherless shard, exactly as the single-engine path does: no
 		// half-duplex flusher runs and enqueue never does a %0, even though
 		// submit routes past it entirely.
@@ -1476,10 +1512,10 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineOptions, bloc
 	}
 
 	if fdOn {
-		// NumShards>1 runs N engines on THIS client, each leasing its own
-		// connection from the pipeline pool. fds[0] is fd, so nothing
+		// FullDuplexPipes>1 runs N engines (pipes) on THIS client, each leasing
+		// its own connection from the pipeline pool. fds[0] is fd, so nothing
 		// downstream needs to know the difference when N==1.
-		n := fdShardCount(config)
+		n := fdPipeCount(config)
 		ap.fds = make([]*fdEngine, 0, n)
 		for i := 0; i < n; i++ {
 			e := newFDEngine(ap, fdClient)

autopipeline_cluster_fd.go

@@ -72,6 +72,7 @@ func newClusterFDRouter(parent *AutoPipeliner, cc *ClusterClient, cfg *AutoPipel
 	child.Unordered = false
 	child.MaxConcurrentBatches = 1
 	child.NumShards = 1
+	child.FullDuplexPipes = 1 // one pipe per master; a parent's count must not multiply per node
 	child.contentSharded = false
 	// Redirect handling: each child's FD engine re-runs a MOVED/ASK (or retryable)
 	// reply through the redirect-aware ClusterClient instead of its own node-local

autopipeline_fd_pipeline.go

@@ -52,8 +52,8 @@ var ErrFDPipelineUnavailable = errors.New(
 var ErrFDPipelineDiverts = errors.New(
 	"redis: FDPipelined got a command that must be diverted off the full-duplex pipe")
 
-// ErrFDPipelineSpansEngines is returned when NumShards > 1 runs several
-// full-duplex engines and the batch's keyed commands hash to different ones.
+// ErrFDPipelineSpansEngines is returned when FullDuplexPipes > 1 runs several
+// full-duplex engines (pipes) and the batch's keyed commands hash to different ones.
 // The batch can only ride one wire, and keeping per-first-key order needs each
 // command on its own key's engine, so it is refused rather than split.
 // Pipeline() falls back to an ordinary pipeline for such a batch.

autopipeline_fd_shards.go

@@ -6,7 +6,7 @@ package redis
 // connection has a throughput ceiling — measured at roughly 920k ops/s on a
 // 14-core client against a local server, for both this client and rueidis. Past
 // that the engine does not saturate, it degrades, because callers queue behind
-// a wire that is already busy. Raising NumShards above 1 under FullDuplex runs
+// a wire that is already busy. Raising FullDuplexPipes above 1 runs
 // several engines on the same client, each holding its own connection, which is
 // the only way past that ceiling.
 //
@@ -30,13 +30,13 @@ package redis
 // on different engines. Cluster sharding never meets this case because Redis
 // Cluster rejects multi-key commands across slots (CROSSSLOT); a standalone
 // server accepts them. Callers that chain such commands should await the
-// first future before submitting the dependent one, or keep NumShards at 1.
+// first future before submitting the dependent one, or keep FullDuplexPipes at 1.
 //
 // Keyless commands are not ordered against anything either: they round-robin.
 // That includes commands that change state for every key (FLUSHDB, FLUSHALL,
 // SWAPDB, SCRIPT FLUSH, FUNCTION FLUSH/RESTORE), so `FlushDB()` followed by
 // `Get(k)` can run the GET first. The same rule applies: await the first
-// future, or keep NumShards at 1.
+// future, or keep FullDuplexPipes at 1.
 //
 // Keys are read with stringArg, the same conversion Ring and the cluster slot
 // hash use, so all three route the same key the same way. It differs from the
@@ -50,8 +50,8 @@ package redis
 //
 // This is an opt-in: before this change, NumShards>1 never ran more than one
 // full-duplex engine, so every existing caller keeps the single ordered stream
-// and only a caller that raises NumShards on a full-duplex autopipeliner
-// accepts per-first-key order.
+// and only a caller that raises the pipe count (FullDuplexPipes, or NumShards as
+// spelled in 9.23.0) on a full-duplex autopipeliner accepts per-first-key order.
 //
 // The cost of that choice is measurable and worth stating: pinning a caller to
 // one engine means an unlucky key waits behind that engine's queue while
@@ -66,20 +66,27 @@ import (
 	"hash/fnv"
 )
 
-// fdShardCount reports how many full-duplex engines to run.
+// fdPipeCount reports how many full-duplex engines (pipes) to run.
 //
-// 0 and 1 both mean a single engine, which is the default and leaves the
-// single-engine path untouched. Sizing is deliberately NOT automatic yet: the
+// FullDuplexPipes is the option; NumShards>1 is honored as its 9.23.0 spelling
+// when FullDuplexPipes is 0 (Validate rejects the two disagreeing). 0 and 1 both
+// mean a single engine, which is the default and leaves the single-engine path
+// untouched. Sizing is deliberately NOT automatic yet: the
 // optimum tracks caller count (roughly one wire per 256-512 callers), and a
 // fixed count above 1 costs low-concurrency callers real throughput — eight
 // engines measured ~30% behind one at 8 callers, because a handful of callers
 // cannot fill eight queues and every batch fragments. Until the count can be
 // derived from observed load, it stays the caller's explicit decision.
-func fdShardCount(cfg *AutoPipelineOptions) int {
-	if !cfg.FullDuplex || cfg.NumShards <= 1 {
+func fdPipeCount(cfg *AutoPipelineOptions) int {
+	switch {
+	case !cfg.FullDuplex:
 		return 1
+	case cfg.FullDuplexPipes > 0:
+		return cfg.FullDuplexPipes
+	case cfg.NumShards > 1:
+		return cfg.NumShards
 	}
-	return cfg.NumShards
+	return 1
 }
 
 // fdFor picks the engine that will carry one command.

autopipeline_fullduplex_test.go

@@ -6906,3 +6906,125 @@ func BenchmarkFDQueueDeepDrain(b *testing.B) {
 		q.pushBatch(refill)
 	}
 }
+
+// FullDuplexPipes chooses the number of full-duplex pipes (engines). NumShards>1
+// is still honored as the 9.23.0 spelling when FullDuplexPipes is 0, and
+// Config() reports the effective pipe count either way. Construction only:
+// engines lease their connection lazily, so no server is needed.
+func TestFullDuplexPipesEngineCount(t *testing.T) {
+	cases := []struct {
+		name string
+		cfg  AutoPipelineOptions
+		want int
+	}{
+		{"default", AutoPipelineOptions{FullDuplex: true}, 1},
+		{"pipes", AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: 3}, 3},
+		{"numshards fallback", AutoPipelineOptions{FullDuplex: true, NumShards: 3}, 3},
+		{"both equal", AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: 3, NumShards: 3}, 3},
+		{"pipes with numshards 1", AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: 2, NumShards: 1}, 2},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			c := NewClient(&Options{Addr: internalTestRedisAddr()})
+			defer c.Close()
+			cfg := tc.cfg
+			ap, err := c.AsyncAutoPipelineWithOptions(&cfg)
+			if err != nil {
+				t.Fatalf("construct: %v", err)
+			}
+			defer ap.Close()
+			if got := len(ap.fds); got != tc.want {
+				t.Fatalf("engines = %d, want %d", got, tc.want)
+			}
+			got := ap.Config()
+			if !got.FullDuplex {
+				t.Fatal("Config().FullDuplex = false, want true")
+			}
+			if got.FullDuplexPipes != tc.want {
+				t.Fatalf("Config().FullDuplexPipes = %d, want %d", got.FullDuplexPipes, tc.want)
+			}
+		})
+	}
+}
+
+func TestFullDuplexPipesValidation(t *testing.T) {
+	cases := []struct {
+		name    string
+		cfg     AutoPipelineOptions
+		wantErr string
+	}{
+		{"negative", AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: -1}, "FullDuplexPipes=-1"},
+		{"disagrees with numshards", AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: 4, NumShards: 2}, "disagree"},
+		// Like the other full-duplex fields, it is ignored without FullDuplex.
+		{"ignored without full duplex", AutoPipelineOptions{FullDuplexPipes: -1}, ""},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.cfg.Validate()
+			if tc.wantErr == "" {
+				if err != nil {
+					t.Fatalf("Validate: %v, want nil", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tc.wantErr) {
+				t.Fatalf("Validate: got %v, want an error containing %q", err, tc.wantErr)
+			}
+		})
+	}
+}
+
+// The pool-size error names the field the caller set.
+func TestFullDuplexPipesNeedsPipelinePool(t *testing.T) {
+	c := NewClient(&Options{Addr: internalTestRedisAddr(), PipelinePoolSize: 2})
+	defer c.Close()
+	ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: 4})
+	if err == nil {
+		ap.Close()
+		t.Fatal("FullDuplexPipes=4 on a pipeline pool of 2 was accepted")
+	}
+	if !strings.Contains(err.Error(), "FullDuplexPipes=4") || !strings.Contains(err.Error(), "PipelinePoolSize gives 2") {
+		t.Fatalf("error should name FullDuplexPipes and the pool size, got: %v", err)
+	}
+}
+
+// When full duplex cannot engage (no pipeline pool), FullDuplexPipes has no
+// effect. Unlike NumShards>1, it must not turn into round-robin half-duplex
+// shards that then demand Unordered on the ordered async face.
+func TestFullDuplexPipesNoEffectWithoutPipelinePool(t *testing.T) {
+	c := NewClient(&Options{Addr: internalTestRedisAddr(), PipelinePoolSize: -1})
+	defer c.Close()
+	ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: 4})
+	if err != nil {
+		t.Fatalf("construct: %v (FullDuplexPipes must not require Unordered when full duplex is off)", err)
+	}
+	defer ap.Close()
+	if ap.Config().FullDuplex {
+		t.Fatal("Config().FullDuplex = true without a pipeline pool")
+	}
+	if got := len(ap.shards); got != 1 {
+		t.Fatalf("half-duplex shards = %d, want 1 (FullDuplexPipes must not shard the half-duplex queue)", got)
+	}
+}
+
+// Cluster full duplex runs one pipe per master. The per-node child config is a
+// copy of the parent's, so a parent FullDuplexPipes must be forced to 1 there;
+// otherwise every master would get that many engines.
+func TestFullDuplexPipesClusterChildForcedToOne(t *testing.T) {
+	cc := NewClusterClient(&ClusterOptions{Addrs: []string{internalTestRedisAddr()}})
+	defer cc.Close()
+	ap, err := cc.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: 4})
+	if err != nil {
+		t.Fatalf("construct: %v", err)
+	}
+	defer ap.Close()
+	if ap.clusterFD == nil {
+		t.Fatal("cluster full duplex did not engage")
+	}
+	if got := ap.clusterFD.childCfg.FullDuplexPipes; got != 1 {
+		t.Fatalf("child FullDuplexPipes = %d, want 1", got)
+	}
+	if got := ap.Config().FullDuplexPipes; got != 1 {
+		t.Fatalf("Config().FullDuplexPipes = %d, want 1 (one pipe per master)", got)
+	}
+}

example/autopipeline-fullduplex-async/README.md

@@ -7,7 +7,7 @@ at once; the result accessors (`Val`, `Result`, `Err`) block until the reply
 lands. Submit a window of commands, then read the results: the connection
 stays full, and that is where full duplex is fastest.
 
-One connection has a throughput ceiling. With `NumShards: N`, one client runs
+One connection has a throughput ceiling. With `FullDuplexPipes: N`, one client runs
 **N full-duplex engines**, each holding its own connection. The pool, the hook
 chain and the invalidation stream stay shared; only the engines multiply. The
 price is ordering: it holds per key, not for the whole submit stream.
@@ -40,10 +40,10 @@ go run .
    until the reply lands; `WaitContext` also returns when its ctx is done.
    Giving up does not cancel the command: it is already on the connection.
 
-**B. Several engines (`NumShards: 4`)**
+**B. Several engines (`FullDuplexPipes: 4`)**
 
 4. **Sizing.** Each engine holds one pipeline-pool connection. Construction
-   fails when `PipelinePoolSize` (default 10) is smaller than `NumShards`. The
+   fails when `PipelinePoolSize` (default 10) is smaller than `FullDuplexPipes`. The
    check covers one autopipeliner: the blocking and async faces share the
    pool, so size it for every full-duplex autopipeliner the client runs.
 5. **Routing by first key.** A command goes to the engine its first key
@@ -53,7 +53,7 @@ go run .
    command (`COPY`, `RENAME`, `MSET`, multi-key `DEL`, `EVAL` with several
    keys) against its other keys; keyless commands (`FLUSHDB`,
    `SCRIPT FLUSH`, ...), which round-robin. Await the first command when
-   another depends on it, or keep `NumShards` at 1.
+   another depends on it, or keep `FullDuplexPipes` at 1.
 7. **Pipelines.** A batch is contiguous on one connection, so it rides an
    engine only when all its keyed commands hash to that engine. A one-key
    batch always does. Engines hash the whole key, so a `{hash tag}` does not
@@ -75,8 +75,8 @@ A. one engine
 2. unawaited SET then GET -> "first"; INCRs -> 1..5 in order
 3. Submit INCRBY 10 -> 15; WaitContext GET -> "first"
 B. several engines
-4. NumShards=4, PipelinePoolSize=2 -> redis: AutoPipelineOptions.NumShards=4 needs a pipeline pool of at least 4 connections (each full-duplex engine holds one), but PipelinePoolSize gives 2; raise Options.PipelinePoolSize or lower NumShards
-   NumShards=4, PipelinePoolSize=4 -> 4 engines, full duplex true
+4. FullDuplexPipes=4, PipelinePoolSize=2 -> redis: AutoPipelineOptions.FullDuplexPipes=4 needs a pipeline pool of at least 4 connections (each full-duplex pipe holds one), but PipelinePoolSize gives 2; raise Options.PipelinePoolSize or lower FullDuplexPipes
+   FullDuplexPipes=4, PipelinePoolSize=4 -> 4 engines, full duplex true
 5. 100 unawaited SET+GET pairs, each pair on its key's engine: all read their write
    connections: 4 (one per engine)
 6. COPY awaited before GET dst -> "copied"
@@ -96,13 +96,13 @@ done
 - `FullDuplexWindow` (default 65536) bounds the commands in flight on each
   connection. It must exceed round-trip time x target rate, or it throttles
   throughput. It costs memory only while commands are really in flight. With
-  `NumShards > 1` every engine has its own window.
+  `FullDuplexPipes > 1` every engine has its own window.
 - `Unordered` and `MaxConcurrentBatches > 1` are rejected with `FullDuplex`:
   replies are matched by position on a connection, which needs one ordered
-  stream per connection. Use `NumShards` for more connections.
-- **Choosing `NumShards`.** More engines help only when one connection is
+  stream per connection. Use `FullDuplexPipes` for more connections.
+- **Choosing `FullDuplexPipes`.** More engines help only when one connection is
   saturated. With few commands in flight, each batch fragments across the
-  engines and throughput drops. In the measurements behind the `NumShards`
+  engines and throughput drops. In the measurements behind the `FullDuplexPipes`
   doc (10-command pipelines on a 14-core client), 8 engines were 30% slower
   than 1 at 8 callers, broke even at about 128, and were 2.2x faster at 1024.
   There is no automatic sizing yet. More engines also widen the latency

example/autopipeline-fullduplex-async/main.go

@@ -1,11 +1,11 @@
 // Command autopipeline-fullduplex-async shows the deferred (async)
 // autopipeliner in full-duplex mode, first on one connection, then on
-// several (NumShards > 1).
+// several (FullDuplexPipes > 1).
 //
 // Calls return at once; the result accessors (Val/Result/Err) block until the
 // reply lands. Submitting a window of commands before reading any result keeps
 // the connection full, which is where full duplex is fastest. One connection
-// has a throughput ceiling; NumShards runs several engines on one client, each
+// has a throughput ceiling; FullDuplexPipes runs several engines on one client, each
 // holding its own connection, at the cost of ordering only per key.
 //
 // Part A, one engine:
@@ -155,23 +155,23 @@ func severalEngines(ctx context.Context, admin *redis.Client) {
 	// covers one autopipeliner: the blocking and async faces share the pool,
 	// so size it for every full-duplex autopipeliner the client runs.
 	small := redis.NewClient(&redis.Options{Addr: addr(), PipelinePoolSize: 2})
-	_, err := small.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true, NumShards: engines})
+	_, err := small.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: engines})
 	small.Close()
 	if err == nil {
-		fatalf("NumShards %d on a pipeline pool of 2 was accepted", engines)
+		fatalf("FullDuplexPipes %d on a pipeline pool of 2 was accepted", engines)
 	}
-	fmt.Printf("4. NumShards=%d, PipelinePoolSize=2 -> %v\n", engines, err)
+	fmt.Printf("4. FullDuplexPipes=%d, PipelinePoolSize=2 -> %v\n", engines, err)
 
 	const name = "fd-engines-example"
 	rdb := redis.NewClient(&redis.Options{Addr: addr(), ClientName: name, PipelinePoolSize: engines})
 	defer rdb.Close()
-	ap, err := rdb.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true, NumShards: engines})
+	ap, err := rdb.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: engines})
 	if err != nil {
 		fatalf("AsyncAutoPipelineWithOptions: %v", err)
 	}
 	defer ap.Close()
-	fmt.Printf("   NumShards=%d, PipelinePoolSize=%d -> %d engines, full duplex %v\n",
-		engines, engines, ap.Config().NumShards, ap.Config().FullDuplex)
+	fmt.Printf("   FullDuplexPipes=%d, PipelinePoolSize=%d -> %d engines, full duplex %v\n",
+		engines, engines, ap.Config().FullDuplexPipes, ap.Config().FullDuplex)
 
 	// 5. Routing by first key. These SET/GET pairs are submitted without
 	// waiting in between; each pair shares a key, so both commands take the
@@ -195,7 +195,7 @@ func severalEngines(ctx context.Context, admin *redis.Client) {
 	//   - a multi-key command against its other keys: COPY src dst is routed by
 	//     src, so a later GET dst may run first;
 	//   - keyless commands (FLUSHDB, SCRIPT FLUSH, ...), which round-robin.
-	// When one depends on another, await the first. NumShards 1 keeps the
+	// When one depends on another, await the first. FullDuplexPipes 1 keeps the
 	// whole submit order.
 	ap.Set(ctx, "fde:src", "copied", 0)
 	if err := ap.Copy(ctx, "fde:src", "fde:dst", 0, true).Err(); err != nil { // await before the dependent GET
@@ -270,7 +270,7 @@ func bench(ctx context.Context, admin *redis.Client, n, goroutines, window int)
 	name := fmt.Sprintf("fd-bench-%d-%d", n, goroutines)
 	rdb := redis.NewClient(&redis.Options{Addr: addr(), ClientName: name, PipelinePoolSize: engines})
 	defer rdb.Close()
-	ap, err := rdb.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true, NumShards: n})
+	ap, err := rdb.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true, FullDuplexPipes: n})
 	if err != nil {
 		fatalf("bench: %v", err)
 	}

example/autopipeline-fullduplex/README.md

@@ -9,7 +9,7 @@ so there is no request-response wait between batches.
 
 This example uses the blocking face (`AutoPipeline`): each call blocks until
 its reply lands, like a plain client. For the async face, and for several
-connections (`NumShards > 1`), see
+connections (`FullDuplexPipes > 1`), see
 [`autopipeline-fullduplex-async`](../autopipeline-fullduplex-async).
 
 ## Run

example/autopipeline/README.md

@@ -83,4 +83,4 @@ request-response wait between batches. Two examples cover it:
   and pipelines on the full-duplex connection.
 - [`autopipeline-fullduplex-async`](../autopipeline-fullduplex-async): the
   async face, `Submit` and submission windows, on one connection and on
-  several (`NumShards > 1`).
+  several (`FullDuplexPipes > 1`).