diff --git a/autopipeline.go b/autopipeline.go index c59ed825..0dd55d9b 100644 --- a/autopipeline.go +++ b/autopipeline.go @@ -16,6 +16,7 @@ import ( "golang.org/x/sys/cpu" "github.com/redis/go-redis/v9/internal" + "github.com/redis/go-redis/v9/internal/otel" "github.com/redis/go-redis/v9/internal/pool" ) @@ -495,6 +496,22 @@ type cmdableClient interface { // one goroutine wake-up apiece). type apBatch struct { done chan struct{} + // fdAttempts is how many times the full-duplex engine issued a PIPELINED + // command (1, plus one per connection-error replay), stamped by the reader + // before it completes the command, so closing done publishes it. The + // pipeline retry reads it to charge those executions against MaxRetries. + // Zero on every other path. + fdAttempts int + // fdConn is the held connection that carried a PIPELINED command's reply, + // stamped with fdAttempts, for the pipeline duration metric. + fdConn *pool.Conn + // fdGroup marks the commands of one FD pipeline batch: every batch of the + // pipeline points at the batch of its first command. The Close-time flush + // uses it to run the pipeline as one, with one whole-batch retry. + fdGroup *apBatch + // fdFlushed is set when the Close-time flush ran this pipelined command + // through the pooled pipeline, which recorded the pipeline metric itself. + fdFlushed bool // closed makes close() idempotent: on the async faces the dispatch closes // the batch at the innermost exec seam (under the user hooks, so a hook // reading a result after next() does not block on a channel its own @@ -3440,15 +3457,151 @@ func (ap *AutoPipeliner) calculateDelay(queueLen int) time.Duration { // Pipeline returns a new pipeline that uses the underlying pipeliner. // This allows you to create a traditional pipeline from an autopipeliner. +// +// On a FULL-DUPLEX autopipeliner the pipeline runs on the HELD connection +// rather than taking one of its own: see fdPipelineExec. Without that, a +// pipeline built from an autopipeliner would quietly leave the pipe, take a +// pooled connection per Exec, and lose the coalescing that is the whole point +// of the engine — measured at 150 active sockets and less than a third of the +// throughput of the same commands submitted through the engine. func (ap *AutoPipeliner) Pipeline() Pipeliner { + if ap.fd != nil { + pipe := Pipeline{exec: pipelineExecer(ap.fdPipelineHooked)} + pipe.init() + return &pipe + } return ap.pipeliner.Pipeline() } +// fdPipelineExec runs a whole pipeline through the full-duplex engine as one +// contiguous batch, and falls back to a pooled pipeline when the batch cannot +// ride the held connection. +// +// Contiguity is what makes this safe: FDPipelined admits the batch all-or- +// nothing, so the commands occupy consecutive in-flight slots and their replies +// come back in submit order. A pipeline's internal ordering therefore holds +// exactly as it does on a dedicated connection. +// +// The fallback covers the cases the engine must divert — a blocking command, a +// command with its own read timeout, anything the divert predicate claims, or a +// batch larger than the whole submit queue. Those go to the ordinary pipeline +// path, which is where they would have gone before this existed. +// +// fdPipelineExec is the TERMINAL of the client's ProcessPipelineHook chain +// (see fdPipelineHooked), the place processPipeline takes in an ordinary +// pipeline. So hooks wrap the whole execution, fallback and retry included, +// exactly once, and the pooled paths below call the hookless processPipeline. +// +// On the FD path it records what an ordinary pipeline records: one pipeline +// duration and, on failure, one pipeline-level error, instead of the reader's +// per-command metrics. +// +// A Limiter is asked per WRITE CHUNK on the FD path, not once per batch: a +// chunk can mix this batch with other callers' commands, so a batch longer +// than MaxBatchSize (or MaxBatchBytes) can be partly admitted. An ordinary +// pipeline makes one decision per attempt. +// +// Also: +// +// - A retryable reply (LOADING and the like) on the FIRST command: an +// ordinary pipeline retries the whole batch, in order. FDPipelined settles +// retryable replies inline, so the batch is re-run the ordinary way, unless +// it holds a NoRetry command, which an ordinary pipeline does not retry +// either. A retryable reply on a later command is returned, as an ordinary +// pipeline does. +// +// fdPipelineHooked is Pipeline().Exec on a full-duplex autopipeliner: the +// client's pipeline hooks around fdPipelineExec. +// +// ContextTimeoutEnabled keeps the pooled path: an ordinary pipeline bounds its +// socket I/O by the caller's ctx, and the FD wait cannot abandon an admitted +// batch on a shared connection. +func (ap *AutoPipeliner) fdPipelineHooked(ctx context.Context, cmds []Cmder) error { + c := ap.fd.client + if c.opt.ContextTimeoutEnabled { + return ap.pipeliner.processPipelineHook(ctx, cmds) + } + return c.wrapPipelineHooks(ap.fdPipelineExec)(ctx, cmds) +} + +func (ap *AutoPipeliner) fdPipelineExec(ctx context.Context, cmds []Cmder) error { + opt := ap.fd.client.opt + var start time.Time + if otel.GetPipelineOperationDurationCallback() != nil { + start = time.Now() + } + res, err := ap.fdPipelined(ctx, cmds) + used, cn := res.attempts, res.cn + // An ordinary pipeline allows MaxRetries+1 executions. The FD engine may + // already have issued the batch more than once (a connection-error replay), + // so the re-run gets what is left. + remaining := opt.MaxRetries - used + ineligible, retry := false, false + switch { + case errors.Is(err, ErrFDPipelineDiverts), + errors.Is(err, ErrFDPipelineUnavailable), + errors.Is(err, ErrFDPipelineTooLarge): + // Not a command failure: the batch is simply not eligible. + ineligible = true + case len(cmds) > 0 && remaining >= 0 && !cmdsContainNoRetry(cmds): + // Only a batch the engine issued exactly once. After a connection-error + // replay, the replay already was this batch's retry: it re-ran the unread + // tail, which an ordinary pipeline re-runs as its one whole-batch retry + // after the same network error. A whole-batch re-run on top would run + // that tail a third time (a mutation beyond what an ordinary pipeline + // does), so the first command's stale retryable reply is returned instead. + // The Close-time flush stamps fdAttempts the same way, so a flushed batch + // is not re-run either. + // + // A transport or protocol failure anywhere in the batch rules the + // re-run out too. The engine did not replay that command, since it may + // already have run, and an ordinary pipeline returns that read error + // instead of retrying. + if e := cmds[0].rawErr(); used == 1 && isRedisError(e) && shouldRetry(e, false) && + fdPipelineTransportErr(cmds) == nil { + retry = true + } + } + if !ineligible && !retry { + // Run on the FD path, or refused before admission. A batch the + // Close-time flush ran in full was measured by that pooled pipeline. + if res.measured() { + ap.fdPipelineMetrics(ctx, start, cmds, max(used, 1), cn) + } + return err + } + // Clear the per-command errors FDPipelined may have stamped, then run the + // batch the ordinary way so the caller sees one authoritative outcome. + for _, cmd := range cmds { + cmd.SetErr(nil) + } + // Hookless processPipeline: this is already the terminal of the hook + // chain. The pooled run records its own pipeline metric. + if ineligible { + return ap.fd.client.processPipeline(ctx, cmds) + } + // The re-run runs remaining+1 times at most, so FD attempts plus re-runs + // never exceed MaxRetries+1, after the backoff an ordinary pipeline sleeps + // before its next retry. Inside the hook chain, like generalProcessPipeline's + // own retries. It continues this operation's measurement (start and the FD + // attempts), so one pipeline metric covers the whole operation. + c := ap.fd.client + if serr := internal.Sleep(ctx, c.retryBackoff(used)); serr != nil { + setCmdsErr(cmds, serr) + ap.fdPipelineMetrics(ctx, start, cmds, used, cn) + return serr + } + return c.processPipelineRetriesAfter(ctx, cmds, remaining, start, used) +} + // Pipelined executes a function in a pipeline context. // This is a convenience method that creates a pipeline, executes the function, // and returns the results. +// +// Uses Pipeline, so on a full-duplex autopipeliner this also runs on the held +// connection rather than taking one of its own. func (ap *AutoPipeliner) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) { - return ap.pipeliner.Pipeline().Pipelined(ctx, fn) + return ap.Pipeline().Pipelined(ctx, fn) } // TxPipelined executes a function in a transaction pipeline context. diff --git a/autopipeline_fd_batch_backpressure_test.go b/autopipeline_fd_batch_backpressure_test.go new file mode 100644 index 00000000..a67588d6 --- /dev/null +++ b/autopipeline_fd_batch_backpressure_test.go @@ -0,0 +1,405 @@ +package redis + +import ( + "context" + "runtime" + "strings" + "sync" + "testing" + "time" +) + +// fdBackpressureEngine is an engine with no writer and a submit queue of +// capacity 10 holding 9 commands, so a 5-command batch cannot be admitted. +func fdBackpressureEngine(t *testing.T) *fdEngine { + t.Helper() + apCtx, apCancel := context.WithCancel(context.Background()) + t.Cleanup(apCancel) + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}, ctx: apCtx}, + client: &Client{baseClient: &baseClient{opt: &Options{}}}, + q: newFDQueue(10), + } + for i := 0; i < 9; i++ { + fd.q.push(fdReq{cmd: NewStatusCmd(context.Background(), "set", "k", "v"), batch: newAPBatch()}) + } + return fd +} + +// startBlockedBatch submits a 5-command batch that must wait for room and +// returns a func that cancels it and waits for it to return. +func startBlockedBatch(t *testing.T, fd *fdEngine) (stop func()) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + cmds := make([]Cmder, 5) + for i := range cmds { + cmds[i] = NewStatusCmd(ctx, "set", "k", "v") + } + done := make(chan struct{}) + go func() { + defer close(done) + if b, err := fd.submitBatch(ctx, cmds); b != nil || err != nil { + t.Errorf("submitBatch = (%v, %v); want a ctx rejection", b, err) + } + }() + time.Sleep(50 * time.Millisecond) // let it park on the room signal + return func() { + cancel() + <-done + } +} + +// submitBatchParked reports whether the goroutine running submitBatch is +// parked in its select. A spinning loop never parks: it finds the room signal +// it just put back and runs on. +func submitBatchParked(t *testing.T) bool { + t.Helper() + buf := make([]byte, 1<<20) + buf = buf[:runtime.Stack(buf, true)] + for _, g := range strings.Split(string(buf), "\n\n") { + if strings.Contains(g, "(*fdEngine).submitBatch(") { + header, _, _ := strings.Cut(g, "\n") + return strings.Contains(header, "[select") + } + } + t.Fatal("no goroutine is running submitBatch") + return false +} + +// TestFDSubmitBatchDoesNotSpinWhileBlocked pins that a batch waiting for more +// room than one wake frees does not spin. It used to put the room signal back +// whenever any slot was free, take it straight back, fail to fit, and loop, +// burning a core for as long as the writer was blocked. +func TestFDSubmitBatchDoesNotSpinWhileBlocked(t *testing.T) { + fd := fdBackpressureEngine(t) + stop := startBlockedBatch(t, fd) + defer stop() + + wakeHead(t, fd.q) // one slot free, the batch needs five + time.Sleep(20 * time.Millisecond) + running := 0 + const samples = 20 + for i := 0; i < samples; i++ { + if !submitBatchParked(t) { + running++ + } + time.Sleep(5 * time.Millisecond) + } + if running > 1 { + t.Fatalf("the waiting batch was running in %d of %d samples: it is spinning", running, samples) + } +} + +// wakeHead signals the head of the holder line the way a take would. +func wakeHead(t *testing.T, q *fdQueue) { + t.Helper() + q.mu.Lock() + head := q.headWakeLocked() + q.mu.Unlock() + if head == nil { + t.Fatal("no holder is waiting") + } + signalHolder(head) +} + +// holderCount reports how many batches are waiting with a reservation. +func holderCount(q *fdQueue) int { + q.mu.Lock() + defer q.mu.Unlock() + return len(q.holders) +} + +// waitHolders blocks until n batches hold a reservation. +func waitHolders(t *testing.T, q *fdQueue, n int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for holderCount(q) != n { + if time.Now().After(deadline) { + t.Fatalf("holders = %d, want %d", holderCount(q), n) + } + time.Sleep(time.Millisecond) + } +} + +// TestFDSubmitBatchWakesOnlyTheHead pins the direct wake: a take signals the +// head of the holder line on its own channel, and a holder behind it is not +// woken at all, so no wake is shared or passed between holders. +func TestFDSubmitBatchWakesOnlyTheHead(t *testing.T) { + fd := fdBackpressureEngine(t) + headID, headWake := fd.q.hold(1) // an earlier holder, standing in for a head + stop := startBlockedBatch(t, fd) // the batch lines up behind it + defer stop() + waitHolders(t, fd.q, 2) + + fd.q.takeInto(nil, 1) + select { + case <-headWake: + case <-time.After(time.Second): + t.Fatal("a take did not wake the head of the line") + } + time.Sleep(20 * time.Millisecond) + if !submitBatchParked(t) { + t.Fatal("the batch behind the head was woken by a take meant for the head") + } + + // The head leaves: the batch becomes the head and is woken to try. + fd.q.unhold(headID) + time.Sleep(20 * time.Millisecond) + if n := holderCount(fd.q); n != 1 { + t.Fatalf("holders after unhold = %d, want 1", n) + } +} + +// TestFDSubmitBatchHoldersAreFIFO pins that holders are admitted in arrival +// order: small batches that arrive behind a large one cannot take the room +// the writer frees for the large one, even though each alone would fit, and +// each is admitted in turn once it reaches the head of the line. +func TestFDSubmitBatchHoldersAreFIFO(t *testing.T) { + apCtx, apCancel := context.WithCancel(context.Background()) + defer apCancel() + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}, ctx: apCtx}, + client: &Client{baseClient: &baseClient{opt: &Options{}}}, + q: newFDQueue(10), + } + for i := 0; i < 10; i++ { + fd.q.push(fdReq{cmd: NewStatusCmd(context.Background(), "set", "k", "v"), batch: newAPBatch()}) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + mk := func(n int) []Cmder { + cmds := make([]Cmder, n) + for i := range cmds { + cmds[i] = NewStatusCmd(ctx, "set", "k", "v") + } + return cmds + } + order := make(chan string, 3) + submit := func(name string, cmds []Cmder) { + b, err := fd.submitBatch(ctx, cmds) + if err != nil || len(b) != len(cmds) { + t.Errorf("%s: submitBatch = (%d, %v)", name, len(b), err) + } + order <- name + } + go submit("big", mk(10)) + waitHolders(t, fd.q, 1) + go submit("small1", mk(1)) + waitHolders(t, fd.q, 2) + go submit("small2", mk(1)) + waitHolders(t, fd.q, 3) + + // Drain one at a time, as a writer with a one-command wave would. After + // each take a small batch alone would fit; neither may be admitted. + for i := 0; i < 10; i++ { + fd.q.takeInto(nil, 1) + time.Sleep(5 * time.Millisecond) + if i < 9 { + select { + case name := <-order: + t.Fatalf("%q admitted after %d takes, ahead of the head of the line", name, i+1) + default: + } + } + } + next := func(want string) { + t.Helper() + select { + case got := <-order: + if got != want { + t.Fatalf("admitted %q, want %q", got, want) + } + case <-time.After(time.Second): + t.Fatalf("%q was not admitted", want) + } + } + next("big") // 10 live again + fd.q.takeInto(nil, 1) + next("small1") + fd.q.takeInto(nil, 1) + next("small2") +} + +// TestFDSubmitBatchCloseReleasesAllHolders pins that closing the queue wakes +// every holder, not just the head: a holder behind the head has no other +// room signal, and Close must not leave it waiting for its ctx. +func TestFDSubmitBatchCloseReleasesAllHolders(t *testing.T) { + fd := fdBackpressureEngine(t) + ctx := context.Background() + done := make(chan error, 2) + for _, n := range []int{5, 5} { + cmds := make([]Cmder, n) + for i := range cmds { + cmds[i] = NewStatusCmd(ctx, "set", "k", "v") + } + go func() { + b, err := fd.submitBatch(ctx, cmds) + if b != nil || err != nil { + t.Errorf("submitBatch = (%v, %v); want a submit-time rejection", b, err) + } + done <- cmds[0].Err() + }() + } + waitHolders(t, fd.q, 2) + + fd.q.closeQueue() + for i := 0; i < 2; i++ { + select { + case err := <-done: + if err != ErrClosed { + t.Fatalf("holder error = %v, want ErrClosed", err) + } + case <-time.After(time.Second): + t.Fatal("a holder was not released by closeQueue") + } + } +} + +// TestFDSubmitBatchReservesRoom pins that a refused batch holds its slots: the +// room the writer frees accumulates for it instead of going to single +// submitters, and the singles resume once the batch is in. +func TestFDSubmitBatchReservesRoom(t *testing.T) { + fd := fdBackpressureEngine(t) // capacity 10, 9 queued + ctx := context.Background() + cmds := make([]Cmder, 5) + for i := range cmds { + cmds[i] = NewStatusCmd(ctx, "set", "k", "v") + } + type result struct { + batches []*apBatch + err error + } + done := make(chan result, 1) + go func() { + b, err := fd.submitBatch(ctx, cmds) + done <- result{b, err} + }() + time.Sleep(50 * time.Millisecond) + + // One slot is free, but the batch has reserved five: a single is refused. + single := fdReq{cmd: NewStatusCmd(ctx, "set", "k", "v"), batch: newAPBatch()} + if res := fd.q.push(single); res != fdPushFull { + t.Fatalf("push with a batch waiting = %v; want fdPushFull (slots reserved)", res) + } + + // The writer takes four. The batch now fits (5 live + 5), and the take + // wakes it through batchRoom. + if got := len(fd.q.takeInto(nil, 4)); got != 4 { + t.Fatalf("takeInto = %d, want 4", got) + } + select { + case r := <-done: + if r.err != nil || len(r.batches) != 5 { + t.Fatalf("submitBatch = (%d batches, %v); want 5 batches, nil", len(r.batches), r.err) + } + case <-time.After(time.Second): + t.Fatal("the batch was not admitted after the writer freed its room") + } + if d := fd.q.depth(); d != 10 { + t.Fatalf("depth after admission = %d, want 10", d) + } + if fd.q.roomFor(1) { + t.Fatal("roomFor(1) with a full queue and no reservation = true") + } + + // The reservation is consumed: once the writer frees a slot, a single fits. + fd.q.takeInto(nil, 1) + if res := fd.q.push(single); res != fdPushOK { + t.Fatalf("push after admission = %v; want fdPushOK (reservation released)", res) + } +} + +// TestFDSubmitBatchReleasesReservationOnCancel pins that a batch that gives up +// gives its slots back and wakes a single its reservation was holding back. +func TestFDSubmitBatchReleasesReservationOnCancel(t *testing.T) { + fd := fdBackpressureEngine(t) + stop := startBlockedBatch(t, fd) + + single := fdReq{cmd: NewStatusCmd(context.Background(), "set", "k", "v"), batch: newAPBatch()} + if res := fd.q.push(single); res != fdPushFull { + t.Fatalf("push with a batch waiting = %v; want fdPushFull", res) + } + stop() // ctx cancelled: the batch is rejected and must unhold + + select { + case <-fd.q.roomCh(): + case <-time.After(time.Second): + t.Fatal("unhold did not wake the single the reservation held back") + } + if res := fd.q.push(single); res != fdPushOK { + t.Fatalf("push after the batch gave up = %v; want fdPushOK", res) + } +} + +// TestFDSubmitBatchNotStarvedBySingles is the end-to-end shape of the fix: a +// writer that frees two slots per take, singles that refill every free slot +// as fast as they can, and a batch longer than a take. Without a reservation +// the batch never sees its whole length free at once and waits until its ctx +// expires. +func TestFDSubmitBatchNotStarvedBySingles(t *testing.T) { + apCtx, apCancel := context.WithCancel(context.Background()) + defer apCancel() + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}, ctx: apCtx}, + client: &Client{baseClient: &baseClient{opt: &Options{}}}, + q: newFDQueue(10), + } + stop := make(chan struct{}) + var wg sync.WaitGroup + + // The writer: two slots per take, continuously. + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + fd.q.takeInto(nil, 2) + time.Sleep(200 * time.Microsecond) + } + }() + // The singles: refill as submit() does, waiting on room when refused. + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + req := fdReq{cmd: NewStatusCmd(context.Background(), "set", "k", "v"), batch: newAPBatch()} + for { + select { + case <-stop: + return + default: + } + if fd.q.push(req) == fdPushOK { + continue + } + select { + case <-fd.q.roomCh(): + if fd.q.roomFor(1) { + fd.q.signalRoom() + } + case <-stop: + return + } + } + }() + } + time.Sleep(20 * time.Millisecond) // let the singles saturate the queue + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + cmds := make([]Cmder, 8) // four times a take + for i := range cmds { + cmds[i] = NewStatusCmd(ctx, "set", "k", "v") + } + batches, err := fd.submitBatch(ctx, cmds) + close(stop) + wg.Wait() + if err != nil || len(batches) != 8 { + t.Fatalf("submitBatch under single-command load = (%d batches, %v, cmd err %v); want admission", + len(batches), err, cmds[0].Err()) + } +} diff --git a/autopipeline_fd_failqueue_metrics_test.go b/autopipeline_fd_failqueue_metrics_test.go new file mode 100644 index 00000000..4ef8ab60 --- /dev/null +++ b/autopipeline_fd_failqueue_metrics_test.go @@ -0,0 +1,55 @@ +package redis + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/redis/go-redis/v9/internal/pool" +) + +// TestFDFailQueueSkipsPipelinedErrorMetric pins that failQueue, like +// failReqs, reports no per-command error for a pipelined command. +// fdPipelineExec reports the batch once, as an ordinary pipeline does, so a +// per-command report here would count the same failure twice. A plain +// command in the same backlog is still reported. +func TestFDFailQueueSkipsPipelinedErrorMetric(t *testing.T) { + var calls atomic.Int64 + pool.SetAllMetricCallbacks(&pool.MetricCallbacks{ + Error: func(context.Context, string, *pool.Conn, string, bool, int) { + calls.Add(1) + }, + }) + defer pool.SetAllMetricCallbacks(nil) + + ctx := context.Background() + fd := &fdEngine{client: &Client{baseClient: &baseClient{opt: &Options{}}}, q: newFDQueue(4)} + group := newAPBatch() + group.fdGroup = group + var batches []*apBatch + for i := 0; i < 3; i++ { + b := group + if i > 0 { + b = newAPBatch() + b.fdGroup = group + } + batches = append(batches, b) + fd.q.push(fdReq{cmd: NewStatusCmd(ctx, "set", "k", "v"), batch: b, ctx: ctx, attempts: 1, pipelined: true}) + } + plain := newAPBatch() + batches = append(batches, plain) + fd.q.push(fdReq{cmd: NewStatusCmd(ctx, "set", "k", "v"), batch: plain, ctx: ctx, attempts: 1}) + + fd.failQueue(ErrClosed) + + if got := calls.Load(); got != 1 { + t.Fatalf("error callback invoked %d times, want 1 (the plain command only)", got) + } + for i, b := range batches { + select { + case <-b.done: + default: + t.Fatalf("req %d not completed", i) + } + } +} diff --git a/autopipeline_fd_flush_budget_test.go b/autopipeline_fd_flush_budget_test.go new file mode 100644 index 00000000..24fd250e --- /dev/null +++ b/autopipeline_fd_flush_budget_test.go @@ -0,0 +1,83 @@ +package redis + +import ( + "context" + "errors" + "testing" +) + +// fdBudgetProbe is a stub engine whose Close flush records the retry bound of +// each pipeline it runs. +func fdBudgetProbe() (*fdEngine, *[]int) { + var bounds []int + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}}, + client: &Client{baseClient: &baseClient{opt: &Options{MaxRetries: 3}}}, + maxBatch: 8, + runPipeline: func(_ context.Context, _ []Cmder, maxRetries int) error { + bounds = append(bounds, maxRetries) + return nil + }, + } + return fd, &bounds +} + +// One FD pipeline flushed on Close can carry commands with different attempt +// counts. It runs as one pipeline, so it gets the smallest remaining budget +// of its commands: the budget of its first command let a later command with +// more attempts run past its MaxRetries+1 executions. +func TestFDCloseFlushUsesSmallestBudgetOfAPipeline(t *testing.T) { + ctx := context.Background() + fd, bounds := fdBudgetProbe() + carry := fdPipeGroupReqs(ctx, 1, 3) // budgets 3 and 1 with MaxRetries 3 + if err := fd.flushCarryBudgeted(ctx, carry); err != nil { + t.Fatalf("flushCarryBudgeted: %v", err) + } + if got := *bounds; len(got) != 1 || got[0] != 1 { + t.Fatalf("pipeline retry bounds %v, want [1] (the smallest budget)", got) + } +} + +// When any command of the pipeline has spent its budget, the pipeline does not +// run: running it would execute that command again. +func TestFDCloseFlushFailsPipelineWithSpentBudget(t *testing.T) { + ctx := context.Background() + fd, bounds := fdBudgetProbe() + carry := fdPipeGroupReqs(ctx, 1, 5) // the second has run past MaxRetries+1 + if err := fd.flushCarryBudgeted(ctx, carry); err != nil { + t.Fatalf("flushCarryBudgeted: %v", err) + } + if got := *bounds; len(got) != 0 { + t.Fatalf("the pipeline ran with bounds %v; a command in it had spent its budget", got) + } + for i, r := range carry { + if !errors.Is(r.cmd.Err(), errFDRetryBudgetExhausted) { + t.Fatalf("command %d err=%v, want errFDRetryBudgetExhausted", i, r.cmd.Err()) + } + } +} + +// A command in front of an FD pipeline is not part of that pipeline's budget +// group. With the same attempt count as the pipeline's first command it used +// to join the group, so a later pipeline command that had spent its budget +// failed the unrelated command too. +func TestFDCloseFlushKeepsOtherCommandsOutOfAPipelineBudget(t *testing.T) { + ctx := context.Background() + fd, bounds := fdBudgetProbe() + other := fdReq{cmd: NewStatusCmd(ctx, "set", "o", "v"), batch: newAPBatch(), attempts: 2} + carry := append([]fdReq{other}, fdPipeGroupReqs(ctx, 2, 5)...) // the pipeline's 2nd command has no budget left + if err := fd.flushCarryBudgeted(ctx, carry); err != nil { + t.Fatalf("flushCarryBudgeted: %v", err) + } + if got := *bounds; len(got) != 1 || got[0] != 2 { + t.Fatalf("retry bounds %v, want [2]: the other command runs alone with its own budget", got) + } + if err := carry[0].cmd.Err(); err != nil { + t.Fatalf("other command err=%v, want nil", err) + } + for i, r := range carry[1:] { + if !errors.Is(r.cmd.Err(), errFDRetryBudgetExhausted) { + t.Fatalf("pipeline command %d err=%v, want errFDRetryBudgetExhausted", i, r.cmd.Err()) + } + } +} diff --git a/autopipeline_fd_pipeline.go b/autopipeline_fd_pipeline.go new file mode 100644 index 00000000..f04ce3af --- /dev/null +++ b/autopipeline_fd_pipeline.go @@ -0,0 +1,405 @@ +package redis + +import ( + "context" + "errors" + "time" + + "github.com/redis/go-redis/v9/internal/otel" + "github.com/redis/go-redis/v9/internal/pool" +) + +// Batch submission on the full-duplex wire. +// +// WHY THIS EXISTS. AutoPipeliner.Pipeline() delegates to the underlying +// client, so a pipeline runs through processPipeline on a POOLED connection and +// never touches the full-duplex engine — the engine's only entry point is +// submit(), which takes one Cmder. That leaves go-redis without the shape +// rueidis offers as DoMulti: hand over N commands at once, block until every +// reply lands, on ONE multiplexed connection, spending no extra goroutines. +// +// The submit queue already has the primitive for it. fdQueue.pushBatch appends +// a whole slice under one lock, and because the queue is FIFO and the writer can +// only take contiguous prefixes, the commands stay ADJACENT on the wire — no +// other caller's command can land between them. That is a stronger guarantee +// than issuing N single commands gives, and it is what makes this a pipeline +// rather than a loop. +// +// Each command still carries its OWN completion batch. One shared batch would +// complete on the first reply, since the reader closes the batch per command, so +// the caller would return before the rest had landed. + +// ErrFDPipelineUnavailable is returned when batch submission is requested on an +// autopipeliner that has no full-duplex engine. It is deliberately an error +// rather than a silent fallback to the pooled pipeline path: a caller asking for +// this wants the commands on the FD wire, and quietly running them somewhere +// else would be measured (or relied upon) as something it is not. +// +// Batch submission is STANDALONE ONLY. A full-duplex autopipeliner on a +// ClusterClient reports Config().FullDuplex true but runs one engine per node, +// and a pipeline can span nodes, so it gets this error and its Pipeline() keeps +// using the cluster pipeline, as it did before FD batches existed. +var ErrFDPipelineUnavailable = errors.New( + "redis: FDPipelined requires a full-duplex autopipeliner on a standalone Client") + +// ErrFDPipelineDiverts is returned when a batch contains a command the engine +// cannot stream (blocking, per-command read timeout, runs-outside-pipeline, +// HIMPORT, or one the caller's own mustDivert rejects). Such a command has to +// leave the pipe, which would break the batch's contiguity, so the whole call is +// refused instead of being silently split. A NoRetry command (e.g. RawWriteTo) +// is refused too: it makes the batch unreplayable as a unit, which the pipe's +// per-command recovery cannot honor. +var ErrFDPipelineDiverts = errors.New( + "redis: FDPipelined got a command that must be diverted off the full-duplex pipe") + +// FDPipelined submits cmds as one contiguous batch on the full-duplex +// connection and blocks until every reply has landed. It returns the first +// command error, Nil included, the same as Pipeline.Exec. +// +// Two differences from an ordinary pipeline. Retryable replies (LOADING and +// the like) are not retried here; Pipeline() retries the whole batch on top of +// this, see fdPipelineExec. And once the batch is admitted, ctx no longer bounds +// the wait: the commands share the held connection with other callers, so they +// cannot be abandoned mid-stream. ReadTimeout still bounds each reply. +// +// Compared with the two shapes that already exist: +// +// - Pipeline().Exec() batches up front and blocks, but on a pooled +// connection: a separate socket, its own round trip, no FD engine. +// - Submit()/the async face keeps N commands in flight on the FD wire with +// individual completion, but the caller issues them one at a time and they +// may interleave with other callers' commands. +// +// This is the third: batched up front, contiguous, blocking, on the FD wire. +// +// The commands are still completed individually by the reader, so after this +// returns each cmd carries its own result and error. +// +// Hooks and metrics match an ordinary pipeline: the client's +// ProcessPipelineHook chain runs around the batch (per-command ProcessHooks do +// not, as in an ordinary pipeline), and an OTel recorder gets one pipeline +// duration and, on failure, one pipeline-level error rather than per-command +// metrics. A Limiter is the exception: the FD writer asks it once per write +// chunk, and a chunk can mix this batch with other callers' commands, so a +// batch longer than MaxBatchSize (or MaxBatchBytes) can be partly admitted. +// +// EXPERIMENTAL: this API is subject to change, use with caution. +func (ap *AutoPipeliner) FDPipelined(ctx context.Context, cmds []Cmder) error { + if ap.fd == nil { + return ErrFDPipelineUnavailable + } + return ap.fd.client.wrapPipelineHooks(func(ctx context.Context, cmds []Cmder) error { + var start time.Time + if otel.GetPipelineOperationDurationCallback() != nil { + start = time.Now() + } + res, err := ap.fdPipelined(ctx, cmds) + if res.measured() { + ap.fdPipelineMetrics(ctx, start, cmds, max(res.attempts, 1), res.cn) + } + return err + })(ctx, cmds) +} + +// wrapPipelineHooks folds the client's current hooks around next, in the same +// order the hook state builds its pipeline chain, so next runs exactly where +// processPipeline runs in an ordinary pipeline. It reads the live hook +// snapshot, so a hook added after the autopipeliner was built is honored. +func (hs *hooksMixin) wrapPipelineHooks(next ProcessPipelineHook) ProcessPipelineHook { + st := hs.state.Load() + for i := len(st.slice) - 1; i >= 0; i-- { + if wrapped := st.slice[i].ProcessPipelineHook(next); wrapped != nil { + next = wrapped + } + } + return next +} + +// fdPipelineMetrics records an FD batch the way generalProcessPipeline records +// an ordinary pipeline: one duration with the command count and attempts, and +// one error when the pipeline as a whole failed. The reader skips its +// per-command metrics for pipelined commands, so nothing is counted twice. +// +// attempts counts the times the batch was issued. One difference from an +// ordinary pipeline: connection-lease retries are not counted. The lease +// belongs to the engine, not to the batch; a batch queued while the engine +// retries a lease waits and is issued once. So a batch issued after lease +// retries reports 1 attempt, and a batch failed because the lease retries ran +// out reports 0 retries. +func (ap *AutoPipeliner) fdPipelineMetrics(ctx context.Context, start time.Time, cmds []Cmder, attempts int, cn *pool.Conn) { + perr := fdPipelineLevelErr(cmds) + db := ap.fd.client.opt.DB + ap.fd.emitMetricsGuarded(ctx, func() { + // A zero start means no duration callback existed when the operation + // began (a recorder installed mid-flight). Record no duration then, as + // an ordinary pipeline does, rather than time.Since(time.Time{}). + if cb := otel.GetPipelineOperationDurationCallback(); cb != nil && !start.IsZero() { + cb(ctx, time.Since(start), "PIPELINE", len(cmds), attempts, perr, cn, db) + } + if perr != nil { + if errorCallback := pool.GetMetricErrorCallback(); errorCallback != nil { + errorType, statusCode, isInternal := classifyCommandErrorGuarded(perr) + errorCallback(ctx, errorType, cn, statusCode, isInternal, attempts-1) + } + } + }) +} + +// fdPipelineLevelErr is the error an ordinary pipeline would report for the +// batch: the first transport (non-Redis) error, else the first command's reply +// error. pipelineReadCmds returns exactly that. +func fdPipelineLevelErr(cmds []Cmder) error { + if err := fdPipelineTransportErr(cmds); err != nil { + return err + } + if len(cmds) > 0 { + return cmds[0].rawErr() + } + return nil +} + +// fdPipelineTransportErr returns the first error in cmds that is not a reply +// from the server: a connection, protocol or push-drain failure, classified +// as the reader classifies a fatal reply (fdReplyIsFatal). nil when every +// command got a reply. +func fdPipelineTransportErr(cmds []Cmder) error { + for _, cmd := range cmds { + if e := cmd.rawErr(); e != nil && (errors.Is(e, errFDPushDrainFailed) || !isRedisError(e)) { + return e + } + } + return nil +} + +// fdPipeResult is what fdPipelined reports about a batch it ran. +type fdPipeResult struct { + // attempts is how many times the engine issued the batch: 1, plus one per + // connection-error replay (the highest across its commands); 0 when nothing + // was admitted. + attempts int + // cn is the held connection its replies came back on. + cn *pool.Conn + // flushed reports that the Close-time flush ran every batch of it through + // the pooled pipeline, which already recorded the pipeline metric. + flushed bool + // rejected reports that it was refused before admission (closed, or ctx + // done), so nothing ran. + rejected bool +} + +// measured reports whether the caller records the pipeline metric: for a +// batch that ran on the FD wire, or one refused before admission (an +// ordinary pipeline records that case when withPipelineConn fails), but not +// for one the Close-time flush measured in full. +func (r fdPipeResult) measured() bool { + return !r.flushed && (r.attempts > 0 || r.rejected) +} + +// fdPipelineAllFlushed reports whether the Close-time flush ran every batch of +// a pipeline. Only then did the pooled flush record the whole pipeline. When +// it ran only an unread tail, it recorded that tail, and the FD metric still +// covers the whole operation. +func fdPipelineAllFlushed(batches []*apBatch) bool { + if len(batches) == 0 { + return false + } + for _, b := range batches { + if !b.fdFlushed { + return false + } + } + return true +} + +// fdPipelined runs an eligible batch on the FD path. +func (ap *AutoPipeliner) fdPipelined(ctx context.Context, cmds []Cmder) (fdPipeResult, error) { + if len(cmds) == 0 { + return fdPipeResult{}, nil + } + // A redirect-aware engine is a cluster node child. It is reachable (the + // child is the node client's cached async autopipeliner, which + // ForEachMaster hands out), but it follows MOVED/ASK by re-running ONE + // command off the pipe, which would let the rest of the pipeline run ahead + // of it. Such a batch keeps the node client's ordinary pipeline, as before. + if ap.fd == nil || ap.fd.redirectAware { + return fdPipeResult{}, ErrFDPipelineUnavailable + } + // Refuse a batch containing anything that would leave the pipe. Checked for + // EVERY command before anything is submitted, so the call either goes as one + // contiguous unit or not at all — a half-submitted batch would have the + // diverted command complete out of order relative to its neighbours. + for _, cmd := range cmds { + if cmd == nil { + return fdPipeResult{}, ErrFDPipelineDiverts + } + if cmd.readTimeout() != nil || runsOutsidePipeline(cmd.Name()) || + isBlockingCmd(cmd) || isHImportCmd(cmd) || + (ap.mustDivert != nil && ap.mustDivert(ctx, cmd)) { + return fdPipeResult{}, ErrFDPipelineDiverts + } + // A NoRetry command makes the whole batch unreplayable: an ordinary + // pipeline holding one retries nothing after a connection error. The FD + // tail recovery replays per command, up to the first sent NoRetry one, + // so the commands before it could run twice. Keep such a batch off the + // pipe. + if fdNoRetrySafe(cmd) { + return fdPipeResult{}, ErrFDPipelineDiverts + } + } + batches, err := ap.fd.submitBatch(ctx, cmds) + if err != nil { + return fdPipeResult{}, err + } + if batches == nil { + // Submit-time rejection (closed, or ctx expired while backpressured). + // Every command carries its own error; report the first. + return fdPipeResult{rejected: true}, cmdsFirstErr(cmds) + } + // Mirror submit()'s contract on the deferred face: the batch is installed on + // the command so its result accessors self-gate, which matters for a caller + // that keeps the Cmder after this returns. + if !ap.blocking { + for i, cmd := range cmds { + cmd.setReady(batches[i]) + } + } + var first error + res := fdPipeResult{attempts: 1} + for i, cmd := range cmds { + // AutoFuture.Wait carries the executor-goroutine self-deadlock guard, so + // waiting through it rather than on batch.done keeps a pipeline hook that + // calls this from behaving differently than it does elsewhere. + if werr := (AutoFuture{cmd: cmd, batch: batches[i]}).Wait(); werr != nil && first == nil { + first = werr + } + // Wait returned, so the reader's stamp is visible. + if a := batches[i].fdAttempts; a > res.attempts { + res.attempts = a + } + if c := batches[i].fdConn; c != nil { + res.cn = c + } + } + res.flushed = fdPipelineAllFlushed(batches) + // A transport failure comes first, as in processPipeline: it tells the + // caller the later commands have no known result. + if err := fdPipelineTransportErr(cmds); err != nil { + return res, err + } + return res, first +} + +// submitBatch enqueues cmds as one contiguous run and returns their completion +// batches, one per command. A nil slice with a nil error means submit-time +// rejection, with each command's own error already set (same convention as +// submit()'s completedBatch return). +// +// Admission is all-or-nothing: pushBatch either takes the whole slice or takes +// none of it, so the run cannot be split across two waves by a queue that fills +// halfway through. +// +// A refused batch holds its slots while it waits (fdQueue.hold). Without that a +// batch longer than the writer's wave could starve: each take frees at most a +// wave, single submitters refill any free slot, and the batch never sees its +// whole length free at once. With the reservation the freed room accumulates +// for the batch, and the singles resume once it is admitted (or gives up). +// Holders are admitted in arrival order: a later, smaller batch that could be +// admitted with the room the head is waiting for would otherwise starve the +// head. +func (fd *fdEngine) submitBatch(ctx context.Context, cmds []Cmder) ([]*apBatch, error) { + // A batch larger than the queue itself can NEVER be admitted, and the + // room-wait loop below would spin forever waiting for space that cannot + // exist. Report it instead of hanging: the caller can split, or raise + // FullDuplexWindow. + if n := fd.q.capacity(); n > 0 && len(cmds) > n { + return nil, ErrFDPipelineTooLarge + } + if fd.ap.isClosed() { + setCmdsErr(cmds, ErrClosed) + return nil, nil + } + // No per-command ProcessHooks here: the batch runs inside the client's + // ProcessPipelineHook chain (FDPipelined, fdPipelineHooked), and an ordinary + // pipeline runs only that chain, not a ProcessHook per command. + reqs := make([]fdReq, len(cmds)) + batches := make([]*apBatch, len(cmds)) + for i, cmd := range cmds { + b := newAPBatch() // never pooled: pooled batches are the blocking face's single-waiter signal + b.fdAttempts = 1 // the first issue; replays and the reader raise it + b.fdGroup = batches[0] + if i == 0 { + b.fdGroup = b + } + batches[i] = b + reqs[i] = fdReq{cmd: cmd, batch: b, ctx: ctx, attempts: 1, pipelined: true} + } + + // held: this batch holds len(reqs) reserved slots under id and waits on + // wake, which the queue signals only while this batch is the head of the + // holder line (and on close). Every exit that does not admit the batch must + // give the slots back, after the RLock is dropped. + held, id := false, uint64(0) + var wake <-chan struct{} + reject := func(err error) ([]*apBatch, error) { + fd.submitMu.RUnlock() + if held { + fd.q.unhold(id) + } + setCmdsErr(cmds, err) + return nil, nil + } + fd.submitMu.RLock() + for { + if fd.closed { + return reject(ErrClosed) + } + // A done ctx is refused before admission, not only while waiting for + // room: once admitted the batch runs, and an ordinary pipeline rejects a + // canceled call before writing anything. + if cerr := ctx.Err(); cerr != nil { + return reject(cerr) + } + var res fdPushResult + if held { + res = fd.q.pushHeld(reqs, id) // consumes the reservation on success + } else { + res = fd.q.pushBatch(reqs) + } + switch res { + case fdPushOK: + fd.submitMu.RUnlock() + return batches, nil + case fdPushClosed: + return reject(ErrClosed) + } + // Queue full. Reserve the batch's slots, then retry at once rather than + // sleeping: a take between the refused push and the hold saw no holder + // to wake, and if it emptied the queue there would be no later take to + // wake this batch. The retry under the reservation either fits, or + // observes a queue that still holds work (so a take, which wakes the + // head, is guaranteed), or finds another holder ahead (whose departure + // wakes the new head). + if !held { + id, wake = fd.q.hold(len(reqs)) + held = true + continue + } + // Wait to be woken as the head of the line (a take freed room, or the + // previous head left), then retry the whole batch. The wake is this + // batch's own cap-1 channel, so it cannot be taken by another waiter and + // a wake that lands while this batch is mid-retry is kept for the next + // wait. Same release conditions as submit()'s backpressure wait. + select { + case <-wake: + case <-ctx.Done(): + return reject(ctx.Err()) + case <-fd.ap.ctx.Done(): + return reject(ErrClosed) + } + } +} + +// ErrFDPipelineTooLarge is returned when a batch cannot fit the submit queue +// even when the queue is empty. +var ErrFDPipelineTooLarge = errors.New( + "redis: FDPipelined batch is larger than the full-duplex submit queue") diff --git a/autopipeline_fd_pipeline_csc_guard_test.go b/autopipeline_fd_pipeline_csc_guard_test.go new file mode 100644 index 00000000..ed2c1eef --- /dev/null +++ b/autopipeline_fd_pipeline_csc_guard_test.go @@ -0,0 +1,84 @@ +package redis + +import ( + "context" + "errors" + "strconv" + "testing" + "time" +) + +// An FD pipeline on a client-side-caching client refuses the commands that +// would change the held connection's state (SELECT, AUTH, HELLO with a +// protocol, RESET, CLIENT TRACKING), as an ordinary pipeline does. They leave +// the pipe (runsOutsidePipeline), so Pipeline() falls back to the ordinary +// pipeline, whose CSC guard rejects them, and FDPipelined refuses the batch. +// The held connection stays on DB 0. +func TestFDPipelineRejectsCSCStateCommands(t *testing.T) { + ctx := context.Background() + if err := probeRedis(internalTestRedisAddr()); err != nil { + t.Skipf("no redis: %v", err) + } + c := NewClient(&Options{ + Addr: internalTestRedisAddr(), + Protocol: 3, + ClientSideCacheConfig: &ClientSideCacheConfig{MaxEntries: 128}, + }) + defer c.Close() + if err := c.Ping(ctx).Err(); err != nil { + t.Skipf("no redis: %v", err) + } + if !c.autopipelineCSCActive() { + t.Skip("client-side caching did not attach (server lacks CLIENT TRACKING?)") + } + ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true}) + if err != nil { + t.Fatalf("AsyncAutoPipeline: %v", err) + } + defer ap.Close() + if ap.fd == nil { + t.Fatal("full-duplex engine not active") + } + + key := "fd-csc-guard:" + strconv.FormatInt(time.Now().UnixNano(), 10) + plain := NewClient(&Options{Addr: internalTestRedisAddr()}) + defer plain.Close() + if err := plain.Set(ctx, key, "db0", 0).Err(); err != nil { + t.Fatalf("seed: %v", err) + } + defer plain.Del(ctx, key) + + for _, tc := range []struct { + args []interface{} + want error + }{ + {[]interface{}{"select", 1}, errSelectWithCSC}, + {[]interface{}{"SELECT", 1}, errSelectWithCSC}, + {[]interface{}{"auth", "user", "pass"}, errAuthWithCSC}, + {[]interface{}{"hello", 2}, errHelloWithCSC}, + {[]interface{}{"reset"}, errResetWithCSC}, + {[]interface{}{"client", "tracking", "off"}, errClientTrackingWithCSC}, + } { + pipe := ap.Pipeline() + cmd := pipe.Do(ctx, tc.args...) + if _, err := pipe.Exec(ctx); !errors.Is(err, tc.want) { + t.Fatalf("Pipeline %v: Exec err=%v, want %v", tc.args, err, tc.want) + } + if !errors.Is(cmd.Err(), tc.want) { + t.Fatalf("Pipeline %v: cmd err=%v, want %v", tc.args, cmd.Err(), tc.want) + } + if err := ap.FDPipelined(ctx, []Cmder{NewCmd(ctx, tc.args...)}); !errors.Is(err, ErrFDPipelineDiverts) { + t.Fatalf("FDPipelined %v: err=%v, want ErrFDPipelineDiverts", tc.args, err) + } + } + + // The held connection must still be on DB 0. + pipe := ap.Pipeline() + get := pipe.Get(ctx, key) + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("GET after the refused SELECT: %v", err) + } + if v := get.Val(); v != "db0" { + t.Fatalf("GET %s = %q on the FD pipe, want db0: the connection changed DB", key, v) + } +} diff --git a/autopipeline_fd_pipeline_errs_test.go b/autopipeline_fd_pipeline_errs_test.go new file mode 100644 index 00000000..8cbdfc1d --- /dev/null +++ b/autopipeline_fd_pipeline_errs_test.go @@ -0,0 +1,67 @@ +package redis + +import ( + "context" + "strings" + "testing" + "time" +) + +// An FD pipeline whose connection fails after an early Redis error returns the +// transport failure, as an ordinary pipeline does: it tells the caller that +// the later commands have no known result. It used to return the early +// WRONGTYPE. +func TestFDPipelineReturnsTransportErrorOverRedisError(t *testing.T) { + srv := newFDDropScript(t, func(conn, idx int, name string) (string, bool) { + if conn == 0 && name == "get" { + return "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n", false + } + if conn == 0 { + return "", true // SET executed, then the connection drops + } + return "", false + }) + ap := fdPipelineTestAP(t, &Options{Addr: srv.ln.Addr().String(), MaxRetries: -1}) + + ctx := context.Background() + pipe := ap.Pipeline() + get := pipe.Get(ctx, "k") + pipe.Set(ctx, "k2", "v", 0) + _, err := pipe.Exec(ctx) + if err == nil || isRedisError(err) { + t.Fatalf("Exec err=%v, want the transport failure", err) + } + if e := get.Err(); e == nil || !strings.Contains(e.Error(), "WRONGTYPE") { + t.Fatalf("GET err=%v, want its own WRONGTYPE", e) + } +} + +// A retryable first reply must not re-run the batch when a later reply failed +// with a protocol error, which the engine does not replay because that command +// may already have run. The re-run executed SET twice; an ordinary pipeline +// returns the read error and does not retry. +func TestFDPipelineNoRerunAfterUnreplayedFailure(t *testing.T) { + srv := newFDDropScript(t, func(conn, idx int, name string) (string, bool) { + if conn == 0 && name == "get" { + return "-LOADING Redis is loading the dataset in memory\r\n", false + } + if conn == 0 && name == "set" { + return "?malformed\r\n", false // SET executed, then a protocol fault + } + return "", false + }) + ap := fdPipelineTestAP(t, &Options{Addr: srv.ln.Addr().String(), MaxRetries: 3, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond}) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Get(ctx, "k") + pipe.Set(ctx, "k2", "v", 0) + _, err := pipe.Exec(ctx) + if n := srv.count("set"); n != 1 { + t.Fatalf("SET executed %d times, want 1", n) + } + if err == nil || isRedisError(err) { + t.Fatalf("Exec err=%v, want the protocol failure", err) + } +} diff --git a/autopipeline_fd_pipeline_hooks_test.go b/autopipeline_fd_pipeline_hooks_test.go new file mode 100644 index 00000000..3032718a --- /dev/null +++ b/autopipeline_fd_pipeline_hooks_test.go @@ -0,0 +1,202 @@ +package redis + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/redis/go-redis/v9/internal/otel" +) + +// Pipeline hooks and pipeline metrics on the full-duplex path. A pipeline +// batch runs inside the client's ProcessPipelineHook chain, exactly once, and +// records one pipeline duration, as an ordinary pipeline does, while staying +// on the held connection. + +// fdRodePipe reports whether cmd ran as a full-duplex pipeline batch: only +// that path installs a batch the reader stamps with an attempt count. +func fdRodePipe(cmd Cmder) bool { + var b *apBatch + switch c := cmd.(type) { + case *StatusCmd: + b = c.ready.Load() + case *StringCmd: + b = c.ready.Load() + } + return b != nil && b.fdAttempts > 0 +} + +// fdWarm runs one pipeline so the held connection is leased before a test +// measures: the lease's handshake (HELLO) goes through the client's process +// path once, which is connection setup, not a pipelined command. +func fdWarm(t *testing.T, ap *AutoPipeliner) { + t.Helper() + pipe := ap.Pipeline() + pipe.Get(context.Background(), "warm") + if _, err := pipe.Exec(context.Background()); err != nil && !errors.Is(err, Nil) { + t.Fatalf("warm-up Exec: %v", err) + } +} + +// fdPipeHooksCounter counts pipeline and per-command hook calls, and records what a +// pipeline hook sees in the last command after next() returns. +type fdPipeHooksCounter struct { + pipelines, processes atomic.Int64 + afterNext atomic.Value // string +} + +func (h *fdPipeHooksCounter) DialHook(next DialHook) DialHook { return next } +func (h *fdPipeHooksCounter) ProcessHook(next ProcessHook) ProcessHook { + return func(ctx context.Context, cmd Cmder) error { + h.processes.Add(1) + return next(ctx, cmd) + } +} +func (h *fdPipeHooksCounter) ProcessPipelineHook(next ProcessPipelineHook) ProcessPipelineHook { + return func(ctx context.Context, cmds []Cmder) error { + h.pipelines.Add(1) + err := next(ctx, cmds) + if s, ok := cmds[len(cmds)-1].(*StringCmd); ok { + h.afterNext.Store(s.Val()) + } + return err + } +} + +func TestFDPipelineHooksWrapTheFDBatch(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + fdWarm(t, ap) + h := &fdPipeHooksCounter{} + ap.pipeliner.(*Client).AddHook(h) + + ctx := context.Background() + pipe := ap.Pipeline() + set := pipe.Set(ctx, "k", "v", 0) + get := pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if !fdRodePipe(set) || !fdRodePipe(get) { + t.Fatal("a hook took the pipeline off the FD path") + } + if n := h.pipelines.Load(); n != 1 { + t.Fatalf("ProcessPipelineHook ran %d times, want 1", n) + } + if n := h.processes.Load(); n != 0 { + t.Fatalf("ProcessHook ran %d times for a pipeline, want 0 (as an ordinary pipeline)", n) + } + if v, _ := h.afterNext.Load().(string); v != "v" { + t.Fatalf("the hook saw %q after next(), want the GET result", v) + } +} + +func TestFDPipelinedRunsPipelineHooks(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + fdWarm(t, ap) + h := &fdPipeHooksCounter{} + ap.pipeliner.(*Client).AddHook(h) + + ctx := context.Background() + get := NewStringCmd(ctx, "get", "k") + if err := ap.FDPipelined(ctx, []Cmder{NewStatusCmd(ctx, "set", "k", "v"), get}); err != nil { + t.Fatalf("FDPipelined: %v", err) + } + if !fdRodePipe(get) { + t.Fatal("FDPipelined did not run on the FD path") + } + if h.pipelines.Load() != 1 || h.processes.Load() != 0 { + t.Fatalf("hooks: pipeline=%d process=%d, want 1 and 0", h.pipelines.Load(), h.processes.Load()) + } +} + +// A batch that cannot ride the pipe runs the ordinary pipeline INSIDE the same +// hook chain, so the hooks still run once, not once around the refusal and +// again around the fallback. +func TestFDPipelineHooksRunOnceOnFallback(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + h := &fdPipeHooksCounter{} + ap.pipeliner.(*Client).AddHook(h) + + ctx := context.Background() + var buf strings.Builder + pipe := ap.Pipeline() + set := pipe.Set(ctx, "k", "v", 0) + _ = pipe.Process(ctx, NewRawWriteToCmd(ctx, &buf, "get", "k")) // NoRetry: ordinary path + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if fdRodePipe(set) { + t.Fatal("a batch with a NoRetry command rode the FD path") + } + if n := h.pipelines.Load(); n != 1 { + t.Fatalf("ProcessPipelineHook ran %d times across the fallback, want 1", n) + } +} + +// The whole-batch retry runs inside the hook chain, as generalProcessPipeline's +// retries do, so the hooks see one pipeline. +func TestFDPipelineHooksRunOnceAcrossRetry(t *testing.T) { + srv := newFDStateServer(t) + srv.loading.Store(1) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr(), MaxRetries: 3, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond}) + h := &fdPipeHooksCounter{} + ap.pipeliner.(*Client).AddHook(h) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + get := pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if v, err := get.Result(); err != nil || v != "v" { + t.Fatalf("get: v=%q err=%v", v, err) + } + if n := h.pipelines.Load(); n != 1 { + t.Fatalf("ProcessPipelineHook ran %d times across the retry, want 1", n) + } +} + +// An OTel recorder gets one pipeline duration for an FD batch and no +// per-command durations, as for an ordinary pipeline; the batch stays on the +// FD path. Both entry points. +func TestFDPipelineRecordsPipelineMetricOnFD(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + fdWarm(t, ap) + rec := &fdPipelineRecorder{} + otel.SetGlobalRecorder(rec) + defer otel.SetGlobalRecorder(nil) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + get := pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if !fdRodePipe(get) { + t.Fatal("an OTel recorder took the pipeline off the FD path") + } + if rec.pipelines.Load() != 1 || rec.cmdCount.Load() != 2 { + t.Fatalf("pipeline metric: count=%d cmdCount=%d, want 1 and 2", rec.pipelines.Load(), rec.cmdCount.Load()) + } + if n := rec.opDurations.Load(); n != 0 { + t.Fatalf("%d per-command durations recorded for a pipeline, want 0", n) + } + + get2 := NewStringCmd(ctx, "get", "k") + if err := ap.FDPipelined(ctx, []Cmder{get2}); err != nil { + t.Fatalf("FDPipelined: %v", err) + } + if rec.pipelines.Load() != 2 || rec.opDurations.Load() != 0 { + t.Fatalf("after FDPipelined: pipelines=%d per-command=%d, want 2 and 0", rec.pipelines.Load(), rec.opDurations.Load()) + } +} diff --git a/autopipeline_fd_pipeline_late_recorder_test.go b/autopipeline_fd_pipeline_late_recorder_test.go new file mode 100644 index 00000000..749b45ca --- /dev/null +++ b/autopipeline_fd_pipeline_late_recorder_test.go @@ -0,0 +1,79 @@ +package redis + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/redis/go-redis/v9/internal/otel" + "github.com/redis/go-redis/v9/internal/pool" +) + +// fdDurationRecorder records the longest pipeline duration it was given. +type fdDurationRecorder struct { + fdOtelRecorder + calls atomic.Int64 + max atomic.Int64 +} + +func (r *fdDurationRecorder) RecordPipelineOperationDuration(_ context.Context, d time.Duration, _ string, _, _ int, _ error, _ *pool.Conn, _ int) { + r.calls.Add(1) + if int64(d) > r.max.Load() { + r.max.Store(int64(d)) + } +} + +// A recorder installed while an FD pipeline runs finds the operation without +// a start time: the start is read only when a duration callback exists. +// fdPipelineMetrics used to report time.Since(time.Time{}), a duration of +// centuries. It now records no duration for that operation, as an ordinary +// pipeline does (it reads the callback once, at the start). +func TestFDPipelineMetricsSkipsDurationWithoutStart(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + rec := &fdDurationRecorder{} + otel.SetGlobalRecorder(rec) + defer otel.SetGlobalRecorder(nil) + + ctx := context.Background() + cmds := []Cmder{NewStatusCmd(ctx, "set", "k", "v")} + ap.fdPipelineMetrics(ctx, time.Time{}, cmds, 1, nil) + if n := rec.calls.Load(); n != 0 { + t.Fatalf("recorded %d duration(s) for an operation with no start (max %v)", n, time.Duration(rec.max.Load())) + } + ap.fdPipelineMetrics(ctx, time.Now(), cmds, 1, nil) + if n := rec.calls.Load(); n != 1 { + t.Fatalf("recorded %d duration(s) for an operation with a start, want 1", n) + } +} + +// The pooled re-run continues an FD operation. If that operation began with +// no duration callback (zero start) and a recorder was installed before the +// re-run, generalProcessPipelineFrom started the clock at the re-run: the +// duration covered only the pooled tail while the count included the FD +// attempt. It now records no duration, as the FD path does. +func TestPipelineRetriesAfterSkipsDurationWithoutStart(t *testing.T) { + srv := newFDStateServer(t) + c := NewClient(&Options{Addr: srv.addr(), Protocol: 2, DisableIdentity: true}) + defer c.Close() + rec := &fdDurationRecorder{} + otel.SetGlobalRecorder(rec) + defer otel.SetGlobalRecorder(nil) + + ctx := context.Background() + cmds := []Cmder{NewStatusCmd(ctx, "set", "k", "v")} + if err := c.processPipelineRetriesAfter(ctx, cmds, 0, time.Time{}, 1); err != nil { + t.Fatalf("processPipelineRetriesAfter: %v", err) + } + if n := rec.calls.Load(); n != 0 { + t.Fatalf("recorded %d duration(s) for a continued operation with no start", n) + } + cmds = []Cmder{NewStatusCmd(ctx, "set", "k", "v")} + if err := c.processPipelineRetriesAfter(ctx, cmds, 0, time.Now(), 1); err != nil { + t.Fatalf("processPipelineRetriesAfter: %v", err) + } + if n := rec.calls.Load(); n != 1 { + t.Fatalf("recorded %d duration(s) for a continued operation with a start, want 1", n) + } +} diff --git a/autopipeline_fd_pipeline_metrics_test.go b/autopipeline_fd_pipeline_metrics_test.go new file mode 100644 index 00000000..49945dee --- /dev/null +++ b/autopipeline_fd_pipeline_metrics_test.go @@ -0,0 +1,120 @@ +package redis + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/redis/go-redis/v9/internal/otel" +) + +// processPipelineRetriesAfter continues an operation that already ran, so its +// retries continue the backoff too. It restarted at the first step, so after +// an FD attempt the pooled re-run's later retries slept like early ones. With +// 20 prior attempts the next backoff is at MaxRetryBackoff; the first step is +// at most 3*MinRetryBackoff. +func TestPipelineRetriesAfterContinueBackoff(t *testing.T) { + srv := newFDStateServer(t) + srv.loading.Store(1) // the first run gets LOADING, the retry succeeds + c := NewClient(&Options{ + Addr: srv.addr(), Protocol: 2, DisableIdentity: true, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: 200 * time.Millisecond, + }) + defer c.Close() + + ctx := context.Background() + cmd := NewStatusCmd(ctx, "set", "k", "v") + start := time.Now() + if err := c.processPipelineRetriesAfter(ctx, []Cmder{cmd}, 1, start, 20); err != nil { + t.Fatalf("processPipelineRetriesAfter: %v", err) + } + if elapsed := time.Since(start); elapsed < 100*time.Millisecond { + t.Fatalf("the retry slept %v; after 20 prior attempts the backoff should be at its 200ms cap", elapsed) + } +} + +// A pipeline rejected before admission (its ctx already done) records one +// pipeline metric, as an ordinary pipeline does when withPipelineConn fails. +// The FD path recorded nothing. +func TestFDPipelineRecordsRejectedPipelineMetric(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + rec := &fdPipelineRecorder{} + otel.SetGlobalRecorder(rec) + defer otel.SetGlobalRecorder(nil) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + if _, err := pipe.Exec(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Exec err=%v, want context.Canceled", err) + } + if n := rec.pipelines.Load(); n != 1 { + t.Fatalf("Pipeline().Exec: pipeline duration recorded %d times, want 1", n) + } + + cmds := []Cmder{NewStatusCmd(ctx, "set", "k", "v")} + if err := ap.FDPipelined(ctx, cmds); !errors.Is(err, context.Canceled) { + t.Fatalf("FDPipelined err=%v, want context.Canceled", err) + } + if n := rec.pipelines.Load(); n != 2 { + t.Fatalf("FDPipelined: pipeline duration recorded %d times in total, want 2", n) + } +} + +// A batch the Close-time flush never runs keeps its own metric: the flush +// marked every batch before running any, so when the carry hit a dead +// endpoint and the fresh queue was failed, fdPipelineExec skipped the failed +// batch's metric and nothing recorded it. +func TestFDShutdownFlushMarksOnlyBatchesItRuns(t *testing.T) { + ctx := context.Background() + dead := errors.New("dial tcp: connection refused") + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}}, + client: &Client{baseClient: &baseClient{opt: &Options{MaxRetries: 3}}}, + maxBatch: 8, + q: newFDQueue(8), + runPipeline: func(_ context.Context, cmds []Cmder, _ int) error { + setCmdsErr(cmds, dead) + return dead + }, + } + carry := fdPipeGroupReqs(ctx, 1) + fresh := fdPipeGroupReqs(ctx, 1) + carry[0].batch.fdAttempts, fresh[0].batch.fdAttempts = 1, 1 // as submitBatch stamps them + fd.q.push(fresh[0]) + fd.shutdownFlush(ctx, carry) + + if !carry[0].batch.fdFlushed { + t.Fatal("the carried batch ran in the flush but is not marked flushed") + } + if carry[0].batch.fdAttempts < 2 { + t.Fatalf("the carried batch ran again in the flush but reports %d issue(s)", carry[0].batch.fdAttempts) + } + if fresh[0].batch.fdFlushed { + t.Fatal("the fresh batch never ran (the carry hit a dead endpoint) but is marked flushed; its failure would record no metric") + } + // Its attempt count must not count an issue that never happened. + if a := fresh[0].batch.fdAttempts; a != 1 { + t.Fatalf("the fresh batch never ran but reports %d issues, want 1", a) + } +} + +// Only a pipeline the flush ran in full was measured by the flush. When it ran +// only the unread tail, the FD metric still covers the whole operation. +func TestFDPipelineAllFlushed(t *testing.T) { + a, b := newAPBatch(), newAPBatch() + b.fdFlushed = true + if fdPipelineAllFlushed([]*apBatch{a, b}) { + t.Fatal("a pipeline with only its tail flushed reported as flushed in full") + } + a.fdFlushed = true + if !fdPipelineAllFlushed([]*apBatch{a, b}) { + t.Fatal("a pipeline flushed in full reported as not flushed") + } + if fdPipelineAllFlushed(nil) { + t.Fatal("an empty pipeline reported as flushed") + } +} diff --git a/autopipeline_fd_pipeline_replay_test.go b/autopipeline_fd_pipeline_replay_test.go new file mode 100644 index 00000000..3bb2491b --- /dev/null +++ b/autopipeline_fd_pipeline_replay_test.go @@ -0,0 +1,150 @@ +package redis + +import ( + "bufio" + "context" + "errors" + "io" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// Replay and recovery tests for full-duplex pipeline batches. Commands are +// parsed with readRESPCommand from internal_maint_notif_test.go. + +// fdDropScript is a scripted RESP2 server for replay tests. Data connections +// are numbered in the order their first GET/SET arrives; step decides, per +// command on a connection, whether to reply normally, reply with LOADING, or +// close the connection. It counts executions per command name. +type fdDropScript struct { + ln net.Listener + conns atomic.Int64 + calls sync.Map // name -> *atomic.Int64 + step func(conn, cmdIdx int, name string) (reply string, drop bool) +} + +func (s *fdDropScript) count(name string) int64 { + v, _ := s.calls.LoadOrStore(name, new(atomic.Int64)) + return v.(*atomic.Int64).Load() +} + +func newFDDropScript(t *testing.T, step func(conn, cmdIdx int, name string) (string, bool)) *fdDropScript { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + s := &fdDropScript{ln: ln, step: step} + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func(c net.Conn) { + defer c.Close() + rd := bufio.NewReader(c) + conn, idx := -1, 0 + for { + args, err := readRESPCommand(rd) + if err != nil { + return + } + name := strings.ToLower(args[0]) + if name == "hello" { + _, _ = io.WriteString(c, "*4\r\n$6\r\nserver\r\n$5\r\nredis\r\n$5\r\nproto\r\n:2\r\n") + continue + } + if name != "get" && name != "set" { + _, _ = io.WriteString(c, "+OK\r\n") + continue + } + if conn < 0 { + conn = int(s.conns.Add(1)) - 1 + } + v, _ := s.calls.LoadOrStore(name, new(atomic.Int64)) + v.(*atomic.Int64).Add(1) + reply, drop := s.step(conn, idx, name) + idx++ + if drop { + return + } + if reply == "" { + if name == "set" { + reply = "+OK\r\n" + } else { + reply = "$-1\r\n" + } + } + if _, err := io.WriteString(c, reply); err != nil { + return + } + } + }(c) + } + }() + return s +} + +// A pipeline holding a NoRetry command must not be replayed as a unit after a +// connection error, as an ordinary pipeline is not. The FD tail recovery +// replayed the commands before the NoRetry one, so here the SET ran twice. +// Such a batch now stays off the pipe. +func TestFDPipelineNoRetryBatchNotReplayed(t *testing.T) { + srv := newFDDropScript(t, func(conn, idx int, name string) (string, bool) { + return "", conn == 0 && idx == 0 // lose the first connection on its first command + }) + ap := fdPipelineTestAP(t, &Options{Addr: srv.ln.Addr().String(), MaxRetries: 3, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond}) + + ctx := context.Background() + var buf strings.Builder + batch := []Cmder{NewStatusCmd(ctx, "set", "k", "v"), NewRawWriteToCmd(ctx, &buf, "get", "x")} + if err := ap.FDPipelined(ctx, batch); !errors.Is(err, ErrFDPipelineDiverts) { + t.Fatalf("FDPipelined with a NoRetry command: err=%v, want ErrFDPipelineDiverts", err) + } + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + _ = pipe.Process(ctx, NewRawWriteToCmd(ctx, &buf, "get", "x")) + _, _ = pipe.Exec(ctx) + if n := srv.count("set"); n != 1 { + t.Fatalf("SET executed %d times; a pipeline holding a NoRetry command must not be replayed", n) + } +} + +// Replays the FD engine spends on a command count against the pipeline +// budget even when that command then FAILS instead of completing. Here, with +// MaxRetries:1, the SET gets LOADING, the connection drops before the GET's +// reply, the GET is replayed and that connection drops too. That is two +// executions of the GET, the whole budget. The failed GET used to carry no +// attempt count, so the pooled re-run ran it a third time. +func TestFDPipelineRetryCountsFailedReplays(t *testing.T) { + srv := newFDDropScript(t, func(conn, idx int, name string) (string, bool) { + switch { + case conn == 0 && name == "set": + return "-LOADING Redis is loading the dataset in memory\r\n", false + case conn <= 1 && name == "get": + return "", true // drop before replying, on the first and the replay connection + } + return "", false + }) + ap := fdPipelineTestAP(t, &Options{Addr: srv.ln.Addr().String(), MaxRetries: 1, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond}) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + pipe.Get(ctx, "k") + _, err := pipe.Exec(ctx) + if n := srv.count("get"); n != 2 { + t.Fatalf("GET executed %d times with MaxRetries:1, want 2 (err=%v)", n, err) + } + if err == nil { + t.Fatal("Exec succeeded, want the failure once the budget is spent") + } +} diff --git a/autopipeline_fd_pipeline_retry_test.go b/autopipeline_fd_pipeline_retry_test.go new file mode 100644 index 00000000..bc70e25e --- /dev/null +++ b/autopipeline_fd_pipeline_retry_test.go @@ -0,0 +1,285 @@ +package redis + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/redis/go-redis/v9/internal/otel" + "github.com/redis/go-redis/v9/internal/pool" + "github.com/redis/go-redis/v9/internal/proto" +) + +// The whole-batch retry and its interaction with the engine's own replays. + +// After a connection-error replay the replay already was the batch's retry, so +// a stale retryable reply on the first command must not trigger a pooled +// re-run on top. Here SET gets LOADING, the connection drops before GET's +// reply, and the engine replays GET. The re-run used to run GET a third time, +// where an ordinary pipeline runs it twice (once, then once in its retry). +func TestFDPipelineNoRerunAfterReplay(t *testing.T) { + srv := newFDDropScript(t, func(conn, idx int, name string) (string, bool) { + if conn == 0 { + if name == "set" { + return "-LOADING Redis is loading the dataset in memory\r\n", false + } + return "", true // GET executed, then the connection drops + } + return "", false + }) + ap := fdPipelineTestAP(t, &Options{Addr: srv.ln.Addr().String(), MaxRetries: 3, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond}) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + pipe.Get(ctx, "k") + _, err := pipe.Exec(ctx) + if n := srv.count("get"); n != 2 { + t.Fatalf("GET executed %d times, want 2 (once, then the engine's replay)", n) + } + if err == nil || !strings.Contains(err.Error(), "LOADING") { + t.Fatalf("Exec err=%v, want the first command's LOADING", err) + } +} + +// The Close-time flush runs a pipelined batch through the pooled pipeline with +// its own retry loop, so the batch must count as already retried: otherwise +// fdPipelineExec saw one issue and a stale LOADING on the first command, and +// ran the batch again after Close. +func TestFDShutdownFlushMarksPipelinedBatchesRetried(t *testing.T) { + ctx := context.Background() + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}}, + client: &Client{baseClient: &baseClient{opt: &Options{MaxRetries: 3}}}, + maxBatch: 8, + runPipeline: func(_ context.Context, cmds []Cmder, _ int) error { + cmds[0].SetErr(proto.RedisError("LOADING Redis is loading the dataset in memory")) + return nil + }, + } + set := NewStatusCmd(ctx, "set", "k", "v") + b := newAPBatch() + b.fdAttempts = 1 + fd.shutdownFlush(ctx, []fdReq{{cmd: set, batch: b, attempts: 1, pipelined: true}}) + + select { + case <-b.done: + case <-time.After(5 * time.Second): + t.Fatal("the flushed command never completed") + } + if b.fdAttempts < 2 { + t.Fatalf("flushed pipelined batch reports %d issue(s); fdPipelineExec would re-run it after Close", b.fdAttempts) + } + if !b.fdFlushed { + t.Fatal("flushed pipelined batch is not marked flushed; fdPipelineExec would record its pipeline metric twice") + } +} + +// fdFlushProbe is a stub engine whose flush records the chunk sizes it runs. +func fdFlushProbe(maxBatch int) (*fdEngine, *[]int) { + var chunks []int + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}}, + client: &Client{baseClient: &baseClient{opt: &Options{MaxRetries: 3}}}, + maxBatch: maxBatch, + runPipeline: func(_ context.Context, cmds []Cmder, _ int) error { + chunks = append(chunks, len(cmds)) + return nil + }, + } + return fd, &chunks +} + +// fdPipeGroupReqs builds n pipelined requests of one FD pipeline batch, the +// way submitBatch stamps them, with the given attempt counts. +func fdPipeGroupReqs(ctx context.Context, attempts ...int) []fdReq { + reqs := make([]fdReq, len(attempts)) + var first *apBatch + for i, a := range attempts { + b := newAPBatch() + if first == nil { + first = b + } + b.fdGroup = first + reqs[i] = fdReq{cmd: NewStatusCmd(ctx, "set", "k"+itoa(i), "v"), batch: b, attempts: a, pipelined: true} + } + return reqs +} + +// The Close-time flush runs an FD pipeline batch as ONE pipeline, so its +// whole-batch retry covers all of it. It used to split the batch at +// MaxBatchSize and retry each chunk on its own. Ordinary commands around it are +// still chunked, without being mixed into the pipeline. +func TestFDShutdownFlushKeepsPipelineWhole(t *testing.T) { + ctx := context.Background() + fd, chunks := fdFlushProbe(2) + carry := append(fdPipeGroupReqs(ctx, 1, 1, 1, 1, 1), + fdReq{cmd: NewStatusCmd(ctx, "get", "a"), batch: newAPBatch(), attempts: 1}, + fdReq{cmd: NewStatusCmd(ctx, "get", "b"), batch: newAPBatch(), attempts: 1}) + fd.shutdownFlush(ctx, carry) + if got := *chunks; len(got) != 2 || got[0] != 5 || got[1] != 2 { + t.Fatalf("flush chunks %v, want [5 2]: the 5-command pipeline whole, then the ordinary commands", got) + } +} + +// A pipeline batch whose commands carry different attempt counts (a partial +// replay) is still flushed as one pipeline, not split by attempt count. +func TestFDShutdownFlushKeepsPipelineWholeAcrossAttempts(t *testing.T) { + ctx := context.Background() + fd, chunks := fdFlushProbe(8) + fd.shutdownFlush(ctx, fdPipeGroupReqs(ctx, 2, 2, 1)) + if got := *chunks; len(got) != 1 || got[0] != 3 { + t.Fatalf("flush chunks %v, want [3]: one pipeline despite mixed attempt counts", got) + } +} + +// A pipeline the writer took only a prefix of, before the session failed, has +// that prefix in the carry and its tail still queued. Close flushes it as one +// pipeline with one retry budget, the smallest of its commands', not as two +// pipelines where the tail gets a full budget of its own. +func TestFDShutdownFlushJoinsPipelineSplitAcrossCarryAndQueue(t *testing.T) { + ctx := context.Background() + type run struct{ n, maxRetries int } + var runs []run + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}}, + client: &Client{baseClient: &baseClient{opt: &Options{MaxRetries: 3}}}, + maxBatch: 8, + q: newFDQueue(8), + runPipeline: func(_ context.Context, cmds []Cmder, maxRetries int) error { + runs = append(runs, run{len(cmds), maxRetries}) + return nil + }, + } + reqs := fdPipeGroupReqs(ctx, 2, 2, 1, 1, 1) // prefix replayed once, tail never taken + if res := fd.q.pushBatch(reqs[2:]); res != fdPushOK { + t.Fatalf("pushBatch = %v", res) + } + fd.shutdownFlush(ctx, reqs[:2]) + + want := run{5, fdCarryRemainingRetries(2, 3)} + if len(runs) != 1 || runs[0] != want { + t.Fatalf("flush runs %+v, want one %+v: the whole pipeline at the prefix's remaining budget", runs, want) + } +} + +// fdRetryMetricRecorder captures pipeline-duration records. +type fdRetryMetricRecorder struct { + fdOtelRecorder + calls atomic.Int64 + attempts atomic.Int64 + duration atomic.Int64 + lastErr atomic.Value +} + +func (r *fdRetryMetricRecorder) RecordPipelineOperationDuration(_ context.Context, d time.Duration, _ string, _ int, attempts int, err error, _ *pool.Conn, _ int) { + r.calls.Add(1) + r.attempts.Store(int64(attempts)) + r.duration.Store(int64(d)) + if err != nil { + r.lastErr.Store(err) + } +} + +// A whole-batch re-run continues the FD attempt's measurement: one pipeline +// metric, counting the FD attempt and covering its time. The pooled re-run +// used to open its own measurement, recording one attempt and only its own +// duration. +func TestFDPipelineRetryMetricCoversTheWholeOperation(t *testing.T) { + srv := newFDStateServer(t) + srv.loading.Store(1) + srv.delay = 30 * time.Millisecond + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr(), MaxRetries: 3, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond}) + rec := &fdRetryMetricRecorder{} + otel.SetGlobalRecorder(rec) + defer otel.SetGlobalRecorder(nil) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if n := rec.calls.Load(); n != 1 { + t.Fatalf("%d pipeline metrics for one operation, want 1", n) + } + if a := rec.attempts.Load(); a != 2 { + t.Fatalf("attempts=%d, want 2 (the FD attempt and the re-run)", a) + } + // Both attempts wait srv.delay, so one operation spans at least two of them. + if d := time.Duration(rec.duration.Load()); d < 2*srv.delay { + t.Fatalf("duration %v excludes the FD attempt (each attempt waits %v)", d, srv.delay) + } +} + +// A ctx that ends during the backoff before the re-run still records the +// operation, with the ctx error. +func TestFDPipelineRetryMetricOnBackoffCancel(t *testing.T) { + srv := newFDStateServer(t) + srv.loading.Store(1) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr(), MaxRetries: 3, + MinRetryBackoff: time.Second, MaxRetryBackoff: time.Second}) + rec := &fdRetryMetricRecorder{} + otel.SetGlobalRecorder(rec) + defer otel.SetGlobalRecorder(nil) + + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + if _, err := pipe.Exec(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Exec err=%v, want the ctx deadline", err) + } + if n := rec.calls.Load(); n != 1 { + t.Fatalf("%d pipeline metrics, want 1 even though ctx ended in the backoff", n) + } + if e, _ := rec.lastErr.Load().(error); !errors.Is(e, context.DeadlineExceeded) { + t.Fatalf("metric error %v, want the ctx deadline", e) + } +} + +// Live recovery fails an FD pipeline as one. When part of it fails (its +// budget is spent, or the error is not retryable), the rest of it must not +// run: neither its members kept for replay nor the tail the writer had not yet +// taken from the queue. It used to send that tail on the next connection, so +// Exec reported a transport error for a pipeline whose later commands ran. +func TestFDRecoveryFailsTheWholePipeline(t *testing.T) { + ctx := context.Background() + fd := &fdEngine{ + ap: &AutoPipeliner{config: &AutoPipelineOptions{}}, + client: &Client{baseClient: &baseClient{opt: &Options{}}}, + q: newFDQueue(16), + } + pipe := fdPipeGroupReqs(ctx, 2, 1, 1, 1, 1) // [0] spent, [1] kept, [2:] still queued + other := fdPipeGroupReqs(ctx, 1) // another pipeline, kept + single := fdReq{cmd: NewStatusCmd(ctx, "get", "s"), batch: newAPBatch(), attempts: 1} + if res := fd.q.pushBatch(pipe[2:]); res != fdPushOK { + t.Fatalf("pushBatch = %v", res) + } + fd.q.push(single) + + dead := errors.New("read: connection reset by peer") + keep := fd.failPipelines(pipe[:1], []fdReq{pipe[1], other[0]}, dead) + + if len(keep) != 1 || keep[0].batch != other[0].batch { + t.Fatalf("kept %d requests, want only the other pipeline's", len(keep)) + } + for i, r := range pipe { + select { + case <-r.batch.done: + default: + t.Fatalf("pipeline command %d was not failed with the rest of its pipeline", i) + } + if !errors.Is(r.cmd.Err(), dead) { + t.Fatalf("pipeline command %d: err %v, want %v", i, r.cmd.Err(), dead) + } + } + if d := fd.q.depth(); d != 1 { + t.Fatalf("queue depth %d after the failure, want 1 (only the unrelated command)", d) + } +} diff --git a/autopipeline_fd_pipeline_test.go b/autopipeline_fd_pipeline_test.go new file mode 100644 index 00000000..c84d65ff --- /dev/null +++ b/autopipeline_fd_pipeline_test.go @@ -0,0 +1,465 @@ +package redis + +import ( + "bufio" + "context" + "errors" + "io" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/redis/go-redis/v9/internal/otel" + "github.com/redis/go-redis/v9/internal/pool" +) + +// fdStateServer is a minimal stateful RESP2 server for the FD pipeline tests. +// SET stores, GET reads ($-1 when missing), HELLO gets a minimal map, anything +// else +OK. It can answer the first N SETs with -LOADING (not applied), counts +// SETs, and can delay every GET/SET reply. Commands are parsed with readRESPCommand from +// internal_maint_notif_test.go. +type fdStateServer struct { + ln net.Listener + mu sync.Mutex + kv map[string]string + loading atomic.Int64 // SETs still to answer with -LOADING (not applied) + setCalls atomic.Int64 // SETs received, LOADING and dropped ones included + // dropFirst closes the connection on the first SET instead of replying, + // so the client sees the write land and the reply never arrive. + dropFirst atomic.Bool + delay time.Duration + wg sync.WaitGroup + conns []net.Conn +} + +func newFDStateServer(t *testing.T) *fdStateServer { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + s := &fdStateServer{ln: ln, kv: map[string]string{}} + s.wg.Add(1) + go func() { + defer s.wg.Done() + for { + c, err := ln.Accept() + if err != nil { + return + } + s.mu.Lock() + s.conns = append(s.conns, c) + s.mu.Unlock() + s.wg.Add(1) + go func() { + defer s.wg.Done() + s.serve(c) + }() + } + }() + t.Cleanup(s.close) + return s +} + +func (s *fdStateServer) addr() string { return s.ln.Addr().String() } + +func (s *fdStateServer) close() { + _ = s.ln.Close() + s.mu.Lock() + for _, c := range s.conns { + _ = c.Close() + } + s.mu.Unlock() + s.wg.Wait() +} + +func (s *fdStateServer) serve(c net.Conn) { + defer c.Close() + rd := bufio.NewReader(c) + for { + args, err := readRESPCommand(rd) + if err != nil { + return + } + name := "" + if len(args) > 0 { + name = strings.ToLower(args[0]) + } + var reply string + switch { + case name == "hello": + reply = "*4\r\n$6\r\nserver\r\n$5\r\nredis\r\n$5\r\nproto\r\n:2\r\n" + case name == "set" && len(args) >= 3: + s.setCalls.Add(1) + if s.dropFirst.CompareAndSwap(true, false) { + return // closes c: the reply never arrives + } + if s.loading.Add(-1) >= 0 { + reply = "-LOADING Redis is loading the dataset in memory\r\n" + break + } + s.mu.Lock() + s.kv[args[1]] = args[2] + s.mu.Unlock() + reply = "+OK\r\n" + case name == "get" && len(args) == 2: + s.mu.Lock() + v, ok := s.kv[args[1]] + s.mu.Unlock() + if ok { + reply = "$" + itoa(len(v)) + "\r\n" + v + "\r\n" + } else { + reply = "$-1\r\n" + } + default: + reply = "+OK\r\n" + } + if (name == "get" || name == "set") && s.delay > 0 { + time.Sleep(s.delay) + } + if _, err := io.WriteString(c, reply); err != nil { + return + } + } +} + +func fdPipelineTestAP(t *testing.T, opt *Options) *AutoPipeliner { + t.Helper() + opt.Protocol = 2 + opt.DisableIdentity = true + if opt.PipelinePoolSize == 0 { + opt.PipelinePoolSize = 2 + } + c := NewClient(opt) + t.Cleanup(func() { _ = c.Close() }) + ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true}) + if err != nil { + t.Fatalf("AsyncAutoPipeline: %v", err) + } + t.Cleanup(func() { _ = ap.Close() }) + if ap.fd == nil { + t.Fatal("full-duplex engine not active") + } + return ap +} + +// Pipeline.Exec returns the first command error, redis.Nil included. The FD +// path used to drop Nil and return nil for a pipeline whose only error is a +// missing key. +func TestFDPipelineExecReturnsNil(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + + ctx := context.Background() + pipe := ap.Pipeline() + get := pipe.Get(ctx, "missing") + if _, err := pipe.Exec(ctx); !errors.Is(err, Nil) { + t.Fatalf("Exec err=%v, want redis.Nil", err) + } + if !errors.Is(get.Err(), Nil) { + t.Fatalf("get err=%v, want redis.Nil", get.Err()) + } +} + +// A retryable reply on the first command retries the whole pipeline, in +// order, as an ordinary pipeline does. The FD path used to retry the SET alone +// on another connection while the GET completed from the first attempt, so +// the GET returned the value from before the SET. +func TestFDPipelineExecRetriesWholeBatchInOrder(t *testing.T) { + srv := newFDStateServer(t) + srv.loading.Store(1) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr(), MaxRetries: 3}) + + ctx := context.Background() + pipe := ap.Pipeline() + set := pipe.Set(ctx, "k", "v", 0) + get := pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if err := set.Err(); err != nil { + t.Fatalf("set: %v", err) + } + if v, err := get.Result(); err != nil || v != "v" { + t.Fatalf("get: v=%q err=%v, want the value the pipeline's SET wrote", v, err) + } + if srv.loading.Load() > 0 { + t.Fatal("the LOADING reply was never sent; the test did not exercise the retry") + } +} + +// With ContextTimeoutEnabled an ordinary pipeline is bounded by the caller's +// deadline. The FD path used to wait for the replies regardless. +func TestFDPipelineExecHonorsContextTimeout(t *testing.T) { + srv := newFDStateServer(t) + srv.delay = time.Second + ap := fdPipelineTestAP(t, &Options{ + Addr: srv.addr(), + ContextTimeoutEnabled: true, + ReadTimeout: 5 * time.Second, + MaxRetries: -1, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + pipe := ap.Pipeline() + pipe.Get(ctx, "k") + start := time.Now() + _, err := pipe.Exec(ctx) + if el := time.Since(start); el > 700*time.Millisecond { + t.Fatalf("Exec took %v with a 100ms deadline (err=%v): the deadline was ignored", el, err) + } + if err == nil { + t.Fatal("Exec returned nil although its deadline passed before the reply") + } +} + +// fdPipeHook counts ProcessPipelineHook invocations and passes everything +// through. +type fdPipeHook struct{ pipelines atomic.Int64 } + +func (h *fdPipeHook) DialHook(next DialHook) DialHook { return next } +func (h *fdPipeHook) ProcessHook(next ProcessHook) ProcessHook { return next } +func (h *fdPipeHook) ProcessPipelineHook(next ProcessPipelineHook) ProcessPipelineHook { + return func(ctx context.Context, cmds []Cmder) error { + h.pipelines.Add(1) + return next(ctx, cmds) + } +} + +// An ordinary Pipeline.Exec runs the ProcessPipelineHook chain once. The FD +// path used to skip it, so pipeline-level tracing and metrics hooks never saw +// these pipelines. +func TestFDPipelineExecRunsPipelineHooks(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + h := &fdPipeHook{} + ap.pipeliner.(*Client).AddHook(h) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if n := h.pipelines.Load(); n != 1 { + t.Fatalf("ProcessPipelineHook ran %d times, want 1", n) + } +} + +// fdPipeCountLimiter admits everything and counts Allow calls. +type fdPipeCountLimiter struct{ calls atomic.Int64 } + +func (l *fdPipeCountLimiter) Allow() error { l.calls.Add(1); return nil } +func (l *fdPipeCountLimiter) ReportResult(error) {} + +// A Limiter does not take the batch off the FD path. The FD writer asks it +// once per write chunk (documented on FDPipelined), so at MaxBatchSize 1 a +// three-command pipeline takes three decisions, where an ordinary pipeline +// takes one per attempt. +func TestFDPipelineWithLimiterStaysOnFD(t *testing.T) { + srv := newFDStateServer(t) + lim := &fdPipeCountLimiter{} + c := NewClient(&Options{ + Addr: srv.addr(), Limiter: lim, + Protocol: 2, DisableIdentity: true, PipelinePoolSize: 2, + }) + t.Cleanup(func() { _ = c.Close() }) + ap, err := c.AsyncAutoPipelineWithOptions(&AutoPipelineOptions{FullDuplex: true, MaxBatchSize: 1}) + if err != nil { + t.Fatalf("AsyncAutoPipeline: %v", err) + } + t.Cleanup(func() { _ = ap.Close() }) + if ap.fd == nil { + t.Fatal("full-duplex engine not active") + } + + ctx := context.Background() + exec := func() ([]*StatusCmd, int64) { + pipe := ap.Pipeline() + cmds := []*StatusCmd{pipe.Set(ctx, "a", "1", 0), pipe.Set(ctx, "b", "2", 0), pipe.Set(ctx, "c", "3", 0)} + before := lim.calls.Load() + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + return cmds, lim.calls.Load() - before + } + exec() // warm: the first Exec also pays for leasing the held connection + cmds, n := exec() + for _, cmd := range cmds { + if !fdRodePipe(cmd) { + t.Fatal("a Limiter took the pipeline off the FD path") + } + } + if n != int64(len(cmds)) { + t.Fatalf("%d Limiter decisions for %d single-command chunks, want one per chunk", n, len(cmds)) + } +} + +// fdPipelineRecorder counts pipeline-duration records. +type fdPipelineRecorder struct { + fdOtelRecorder + pipelines atomic.Int64 + cmdCount atomic.Int64 +} + +func (r *fdPipelineRecorder) RecordPipelineOperationDuration(_ context.Context, _ time.Duration, _ string, n, _ int, _ error, _ *pool.Conn, _ int) { + r.pipelines.Add(1) + r.cmdCount.Store(int64(n)) +} + +// With an OTel recorder installed, an ordinary Pipeline.Exec records one +// pipeline duration. The FD path recorded per-command durations only. +func TestFDPipelineExecRecordsPipelineMetric(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + rec := &fdPipelineRecorder{} + otel.SetGlobalRecorder(rec) + defer otel.SetGlobalRecorder(nil) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if n := rec.pipelines.Load(); n != 1 { + t.Fatalf("pipeline duration recorded %d times, want 1", n) + } + if n := rec.cmdCount.Load(); n != 2 { + t.Fatalf("pipeline duration cmdCount %d, want 2", n) + } +} + +// A pipeline holding a NoRetry command is not retried, even when its first +// command gets a retryable reply: an ordinary pipeline stops on +// cmdsContainNoRetry. The FD path re-ran the batch, so a RawWriteTo command +// wrote its reply into the caller's writer twice. +func TestFDPipelineExecDoesNotRetryNoRetryCommands(t *testing.T) { + srv := newFDStateServer(t) + srv.loading.Store(1) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr(), MaxRetries: 3}) + + ctx := context.Background() + var buf strings.Builder + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + _ = pipe.Process(ctx, NewRawWriteToCmd(ctx, &buf, "get", "missing")) + _, err := pipe.Exec(ctx) + if err == nil || !strings.Contains(err.Error(), "LOADING") { + t.Fatalf("Exec err=%v, want the first command's LOADING (not retried)", err) + } + if got, want := buf.String(), "$-1\r\n"; got != want { + t.Fatalf("NoRetry command wrote %q, want one reply %q", got, want) + } +} + +// A canceled context is refused before any command is written, as an ordinary +// pipeline does: every command carries the context error and nothing runs. +func TestFDPipelineExecRejectsCanceledContext(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + pipe := ap.Pipeline() + set := pipe.Set(ctx, "k", "v", 0) + get := pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Exec err=%v, want context.Canceled", err) + } + for _, cmd := range []Cmder{set, get} { + if !errors.Is(cmd.Err(), context.Canceled) { + t.Fatalf("%s err=%v, want context.Canceled", cmd.Name(), cmd.Err()) + } + } + srv.mu.Lock() + _, wrote := srv.kv["k"] + srv.mu.Unlock() + if wrote { + t.Fatal("the SET reached the server although the context was already canceled") + } +} + +// The whole-batch retry keeps the ordinary pipeline's budget of MaxRetries+1 +// executions, counting the FD attempt as the first. The re-run used to start +// with the full budget again, so MaxRetries:1 allowed three executions. +func TestFDPipelineExecRetryKeepsBudget(t *testing.T) { + srv := newFDStateServer(t) + srv.loading.Store(2) // the FD attempt and one retry both see LOADING + ap := fdPipelineTestAP(t, &Options{ + Addr: srv.addr(), MaxRetries: 1, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond, + }) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + pipe.Get(ctx, "k") + _, err := pipe.Exec(ctx) + if n := srv.setCalls.Load(); n != 2 { + t.Fatalf("pipeline executed %d times with MaxRetries:1, want 2 (err=%v)", n, err) + } + if err == nil || !strings.Contains(err.Error(), "LOADING") { + t.Fatalf("Exec err=%v, want LOADING once the budget is spent", err) + } +} + +// Executions the FD engine spends on its own connection-error replay count +// against the same budget. Here the first write is lost (connection closed), +// the engine replays the batch and gets LOADING: with MaxRetries:1 that is +// both allowed executions, so there is no pooled re-run. The re-run used to +// assume the FD engine had issued the batch exactly once. +func TestFDPipelineExecRetryCountsFDReplays(t *testing.T) { + srv := newFDStateServer(t) + srv.dropFirst.Store(true) + srv.loading.Store(1) + ap := fdPipelineTestAP(t, &Options{ + Addr: srv.addr(), MaxRetries: 1, + MinRetryBackoff: time.Millisecond, MaxRetryBackoff: time.Millisecond, + }) + + ctx := context.Background() + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + pipe.Get(ctx, "k") + _, err := pipe.Exec(ctx) + if n := srv.setCalls.Load(); n != 2 { + t.Fatalf("pipeline executed %d times with MaxRetries:1, want 2 (err=%v)", n, err) + } + if err == nil || !strings.Contains(err.Error(), "LOADING") { + t.Fatalf("Exec err=%v, want LOADING once the budget is spent", err) + } + if srv.dropFirst.Load() { + t.Fatal("the first write was never dropped; the test did not exercise the replay") + } +} + +// A redirect-aware engine (a cluster node child, reachable as the node +// client's cached async autopipeliner) follows MOVED/ASK by re-running one +// command off the pipe. A pipeline must not ride it, or a redirected command +// would run after the rest of its batch. It keeps the ordinary pipeline. +func TestFDPipelineRefusesRedirectAwareEngine(t *testing.T) { + srv := newFDStateServer(t) + ap := fdPipelineTestAP(t, &Options{Addr: srv.addr()}) + ap.fd.redirectAware = true // what the cluster router sets on its children + + ctx := context.Background() + if err := ap.FDPipelined(ctx, []Cmder{NewStatusCmd(ctx, "set", "k", "v")}); !errors.Is(err, ErrFDPipelineUnavailable) { + t.Fatalf("FDPipelined on a redirect-aware engine: err=%v, want ErrFDPipelineUnavailable", err) + } + pipe := ap.Pipeline() + pipe.Set(ctx, "k", "v", 0) + get := pipe.Get(ctx, "k") + if _, err := pipe.Exec(ctx); err != nil { + t.Fatalf("Exec: %v", err) + } + if v, err := get.Result(); err != nil || v != "v" { + t.Fatalf("get: v=%q err=%v", v, err) + } +} diff --git a/autopipeline_fdqueue.go b/autopipeline_fdqueue.go index c4abf12f..7e24409f 100644 --- a/autopipeline_fdqueue.go +++ b/autopipeline_fdqueue.go @@ -16,7 +16,8 @@ import "sync" // // The slice costs one lock and one append per command on the producer side, and // ONE lock plus ONE bulk copy for the whole wave on the writer side — the -// writer's per-command cost disappears. +// writer's per-command cost disappears. pushBatch amortises the producer lock +// across k commands as well. // // The structural win is not the copy, it is the parking. A channel can only // express "wake me on the next arrival", so an accumulating writer was woken @@ -32,6 +33,19 @@ import "sync" // for the mutex. Per-goroutine program order is still exact (a goroutine's push // happens-before its next push), which is what the ordering contract actually // promises; cross-goroutine arrival order under saturation was never defined. +// +// Re-contention is fair enough between single commands, which all need one +// slot, but not between a single command and a BATCH (pushBatch), which needs +// its whole length free at one instant. The writer frees at most a wave per +// take while singles refill any free slot, so under sustained load a batch +// longer than a wave would never see enough room. A batch that is refused +// therefore RESERVES its slots (hold): reserved slots count as occupied for +// every other push, so the writer's takes accumulate room for the batch +// instead of handing it straight back to the singles. Holders form a FIFO +// line and only the HEAD can be admitted (pushHeld): a later, smaller batch +// would otherwise refill the head's room exactly as the singles did. Each +// holder has its own wake channel and only the head is ever signalled, so no +// wake is shared, passed along, or lost between waiters. type fdQueue struct { mu sync.Mutex // buf[head:] is the live queue. takeInto advances head instead of shifting @@ -46,6 +60,20 @@ type fdQueue struct { head int max int // bound; mirrors the old channel capacity + // holders are the batches that were refused for lack of room and are + // waiting to be admitted, in arrival order (see hold/pushHeld). reserved + // is the sum of their sizes; it counts as occupied for push and for a + // non-holding pushBatch, so neither a single command nor a new batch can + // refill the room the writer frees ahead of a waiting batch. Only + // holders[0] can be admitted, and it needs just its own length free: the + // sum of reservations may exceed the capacity, so a rule that made the + // head wait for ALL reservations to fit could never be satisfied. Every + // event that may let the head in (a take, a drain, the previous head + // leaving) signals holders[0].wake; closeQueue signals every holder. + holders []fdHolder + reserved int + nextID uint64 + // parked/wakeAt are the writer's standing request: it is asleep and wants a // signal once the queue holds at least wakeAt entries. Only a submitter that // takes parked from true to false sends the signal, so a wave of arrivals @@ -54,7 +82,7 @@ type fdQueue struct { wakeAt int wake chan struct{} // cap 1: submitter -> parked writer - room chan struct{} // cap 1: writer -> submitters blocked on a full queue + room chan struct{} // cap 1: writer -> single submitters blocked on a full queue closed bool } @@ -69,6 +97,16 @@ var fdQueueReady = func() chan struct{} { return c }() +// fdHolder is a batch waiting for room with a reservation of n slots. The id +// identifies it to pushHeld and unhold. wake (cap 1) is signalled when the +// holder is at the head of the line and may fit, and on close. A holder behind +// the head is never woken by room events: nothing it could do with the wake. +type fdHolder struct { + id uint64 + n int + wake chan struct{} +} + // fdPushResult is the outcome of an enqueue attempt. type fdPushResult int @@ -168,8 +206,9 @@ func (q *fdQueue) reserve(k int) { } } -// signalRoom re-arms the cap-1 room signal. Called only by a submitter that just -// woke from a full queue, to chain the wake to the next one waiting. +// signalRoom re-arms the cap-1 room signal. Called by a submitter that just +// woke from a full queue, to chain the wake to the next one waiting, and by a +// holder whose reservation stops counting against the singles. func (q *fdQueue) signalRoom() { if q == nil { return @@ -180,6 +219,107 @@ func (q *fdQueue) signalRoom() { } } +// signalHolder wakes one holder. Non-blocking on its cap-1 channel, so a +// pending wake is not doubled; nil (no holder) is a no-op. Called OUTSIDE mu. +func signalHolder(ch chan struct{}) { + if ch == nil { + return + } + select { + case ch <- struct{}{}: + default: + } +} + +// headWakeLocked returns the wake channel of the head of the holder line, or +// nil when the line is empty. Caller holds mu; the send happens after unlock. +func (q *fdQueue) headWakeLocked() chan struct{} { + if len(q.holders) == 0 { + return nil + } + return q.holders[0].wake +} + +// signalRoomAndHead releases the waiters a take or drain may have made room +// for: one single on room, and the head of the holder line on its own wake. +// Called OUTSIDE mu with head read under it. +func (q *fdQueue) signalRoomAndHead(head chan struct{}) { + q.signalRoom() + signalHolder(head) +} + +// hold reserves k slots for a batch that was refused for lack of room and +// puts it at the back of the holder line. Until the holder is admitted +// (pushHeld) or gives up (unhold), the slots count as occupied for every other +// push, so the room the writer frees accumulates for the holders. k is bounded +// by the caller to the queue capacity. Returns the holder's id and the channel +// it must wait on; on a nil queue both are zero and the nil channel never +// fires (see the nil-queue contract). +func (q *fdQueue) hold(k int) (id uint64, wake <-chan struct{}) { + if q == nil || k <= 0 { + return 0, nil + } + ch := make(chan struct{}, 1) + q.mu.Lock() + q.nextID++ + id = q.nextID + q.holders = append(q.holders, fdHolder{id: id, n: k, wake: ch}) + q.reserved += k + q.mu.Unlock() + return id, ch +} + +// unhold releases the reservation of a holder that gave up (ctx done, +// shutdown). It wakes a single the reservation may have been holding back, +// and, if the holder was at the head of the line, the new head, which may fit. +// A holder behind the head leaves without a trace: it was never signalled, so +// no wake is lost with it. +func (q *fdQueue) unhold(id uint64) { + if q == nil { + return + } + q.mu.Lock() + wasHead, found := q.dropHolderLocked(id) + wakeSingles := found && q.live()+q.reserved < q.max + var head chan struct{} + if wasHead { + head = q.headWakeLocked() + } + q.mu.Unlock() + if wakeSingles { + q.signalRoom() + } + signalHolder(head) +} + +// dropHolderLocked removes the holder with id from the line and releases +// its slots. Caller holds mu. +func (q *fdQueue) dropHolderLocked(id uint64) (wasHead, found bool) { + for i, h := range q.holders { + if h.id != id { + continue + } + q.reserved -= h.n + q.holders = append(q.holders[:i], q.holders[i+1:]...) + if len(q.holders) == 0 { + q.holders = nil // let a burst's line be collected + } + return i == 0, true + } + return false, false +} + +// roomFor reports whether n more single commands fit, net of reservations. +func (q *fdQueue) roomFor(n int) bool { + if q == nil { + return false + } + q.mu.Lock() + ok := q.max-q.live()-q.reserved >= n + q.mu.Unlock() + return ok +} + // push enqueues one command. The wake signal is sent OUTSIDE the mutex: it is a // non-blocking send on a cap-1 channel, but it can call goready, and holding the // submit mutex across a scheduler operation would convoy every other submitter @@ -193,7 +333,7 @@ func (q *fdQueue) push(req fdReq) fdPushResult { q.mu.Unlock() return fdPushClosed } - if q.live() >= q.max { + if q.live()+q.reserved >= q.max { q.mu.Unlock() return fdPushFull } @@ -205,14 +345,97 @@ func (q *fdQueue) push(req fdReq) fdPushResult { } q.mu.Unlock() if signal { - select { - case q.wake <- struct{}{}: - default: - } + q.signalWake() } return fdPushOK } +// pushBatch enqueues reqs under a SINGLE lock, for callers that already hold a +// run of commands (a pipeline, or a carry tail being returned to the queue). +// All-or-nothing: a partial enqueue would split a pipeline across two flushes +// and, worse, leave the caller unsure which half it owns. Slots reserved by +// waiting holders count as occupied; a refused caller that wants to wait +// should hold its own slots and retry through pushHeld. +func (q *fdQueue) pushBatch(reqs []fdReq) fdPushResult { + if len(reqs) == 0 { + return fdPushOK + } + if q == nil { + return fdPushFull // see the nil-queue contract + } + q.mu.Lock() + if q.closed { + q.mu.Unlock() + return fdPushClosed + } + if q.live()+q.reserved+len(reqs) > q.max { + q.mu.Unlock() + return fdPushFull + } + signal := q.appendLocked(reqs) + q.mu.Unlock() + if signal { + q.signalWake() + } + return fdPushOK +} + +// pushHeld is pushBatch for the holder with id, whose reservation is +// len(reqs) slots. Only the head of the holder line is admitted; anyone else +// is refused (fdPushFull) and waits on its own wake, which fires once it has +// become the head and may fit. The head needs only its own length free: the +// slots behind it are reserved, not occupied, and counting them would let the +// line deadlock once the reservations outgrow the capacity. On admission the +// reservation is consumed, a single the reservation was holding back is woken +// if room remains, and the next holder is woken to try in turn. +func (q *fdQueue) pushHeld(reqs []fdReq, id uint64) fdPushResult { + if q == nil { + return fdPushFull // see the nil-queue contract + } + q.mu.Lock() + if q.closed { + q.mu.Unlock() + return fdPushClosed + } + if len(q.holders) == 0 || q.holders[0].id != id || q.live()+len(reqs) > q.max { + q.mu.Unlock() + return fdPushFull + } + signal := q.appendLocked(reqs) + q.dropHolderLocked(id) + wakeSingles := q.live()+q.reserved < q.max + head := q.headWakeLocked() + q.mu.Unlock() + if signal { + q.signalWake() + } + if wakeSingles { + q.signalRoom() + } + signalHolder(head) + return fdPushOK +} + +// appendLocked appends an admitted run and reports whether the parked writer +// must be woken. Caller holds mu and has checked the bound. +func (q *fdQueue) appendLocked(reqs []fdReq) (signal bool) { + q.reserve(len(reqs)) + q.buf = append(q.buf, reqs...) + signal = q.parked && q.live() >= q.wakeAt + if signal { + q.parked = false + } + return signal +} + +// signalWake sends the writer wake claimed under mu. Outside the lock: see push. +func (q *fdQueue) signalWake() { + select { + case q.wake <- struct{}{}: + default: + } +} + // pushFront returns commands the WRITER already took but could not write this // round (the MaxBatchBytes cap trips mid-wave) to the head of the queue, so they // go out first and FIFO order is exact. Writer-only: there is one writer, it @@ -244,6 +467,23 @@ func (q *fdQueue) pushFront(reqs []fdReq) { // The taken slots are cleared, because the backing array outlives them and a // stale fdReq pins a Cmder, a context and an apBatch. That is 104 bytes per slot // of memclr, against the per-command lock and copy the channel charged. +// headRun counts the leading queued requests for which match is true. With +// takeInto it lets the engine goroutine take the queued tail of a pipeline: +// only that goroutine removes from the head, so the run cannot change in +// between. +func (q *fdQueue) headRun(match func(fdReq) bool) int { + if q == nil { + return 0 + } + q.mu.Lock() + n := 0 + for q.head+n < len(q.buf) && match(q.buf[q.head+n]) { + n++ + } + q.mu.Unlock() + return n +} + func (q *fdQueue) takeInto(dst []fdReq, max int) []fdReq { if q == nil || max <= 0 { return dst @@ -267,13 +507,11 @@ func (q *fdQueue) takeInto(dst []fdReq, max int) []fdReq { q.compact() } q.parked = false // we are awake; stop submitters from signalling + head := q.headWakeLocked() q.mu.Unlock() - // Release anyone blocked on a full queue. Non-blocking on a cap-1 channel, so + // Release anyone blocked on a full queue. Non-blocking on cap-1 channels, so // this is a cheap no-op once a signal is already pending. - select { - case q.room <- struct{}{}: - default: - } + q.signalRoomAndHead(head) return dst } @@ -338,11 +576,9 @@ func (q *fdQueue) drainAll(dst []fdReq) []fdReq { clear(q.buf) q.resetEmpty() q.parked = false + head := q.headWakeLocked() q.mu.Unlock() - select { - case q.room <- struct{}{}: - default: - } + q.signalRoomAndHead(head) return dst } @@ -354,10 +590,17 @@ func (q *fdQueue) closeQueue() { } q.mu.Lock() q.closed = true + // Every holder, not just the head: a holder behind the head is otherwise + // woken only by its ctx, and Close must release it. + wakes := make([]chan struct{}, len(q.holders)) + for i, h := range q.holders { + wakes[i] = h.wake + } q.mu.Unlock() - // Release submitters blocked on a full queue so they observe the closed state. - select { - case q.room <- struct{}{}: - default: + // Release submitters blocked on a full queue, singles and holders alike, so + // they observe the closed state. + q.signalRoom() + for _, ch := range wakes { + signalHolder(ch) } } diff --git a/autopipeline_fdqueue_test.go b/autopipeline_fdqueue_test.go index 8709104f..b4693940 100644 --- a/autopipeline_fdqueue_test.go +++ b/autopipeline_fdqueue_test.go @@ -28,16 +28,19 @@ func TestFDQueueFIFOModel(t *testing.T) { } else if len(model) < 512 { t.Fatalf("seed %d: push refused at depth %d", seed, len(model)) } - case r < 6: // a burst of pushes - for k := 1 + rng.Intn(40); k > 0; k-- { - if q.push(fdQueueTag(next)) != fdPushOK { - if len(model) < 512 { - t.Fatalf("seed %d: push refused at depth %d", seed, len(model)) - } - break + case r < 6: // pushBatch + k := 1 + rng.Intn(40) + batch := make([]fdReq, k) + for i := range batch { + batch[i] = fdQueueTag(next + i) + } + if q.pushBatch(batch) == fdPushOK { + for i := 0; i < k; i++ { + model = append(model, next+i) } - model = append(model, next) - next++ + next += k + } else if len(model)+k <= 512 { + t.Fatalf("seed %d: pushBatch(%d) refused at depth %d", seed, k, len(model)) } default: // take, maybe give some back got := q.takeInto(nil, 1+rng.Intn(64)) @@ -87,8 +90,6 @@ func BenchmarkFDQueueDeepDrain(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { dst = q.takeInto(dst[:0], wave) - for _, r := range refill { - q.push(r) - } + q.pushBatch(refill) } } diff --git a/autopipeline_fullduplex.go b/autopipeline_fullduplex.go index 7d1a25f2..18cb265b 100644 --- a/autopipeline_fullduplex.go +++ b/autopipeline_fullduplex.go @@ -247,6 +247,13 @@ type fdReq struct { // never-sent NoRetry commands recovered from a dead connection's backlog, returning // an error for a command the server never saw. sent bool + // pipelined marks a command submitted as part of an FD pipeline batch + // (submitBatch). The reader settles its retryable replies inline instead of + // diverting them one by one: an individual retry would let later commands + // of the same pipeline run before it. fdPipelineExec retries the whole batch + // instead, the way an ordinary pipeline does. Next to sent, so it fits in + // existing padding and does not grow the struct. + pipelined bool // limReport is the Limiter obligation for the WRITTEN chunk this req closes: // non-nil ONLY on the LAST req of a chunk that Allow() admitted and writeBatch // then wrote cleanly. It rides the in-flight deque so the reply-side outcome @@ -910,8 +917,11 @@ func (fd *fdEngine) submit(ctx context.Context, cmd Cmder) *apBatch { case <-fd.q.roomCh(): // Chain the signal: room is cap-1, so with several submitters blocked only // one is released per take. Whoever wakes re-signals while space remains. - // Confined to this saturated path, so steady state pays nothing. - if fd.q.depth() < fd.q.capacity() { + // Confined to this saturated path, so steady state pays nothing. Room a + // waiting batch has reserved does not count: no single can use it, and + // passing the wake on for it would only spin this chain until the batch + // is admitted (each holder has its own wake channel). + if fd.q.roomFor(1) { fd.q.signalRoom() } case <-ctx.Done(): @@ -1378,7 +1388,9 @@ func (fd *fdEngine) run() { var exhausted []fdReq eligible, exhausted = fdPartitionByBudget(unacked, fd.retryBudget()) if len(exhausted) > 0 { - fd.failReqs(exhausted, aerr) // spent MaxRetries+1 attempts; fail with the real cause + // Spent MaxRetries+1 attempts; fail with the real cause, and with + // them the rest of their pipelines. + eligible = fd.failPipelines(exhausted, eligible, aerr) } // Split the eligible suffix at the first SENT NoRetry command: replay the // prefix and fail that command plus everything ordered after it (a NoRetry @@ -1392,12 +1404,12 @@ func (fd *fdEngine) run() { // and fall through to fail it — a command that may be a sent NoRetry must // never be re-sent when in doubt. if n, scanPanic := fdFirstNoRetrySafe(eligible); !scanPanic && n > 0 { + carry = eligible[:n] if n < len(eligible) { - fd.failReqs(eligible[n:], aerr) + carry = fd.failPipelines(eligible[n:], carry, aerr) } retryAttempts++ fd.sleepBackoff(retryAttempts) - carry = eligible[:n] // Already issued on the failed connection and about to be re-issued; // bump attempts so a later success/failure reports the real // retry_attempts (not always 1). The clean-recycle suffix path @@ -1405,6 +1417,12 @@ func (fd *fdEngine) run() { // replay is a first attempt. for i := range carry { carry[i].attempts++ + // Keep a pipelined command's batch in step, so the pipeline + // retry charges this replay even if the command then fails + // here instead of completing through the reader. + if carry[i].pipelined { + carry[i].batch.fdAttempts = carry[i].attempts + } } continue } @@ -1413,7 +1431,7 @@ func (fd *fdEngine) run() { // remaining unfailed commands, then ALWAYS back off before re-leasing so a // dead server cannot spin this loop. Keep the engine alive to serve new work // when it recovers. - fd.failReqs(eligible, aerr) + fd.failPipelines(eligible, nil, aerr) carry = nil retryAttempts++ fd.sleepBackoff(retryAttempts) @@ -1745,7 +1763,9 @@ func (fd *fdEngine) session(bg context.Context, cn *pool.Conn, carry []fdReq) (u } // A RETRYABLE execution error (not a redirect) may have produced a // partially consumed response, so it stays gated on NoRetry. - if !fdNoRetrySafe(req.cmd) { + // A pipelined command is never diverted alone (see + // fdReq.pipelined); its reply settles inline below. + if !req.pipelined && !fdNoRetrySafe(req.cmd) { // Cluster full-duplex: divert a retryable server reply // (LOADING/READONLY/TRYAGAIN/CLUSTERDOWN/MASTERDOWN/NOREPLICAS/ // max-clients) to the redirect-aware ClusterClient. It consults NO @@ -1802,7 +1822,14 @@ func (fd *fdEngine) session(bg context.Context, cn *pool.Conn, carry []fdReq) (u // here would reach the reader's session-failure recovery BEFORE this req // is advanced, so recovery would re-own the already-consumed reply and // replay it — a mutating command twice. - fd.reportReplyMetrics(octx, req, e, cn) + if req.pipelined { + // A pipelined command is measured with its batch, as in an + // ordinary pipeline (fdPipelineMetrics), not per command. + req.batch.fdAttempts = req.attempts // published by complete() + req.batch.fdConn = cn + } else { + fd.reportReplyMetrics(octx, req, e, cn) + } req.complete() // wake the caller, or hand off to the hook host done++ } @@ -2851,6 +2878,48 @@ func classifyCommandErrorGuarded(err error) (errorType, statusCode string, isInt return classifyCommandError(err) } +// failPipelines fails failed with err, and with it the rest of every FD +// pipeline that has a request in failed: its members in keep and its tail +// still at the head of the queue (the writer may have taken only a prefix). +// A pipeline fails as one, as on a dedicated connection: once part of it has +// failed, the rest must not run. Returns keep without those members. +func (fd *fdEngine) failPipelines(failed, keep []fdReq, err error) []fdReq { + var groups map[*apBatch]struct{} + for _, r := range failed { + if r.pipelined && r.batch != nil && r.batch.fdGroup != nil { + if groups == nil { + groups = make(map[*apBatch]struct{}) + } + groups[r.batch.fdGroup] = struct{}{} + } + } + if groups == nil { + fd.failReqs(failed, err) + return keep + } + failed = failed[:len(failed):len(failed)] // may be a sub-slice of keep's array: append copies + inGroup := func(r fdReq) bool { + if !r.pipelined || r.batch == nil { + return false + } + _, ok := groups[r.batch.fdGroup] + return ok + } + out := make([]fdReq, 0, len(keep)) + for _, r := range keep { + if inGroup(r) { + failed = append(failed, r) + } else { + out = append(out, r) + } + } + if n := fd.q.headRun(inGroup); n > 0 { + failed = fd.q.takeInto(failed, n) + } + fd.failReqs(failed, err) + return out +} + func (fd *fdEngine) failReqs(reqs []fdReq, err error) { // Error-metric parity: commands terminated here (lease failure, retry // exhaustion, a NoRetry tail, Close) never reach the reader's inline @@ -2874,7 +2943,9 @@ func (fd *fdEngine) failReqs(reqs []fdReq, err error) { // reqs unsettled. fdSetErrSafe(reqs[i].cmd, err) } - if errorCallback != nil { + // A pipelined command's failure is reported once for its batch + // (fdPipelineMetrics), as an ordinary pipeline does. + if errorCallback != nil && !reqs[i].pipelined { octx := reqs[i].ctx if octx == nil { octx = context.Background() @@ -2924,6 +2995,21 @@ func (fd *fdEngine) shutdownFlush(bg context.Context, carry []fdReq) { // Drain and close the queue now (before flushing carry) so no new submit lands // mid-flush. fresh commands (attempts == 1) have not run yet. fresh := fd.takeQueue() + // The writer may have taken only a prefix of an FD pipeline before the + // session failed: the prefix is then at the end of carry and the tail at + // the head of fresh. Move the tail into carry so the pipeline flushes as + // one, with one retry budget. + if n := len(carry); n > 0 && len(fresh) > 0 && fdSameGroup(carry[n-1], fresh[0]) { + k := 1 + for k < len(fresh) && fdSameGroup(fresh[k-1], fresh[k]) { + k++ + } + carry = append(carry[:n:n], fresh[:k]...) // copy: carry may share a backing array + fresh = fresh[k:] + } + // The flush below runs pipelined commands through the pooled pipeline with + // its own retry loop; flushReqs counts that as a further issue on each batch + // it runs, so fdPipelineExec does not run them again. // Flush the carried tail honoring EACH command's remaining retry budget across // the Close boundary (flushCarryBudgeted), then the fresh queue at the full // budget. Flushing carry first keeps FIFO order across the two sets. @@ -2939,6 +3025,12 @@ func (fd *fdEngine) shutdownFlush(bg context.Context, carry []fdReq) { _ = fd.flushReqs(bg, fresh, fd.retryBudget()) } +// fdSameGroup reports whether b follows a in the same FD pipeline batch. +func fdSameGroup(a, b fdReq) bool { + return a.pipelined && b.pipelined && a.batch != nil && b.batch != nil && + a.batch.fdGroup != nil && a.batch.fdGroup == b.batch.fdGroup +} + // fdCarryRemainingRetries returns the retry bound for a carried command flushed on // Close. carry commands are POST-BUMP: run() bumps attempts before re-issuing, so a // command carried at attempts=A has completed A-1 executions (contrast the pre-bump @@ -2967,12 +3059,30 @@ func (fd *fdEngine) flushCarryBudgeted(bg context.Context, carry []fdReq) error for i := 0; i < len(carry); { a := carry[i].attempts j := i + 1 - for j < len(carry) && carry[j].attempts == a { - j++ + if carry[i].pipelined { + // One FD pipeline batch, even if its commands carry different + // attempt counts, so flushReqs can run it as one pipeline. + for j < len(carry) && fdSameGroup(carry[j-1], carry[j]) { + j++ + } + } else { + // A run of other commands with the same attempt count. It stops at + // a pipeline batch, so no command shares a pipeline's budget. + for j < len(carry) && !carry[j].pipelined && carry[j].attempts == a { + j++ + } } group := carry[i:j] i = j + // The group runs as one pipeline, so it gets the smallest remaining + // budget of its commands: none may run past its MaxRetries+1 + // executions. A command with more budget left may therefore get fewer + // retries than it alone would, as in an ordinary pipeline, whose retry + // budget covers the whole batch. rem := fdCarryRemainingRetries(a, mr) + for k := range group { + rem = min(rem, fdCarryRemainingRetries(group[k].attempts, mr)) + } if rem < 0 { fd.failReqs(group, errFDRetryBudgetExhausted) // budget spent; do not re-run continue @@ -3033,23 +3143,51 @@ func (fd *fdEngine) flushReqs(bg context.Context, reqs []fdReq, maxRetries int) byteLimit := int64(fd.ap.config.MaxBatchBytes) // 0 = disabled for i < len(reqs) { end := fdBatchEnd(reqs, i, fd.maxBatch, byteLimit) - // Do not mix retry policies in one chunk: generalProcessPipeline disables - // retries for the WHOLE chunk if any command is NoRetry (cmdsContainNoRetry). - // That would strip retryable commands in the same accepted backlog of their - // budget. Break the chunk at the first NoRetry-policy change so a NoRetry - // command (e.g. RawWriteToCmd) is isolated from its retryable neighbors, like - // the half-duplex dispatcher's contiguous retry-policy runs. The clamp starts - // at i+1, so end stays > i and the chunk is never empty (no infinite loop). - policy := reqs[i].cmd.NoRetry() - for k := i + 1; k < end; k++ { - if reqs[k].cmd.NoRetry() != policy { - end = k - break + if reqs[i].pipelined { + // An FD pipeline batch flushes as ONE pipeline, whatever its size, as + // Pipeline.Exec sends it: its first command's retryable reply then + // retries the whole batch together, not only the first chunk. + end = i + 1 + for end < len(reqs) && fdSameGroup(reqs[end-1], reqs[end]) { + end++ + } + } else { + // A chunk of ordinary commands stops where a pipeline batch starts, + // so the pipeline is not split across chunks. + for k := i + 1; k < end; k++ { + if reqs[k].pipelined { + end = k + break + } + } + // Do not mix retry policies in one chunk: generalProcessPipeline + // disables retries for the WHOLE chunk if any command is NoRetry + // (cmdsContainNoRetry). That would strip retryable commands in the same + // accepted backlog of their budget. Break the chunk at the first + // NoRetry-policy change so a NoRetry command (e.g. RawWriteToCmd) is + // isolated from its retryable neighbors, like the half-duplex + // dispatcher's contiguous retry-policy runs. The clamp starts at i+1, + // so end stays > i and the chunk is never empty (no infinite loop). + policy := reqs[i].cmd.NoRetry() + for k := i + 1; k < end; k++ { + if reqs[k].cmd.NoRetry() != policy { + end = k + break + } } } cmds := make([]Cmder, end-i) for j := i; j < end; j++ { cmds[j-i] = reqs[j].cmd + // The pooled pipeline below runs this chunk again and records its + // pipeline metric. Stamped here, where the chunk runs: one more + // issue (so fdPipelineExec does not re-run it), and flushed. A batch + // that never runs (a dead endpoint fails the rest) keeps its issue + // count and stays unmarked, so fdPipelineExec records its failure. + if reqs[j].pipelined && reqs[j].batch != nil { + reqs[j].batch.fdAttempts = reqs[j].attempts + 1 + reqs[j].batch.fdFlushed = true + } } // Initialize the flush with a request's own context (cancellation removed), not // the engine's background context: if this flush initializes a fresh pooled @@ -3104,7 +3242,9 @@ func (fd *fdEngine) failQueue(err error) { // fdSetErrSafe: same hazard as failReqs — a panicking custom Cmder must // not escape on the sole fd.run goroutine with no outer recover. fdSetErrSafe(r.cmd, err) - if errorCallback != nil { + // A pipelined command's failure is reported once for its batch + // (fdPipelineMetrics), as in failReqs. + if errorCallback != nil && !r.pipelined { if !classified { errorType, statusCode, isInternal = classifyCommandErrorGuarded(err) classified = true diff --git a/redis.go b/redis.go index d7bd36cd..42e186a1 100644 --- a/redis.go +++ b/redis.go @@ -2054,6 +2054,26 @@ func pipelineErrShouldStamp(err error) bool { func (c *baseClient) generalProcessPipeline( ctx context.Context, cmds []Cmder, p pipelineProcessor, operationName string, maxRetries int, +) error { + return c.generalProcessPipelineFrom(ctx, cmds, p, operationName, maxRetries, time.Time{}, 0) +} + +// processPipelineRetriesAfter is processPipelineRetries continuing an operation +// that already ran: start and priorAttempts carry over into the one pipeline +// metric, so a full-duplex attempt re-run on the pooled path is measured as a +// single operation, the way an ordinary pipeline's own retries are. +func (c *baseClient) processPipelineRetriesAfter(ctx context.Context, cmds []Cmder, maxRetries int, start time.Time, priorAttempts int) error { + if err := c.generalProcessPipelineFrom(ctx, cmds, c.pipelineProcessCmds, "PIPELINE", maxRetries, start, priorAttempts); err != nil { + return err + } + return cmdsFirstErr(cmds) +} + +// generalProcessPipelineFrom is generalProcessPipeline with its measurement +// started at start (zero: now) and priorAttempts already spent. +func (c *baseClient) generalProcessPipelineFrom( + ctx context.Context, cmds []Cmder, p pipelineProcessor, operationName string, maxRetries int, + start time.Time, priorAttempts int, ) error { // Pipeline commands never pass through process, so apply the same CSC state // guard here. initConn's internal client is exempt. @@ -2067,16 +2087,29 @@ func (c *baseClient) generalProcessPipeline( var operationStart time.Time pipelineOpDurationCallback := otel.GetPipelineOperationDurationCallback() if pipelineOpDurationCallback != nil { - operationStart = time.Now() + operationStart = start + if operationStart.IsZero() { + if priorAttempts > 0 { + // A continued operation (a full-duplex attempt) that began + // with no duration callback: its start is unknown, so record + // no duration, as the FD path does. + pipelineOpDurationCallback = nil + } else { + operationStart = time.Now() + } + } } var lastConn *pool.Conn - totalAttempts := 0 + totalAttempts := priorAttempts var lastErr error for attempt := 0; attempt <= maxRetries; attempt++ { totalAttempts++ if attempt > 0 { - if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil { + // Continue the backoff of an operation that already ran + // priorAttempts times (a full-duplex attempt), as its own retries + // would: 0 for an ordinary pipeline. + if err := internal.Sleep(ctx, c.retryBackoff(priorAttempts+attempt)); err != nil { setCmdsErr(cmds, err) if pipelineOpDurationCallback != nil { operationDuration := time.Since(operationStart)