1
0
mirror of https://github.com/redis/go-redis.git synced 2026-10-22 19:49:00 +03:00

feat(autopipeline): run Pipeline on the full-duplex connection (#4021)

* feat(autopipeline): add fdQueue.pushBatch for FD batches

FDPipelined submits a pipeline as one contiguous run; pushBatch enqueues
it under a single lock, all or nothing.

* feat(autopipeline): run Pipeline on the full-duplex connection

ap.Pipeline() delegated to the client's pipeliner, so a pipeline built
from a full-duplex autopipeliner quietly left the held connection, took a
pooled one per Exec, and lost the cross-caller coalescing the engine
exists for. It now submits the batch to the engine as one contiguous run.

FDPipelined does the submission: the batch is admitted all-or-nothing, so
its commands occupy consecutive in-flight slots and their replies come
back in submit order, which keeps a pipeline's internal ordering exactly
as a dedicated connection would. Batches that cannot ride the wire -- a
blocking command, a per-command read timeout, anything the divert
predicate claims, or a batch larger than the queue -- fall back to the
ordinary pipeline path.

Measured on a 14-core client against a local server, 10 commands per
Exec, 3 passes:

  callers   pooled       on the engine   active sockets
      256   586,557        871,469       149 -> 1
     1024   516,042        785,191       150 -> 1

So +48.6% and +52.2%, on one socket instead of 149.

Above ~2048 callers a single engine saturates its submit queue and
admission serializes; spreading pipelines across several auto-pipeliners
avoids that (1.6M ops/s across eight), which is an argument for a
supported multi-connection full-duplex mode rather than for this commit.

TxPipeline is deliberately untouched: MULTI/EXEC needs connection
affinity for the whole transaction, which the engine does not offer.

* fix(autopipeline): keep FD pipeline Exec semantics

Pipeline() on a full-duplex autopipeliner runs the batch on the held
connection. Three things differed from an ordinary pipeline:

- A retryable reply (LOADING and the like) was retried per command off
  the reader, so later commands of the pipeline could run before it.
  Pipelined commands now settle retryable replies inline, and a
  retryable reply on the first command re-runs the whole batch the
  ordinary way, as processPipeline does.
- With ContextTimeoutEnabled the caller's deadline did not bound the
  wait. Such clients keep the pooled pipeline path.
- Exec dropped redis.Nil. It now returns the first error, Nil included.

FDPipelined is documented as standalone only.

* fix(autopipeline): close FD pipeline Exec gaps

Pipeline() on a full-duplex autopipeliner still differed from an
ordinary pipeline:

- Hooks, a Limiter or an OTel recorder: an ordinary pipeline runs the
  ProcessPipelineHook chain, takes one Limiter decision per batch and
  records one pipeline duration. The FD path did none of that, so such
  clients now keep the pooled pipeline. Running the FD path under these
  is a follow-up.
- The whole-batch retry on a retryable first reply re-ran pipelines
  holding a NoRetry command. It now stops on cmdsContainNoRetry, as
  generalProcessPipeline does.
- That retry also restarted the full MaxRetries budget, so MaxRetries:1
  allowed three executions. The FD attempt now counts as the first: the
  re-run gets MaxRetries-1 retries, after one backoff.
- A canceled ctx was only checked while waiting for queue room, so a
  pre-canceled Exec still ran. submitBatch now refuses a done ctx before
  admission.

* fix(autopipeline): count FD replays in the retry budget

The whole-batch retry assumed the FD engine issued the pipeline once.
After a connection-error replay it had issued it twice, so a lost first
write, LOADING on the replay and the pooled re-run made three executions
under MaxRetries:1, where an ordinary pipeline stops at two.

The reader now stamps each pipelined command's attempt count on its
batch before completing it, fdPipelined reports the highest, and the
re-run gets MaxRetries minus that, or none when it is spent.

* fix(autopipeline): keep FD pipelines off cluster children

A cluster node's FD child is the node client's cached async
autopipeliner, so ForEachMaster hands it to users, and its engine is
redirect-aware: it follows MOVED/ASK by re-running one command off the
pipe. A Pipeline() on it would let the rest of the batch run ahead of a
redirected command. FDPipelined now refuses redirect-aware engines, so
such a pipeline keeps the node client's ordinary pipeline, as before.

* fix(autopipeline): keep NoRetry batches off the FD pipe

An ordinary pipeline holding a NoRetry command retries nothing after a
connection error. The FD tail recovery replays per command, up to the
first sent NoRetry one, so a pipeline like SET k; RawWriteTo(GET x)
that lost its connection ran the SET twice. FDPipelined now refuses a
batch with a NoRetry command, and Pipeline() runs it the ordinary way.

The pipeline retry budget read fdAttempts, which only the reader
stamped. A command that was replayed and then failed carried no count,
so a pooled re-run could exceed MaxRetries. The batch is stamped at
submit and on every replay.

* fix(autopipeline): run pipeline hooks and metrics on FD

Pipeline() and FDPipelined kept away from the full-duplex path whenever
hooks, a Limiter or an OTel recorder were set, or bypassed them. They
now run on it:

- The batch runs inside the client's ProcessPipelineHook chain, with
  fdPipelineExec as its terminal where processPipeline sits in an
  ordinary pipeline. Fallback and the whole-batch retry happen inside
  the chain, so hooks run once. Batched commands get no per-command
  ProcessHook, as in an ordinary pipeline.
- An FD batch records one pipeline duration and, on failure, one
  pipeline-level error; the reader skips its per-command metrics for
  batched commands.
- A Limiter stays asked per write chunk, which can partly admit a batch
  longer than MaxBatchSize. Documented on FDPipelined.

Only ContextTimeoutEnabled still keeps the pooled pipeline.

* fix(autopipeline): no FD pipeline re-run after a replay

The whole-batch retry re-ran a pipeline whose first command got a
retryable reply even when the engine had already replayed its unread
tail after a connection error, so a command in that tail could run
three times where an ordinary pipeline runs it twice. It now re-runs
only a batch the engine issued once. The Close-time flush stamps the
batch as issued again, so a flushed batch is not re-run either.

The re-run continues the FD attempt's measurement (start and attempts)
through processPipelineRetriesAfter, so one pipeline metric covers the
whole operation, and a ctx that ends in the backoff still records it.

* fix(autopipeline): keep flushed FD pipelines whole

When Close wins against an admitted FD pipeline, shutdownFlush sends
its commands through flushReqs. That split them at MaxBatchSize and at
attempt boundaries, so a retryable first reply retried only the first
chunk. Every batch of one FD pipeline now points at its first batch
(fdGroup), and the flush keeps that run in one chunk, so the pooled
retry re-runs the pipeline as a whole.

The pooled flush already records the pipeline metric. fdPipelined now
reports that the flush ran the batch, and FDPipelined/fdPipelineExec
skip their own metric in that case. failQueue skips the per-command
error for pipelined commands, as failReqs does, since fdPipelineExec
reports the batch once.

* fix(autopipeline): FD batch errors and queue wait

An FD pipeline now returns a transport or protocol failure before any
Redis error, as processPipeline does, so a caller that checks only
Exec's error sees that later commands have no known result.

The whole-batch re-run for a retryable first reply is also ruled out
when any command in the batch failed that way. The engine does not
replay such a command, since it may already have run, so the re-run
could execute it twice; an ordinary pipeline does not retry there.

A batch waiting for more room than one writer take freed put the room
signal back, took it straight back, failed to fit, and spun holding
submitMu.RLock. It now passes a wake it cannot use on once per take,
which keeps the wake chain for smaller waiters without spinning.

* fix(autopipeline): FD pipeline metrics and backoff

The pooled re-run after an FD attempt restarted its retry backoff at the
first step. generalProcessPipelineFrom now continues it from the prior
attempts, so a pipeline sleeps the same steps an ordinary one does.

A pipeline refused before admission (ctx done, or closed) recorded no
pipeline metric; an ordinary pipeline records one when withPipelineConn
fails. It now records one with a single attempt.

The Close flush marked every batch flushed before running any, so a
batch it never ran (a dead endpoint fails the rest) recorded nothing.
It now marks a batch where its chunk runs. And the FD metric is skipped
only when the flush ran the whole pipeline: when it ran only an unread
tail, the FD metric still covers the whole operation.

* fix(autopipeline): skip FD duration with no start

The start time is read only when a pipeline duration callback exists.
A recorder installed while an FD pipeline ran found a zero start, and
fdPipelineMetrics reported time.Since(time.Time{}), a duration of
centuries. It now records no duration for that operation, as an
ordinary pipeline does, which reads the callback once at the start.

Also pin that CSC state commands (SELECT, AUTH, HELLO with a protocol,
RESET, CLIENT TRACKING) never ride the FD pipe: they leave it as
runsOutsidePipeline commands, so Pipeline() falls back to the ordinary
pipeline and its CSC guard, and FDPipelined refuses the batch.

* fix(autopipeline): Close flush budget and retry timing

An FD pipeline flushed on Close can carry commands with different
attempt counts, and runs as one pipeline. It took the retry budget of
its first command, so a later command with more attempts could run past
its MaxRetries+1 executions. It now gets the smallest budget of its
commands, and fails with errFDRetryBudgetExhausted when any has none
left. A command with more budget may get fewer retries than it alone
would, as in an ordinary pipeline, whose budget covers the whole batch.

The pooled re-run of an FD operation that began with no duration
callback (zero start) started its clock at the re-run when a recorder
was installed in between, so the duration covered only the tail while
the count included the FD attempt. generalProcessPipelineFrom now
records no duration for a continued operation with no start.

* fix(autopipeline): keep flush budget groups separate

flushCarryBudgeted grew a group while the next command had the same
attempt count or belonged to the same FD pipeline, so a command in front
of a pipeline joined that pipeline's budget group. The group's smallest
budget then applied to it, and a pipeline command with no budget left
failed the unrelated command too. A group is now one FD pipeline, or a
run of other commands with the same attempt count, never both.

* fix(autopipeline): stamp flush issues where they run

shutdownFlush stamped every queued pipelined batch with one more issue
before either flush ran. When the carry flush hit a dead endpoint the
fresh queue was only failed, never issued, yet its callers saw an issue
count of 2 and reported a retry that never happened. flushReqs now
stamps the issue, with fdFlushed, where it runs a chunk. A batch that
never runs keeps its count; the transport error it is failed with
already rules out a re-run in fdPipelineExec.

* fix(autopipeline): flush a split FD pipeline whole

When the writer took only a prefix of an FD pipeline before the session
failed, Close found the prefix in the carry and the tail still queued,
and flushed them as two pooled pipelines. A retryable first reply then
retried only the prefix, and the tail ran under a full budget of its
own. shutdownFlush now moves the queued tail into the carry, so the
pipeline runs as one at the smallest remaining budget of its commands.

* docs(autopipeline): note lease retries in FD metrics

A full-duplex pipeline's attempts count the times the batch was issued.
Connection-lease retries belong to the engine: a batch queued during
them waits and is issued once, so they are not counted, where an
ordinary pipeline counts each failed connection acquisition. Say so on
fdPipelineMetrics.

* fix(autopipeline): reserve queue room for FD batches (#4057)

* fix(autopipeline): reserve queue room for FD batches

* fix(autopipeline): FD batch head-of-line starvation

* fix(autopipeline): remove unused function

* fix(autopipeline): fail a split FD pipeline whole

Live connection-error recovery partitions the unacked tail per command
(budget spent, after a sent NoRetry, not replayable) and fails part of
it. When the writer had taken only a prefix of an FD pipeline, the rest
of that pipeline, kept for replay or still queued, ran on the next
connection, and Exec reported a transport error for a pipeline whose
later commands executed. A pipeline on a dedicated connection sends
nothing more once it has failed. failPipelines now fails the whole
pipeline: its members kept for replay and its tail at the queue head.

---------

Co-authored-by: Vladislav Kotsev <vladygk8@gmail.com>
This commit is contained in:
Nedyalko Dyakov
2026-10-02 16:15:05 +03:00
committed by GitHub
parent 01ba8170c4
commit 349bcf47a9
17 changed files with 3032 additions and 62 deletions

View File

@@ -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.

View File

@@ -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())
}
}

View File

@@ -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)
}
}
}

View File

@@ -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())
}
}
}

405
autopipeline_fd_pipeline.go Normal file
View File

@@ -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")

View File

@@ -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)
}
}

View File

@@ -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)
}
}

View File

@@ -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())
}
}

View File

@@ -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)
}
}

View File

@@ -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")
}
}

View File

@@ -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")
}
}

View File

@@ -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)
}
}

View File

@@ -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)
}
}

View File

@@ -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)
}
}

View File

@@ -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)
}
}

View File

@@ -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

View File

@@ -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)