Skip to content

Fix pipeline command accounting: flush trailing window + exact TotalOps #115

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fcostaoliveira merged 6 commits into from
Jul 13, 2026
Merged

Fix pipeline command accounting: flush trailing window + exact TotalOps #115

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
45bad8c
fix(pipeline): measure latency from send time so no op is dropped fro…
fcostaoliveira Jul 13, 2026
b1d4c6e
fix(pipeline): flush trailing window + exact TotalOps; correct pipeli…
fcostaoliveira Jul 13, 2026
fc3179d
refactor: extract logFlushError + startRedisContainer to satisfy qual…
fcostaoliveira Jul 13, 2026
e7c67fd
refactor(test): share runFTSBReadJSON helper to remove run+read dupli…
fcostaoliveira Jul 13, 2026
81a6d03
refactor: dedup remaining blocks flagged by Sonar
fcostaoliveira Jul 13, 2026
bac0131
Address 16-way adversarial review: fatal-log wording, empty-input gua…
fcostaoliveira Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 45 additions & 32 deletions benchmark_runner/benchmark_runner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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()
Expand All @@ -134,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)
Expand All @@ -153,11 +159,19 @@ func (b *BenchmarkRunner) GetMeasuredRatiosMap() map[string]interface{} {
/////////
configs := map[string]interface{}{}

totalOps := b.totalHistogram.TotalCount()
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)
// 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))
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
Expand All @@ -174,23 +188,29 @@ 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
/////////
configs := 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
writeCount, setupWriteCount, readCount, readCursorCount, updateCount, deleteCount, totalOps := l.overallCounts()
txTotalBytes := atomic.LoadUint64(&l.txTotalBytes)
rxTotalBytes := atomic.LoadUint64(&l.rxTotalBytes)

Expand Down Expand Up @@ -515,6 +535,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()))

Expand Down Expand Up @@ -586,16 +607,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()
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
writeCount, setupWriteCount, readCount, readCursorCount, updateCount, deleteCount, totalOps := l.overallCounts()
txTotalBytes := atomic.LoadUint64(&l.txTotalBytes)
rxTotalBytes := atomic.LoadUint64(&l.rxTotalBytes)
totalErrors := atomic.LoadUint64(&l.totalErrors)
Expand Down Expand Up @@ -698,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)
Expand Down
193 changes: 78 additions & 115 deletions benchmark_runner/redisearch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,31 +9,53 @@ import (
"time"
)

func TestFTSBWithDuration(t *testing.T) {
entries, err := os.ReadDir("../bin")
if err != nil {
t.Fatalf("Failed to read bin/ directory: %v", err)
}

t.Log("Listing bin/ contents:")
for _, entry := range entries {
t.Logf(" - %s", entry.Name())
}
// 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...")
dockerRun := exec.Command("docker", "run", "--rm", "-d", "-p", "6379:6379", "redis:8.4")
containerIDRaw, err := dockerRun.Output()
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)
}

// 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 {
t.Fatalf("Failed to read bin/ directory: %v", err)
}

t.Log("Listing bin/ contents:")
for _, entry := range entries {
t.Logf(" - %s", entry.Name())
}
startRedisContainer(t)

t.Log("Running ftsb_redisearch with --duration=5s")
start := time.Now()
Expand All @@ -58,38 +80,11 @@ 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"
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 {
Expand All @@ -102,37 +97,11 @@ 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"
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"`
Expand All @@ -152,21 +121,42 @@ func TestFTSBWithNoLimitNoDuration(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)
// 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) {
startRedisContainer(t)

data := runFTSBReadJSON(t, "../testdata/results.pipeline_tail.json",
"--input", "../testdata/minimal.csv", "--pipeline", "3")

var parsed struct {
Totals struct {
TotalOps int `json:"TotalOps"`
TxBytes uint64 `json:"TxBytes"`
RxBytes uint64 `json:"RxBytes"`
} `json:"Totals"`
}
containerID := strings.TrimSpace(string(containerIDRaw))
t.Cleanup(func() {
t.Log("Stopping Redis container...")
exec.Command("docker", "stop", containerID).Run()
})
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)
}
// 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)")
}
}

t.Log("Waiting for Redis to be ready...")
time.Sleep(2 * time.Second)
func TestFTSBErrorAndTimeoutTracking(t *testing.T) {
startRedisContainer(t)

t.Log("Running ftsb_redisearch with normal operation (should have 0 errors)")
jsonPath := "../testdata/results.errors.json"
Expand Down Expand Up @@ -393,20 +383,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"
Expand Down Expand Up @@ -550,21 +527,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",
Expand Down
Loading
Loading