mirror of
https://github.com/redis/go-redis.git
synced 2026-10-24 07:59:20 +03:00
f813adcb85
* feat(autopipeline): spike N full-duplex engines per client
SPIKE, not for merge. Answers whether the ~2x measured for
"multi-connection full duplex" is a property of N engines or an artifact
of the benchmark harness, which fakes multi-conn by building N separate
*redis.Client instances and round-robining across them. That holds six
sockets per client to do I/O on one -- 48 for eight wires -- and keeps
the routing outside the library, so key routing could not be tested.
NumShards>1 under FullDuplex now means N engines on ONE client: one pool,
one hook chain, one invalidation stream, N held wires. Commands route by
key hash, so same-key commands keep one wire and per-key order holds
without an Unordered opt-in. Keyless commands round-robin. A whole
FDPipelined batch pins to one engine, since splitting it would lose the
internal order a pipeline exists to guarantee.
The change is TWO-SITED, which is the finding. Relaxing the Validate
rejection alone silently turns full duplex OFF: fdOn is separately gated
on nShards == 1, so NumShards>1 fell through to half-duplex round-robin
shards, which are unordered, and a same-key write-then-read returned
redis.Nil. That is exactly what the deleted error warned about ("reject
it rather than silently falling back to half-duplex"). Both the
validation and the construction gate have to move together, with
nShards forced back to 1 so the shard array stays flusherless.
Measured at 1024 callers, 10 commands per Exec, one binary:
proxy, 8 clients 1,686,080 ops/s 48 held sockets
in-client, 8 1,791,369 ops/s 12 held sockets
Known gap before this could ship: engines lease from the pipeline pool
and DefaultPipelinePoolSize is 10, so more than ~10 engines spill to the
main pool. A real feature must size that pool from the engine count.
cmdFirstKey is also deliberately crude (args[1]) rather than using
declared key positions.
* refactor(autopipeline): document FD sharding, guard the pool
Drops the exploratory framing from the multi-connection full-duplex code
and closes two of its three known gaps.
Key extraction no longer assumes args[1]. It calls
cmdFirstKeyPosWithInfo(cmd, nil), the same helper client-side caching
uses, so keyless commands, the static command table, and eval/evalsha
variants whose key position depends on the runtime numkeys argument all
route correctly instead of hashing a non-key argument.
Engine count is now checked against the pipeline pool. Each engine holds
one pooled connection for as long as it runs, and DefaultPipelinePoolSize
is 10, so 11 or more engines left the surplus permanently spilling onto
the main pool and competing with ordinary commands. Construction fails
with a message naming PipelinePoolSize instead.
The two-sited rule is now stated at both sites: relaxing the Validate
rejection without relaxing the fdOn gate leaves full duplex silently OFF
for NumShards>1, so commands fall through to unordered half-duplex shards
and a same-key write-then-read can read a stale value. Tests cover it
from both directions -- reply correlation and per-key ordering across
eight engines, and Config() reporting FullDuplex true rather than having
fallen back.
Still open, and why this stays draft: NumShards:0 means one engine and
sizing is not automatic. A fixed count above 1 costs low-concurrency
callers about 30% (eight engines at eight callers), so the count should
be derived from observed load, roughly one wire per 256-512 callers.
* fix(autopipeline): fix FD engine routing edges
- The NumShards>1 ordering check exempted any requested FullDuplex.
Without a pipeline pool full duplex does not engage and construction
falls back to round-robin shards, which need Unordered. The exemption
now follows the effective state.
- Keys are read with stringArg, as Ring does, so []byte, *string and
numeric keys route like the string form, and an empty key routes by
hash instead of round-robin.
- Engine index uses uint32 modulo; converting to int first goes
negative on 32-bit platforms.
- Document that N engines keep order per first key only, so multi-key
commands are not ordered across engines, and that NumShards>1 under
full duplex is new, so the mode is opt-in.
* docs(autopipeline): FD engine ordering and window
Document that with NumShards > 1 keyless commands, including ones that
change every key (FLUSHDB, SWAPDB), round-robin and are not ordered
against keyed commands; that keys route through stringArg like Ring and
the cluster slot hash; and that FullDuplexWindow applies per engine.
* docs(autopipeline): pool check is per autopipeliner
The NumShards pipeline-pool check counts only the autopipeliner being
built. The blocking and async faces and WithTimeout clones share one
pipeline pool, so it must hold the sum of their engines. Say so; a
client-wide engine count is a follow-up.
* fix(command): route raw FCALL by its first key
FCALL has the EVAL layout (name, function, numkeys, keys...), but the key
resolver only special-cased EVAL. The typed FCall sets its key position;
a raw Do/Process FCALL fell back to position 1 and hashed the function
name. That misrouted it on Ring (wrong shard), on cluster (a MOVED away)
and across full-duplex engines (a following GET of its key could run
first). A zero-key FCALL now resolves as keyless, like a zero-key EVAL.
* docs(autopipeline): document multi-engine NumShards
The exported FullDuplex and NumShards docs still described a single
full-duplex stream and said NumShards > 1 needs Unordered. Describe the
multi-engine mode: engines routed by first key, per-first-key order
only, per-engine window, and the pipeline pool sizing.
* fix(autopipeline): keep FD batches on their keys' engine
With several full-duplex engines, FDPipelined sent the whole batch to
its first command's engine. A later command in the batch whose key
hashes elsewhere could then run ahead of an earlier, unawaited command
for that key on that key's engine.
A batch now rides an engine only when every keyed command hashes to it
(keyless commands ride along). Otherwise FDPipelined returns
ErrFDPipelineSpansEngines and Pipeline() runs an ordinary pipeline.
* docs(autopipeline): note key types in engine routing
With several full-duplex engines a command is routed by a hash of its
first key in string form (fmt for non-string arguments, as Ring does),
so the same Redis key passed as different Go types (true and "1") can
land on different engines. Say so on NumShards.
* test(autopipeline): use the configured test address
The full-duplex shard tests dialed 127.0.0.1:6379 directly and skipped
under the REDIS_PORT override. Use apTestAddr and internalTestRedisAddr
like the other autopipeline tests.
* fix(command): route raw XGROUP and ZUNION by key
The raw-key resolver fell back to args[1] for XGROUP, whose key follows
the subcommand, and for ZUNION, ZINTER and ZDIFF, whose key follows
numkeys. With several full-duplex engines a raw command could then run
on a different engine from earlier writes to the same key, and Ring
picked the wrong shard. The typed constructors already set position 2.
* fix(command): route raw keyed commands by key
The raw-key resolver fell back to args[1] for SDIFFCARD, SUNIONCARD,
TS.NRANGE and TS.NREVRANGE (key after numkeys) and HIMPORT SET (key
after the subcommand), where the typed constructors set position 2.
HSCAN, SSCAN and ZSCAN with a hash tag in MATCH routed by the pattern,
a hint meant for the keyless SCAN. The key decides the node: a cluster
scan took a MOVED hop, a Ring scan read the wrong shard, and a
multi-engine full-duplex scan could pass earlier writes to its key.
* docs(autopipeline): note the cost of FD engines
NumShards is a fixed engine count, and several engines fragment small
batches. Give the measured trade on NumShards: 8 engines were 30%
slower than 1 at 8 callers, broke even at about 128, and were 2.2x
faster at 1024.
96 lines
3.5 KiB
Go
96 lines
3.5 KiB
Go
package redis
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
)
|
|
|
|
// Raw forms (Do / NewCmd) of the commands whose key is not args[1] carry no
|
|
// declared key position, so the shared resolver must find the key itself.
|
|
// Without that it fell back to args[1]: raw XREAD hashed "streams" and raw
|
|
// LMPOP hashed its numkeys, which routes to the wrong Ring shard, the wrong
|
|
// full-duplex engine, or a cold cluster slot.
|
|
func TestRawCommandsRouteByTheirKey(t *testing.T) {
|
|
ctx := context.Background()
|
|
const K = "KEY_SENTINEL"
|
|
cases := [][]interface{}{
|
|
{"xread", "streams", K, "0"},
|
|
{"XREAD", "COUNT", 1, "STREAMS", K, "0"},
|
|
{"xread", "count", 1, "block", 0, "streams", K, "b", "0", "0"},
|
|
{"xreadgroup", "group", "g", "c", "streams", K, ">"},
|
|
{"xreadgroup", "group", "streams", "streams", "count", 1, "noack", "streams", K, ">"},
|
|
{"object", "encoding", K},
|
|
{"xinfo", "stream", K},
|
|
{"xgroup", "create", K, "g", "$"},
|
|
{"XGROUP", "CREATE", K, "g", "$", "MKSTREAM"},
|
|
{"xgroup", "setid", K, "g", "0"},
|
|
{"xgroup", "destroy", K, "g"},
|
|
{"xgroup", "createconsumer", K, "g", "c"},
|
|
{"xgroup", "delconsumer", K, "g", "c"},
|
|
{"bitop", "and", K, "b"},
|
|
{"lmpop", 1, K, "left"},
|
|
{"zmpop", 1, K, "min"},
|
|
{"sintercard", 1, K},
|
|
{"zintercard", 2, K, "b"},
|
|
{"zunion", 1, K},
|
|
{"zinter", 2, K, "b", "withscores"},
|
|
{"zdiff", 2, K, "b"},
|
|
{"sdiffcard", 1, K, "limit", 0},
|
|
{"sunioncard", 2, K, "b"},
|
|
{"ts.nrange", 1, K, "-", "+"},
|
|
{"TS.NREVRANGE", 2, K, "b", "-", "+"},
|
|
{"himport", "set", K, "fs", "v"},
|
|
{"HIMPORT", "SET", K, "fs", "v"},
|
|
{"blmpop", 0, 1, K, "left"},
|
|
{"bzmpop", 0, 1, K, "min"},
|
|
{"migrate", "h", 1, K, 0, 1000},
|
|
{"migrate", "h", 1, "", 0, 1000, "copy", "keys", K, "b"},
|
|
}
|
|
for _, args := range cases {
|
|
cmd := NewCmd(ctx, args...)
|
|
want := -1
|
|
for i, a := range args {
|
|
if s, ok := a.(string); ok && s == K {
|
|
want = i
|
|
break
|
|
}
|
|
}
|
|
if pos := cmdFirstKeyPosWithInfo(cmd, nil); pos != want {
|
|
t.Errorf("%v: resolver picks position %d, key is at %d", args, pos, want)
|
|
}
|
|
}
|
|
|
|
// Subcommand help has no key.
|
|
for _, args := range [][]interface{}{{"object", "help"}, {"xinfo", "help"}, {"xgroup", "help"}} {
|
|
if pos := cmdFirstKeyPosWithInfo(NewCmd(ctx, args...), nil); pos != 0 {
|
|
t.Errorf("%v: resolver picks position %d, want 0 (keyless)", args, pos)
|
|
}
|
|
}
|
|
}
|
|
|
|
// A keyed scan is routed by its key. A hash tag in MATCH must not move it:
|
|
// the key decides the node, so routing by the pattern sent a cluster scan to
|
|
// the wrong node (MOVED), a Ring scan to the wrong shard (no results), and a
|
|
// multi-engine full-duplex scan past earlier writes to the key. Only the
|
|
// keyless SCAN routes by the pattern's tag.
|
|
func TestKeyedScansRouteByTheirKey(t *testing.T) {
|
|
ctx := context.Background()
|
|
var got Cmder
|
|
c := cmdable(func(_ context.Context, cmd Cmder) error { got = cmd; return nil })
|
|
for name, run := range map[string]func(){
|
|
"hscan": func() { c.HScan(ctx, "k", 0, "{other}*", 0) },
|
|
"hscan novalues": func() { c.HScanNoValues(ctx, "k", 0, "{other}*", 0) },
|
|
"sscan": func() { c.SScan(ctx, "k", 0, "{other}*", 0) },
|
|
"zscan": func() { c.ZScan(ctx, "k", 0, "{other}*", 0) },
|
|
} {
|
|
run()
|
|
if pos := cmdFirstKeyPosWithInfo(got, nil); pos != 1 {
|
|
t.Errorf("%s with a tagged MATCH: routed by position %d (%v), want 1 (the key)", name, pos, got.stringArg(pos))
|
|
}
|
|
}
|
|
c.Scan(ctx, 0, "{tag}*", 0)
|
|
if pos := cmdFirstKeyPosWithInfo(got, nil); pos != 3 {
|
|
t.Errorf("keyless scan with a tagged MATCH: routed by position %d, want 3 (the pattern)", pos)
|
|
}
|
|
}
|