From 16a7904db0975f2ddef7cedf77d58b3740fe615e Mon Sep 17 00:00:00 2001 From: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:25:55 +0300 Subject: [PATCH] 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 | 3 +- autopipeline.go | 108 ++++++++++------ autopipeline_cluster_fd.go | 1 + autopipeline_fd_pipeline.go | 4 +- autopipeline_fd_shards.go | 29 +++-- autopipeline_fullduplex_test.go | 122 ++++++++++++++++++ .../autopipeline-fullduplex-async/README.md | 20 +-- example/autopipeline-fullduplex-async/main.go | 20 +-- example/autopipeline-fullduplex/README.md | 2 +- example/autopipeline/README.md | 2 +- 10 files changed, 239 insertions(+), 72 deletions(-) diff --git a/.claude/specs/full-duplex.md b/.claude/specs/full-duplex.md index 42c75bdb..2b409ae6 100644 --- a/.claude/specs/full-duplex.md +++ b/.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 diff --git a/autopipeline.go b/autopipeline.go index be802c85..9ee3ae42 100644 --- a/autopipeline.go +++ b/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) diff --git a/autopipeline_cluster_fd.go b/autopipeline_cluster_fd.go index 4b007ce5..a7694886 100644 --- a/autopipeline_cluster_fd.go +++ b/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 diff --git a/autopipeline_fd_pipeline.go b/autopipeline_fd_pipeline.go index 71bd5745..fa0eca65 100644 --- a/autopipeline_fd_pipeline.go +++ b/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. diff --git a/autopipeline_fd_shards.go b/autopipeline_fd_shards.go index a0822c45..49f7fe6d 100644 --- a/autopipeline_fd_shards.go +++ b/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. diff --git a/autopipeline_fullduplex_test.go b/autopipeline_fullduplex_test.go index 68ef532b..64adb648 100644 --- a/autopipeline_fullduplex_test.go +++ b/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) + } +} diff --git a/example/autopipeline-fullduplex-async/README.md b/example/autopipeline-fullduplex-async/README.md index 2c86183a..55f46b5a 100644 --- a/example/autopipeline-fullduplex-async/README.md +++ b/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 diff --git a/example/autopipeline-fullduplex-async/main.go b/example/autopipeline-fullduplex-async/main.go index c02e5df0..0202e62d 100644 --- a/example/autopipeline-fullduplex-async/main.go +++ b/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) } diff --git a/example/autopipeline-fullduplex/README.md b/example/autopipeline-fullduplex/README.md index 2852ce12..3a9e8409 100644 --- a/example/autopipeline-fullduplex/README.md +++ b/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 diff --git a/example/autopipeline/README.md b/example/autopipeline/README.md index 8692965a..b6d0d751 100644 --- a/example/autopipeline/README.md +++ b/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`).