From 45bad8c7674193b34990459808e446d4a6a079ba Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 13:37:31 +0100 Subject: [PATCH 1/6] fix(pipeline): measure latency from send time so no op is dropped from TotalOps Latency was computed as endT - pc.start, where pc.start is when a command was buffered into the pipeline window. In a pipeline the last-buffered command is flushed immediately, so its took rounded to 0us; the HDR histograms are New(1, cap, 3) (min 1us), so RecordValue(0) is rejected and that command silently vanishes from TotalOps and the latency histograms (observed: TotalOps=1999 for 2000 commands at --pipeline 4). It also skewed per-command latency: the first-buffered command's latency wrongly included the client-side wait for the window to fill. Measure latency from the batch send time instead (captured just before client.Do) and attribute that single round-trip to every command in the flush, clamped to >=1us. A pipeline is one client round-trip, so this is the correct client-observed latency and it can't be sub-microsecond over TCP. - No continuity impact: --pipeline >1 crashed before #114 (no historical data), and at pipeline=1 sendT equals the command's buffer time, so latency is unchanged. - Validated E2E: TotalOps == 2000 at --pipeline 1, 4, and 10 (was 1999 at 4). - Test asserts pipelined commands share one clamped, non-zero latency. --- cmd/ftsb_redisearch/cmd_processor.go | 19 ++++++++++++++----- cmd/ftsb_redisearch/send_stats_test.go | 17 +++++++++++++---- 2 files changed, 27 insertions(+), 9 deletions(-) diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index 82dcfd2..9aa6f73 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -262,7 +262,6 @@ type pendingCmd struct { 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) { @@ -279,7 +278,6 @@ func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd str redisCmd: cmd, redisKey: key, txBytes: txBytesCount, - start: time.Now(), }) return sendIfRequired(p, client, pending) } @@ -295,6 +293,7 @@ func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([] } var err error + sendT := time.Now() if len(pending) == 1 { // if pipeline is 1 no need to pipeline err = client.Do(pending[0].action) @@ -336,13 +335,23 @@ func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([] } } + // A pipeline is one client round-trip for the whole batch, so attribute the + // same send->reply latency to every command in it. Measuring from each + // command's buffer time instead would fold in client-side queueing and leave + // the last-buffered command at ~0us, which the HDR histogram (min 1us) + // silently drops -- undercounting ops. Clamp to >=1us so no command is ever + // dropped. For pipeline=1 sendT equals the command's buffer time, so latency + // is unchanged. + took := uint64(endT.Sub(sendT).Microseconds()) + if took == 0 { + took = 1 + } 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. + // command records its OWN counts 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) + stat := benchmark_runner.NewStat().AddEntry([]byte(pc.cmdType), []byte(pc.cmdQueryId), uint64(sendT.Unix()), took, hadError, isTimeout, rxBytesCount, pc.txBytes) p.cmdChan <- *stat } diff --git a/cmd/ftsb_redisearch/send_stats_test.go b/cmd/ftsb_redisearch/send_stats_test.go index a45ea27..56f879b 100644 --- a/cmd/ftsb_redisearch/send_stats_test.go +++ b/cmd/ftsb_redisearch/send_stats_test.go @@ -176,10 +176,19 @@ func TestPipelineRecordsPerCommandTxAndDoesNotPanic(t *testing.T) { s1 := <-p.cmdChan s2 := <-p.cmdChan - tx1 := s1.CmdStats()[0].Tx() - tx2 := s2.CmdStats()[0].Tx() + c1 := s1.CmdStats()[0] + c2 := s2.CmdStats()[0] // 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) + if c1.Tx() != 100 || c2.Tx() != 200 { + t.Fatalf("per-command Tx wrong: got [%d %d], want [100 200] (old code recorded [200 200])", c1.Tx(), c2.Tx()) + } + // Latency is measured from the batch send time and clamped to >=1us, so no + // command is ever dropped by the HDR histogram (min 1us); a whole pipeline + // shares one round-trip, so the per-command latencies are equal. + if c1.Latency() < 1 || c2.Latency() < 1 { + t.Fatalf("latency must be clamped to >=1us (else the op is dropped from TotalOps): got [%d %d]", c1.Latency(), c2.Latency()) + } + if c1.Latency() != c2.Latency() { + t.Fatalf("pipelined commands share one round-trip; latencies should be equal: %d != %d", c1.Latency(), c2.Latency()) } } From b1d4c6e428e7d4370edd315d507bee3d96f5f220 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 14:17:57 +0100 Subject: [PATCH 2/6] fix(pipeline): flush trailing window + exact TotalOps; correct pipeline latency Supersedes the earlier latency-only change on this branch. The 7-way adversarial review showed the original premise was wrong: hdrhistogram-go v1.0.1 does NOT drop RecordValue(0) (it is counted), so the latency clamp never prevented any TotalOps undercount. The real op-drops were elsewhere. This commit fixes them: A. Flush the trailing partial pipeline window. connectionProcessor buffered up to `pipeline` commands per slot but never flushed the leftover (< pipeline) at end of input, so the last `rows % pipeline` commands per batch were NEVER sent to Redis or counted -- silent data loss whenever pipeline did not divide the row count (e.g. pipeline=3 over 100 rows -> 99 sent). sendIfRequired is split into a guard + reusable flushPending; connectionProcessor now flushes every non-empty slot before finishing. B. Make TotalOps/overallOpsRate exact via an atomic counter (one increment per recorded command) instead of the HDR histograms' TotalCount(), which drops any latency above the trackable cap (--max-latency-seconds, ~1.05s default -- a cold query or any i/o timeout) and loses increments under concurrent unlocked RecordValue. The histograms remain the source for percentiles only. C. Keep the send-time uniform pipeline latency (one batch round-trip attributed to each command; matches redis-benchmark) but reframe the 1us floor honestly: a network round-trip is never 0us, so a 0 only reflects timer resolution -- it is NOT a TotalOps fix. Also capture sendT after building the pipeline action so client-side slice bookkeeping is not charged as latency. Validated E2E: TotalOps == dbsize == input count at pipeline 1/3/4/8/16/100 (was 99/1980/1920 before). Adds: atomic-counter decoupling unit test, a pipeline=3 integration test asserting exact TotalOps, and reframed latency test. Follow-ups filed separately: histogram RecordValue data race under workers>1; #114 reply-capture unmarshal inflating FT.SEARCH latency. --- benchmark_runner/benchmark_runner.go | 27 +++++++---- benchmark_runner/redisearch_test.go | 49 ++++++++++++++++++++ benchmark_runner/stat_test.go | 33 +++++++++++++ cmd/ftsb_redisearch/cmd_processor.go | 64 ++++++++++++++++++++------ cmd/ftsb_redisearch/send_stats_test.go | 8 ++-- 5 files changed, 153 insertions(+), 28 deletions(-) diff --git a/benchmark_runner/benchmark_runner.go b/benchmark_runner/benchmark_runner.go index 409ba1c..3615051 100644 --- a/benchmark_runner/benchmark_runner.go +++ b/benchmark_runner/benchmark_runner.go @@ -62,6 +62,12 @@ type BenchmarkRunner struct { totalErrors uint64 totalTimeouts uint64 + // totalOps is an exact count of recorded commands, incremented once per + // cmdStat. It is the source of truth for TotalOps/overallOpsRate instead of + // the histograms, whose TotalCount() drops any latency above the trackable + // cap and loses increments under concurrent (unlocked) RecordValue. + totalOps uint64 + // maxLatencySeconds caps the highest trackable latency for every HDR // histogram. Configurable via --max-latency-seconds. Converted to µs at // histogram-allocation time. @@ -106,8 +112,8 @@ type BenchmarkRunner struct { func (b *BenchmarkRunner) GetTotalsMap() map[string]interface{} { configs := map[string]interface{}{} - //TotalOps - configs["TotalOps"] = b.totalHistogram.TotalCount() + //TotalOps (exact atomic count, not histogram-derived) + configs["TotalOps"] = int64(atomic.LoadUint64(&b.totalOps)) //SetupTotalWrites configs["SetupWrites"] = b.setupWriteHistogram.TotalCount() @@ -153,7 +159,7 @@ func (b *BenchmarkRunner) GetMeasuredRatiosMap() map[string]interface{} { ///////// configs := map[string]interface{}{} - totalOps := b.totalHistogram.TotalCount() + totalOps := int64(atomic.LoadUint64(&b.totalOps)) writeRatio := float64(b.writeHistogram.TotalCount()+b.setupWriteHistogram.TotalCount()) / float64(totalOps) readRatio := float64(b.readHistogram.TotalCount()+b.readCursorHistogram.TotalCount()) / float64(totalOps) updateRatio := float64(b.updateHistogram.TotalCount()) / float64(totalOps) @@ -183,14 +189,15 @@ func (l *BenchmarkRunner) GetOverallRatesMap() map[string]interface{} { took := l.end.Sub(l.start) writeCount := l.writeHistogram.TotalCount() setupWriteCount := l.setupWriteHistogram.TotalCount() - totalWriteCount := writeCount + setupWriteCount readCount := l.readHistogram.TotalCount() readCursorCount := l.readCursorHistogram.TotalCount() - totalReadCount := readCount + readCursorCount updateCount := l.updateHistogram.TotalCount() deleteCount := l.deleteHistogram.TotalCount() - totalOps := totalWriteCount + totalReadCount + updateCount + deleteCount + // TotalOps/overallOpsRate come from an exact atomic counter (one increment per + // recorded command), immune to HDR histogram tail-rejection (latency above the + // trackable cap) and concurrent RecordValue drops. + totalOps := int64(atomic.LoadUint64(&l.totalOps)) txTotalBytes := atomic.LoadUint64(&l.txTotalBytes) rxTotalBytes := atomic.LoadUint64(&l.rxTotalBytes) @@ -515,6 +522,7 @@ func (l *BenchmarkRunner) work(b Benchmark, wg *sync.WaitGroup, c *duplexChannel atomic.AddUint64(&l.totalTimeouts, 1) } + atomic.AddUint64(&l.totalOps, 1) _ = l.totalHistogram.RecordValue(int64(cmdStat.Latency())) _ = l.inst_totalHistogram.RecordValue(int64(cmdStat.Latency())) @@ -588,14 +596,15 @@ func (l *BenchmarkRunner) summary() { took := l.end.Sub(l.start) writeCount := l.writeHistogram.TotalCount() setupWriteCount := l.setupWriteHistogram.TotalCount() - totalWriteCount := writeCount + setupWriteCount readCount := l.readHistogram.TotalCount() readCursorCount := l.readCursorHistogram.TotalCount() - totalReadCount := readCount + readCursorCount updateCount := l.updateHistogram.TotalCount() deleteCount := l.deleteHistogram.TotalCount() - totalOps := totalWriteCount + totalReadCount + updateCount + deleteCount + // TotalOps/overallOpsRate come from an exact atomic counter (one increment per + // recorded command), immune to HDR histogram tail-rejection (latency above the + // trackable cap) and concurrent RecordValue drops. + totalOps := int64(atomic.LoadUint64(&l.totalOps)) txTotalBytes := atomic.LoadUint64(&l.txTotalBytes) rxTotalBytes := atomic.LoadUint64(&l.rxTotalBytes) totalErrors := atomic.LoadUint64(&l.totalErrors) diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index e23125d..0471a91 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -152,6 +152,55 @@ func TestFTSBWithNoLimitNoDuration(t *testing.T) { } } +// Regression guard for #113/#115: with a pipeline that does not divide the input +// row count, the trailing partial window must still be flushed (no silent data +// loss) and every command must be counted. minimal.csv has exactly 100 rows and +// 100 % 3 != 0, so the pre-fix code dropped the tail (TotalOps=99, and one HSET +// never reached Redis). +func TestFTSBPipelineTailIsFlushedAndCounted(t *testing.T) { + t.Log("Starting Redis container...") + dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") + containerIDRaw, err := dockerRun.Output() + if err != nil { + t.Fatalf("Failed to start Redis container: %v", err) + } + containerID := strings.TrimSpace(string(containerIDRaw)) + t.Cleanup(func() { + t.Log("Stopping Redis container...") + exec.Command("docker", "stop", containerID).Run() + }) + + t.Log("Waiting for Redis to be ready...") + time.Sleep(2 * time.Second) + + jsonPath := "../testdata/results.pipeline_tail.json" + cmd := exec.Command("../bin/ftsb_redisearch", + "--input", "../testdata/minimal.csv", + "--pipeline", "3", + "--json-out-file", jsonPath, + ) + output, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("Benchmark failed: %v\nOutput: %s", err, string(output)) + } + + data, err := os.ReadFile(jsonPath) + if err != nil { + t.Fatalf("Failed to read json output file: %v", err) + } + var parsed struct { + Totals struct { + TotalOps int `json:"TotalOps"` + } `json:"Totals"` + } + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("Failed to parse JSON output: %v", err) + } + if parsed.Totals.TotalOps != 100 { + t.Errorf("TotalOps = %d, want 100 (pipeline=3 must flush and count the trailing window of a 100-row input)", parsed.Totals.TotalOps) + } +} + func TestFTSBErrorAndTimeoutTracking(t *testing.T) { t.Log("Starting Redis container...") dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") diff --git a/benchmark_runner/stat_test.go b/benchmark_runner/stat_test.go index 2994a78..4dafd5f 100644 --- a/benchmark_runner/stat_test.go +++ b/benchmark_runner/stat_test.go @@ -1,6 +1,7 @@ package benchmark_runner import ( + "sync/atomic" "testing" hdrhistogram "github.com/HdrHistogram/hdrhistogram-go" @@ -63,3 +64,35 @@ func TestGetTotalsMapMapsTxRxCorrectly(t *testing.T) { t.Fatalf("configs[\"RxBytes\"] = %v, want 7 (received bytes)", got) } } + +// TotalOps must come from the exact atomic counter, NOT from the HDR histogram, +// which rejects any latency above its trackable cap (e.g. a slow query or a +// multi-second timeout) and would silently undercount ops. This is the fix for +// the op-drop bug the histogram-derived count had at the high tail. +func TestTotalOpsIsAtomicNotHistogramDerived(t *testing.T) { + h := func() *hdrhistogram.Histogram { return hdrhistogram.New(1, 1_000_000, 3) } + b := &BenchmarkRunner{ + totalHistogram: h(), + setupWriteHistogram: h(), + writeHistogram: h(), + readHistogram: h(), + readCursorHistogram: h(), + updateHistogram: h(), + deleteHistogram: h(), + } + + // A latency above the trackable cap (1_000_000us) is rejected by the + // histogram, so a histogram-derived TotalOps would miss it. + if err := b.totalHistogram.RecordValue(60_000_000); err == nil { + t.Fatal("expected RecordValue above cap to return an error") + } + if hc := b.totalHistogram.TotalCount(); hc != 0 { + t.Fatalf("histogram TotalCount = %d, want 0 (the value was rejected)", hc) + } + + // The atomic counter is the source of truth and is unaffected by that rejection. + atomic.StoreUint64(&b.totalOps, 3) + if got := b.GetTotalsMap()["TotalOps"]; got != int64(3) { + t.Fatalf("TotalOps = %v, want 3 (must be the atomic count, not the histogram's 0)", got) + } +} diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index 9aa6f73..535deee 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -217,6 +217,27 @@ func connectionProcessor(p *processor, rateLimiter *rate.Limiter, useRateLimiter pendingSlots[slotP], _ = sendFlatCmd(p, client, cmdType, cmdQueryId, cmd, docFields, bytelen, pendingSlots[slotP]) } } + + // Flush the trailing partial window(s). Without this, the last + // (rows % pipeline) buffered commands in each slot are never sent to Redis + // or counted -- silent data loss whenever pipeline does not divide the row + // count. flushPending sends whatever is buffered regardless of pipeline size. + if !clusterMode { + if len(pendingSlots[0]) > 0 { + var hadError bool + pendingSlots[0], hadError = flushPending(p, p.vanillaClient, pendingSlots[0]) + if hadError && continueOnErr { + p.reconnectPool() + } + } + } else { + for i := range pendingSlots { + if len(pendingSlots[i]) > 0 { + client, _ := p.vanillaCluster.Client(clusterAddr[i]) + pendingSlots[i], _ = flushPending(p, client, pendingSlots[i]) + } + } + } p.wg.Done() } @@ -283,27 +304,40 @@ func sendFlatCmd(p *processor, client radix.Client, cmdType, cmdQueryId, cmd str } // 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. +// commands; otherwise it buffers and returns. The trailing partial window (fewer +// than `pipeline` commands, e.g. `rows % pipeline` at end of input) is flushed by +// the caller via flushPending -- see connectionProcessor -- so those commands are +// never silently dropped. func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([]pendingCmd, bool) { - hadError := false if len(pending) < pipeline { - return pending, hadError + return pending, false } + return flushPending(p, client, pending) +} - var err error - sendT := time.Now() +// flushPending sends the buffered commands (as a pipeline when >1), records one +// stat per command -- each with its own sent/received bytes and labels, plus the +// shared batch send->reply latency -- and returns the emptied buffer (reusing the +// backing array to avoid churn on the hot path). Callers must guard against an +// empty buffer. +func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pendingCmd, bool) { + hadError := false + + // Build the action BEFORE timing so the latency window covers only the + // round-trip, not client-side slice bookkeeping. + var action radix.Action if len(pending) == 1 { - // if pipeline is 1 no need to pipeline - err = client.Do(pending[0].action) + action = pending[0].action // no need to pipeline a single command } else { actions := make([]radix.CmdAction, len(pending)) for i := range pending { actions[i] = pending[i].action } - err = client.Do(radix.Pipeline(actions...)) + action = radix.Pipeline(actions...) } + + sendT := time.Now() + err := client.Do(action) endT := time.Now() isTimeout := false if err != nil { @@ -337,11 +371,11 @@ func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([] // A pipeline is one client round-trip for the whole batch, so attribute the // same send->reply latency to every command in it. Measuring from each - // command's buffer time instead would fold in client-side queueing and leave - // the last-buffered command at ~0us, which the HDR histogram (min 1us) - // silently drops -- undercounting ops. Clamp to >=1us so no command is ever - // dropped. For pipeline=1 sendT equals the command's buffer time, so latency - // is unchanged. + // command's buffer time instead would fold in client-side queueing (the first + // command would absorb the whole window-fill wait). For pipeline=1 sendT is + // effectively the command's send time, so latency is unchanged. Floor to 1us: + // a real network round-trip is never 0us, so a 0 only reflects sub-microsecond + // timer resolution. took := uint64(endT.Sub(sendT).Microseconds()) if took == 0 { took = 1 diff --git a/cmd/ftsb_redisearch/send_stats_test.go b/cmd/ftsb_redisearch/send_stats_test.go index 56f879b..bd1c09b 100644 --- a/cmd/ftsb_redisearch/send_stats_test.go +++ b/cmd/ftsb_redisearch/send_stats_test.go @@ -182,11 +182,11 @@ func TestPipelineRecordsPerCommandTxAndDoesNotPanic(t *testing.T) { if c1.Tx() != 100 || c2.Tx() != 200 { t.Fatalf("per-command Tx wrong: got [%d %d], want [100 200] (old code recorded [200 200])", c1.Tx(), c2.Tx()) } - // Latency is measured from the batch send time and clamped to >=1us, so no - // command is ever dropped by the HDR histogram (min 1us); a whole pipeline - // shares one round-trip, so the per-command latencies are equal. + // A whole pipeline shares one send->reply round-trip, so per-command latencies + // are equal; the >=1us floor means a sub-microsecond timer reading never + // records a physically-impossible 0us network latency. if c1.Latency() < 1 || c2.Latency() < 1 { - t.Fatalf("latency must be clamped to >=1us (else the op is dropped from TotalOps): got [%d %d]", c1.Latency(), c2.Latency()) + t.Fatalf("latency must be floored to >=1us: got [%d %d]", c1.Latency(), c2.Latency()) } if c1.Latency() != c2.Latency() { t.Fatalf("pipelined commands share one round-trip; latencies should be equal: %d != %d", c1.Latency(), c2.Latency()) From fc3179d7437e0572625b9eacb035aa7d1fc94017 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 14:24:21 +0100 Subject: [PATCH 3/6] refactor: extract logFlushError + startRedisContainer to satisfy quality gate - Extract the pipeline-flush error/timeout logging into logFlushError, dropping flushPending's cognitive complexity below the threshold (behavior unchanged; timeout integration tests still pass). - Extract the duplicated docker-redis setup in the integration tests into a shared startRedisContainer(t) helper (removes the new-code duplication the Sonar gate flagged; all 7 call sites now share one copy). --- benchmark_runner/redisearch_test.go | 94 +++++++--------------------- cmd/ftsb_redisearch/cmd_processor.go | 53 ++++++++-------- 2 files changed, 52 insertions(+), 95 deletions(-) diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index 0471a91..cf93f1e 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -9,6 +9,25 @@ import ( "time" ) +// startRedisContainer starts a throwaway redis:8.4 on :6379, registers cleanup, +// and waits for it to be ready. Shared by the integration tests so the docker +// boilerplate lives in one place. +func startRedisContainer(t *testing.T) { + t.Helper() + t.Log("Starting Redis container...") + containerIDRaw, err := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4").Output() + if err != nil { + t.Fatalf("Failed to start Redis container: %v", err) + } + containerID := strings.TrimSpace(string(containerIDRaw)) + t.Cleanup(func() { + t.Log("Stopping Redis container...") + exec.Command("docker", "stop", containerID).Run() + }) + t.Log("Waiting for Redis to be ready...") + time.Sleep(2 * time.Second) +} + func TestFTSBWithDuration(t *testing.T) { entries, err := os.ReadDir("../bin") if err != nil { @@ -58,20 +77,7 @@ func TestFTSBWithDuration(t *testing.T) { } func TestFTSBWithRequests(t *testing.T) { - t.Log("Starting Redis container...") - dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") - containerIDRaw, err := dockerRun.Output() - if err != nil { - t.Fatalf("Failed to start Redis container: %v", err) - } - containerID := strings.TrimSpace(string(containerIDRaw)) - t.Cleanup(func() { - t.Log("Stopping Redis container...") - exec.Command("docker", "stop", containerID).Run() - }) - - t.Log("Waiting for Redis to be ready...") - time.Sleep(2 * time.Second) + startRedisContainer(t) t.Log("Running ftsb_redisearch with --requests=50000") jsonPath := "../testdata/results.requests.json" @@ -102,20 +108,7 @@ func TestFTSBWithRequests(t *testing.T) { } func TestFTSBWithNoLimitNoDuration(t *testing.T) { - t.Log("Starting Redis container...") - dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") - containerIDRaw, err := dockerRun.Output() - if err != nil { - t.Fatalf("Failed to start Redis container: %v", err) - } - containerID := strings.TrimSpace(string(containerIDRaw)) - t.Cleanup(func() { - t.Log("Stopping Redis container...") - exec.Command("docker", "stop", containerID).Run() - }) - - t.Log("Waiting for Redis to be ready...") - time.Sleep(2 * time.Second) + startRedisContainer(t) t.Log("Running ftsb_redisearch with no --requests or --duration") jsonPath := "../testdata/results.nolimit.json" @@ -158,20 +151,7 @@ func TestFTSBWithNoLimitNoDuration(t *testing.T) { // 100 % 3 != 0, so the pre-fix code dropped the tail (TotalOps=99, and one HSET // never reached Redis). func TestFTSBPipelineTailIsFlushedAndCounted(t *testing.T) { - t.Log("Starting Redis container...") - dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") - containerIDRaw, err := dockerRun.Output() - if err != nil { - t.Fatalf("Failed to start Redis container: %v", err) - } - containerID := strings.TrimSpace(string(containerIDRaw)) - t.Cleanup(func() { - t.Log("Stopping Redis container...") - exec.Command("docker", "stop", containerID).Run() - }) - - t.Log("Waiting for Redis to be ready...") - time.Sleep(2 * time.Second) + startRedisContainer(t) jsonPath := "../testdata/results.pipeline_tail.json" cmd := exec.Command("../bin/ftsb_redisearch", @@ -202,20 +182,7 @@ func TestFTSBPipelineTailIsFlushedAndCounted(t *testing.T) { } func TestFTSBErrorAndTimeoutTracking(t *testing.T) { - t.Log("Starting Redis container...") - dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") - containerIDRaw, err := dockerRun.Output() - if err != nil { - t.Fatalf("Failed to start Redis container: %v", err) - } - containerID := strings.TrimSpace(string(containerIDRaw)) - t.Cleanup(func() { - t.Log("Stopping Redis container...") - exec.Command("docker", "stop", containerID).Run() - }) - - t.Log("Waiting for Redis to be ready...") - time.Sleep(2 * time.Second) + startRedisContainer(t) t.Log("Running ftsb_redisearch with normal operation (should have 0 errors)") jsonPath := "../testdata/results.errors.json" @@ -442,20 +409,7 @@ func TestFTSBWithTimeout(t *testing.T) { } func TestFTSBWithLogFile(t *testing.T) { - t.Log("Starting Redis container...") - dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") - containerIDRaw, err := dockerRun.Output() - if err != nil { - t.Fatalf("Failed to start Redis container: %v", err) - } - containerID := strings.TrimSpace(string(containerIDRaw)) - t.Cleanup(func() { - t.Log("Stopping Redis container...") - exec.Command("docker", "stop", containerID).Run() - }) - - t.Log("Waiting for Redis to be ready...") - time.Sleep(2 * time.Second) + startRedisContainer(t) t.Log("Running ftsb_redisearch with --log-file") logPath := "../testdata/benchmark.log" diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index 535deee..4579b32 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -320,6 +320,33 @@ func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([] // shared batch send->reply latency -- and returns the emptied buffer (reusing the // backing array to avoid churn on the hot path). Callers must guard against an // empty buffer. +// logFlushError logs a pipeline-flush failure, honoring -continue-on-error +// (log-and-continue vs. fatal), and returns whether it was an i/o timeout. A +// flush may mix command types, so the first buffered command is used as the +// summary label. Split out of flushPending to keep that function simple. +func logFlushError(pending []pendingCmd, err error) bool { + rep := pending[0] + isRead := rep.cmdType == "READ" || rep.cmdType == "READ_CURSOR" + logf := log.Printf + if !continueOnErr { + logf = log.Fatalf + } + if isRead { + logf("Received an error with %d command(s) in pipeline, error: %v", len(pending), err) + } else { + logf("Received an error with %s command: %s %s (%d command(s) in pipeline), error: %v", rep.cmdType, rep.redisCmd, rep.redisKey, len(pending), err) + } + if !strings.Contains(err.Error(), "i/o timeout") { + return false + } + if isRead { + 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)) + } + return true +} + func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pendingCmd, bool) { hadError := false @@ -342,31 +369,7 @@ func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pe 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 { - log.Fatalf("Fatal error with %d command(s) in pipeline, error: %v", len(pending), err) - } - } 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) - } - } - - // Log additional timeout-specific message if it's a timeout - if strings.Contains(err.Error(), "i/o timeout") { - isTimeout = true - 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)) - } - } + isTimeout = logFlushError(pending, err) } // A pipeline is one client round-trip for the whole batch, so attribute the From e7c67fdc104e8d44d72140c808f6d6048cfbcdc5 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 14:30:41 +0100 Subject: [PATCH 4/6] refactor(test): share runFTSBReadJSON helper to remove run+read duplication Collapse the duplicated 'run binary + read json result' boilerplate in the JSON-asserting integration tests into a shared runFTSBReadJSON helper (drops the vestigial REDIS_URL env, which ftsb never reads). Pushes new-code duplication below the Sonar gate threshold. --- benchmark_runner/redisearch_test.go | 68 ++++++++++------------------- 1 file changed, 23 insertions(+), 45 deletions(-) diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index cf93f1e..690d65b 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -28,6 +28,23 @@ func startRedisContainer(t *testing.T) { time.Sleep(2 * time.Second) } +// runFTSBReadJSON runs the ftsb_redisearch binary with --json-out-file jsonPath +// plus the given args, then reads and returns the JSON result file, failing the +// test on any error. Shared to avoid duplicating the run+read boilerplate. +func runFTSBReadJSON(t *testing.T, jsonPath string, args ...string) []byte { + t.Helper() + full := append([]string{"--json-out-file", jsonPath}, args...) + output, err := exec.Command("../bin/ftsb_redisearch", full...).CombinedOutput() + if err != nil { + t.Fatalf("Benchmark failed: %v\nOutput: %s", err, string(output)) + } + data, err := os.ReadFile(jsonPath) + if err != nil { + t.Fatalf("Failed to read json output file: %v", err) + } + return data +} + func TestFTSBWithDuration(t *testing.T) { entries, err := os.ReadDir("../bin") if err != nil { @@ -80,22 +97,8 @@ func TestFTSBWithRequests(t *testing.T) { startRedisContainer(t) t.Log("Running ftsb_redisearch with --requests=50000") - jsonPath := "../testdata/results.requests.json" - cmd := exec.Command("../bin/ftsb_redisearch", - "--input", "../testdata/minimal.csv", - "--requests=50000", - "--json-out-file", jsonPath, - ) - cmd.Env = append(os.Environ(), "REDIS_URL=redis://localhost:6379") - output, err := cmd.CombinedOutput() - if err != nil { - t.Fatalf("Benchmark failed: %v\nOutput: %s", err, string(output)) - } - - data, err := os.ReadFile(jsonPath) - if err != nil { - t.Fatalf("Failed to read json output file: %v", err) - } + data := runFTSBReadJSON(t, "../testdata/results.requests.json", + "--input", "../testdata/minimal.csv", "--requests=50000") var parsed map[string]interface{} if err := json.Unmarshal(data, &parsed); err != nil { @@ -111,21 +114,8 @@ func TestFTSBWithNoLimitNoDuration(t *testing.T) { startRedisContainer(t) t.Log("Running ftsb_redisearch with no --requests or --duration") - jsonPath := "../testdata/results.nolimit.json" - cmd := exec.Command("../bin/ftsb_redisearch", - "--input", "../testdata/minimal.csv", - "--json-out-file", jsonPath, - ) - cmd.Env = append(os.Environ(), "REDIS_URL=redis://localhost:6379") - output, err := cmd.CombinedOutput() - if err != nil { - t.Fatalf("Benchmark failed: %v\nOutput: %s", err, string(output)) - } - - data, err := os.ReadFile(jsonPath) - if err != nil { - t.Fatalf("Failed to read json output file: %v", err) - } + data := runFTSBReadJSON(t, "../testdata/results.nolimit.json", + "--input", "../testdata/minimal.csv") var parsed struct { Limit int `json:"Limit"` @@ -153,21 +143,9 @@ func TestFTSBWithNoLimitNoDuration(t *testing.T) { func TestFTSBPipelineTailIsFlushedAndCounted(t *testing.T) { startRedisContainer(t) - jsonPath := "../testdata/results.pipeline_tail.json" - cmd := exec.Command("../bin/ftsb_redisearch", - "--input", "../testdata/minimal.csv", - "--pipeline", "3", - "--json-out-file", jsonPath, - ) - output, err := cmd.CombinedOutput() - if err != nil { - t.Fatalf("Benchmark failed: %v\nOutput: %s", err, string(output)) - } + data := runFTSBReadJSON(t, "../testdata/results.pipeline_tail.json", + "--input", "../testdata/minimal.csv", "--pipeline", "3") - data, err := os.ReadFile(jsonPath) - if err != nil { - t.Fatalf("Failed to read json output file: %v", err) - } var parsed struct { Totals struct { TotalOps int `json:"TotalOps"` From 81a6d0306bfd77cc1cb3632989b381f70b75f792 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 14:36:52 +0100 Subject: [PATCH 5/6] refactor: dedup remaining blocks flagged by Sonar - Collapse two more docker-setup blocks (blank-line variant missed earlier) into startRedisContainer. - Extract BenchmarkRunner.overallCounts() so GetOverallRatesMap and summary share the per-label count + atomic-totalOps gathering instead of duplicating it. Behavior unchanged; TotalOps/overallOpsRate verified E2E. --- benchmark_runner/benchmark_runner.go | 39 ++++++++++++---------------- benchmark_runner/redisearch_test.go | 32 ++--------------------- 2 files changed, 19 insertions(+), 52 deletions(-) diff --git a/benchmark_runner/benchmark_runner.go b/benchmark_runner/benchmark_runner.go index 3615051..625793d 100644 --- a/benchmark_runner/benchmark_runner.go +++ b/benchmark_runner/benchmark_runner.go @@ -180,6 +180,21 @@ func (b *BenchmarkRunner) GetMeasuredRatiosMap() map[string]interface{} { return configs } +// overallCounts returns the per-label recorded command counts (from the +// histograms) and the exact total op count (from the atomic counter, which is +// immune to HDR histogram tail-rejection and concurrent RecordValue drops). +// Shared by GetOverallRatesMap and summary. +func (l *BenchmarkRunner) overallCounts() (writeCount, setupWriteCount, readCount, readCursorCount, updateCount, deleteCount, totalOps int64) { + writeCount = l.writeHistogram.TotalCount() + setupWriteCount = l.setupWriteHistogram.TotalCount() + readCount = l.readHistogram.TotalCount() + readCursorCount = l.readCursorHistogram.TotalCount() + updateCount = l.updateHistogram.TotalCount() + deleteCount = l.deleteHistogram.TotalCount() + totalOps = int64(atomic.LoadUint64(&l.totalOps)) + return +} + func (l *BenchmarkRunner) GetOverallRatesMap() map[string]interface{} { ///////// // Overall Rates @@ -187,17 +202,7 @@ func (l *BenchmarkRunner) GetOverallRatesMap() map[string]interface{} { configs := map[string]interface{}{} took := l.end.Sub(l.start) - writeCount := l.writeHistogram.TotalCount() - setupWriteCount := l.setupWriteHistogram.TotalCount() - readCount := l.readHistogram.TotalCount() - readCursorCount := l.readCursorHistogram.TotalCount() - updateCount := l.updateHistogram.TotalCount() - deleteCount := l.deleteHistogram.TotalCount() - - // TotalOps/overallOpsRate come from an exact atomic counter (one increment per - // recorded command), immune to HDR histogram tail-rejection (latency above the - // trackable cap) and concurrent RecordValue drops. - totalOps := int64(atomic.LoadUint64(&l.totalOps)) + writeCount, setupWriteCount, readCount, readCursorCount, updateCount, deleteCount, totalOps := l.overallCounts() txTotalBytes := atomic.LoadUint64(&l.txTotalBytes) rxTotalBytes := atomic.LoadUint64(&l.rxTotalBytes) @@ -594,17 +599,7 @@ func (l *BenchmarkRunner) work(b Benchmark, wg *sync.WaitGroup, c *duplexChannel // summary prints the summary of statistics from loading func (l *BenchmarkRunner) summary() { took := l.end.Sub(l.start) - writeCount := l.writeHistogram.TotalCount() - setupWriteCount := l.setupWriteHistogram.TotalCount() - readCount := l.readHistogram.TotalCount() - readCursorCount := l.readCursorHistogram.TotalCount() - updateCount := l.updateHistogram.TotalCount() - deleteCount := l.deleteHistogram.TotalCount() - - // TotalOps/overallOpsRate come from an exact atomic counter (one increment per - // recorded command), immune to HDR histogram tail-rejection (latency above the - // trackable cap) and concurrent RecordValue drops. - totalOps := int64(atomic.LoadUint64(&l.totalOps)) + writeCount, setupWriteCount, readCount, readCursorCount, updateCount, deleteCount, totalOps := l.overallCounts() txTotalBytes := atomic.LoadUint64(&l.txTotalBytes) rxTotalBytes := atomic.LoadUint64(&l.rxTotalBytes) totalErrors := atomic.LoadUint64(&l.totalErrors) diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index 690d65b..ef776e3 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -55,21 +55,7 @@ func TestFTSBWithDuration(t *testing.T) { for _, entry := range entries { t.Logf(" - %s", entry.Name()) } - t.Log("Starting Redis container...") - dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") - containerIDRaw, err := dockerRun.Output() - if err != nil { - t.Fatalf("Failed to start Redis container: %v", err) - } - containerID := strings.TrimSpace(string(containerIDRaw)) - - t.Cleanup(func() { - t.Log("Stopping Redis container...") - exec.Command("docker", "stop", containerID).Run() - }) - - t.Log("Waiting for Redis to be ready...") - time.Sleep(2 * time.Second) + startRedisContainer(t) t.Log("Running ftsb_redisearch with --duration=5s") start := time.Now() @@ -531,21 +517,7 @@ func TestFTSBWithLogFileAndTimeout(t *testing.T) { } func TestFTSBWithBatchSize(t *testing.T) { - t.Log("Starting Redis container...") - dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4") - containerIDRaw, err := dockerRun.Output() - if err != nil { - t.Fatalf("Failed to start Redis container: %v", err) - } - containerID := strings.TrimSpace(string(containerIDRaw)) - - t.Cleanup(func() { - t.Log("Stopping Redis container...") - exec.Command("docker", "stop", containerID).Run() - }) - - t.Log("Waiting for Redis to be ready...") - time.Sleep(2 * time.Second) + startRedisContainer(t) t.Log("Running ftsb_redisearch with --batch-size=50 --duration=3s") cmd := exec.Command("../bin/ftsb_redisearch", From bac01319f5425b19778f7de7692101a8d8fc84a6 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 15:25:59 +0100 Subject: [PATCH 6/6] Address 16-way adversarial review: fatal-log wording, empty-input guard, live-total reconcile Fixes from the pre-merge adversarial review (no change to the A/B/C behavior): - logFlushError: restore the historical "Fatal error with" prefix on the -continue-on-error=false path (the refactor had softened it to "Received an error with"). Control flow/exit unchanged; log text now matches master. - GetMeasuredRatiosMap: guard totalOps==0 so empty input yields 0 ratios instead of NaN (NaN is unmarshalable -> json.Marshal aborted the run with no result file). Adds a unit test. Also documents that ratios (histogram numerators over the exact atomic total) may sum to <1.0 on tail-heavy runs -- intended. - Live periodic reporter: derive its total ops from the atomic counter so the progress line reconciles with the final TotalOps/overallOpsRate. - Extract flooredMicros() with a deterministic unit test (the fake-client clamp assertion was timing-dependent). - GetTotalsMap: read TxBytes/RxBytes via atomic.LoadUint64 for consistency. - Move the flushPending doc comment back onto flushPending (the logFlushError extraction had detached it). - Integration test: assert TxBytes>0 and RxBytes>0 (reply capture) on the pipeline-tail run. Validated: empty input no longer crashes (exit 0, JSON written, ratios 0); pipeline-tail + timeout integration tests pass; full suite -race clean. --- benchmark_runner/benchmark_runner.go | 27 +++++++++++------ benchmark_runner/redisearch_test.go | 12 +++++++- benchmark_runner/stat_test.go | 30 +++++++++++++++++++ cmd/ftsb_redisearch/cmd_processor.go | 35 +++++++++++++++------- cmd/ftsb_redisearch/floored_micros_test.go | 27 +++++++++++++++++ 5 files changed, 110 insertions(+), 21 deletions(-) create mode 100644 cmd/ftsb_redisearch/floored_micros_test.go diff --git a/benchmark_runner/benchmark_runner.go b/benchmark_runner/benchmark_runner.go index 625793d..e867f79 100644 --- a/benchmark_runner/benchmark_runner.go +++ b/benchmark_runner/benchmark_runner.go @@ -140,10 +140,10 @@ func (b *BenchmarkRunner) GetTotalsMap() map[string]interface{} { configs["Timeouts"] = atomic.LoadUint64(&b.totalTimeouts) //TotalTxBytes - configs["TxBytes"] = b.txTotalBytes + configs["TxBytes"] = atomic.LoadUint64(&b.txTotalBytes) //TotalRxBytes - configs["RxBytes"] = b.rxTotalBytes + configs["RxBytes"] = atomic.LoadUint64(&b.rxTotalBytes) // //for k, _ := range b.detailedMapHistograms { // fmt.Println(k) @@ -159,11 +159,19 @@ func (b *BenchmarkRunner) GetMeasuredRatiosMap() map[string]interface{} { ///////// configs := map[string]interface{}{} + // Denominator is the exact atomic op count; numerators are per-label histogram + // counts, which drop latencies above the trackable cap. So these ratios can + // sum to slightly < 1.0 on tail-heavy runs -- intended (the total is + // authoritative). Guard totalOps==0 (empty input): otherwise 0/0 = NaN, which + // json.Marshal rejects and would abort the run with no result file written. totalOps := int64(atomic.LoadUint64(&b.totalOps)) - writeRatio := float64(b.writeHistogram.TotalCount()+b.setupWriteHistogram.TotalCount()) / float64(totalOps) - readRatio := float64(b.readHistogram.TotalCount()+b.readCursorHistogram.TotalCount()) / float64(totalOps) - updateRatio := float64(b.updateHistogram.TotalCount()) / float64(totalOps) - deleteRatio := float64(b.deleteHistogram.TotalCount()) / float64(totalOps) + var writeRatio, readRatio, updateRatio, deleteRatio float64 + if totalOps > 0 { + writeRatio = float64(b.writeHistogram.TotalCount()+b.setupWriteHistogram.TotalCount()) / float64(totalOps) + readRatio = float64(b.readHistogram.TotalCount()+b.readCursorHistogram.TotalCount()) / float64(totalOps) + updateRatio = float64(b.updateHistogram.TotalCount()) / float64(totalOps) + deleteRatio = float64(b.deleteHistogram.TotalCount()) / float64(totalOps) + } //MeasuredWriteRatio configs["MeasuredWriteRatio"] = writeRatio @@ -702,14 +710,15 @@ func (l *BenchmarkRunner) report(period time.Duration, start time.Time) { took := now.Sub(prevTime) writeCount := l.writeHistogram.TotalCount() setupWriteCount := l.setupWriteHistogram.TotalCount() - totalWriteCount := writeCount + setupWriteCount readCount := l.readHistogram.TotalCount() readCursorCount := l.readCursorHistogram.TotalCount() - totalReadCount := readCount + readCursorCount updateCount := l.updateHistogram.TotalCount() deleteCount := l.deleteHistogram.TotalCount() - totalOps := totalWriteCount + totalReadCount + updateCount + deleteCount + // Live total from the exact atomic counter so the progress line + // reconciles with the final TotalOps/overallOpsRate (per-label histogram + // sums drop tail-rejected and concurrently-lost ops). + totalOps := int64(atomic.LoadUint64(&l.totalOps)) txTotalBytes := atomic.LoadUint64(&l.txTotalBytes) rxTotalBytes := atomic.LoadUint64(&l.rxTotalBytes) setupWriteRate := calculateRateMetrics(setupWriteCount, prevSetupWriteCount, took) diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index ef776e3..711a12f 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -134,7 +134,9 @@ func TestFTSBPipelineTailIsFlushedAndCounted(t *testing.T) { var parsed struct { Totals struct { - TotalOps int `json:"TotalOps"` + TotalOps int `json:"TotalOps"` + TxBytes uint64 `json:"TxBytes"` + RxBytes uint64 `json:"RxBytes"` } `json:"Totals"` } if err := json.Unmarshal(data, &parsed); err != nil { @@ -143,6 +145,14 @@ func TestFTSBPipelineTailIsFlushedAndCounted(t *testing.T) { if parsed.Totals.TotalOps != 100 { t.Errorf("TotalOps = %d, want 100 (pipeline=3 must flush and count the trailing window of a 100-row input)", parsed.Totals.TotalOps) } + // Byte accounting (#111/#112/#114): sent bytes land in TxBytes, and reply + // bytes are actually captured (RxBytes>0 — the 100 HSET integer replies). + if parsed.Totals.TxBytes == 0 { + t.Errorf("TxBytes = 0, want > 0 (sent bytes must be counted)") + } + if parsed.Totals.RxBytes == 0 { + t.Errorf("RxBytes = 0, want > 0 (reply bytes must be captured)") + } } func TestFTSBErrorAndTimeoutTracking(t *testing.T) { diff --git a/benchmark_runner/stat_test.go b/benchmark_runner/stat_test.go index 4dafd5f..acd578c 100644 --- a/benchmark_runner/stat_test.go +++ b/benchmark_runner/stat_test.go @@ -1,6 +1,7 @@ package benchmark_runner import ( + "math" "sync/atomic" "testing" @@ -96,3 +97,32 @@ func TestTotalOpsIsAtomicNotHistogramDerived(t *testing.T) { t.Fatalf("TotalOps = %v, want 3 (must be the atomic count, not the histogram's 0)", got) } } + +// Empty input (totalOps==0) must yield 0 ratios, not NaN. NaN is unmarshalable +// by encoding/json, so an unguarded 0/0 would abort the run with no result file. +func TestGetMeasuredRatiosMapEmptyInputIsZeroNotNaN(t *testing.T) { + h := func() *hdrhistogram.Histogram { return hdrhistogram.New(1, 1_000_000, 3) } + b := &BenchmarkRunner{ + writeHistogram: h(), + setupWriteHistogram: h(), + readHistogram: h(), + readCursorHistogram: h(), + updateHistogram: h(), + deleteHistogram: h(), + // totalOps left 0 => empty input + } + + configs := b.GetMeasuredRatiosMap() + for _, k := range []string{"MeasuredWriteRatio", "MeasuredReadRatio", "MeasuredUpdateRatio", "MeasuredDeleteRatio"} { + v, ok := configs[k].(float64) + if !ok { + t.Fatalf("%s missing or not float64: %v", k, configs[k]) + } + if math.IsNaN(v) { + t.Fatalf("%s is NaN on empty input (would abort json.Marshal)", k) + } + if v != 0 { + t.Fatalf("%s = %v, want 0 on empty input", k, v) + } + } +} diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index 4579b32..e90ae3e 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -315,11 +315,18 @@ func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([] return flushPending(p, client, pending) } -// flushPending sends the buffered commands (as a pipeline when >1), records one -// stat per command -- each with its own sent/received bytes and labels, plus the -// shared batch send->reply latency -- and returns the emptied buffer (reusing the -// backing array to avoid churn on the hot path). Callers must guard against an -// empty buffer. +// flooredMicros converts a duration to whole microseconds with a 1us floor: a +// real network round-trip is never 0us, so a measured 0 only reflects +// sub-microsecond timer resolution and would otherwise record a physically +// impossible 0us latency. +func flooredMicros(d time.Duration) uint64 { + us := uint64(d.Microseconds()) + if us == 0 { + return 1 + } + return us +} + // logFlushError logs a pipeline-flush failure, honoring -continue-on-error // (log-and-continue vs. fatal), and returns whether it was an i/o timeout. A // flush may mix command types, so the first buffered command is used as the @@ -327,14 +334,18 @@ func sendIfRequired(p *processor, client radix.Client, pending []pendingCmd) ([] func logFlushError(pending []pendingCmd, err error) bool { rep := pending[0] isRead := rep.cmdType == "READ" || rep.cmdType == "READ_CURSOR" + // Preserve the historical prefixes: "Fatal error with" on the aborting path + // (-continue-on-error=false), "Received an error with" when continuing. + prefix := "Received an error with" logf := log.Printf if !continueOnErr { + prefix = "Fatal error with" logf = log.Fatalf } if isRead { - logf("Received an error with %d command(s) in pipeline, error: %v", len(pending), err) + logf("%s %d command(s) in pipeline, error: %v", prefix, len(pending), err) } else { - logf("Received an error with %s command: %s %s (%d command(s) in pipeline), error: %v", rep.cmdType, rep.redisCmd, rep.redisKey, len(pending), err) + logf("%s %s command: %s %s (%d command(s) in pipeline), error: %v", prefix, rep.cmdType, rep.redisCmd, rep.redisKey, len(pending), err) } if !strings.Contains(err.Error(), "i/o timeout") { return false @@ -347,6 +358,11 @@ func logFlushError(pending []pendingCmd, err error) bool { return true } +// flushPending sends the buffered commands (as a pipeline when >1), records one +// stat per command -- each with its own sent/received bytes and labels, plus the +// shared batch send->reply latency -- and returns the emptied buffer (reusing the +// backing array to avoid churn on the hot path). Callers must guard against an +// empty buffer. func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pendingCmd, bool) { hadError := false @@ -379,10 +395,7 @@ func flushPending(p *processor, client radix.Client, pending []pendingCmd) ([]pe // effectively the command's send time, so latency is unchanged. Floor to 1us: // a real network round-trip is never 0us, so a 0 only reflects sub-microsecond // timer resolution. - took := uint64(endT.Sub(sendT).Microseconds()) - if took == 0 { - took = 1 - } + took := flooredMicros(endT.Sub(sendT)) for i := range pending { pc := &pending[i] // AddEntry takes (..., rx, tx): received bytes, then sent bytes. Each diff --git a/cmd/ftsb_redisearch/floored_micros_test.go b/cmd/ftsb_redisearch/floored_micros_test.go new file mode 100644 index 0000000..a3b46c8 --- /dev/null +++ b/cmd/ftsb_redisearch/floored_micros_test.go @@ -0,0 +1,27 @@ +package main + +import ( + "testing" + "time" +) + +// Deterministically guards the 1us latency floor (independent of wall-clock +// timing, unlike the fake-client pipeline test): a sub-microsecond duration +// must never record a physically-impossible 0us network latency. +func TestFlooredMicros(t *testing.T) { + cases := []struct { + d time.Duration + want uint64 + }{ + {0, 1}, // exactly zero -> floored + {500 * time.Nanosecond, 1}, // 0.5us truncates to 0 -> floored + {1500 * time.Nanosecond, 1}, // 1.5us truncates to 1 + {2 * time.Microsecond, 2}, + {1234 * time.Microsecond, 1234}, + } + for _, c := range cases { + if got := flooredMicros(c.d); got != c.want { + t.Errorf("flooredMicros(%v) = %d, want %d", c.d, got, c.want) + } + } +}