From 1622eca36e01f90a0a30acce63823a8ac745d1cb Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 12:59:51 +0100 Subject: [PATCH 1/2] fix(pipeline): correct per-command byte/label accounting and stop --pipeline >1 panic Buffer each pipelined command as a pendingCmd carrying its own send time, sent-byte count, reply receiver, and labels, instead of threading parallel scalar slices through the flush. This fixes several defects on the --pipeline N>1 path (default pipeline=1 was already correct): - Panic: the old code kept a single `replies` slice that connectionProcessor never grew, while cmds/times accumulated to `pipeline`; the flush indexed `replies[pos]` out of range -> "index out of range [1] with length 1", crashing the worker. --pipeline >1 was unusable. Now each command owns its receiver. - Per-command TxBytes: the flush recorded one scalar txBytesCount for every command; now each records its own sent bytes. - Per-command labels: cmdType/cmdQueryId are now per-command, so a mixed pipeline no longer files every command under the flushing command's label. - Reply capture: sendFlatCmd now passes a *interface{} receiver (was a nil interface, so replies were discarded and RxBytes was always 0). getRxLen sizes the captured reply (string/[]byte/[]interface{}/int64, deref *interface{}). - Fix READ_CURSOR label: the histogram switch matched "CURSOR_READ" but the code emits "READ_CURSOR", so cursor reads were dropped from readCursorHistogram. Validated E2E against redis: master crashes at --pipeline 4 (index out of range, 8/2000 keys); this branch completes (2000/2000 keys) with TxBytes=135780 and RxBytes=2000. At pipeline=1 the Tx/ops numbers are byte-identical to master. Closes #113 --- benchmark_runner/benchmark_runner.go | 2 +- cmd/ftsb_redisearch/cmd_processor.go | 178 +++++++++++++++---------- cmd/ftsb_redisearch/send_stats_test.go | 85 +++++++++--- 3 files changed, 172 insertions(+), 93 deletions(-) diff --git a/benchmark_runner/benchmark_runner.go b/benchmark_runner/benchmark_runner.go index 3e1a388..409ba1c 100644 --- a/benchmark_runner/benchmark_runner.go +++ b/benchmark_runner/benchmark_runner.go @@ -559,7 +559,7 @@ func (l *BenchmarkRunner) work(b Benchmark, wg *sync.WaitGroup, c *duplexChannel _ = l.inst_readHistogram.RecordValue(int64(cmdStat.Latency())) break - case "CURSOR_READ": + case "READ_CURSOR": _ = l.readCursorHistogram.RecordValue(int64(cmdStat.Latency())) _ = l.inst_readCursorHistogram.RecordValue(int64(cmdStat.Latency())) diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index 4336f95..fa71aa9 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -147,22 +147,18 @@ func (p *processor) Init(workerNumber int, _ bool, totalWorkers int) { } func connectionProcessor(p *processor, rateLimiter *rate.Limiter, useRateLimiter bool) { - cmdSlots := make([][]radix.CmdAction, 0, 0) - timesSlots := make([][]time.Time, 0, 0) - replies := make([]interface{}, 0, 0) + pendingSlots := make([][]pendingCmd, 0, 0) clusterSlots := make([][2]uint16, 0, 0) clusterAddr := make([]string, 0, 0) clusterAddrLen := 0 slotP := 0 if !clusterMode { - cmdSlots = append(cmdSlots, make([]radix.CmdAction, 0, 0)) - timesSlots = append(timesSlots, make([]time.Time, 0, 0)) + pendingSlots = append(pendingSlots, make([]pendingCmd, 0, 0)) } else { for _, ClusterNode := range p.clusterTopo { for _, slot := range ClusterNode.Slots { clusterSlots = append(clusterSlots, slot) - cmdSlots = append(cmdSlots, make([]radix.CmdAction, 0, 0)) - timesSlots = append(timesSlots, make([]time.Time, 0, 0)) + pendingSlots = append(pendingSlots, make([]pendingCmd, 0, 0)) clusterAddr = append(clusterAddr, ClusterNode.Addr) } } @@ -209,7 +205,7 @@ func connectionProcessor(p *processor, rateLimiter *rate.Limiter, useRateLimiter } if !clusterMode { var hadError bool - cmdSlots[slotP], timesSlots[slotP], hadError = sendFlatCmd(p, p.vanillaClient, cmdType, cmdQueryId, cmd, docFields, bytelen, cmdSlots[slotP], replies, timesSlots[slotP]) + pendingSlots[slotP], hadError = sendFlatCmd(p, p.vanillaClient, cmdType, cmdQueryId, cmd, docFields, bytelen, pendingSlots[slotP]) if hadError && continueOnErr { // Reconnect to get a fresh connection after an error. // This prevents hanging on a broken/half-closed connection @@ -218,101 +214,135 @@ func connectionProcessor(p *processor, rateLimiter *rate.Limiter, useRateLimiter } } else { client, _ := p.vanillaCluster.Client(clusterAddr[slotP]) - cmdSlots[slotP], timesSlots[slotP], _ = sendFlatCmd(p, client, cmdType, cmdQueryId, cmd, docFields, bytelen, cmdSlots[slotP], replies, timesSlots[slotP]) + pendingSlots[slotP], _ = sendFlatCmd(p, client, cmdType, cmdQueryId, cmd, docFields, bytelen, pendingSlots[slotP]) } } p.wg.Done() } +// getRxLen approximates the reply bytes received for a command by sizing the +// value radix unmarshalled into the receiver: bulk strings (string/[]byte), +// arrays ([]interface{} / []string, summed recursively), and integers (decimal +// digit count). A *interface{} receiver is dereferenced; nil/unknown → 0. func getRxLen(v interface{}) (res uint64) { - res = 0 switch x := v.(type) { + case *interface{}: + if x != nil { + res = getRxLen(*x) + } + case string: + res = uint64(len(x)) + case []byte: + res = uint64(len(x)) case []string: for _, i := range x { res += uint64(len(i)) } - case string: - res += uint64(len(x)) - default: - res = 0 + case []interface{}: + for _, e := range x { + res += getRxLen(e) + } + case int64: + res = uint64(len(strconv.FormatInt(x, 10))) } return } -func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd string, docfields []string, txBytesCount uint64, cmds []radix.CmdAction, replies []interface{}, times []time.Time) ([]radix.CmdAction, []time.Time, bool) { - var err error = nil - var rcv interface{} - rxBytesCount := uint64(0) - var radixFlatCmd = radix.Cmd(rcv, cmd, docfields...) - cmds = append(cmds, radixFlatCmd) - replies = append(replies, rcv) - start := time.Now() - times = append(times, start) +// pendingCmd is a single command buffered for the current pipeline window, +// carrying everything needed to record its own stat at flush time. Buffering +// per command (instead of threading parallel scalar values through the flush) +// is what makes pipelined accounting correct: every command keeps its own send +// time, sent-byte count, reply receiver, and labels, rather than inheriting the +// flushing command's values. +type pendingCmd struct { + action radix.CmdAction + reply *interface{} + cmdType string + cmdQueryId string + redisCmd string + redisKey string + txBytes uint64 + start time.Time +} + +func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd string, docfields []string, txBytesCount uint64, pending []pendingCmd) ([]pendingCmd, bool) { + reply := new(interface{}) key := "" if len(docfields) > 0 { key = docfields[0] } - cmds, times, hadError := sendIfRequired(p, client, cmdType, cmdQueryId, cmd, key, cmds, err, times, rxBytesCount, replies, txBytesCount) - return cmds, times, hadError + pending = append(pending, pendingCmd{ + action: radix.Cmd(reply, cmd, docfields...), + reply: reply, + cmdType: cmdType, + cmdQueryId: cmdQueryId, + redisCmd: cmd, + redisKey: key, + txBytes: txBytesCount, + start: time.Now(), + }) + return sendIfRequired(p, client, pending) } -func sendIfRequired(p *processor, client radix.Client, cmdType string, cmdQueryId string, redisCmd string, redisKey string, cmds []radix.CmdAction, err error, times []time.Time, rxBytesCount uint64, replies []interface{}, txBytesCount uint64) ([]radix.CmdAction, []time.Time, bool) { - cmdLen := len(cmds) +// sendIfRequired flushes the buffered pipeline window once it reaches `pipeline` +// commands, records one stat per command (each with its own latency, sent/received +// bytes, and labels), and returns the emptied buffer. Returning `pending[:0]` +// reuses the backing array across windows to avoid churn on the hot path. +func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([]pendingCmd, bool) { hadError := false - if cmdLen >= pipeline { - if cmdLen == 1 { - // if pipeline is 1 no need to pipeline - err = client.Do(cmds[0]) - } else { - err = client.Do(radix.Pipeline(cmds...)) + if len(pending) < pipeline { + return pending, hadError + } + + var err error + if len(pending) == 1 { + // if pipeline is 1 no need to pipeline + err = client.Do(pending[0].action) + } else { + actions := make([]radix.CmdAction, len(pending)) + for i := range pending { + actions[i] = pending[i].action } - endT := time.Now() - isTimeout := false - if err != nil { - hadError = true - - // Always log the error - // For read commands, log full command details; for writes, log only a summary to avoid huge log lines - if cmdType == "READ" || cmdType == "READ_CURSOR" { - if continueOnErr { - log.Println(fmt.Sprintf("Received an error with the following command(s): %v, error: %v", cmds, err)) - } else { - log.Fatal(fmt.Sprintf("Fatal error with the following command(s): %v, error: %v", cmds, err)) - } + err = client.Do(radix.Pipeline(actions...)) + } + endT := time.Now() + isTimeout := false + if err != nil { + hadError = true + // A flush may mix command types; use the first as the summary label. + rep := pending[0] + if rep.cmdType == "READ" || rep.cmdType == "READ_CURSOR" { + if continueOnErr { + log.Printf("Received an error with %d command(s) in pipeline, error: %v", len(pending), err) } else { - if continueOnErr { - log.Println(fmt.Sprintf("Received an error with %s command: %s %s (%d command(s) in pipeline), error: %v", cmdType, redisCmd, redisKey, len(cmds), err)) - } else { - log.Fatal(fmt.Sprintf("Fatal error with %s command: %s %s (%d command(s) in pipeline), error: %v", cmdType, redisCmd, redisKey, len(cmds), err)) - } + log.Fatalf("Fatal error with %d command(s) in pipeline, error: %v", len(pending), err) } - - // Log additional timeout-specific message if it's a timeout - if strings.Contains(err.Error(), "i/o timeout") { - isTimeout = true - if cmdType == "READ" || cmdType == "READ_CURSOR" { - log.Println(fmt.Sprintf("Timeout occurred with the following command(s): %v, continuing execution...", cmds)) - } else { - log.Println(fmt.Sprintf("Timeout occurred with %s command: %s %s (%d command(s) in pipeline), continuing execution...", cmdType, redisCmd, redisKey, len(cmds))) - } + } else { + if continueOnErr { + log.Printf("Received an error with %s command: %s %s (%d command(s) in pipeline), error: %v", rep.cmdType, rep.redisCmd, rep.redisKey, len(pending), err) + } else { + log.Fatalf("Fatal error with %s command: %s %s (%d command(s) in pipeline), error: %v", rep.cmdType, rep.redisCmd, rep.redisKey, len(pending), err) } } - for pos, t := range times { - duration := endT.Sub(t) - took := uint64(duration.Microseconds()) - rcv := replies[pos] - rxBytesCount += getRxLen(rcv) - // AddEntry takes (..., rx, tx): received bytes first, then sent. - // txBytesCount is what we send to Redis, rxBytesCount is the reply. - stat := benchmark_runner.NewStat().AddEntry([]byte(cmdType), []byte(cmdQueryId), uint64(t.Unix()), took, hadError, isTimeout, rxBytesCount, txBytesCount) - p.cmdChan <- *stat + + // Log additional timeout-specific message if it's a timeout + if strings.Contains(err.Error(), "i/o timeout") { + isTimeout = true + log.Printf("Timeout occurred (%d command(s) in pipeline), continuing execution...", len(pending)) } - cmds = nil - cmds = make([]radix.CmdAction, 0, 0) - times = nil - times = make([]time.Time, 0, 0) } - return cmds, times, hadError + + for i := range pending { + pc := &pending[i] + took := uint64(endT.Sub(pc.start).Microseconds()) + // AddEntry takes (..., rx, tx): received bytes, then sent bytes. Each + // command records its OWN counts, times, and labels. + rxBytesCount := getRxLen(pc.reply) + stat := benchmark_runner.NewStat().AddEntry([]byte(pc.cmdType), []byte(pc.cmdQueryId), uint64(pc.start.Unix()), took, hadError, isTimeout, rxBytesCount, pc.txBytes) + p.cmdChan <- *stat + } + + return pending[:0], hadError } // ProcessBatch reads eventsBatches which contain rows of databuild for FT.ADD redis command string diff --git a/cmd/ftsb_redisearch/send_stats_test.go b/cmd/ftsb_redisearch/send_stats_test.go index cbd3e09..a45ea27 100644 --- a/cmd/ftsb_redisearch/send_stats_test.go +++ b/cmd/ftsb_redisearch/send_stats_test.go @@ -3,15 +3,15 @@ package main import ( "errors" "testing" - "time" "github.com/RediSearch/ftsb/benchmark_runner" radix "github.com/mediocregopher/radix/v3" ) -// fakeClient is a radix.Client whose Do returns a canned error (nil = success), -// so the recording path in sendFlatCmd/sendIfRequired can be exercised without -// a real Redis. +// fakeClient is a radix.Client whose Do returns a canned error (nil = success). +// It does not populate command receivers, so the recording path in +// sendFlatCmd/sendIfRequired can be exercised without a real Redis (received +// bytes are therefore 0 in these unit tests; reply capture is covered E2E). type fakeClient struct { calls int err error @@ -21,8 +21,7 @@ func (f *fakeClient) Do(a radix.Action) error { f.calls++; return f.err } func (f *fakeClient) Close() error { return nil } // Regression guard for issue #111: the bytes we SEND to Redis (txBytesCount) -// must be recorded as Tx(), not Rx(). Before the fix the AddEntry arguments -// were swapped, so sent bytes were reported under RxBytes and TxBytes was 0. +// must be recorded as Tx(), not Rx(). func TestSendFlatCmdRecordsSentBytesAsTx(t *testing.T) { // pipeline=1 forces sendIfRequired to flush on the first command. Set it // explicitly (rather than trusting the flag default) so a stray global @@ -34,10 +33,9 @@ func TestSendFlatCmdRecordsSentBytesAsTx(t *testing.T) { p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)} const txBytesCount = uint64(4096) // request/sent bytes for this command - _, _, hadError := sendFlatCmd( + _, hadError := sendFlatCmd( p, &fakeClient{}, "WRITE", "w1", "HSET", - []string{"doc:1", "vec", "payload"}, txBytesCount, - make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0), + []string{"doc:1", "vec", "payload"}, txBytesCount, nil, ) if hadError { t.Fatal("unexpected error from fake client") @@ -51,9 +49,9 @@ func TestSendFlatCmdRecordsSentBytesAsTx(t *testing.T) { if got := entries[0].Tx(); got != txBytesCount { t.Fatalf("Tx() = %d, want %d (sent bytes must land in Tx, not Rx)", got, txBytesCount) } - // Replies are discarded (rcv is a nil interface), so received bytes are 0. + // The fake client never populates the receiver, so received bytes are 0. if got := entries[0].Rx(); got != 0 { - t.Fatalf("Rx() = %d, want 0 (replies are not captured)", got) + t.Fatalf("Rx() = %d, want 0 (fake client does not populate replies)", got) } } @@ -61,14 +59,29 @@ func TestGetRxLen(t *testing.T) { if got := getRxLen("abc"); got != 3 { t.Fatalf("getRxLen(string) = %d, want 3", got) } + if got := getRxLen([]byte("abcd")); got != 4 { + t.Fatalf("getRxLen([]byte) = %d, want 4", got) + } if got := getRxLen([]string{"ab", "cde"}); got != 5 { t.Fatalf("getRxLen([]string) = %d, want 5", got) } + // Arrays are summed recursively (e.g. an FT.SEARCH reply of mixed elements). + if got := getRxLen([]interface{}{"ab", []byte("cd"), int64(100)}); got != 2+2+3 { + t.Fatalf("getRxLen([]interface{}) = %d, want 7", got) + } + if got := getRxLen(int64(12345)); got != 5 { + t.Fatalf("getRxLen(int64) = %d, want 5", got) + } + // The production path stores a *interface{} receiver; it must be dereferenced. + var boxed interface{} = []byte("hello") + if got := getRxLen(&boxed); got != 5 { + t.Fatalf("getRxLen(*interface{}) = %d, want 5", got) + } if got := getRxLen(nil); got != 0 { t.Fatalf("getRxLen(nil) = %d, want 0", got) } if got := getRxLen(42); got != 0 { - t.Fatalf("getRxLen(int) = %d, want 0", got) + t.Fatalf("getRxLen(int) = %d, want 0 (untyped int not a RESP reply type)", got) } } @@ -82,10 +95,9 @@ func TestSendFlatCmdRecordsErrorStatWithCorrectTx(t *testing.T) { p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)} const txBytesCount = uint64(128) - _, _, hadError := sendFlatCmd( + _, hadError := sendFlatCmd( p, &fakeClient{err: errors.New("connection refused")}, "WRITE", "w1", "HSET", - []string{"doc:1"}, txBytesCount, - make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0), + []string{"doc:1"}, txBytesCount, nil, ) if !hadError { t.Fatal("expected hadError=true when client.Do returns an error") @@ -111,10 +123,9 @@ func TestSendFlatCmdMarksTimeout(t *testing.T) { defer func() { pipeline, continueOnErr = savedPipeline, savedContinue }() p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)} - _, _, hadError := sendFlatCmd( + _, hadError := sendFlatCmd( p, &fakeClient{err: errors.New("dial tcp 127.0.0.1:6379: i/o timeout")}, "READ", "r1", "FT.SEARCH", - []string{"idx"}, 64, - make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0), + []string{"idx"}, 64, nil, ) if !hadError { t.Fatal("expected hadError=true on timeout") @@ -134,3 +145,41 @@ func TestSendFlatCmdMarksTimeout(t *testing.T) { t.Fatalf("Tx() = %d, want 64 (sent bytes recorded even on timeout)", got) } } + +// Regression guard for issue #113: with --pipeline > 1, buffering must not panic +// (the old code indexed a length-1 replies slice with the flush position), and +// each command in a flush must record its OWN sent-byte count (the old code +// applied the flushing command's single txBytesCount to every entry). +func TestPipelineRecordsPerCommandTxAndDoesNotPanic(t *testing.T) { + savedPipeline, savedContinue := pipeline, continueOnErr + pipeline, continueOnErr = 2, true + defer func() { pipeline, continueOnErr = savedPipeline, savedContinue }() + + p := &processor{cmdChan: make(chan benchmark_runner.Stat, 2)} + client := &fakeClient{} + + var pending []pendingCmd + pending, _ = sendFlatCmd(p, client, "WRITE", "w1", "HSET", []string{"doc:1"}, 100, pending) + if len(pending) != 1 { + t.Fatalf("with pipeline=2, first command should buffer (len 1), got %d", len(pending)) + } + select { + case <-p.cmdChan: + t.Fatal("no stat should be emitted before the pipeline window is full") + default: + } + + pending, _ = sendFlatCmd(p, client, "WRITE", "w2", "HSET", []string{"doc:2"}, 200, pending) + if len(pending) != 0 { + t.Fatalf("after flush the buffer should be empty, got %d", len(pending)) + } + + s1 := <-p.cmdChan + s2 := <-p.cmdChan + tx1 := s1.CmdStats()[0].Tx() + tx2 := s2.CmdStats()[0].Tx() + // Order is preserved: first buffered command recorded first. + if tx1 != 100 || tx2 != 200 { + t.Fatalf("per-command Tx wrong: got [%d %d], want [100 200] (old code recorded [200 200])", tx1, tx2) + } +} From ee3944bab8073f3eb5193c7e6f3b0355a391048d Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 13:03:49 +0100 Subject: [PATCH 2/2] fix(pipeline): restore command/key details in timeout log line The refactor collapsed the timeout log message to a bare count, dropping the command and key. That broke integration tests (TestFTSBWithTimeout, TestFTSBWithLogFileAndTimeout) which assert the log contains 'Timeout occurred' + the command ('DEBUG') + key ('SLEEP') on one line. Restore the per-command detail using the representative pending command. --- cmd/ftsb_redisearch/cmd_processor.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index fa71aa9..82dcfd2 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -328,7 +328,11 @@ func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([] // Log additional timeout-specific message if it's a timeout if strings.Contains(err.Error(), "i/o timeout") { isTimeout = true - log.Printf("Timeout occurred (%d command(s) in pipeline), continuing execution...", len(pending)) + if rep.cmdType == "READ" || rep.cmdType == "READ_CURSOR" { + log.Printf("Timeout occurred with %d command(s) in pipeline, continuing execution...", len(pending)) + } else { + log.Printf("Timeout occurred with %s command: %s %s (%d command(s) in pipeline), continuing execution...", rep.cmdType, rep.redisCmd, rep.redisKey, len(pending)) + } } }