diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index 711a12f..5e3d4d0 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -638,3 +638,49 @@ func TestFTSBLatencyCapHighOptIn(t *testing.T) { t.Errorf("Expected q100 >= 1500 ms with --max-latency-seconds=60 (DEBUG SLEEP 2 should record ~2000 ms), got %.2f ms", q100) } } + +// Regression guard for #121: a stalled output consumer must not prevent the run +// from completing and writing its --json-out-file result. Output goes to a pipe +// that is never drained; the fast reporting period fills the 64KB pipe buffer +// mid-run. Pre-fix, the blocking progress log wedged the process and no result +// file was written (the test times out). +func TestFTSBCompletesWithStalledOutputConsumer(t *testing.T) { + startRedisContainer(t) + + pr, pw, err := os.Pipe() + if err != nil { + t.Fatalf("pipe: %v", err) + } + defer pr.Close() // held open, never read -> the pipe fills and stays full + + jsonPath := "../testdata/results.stalled.json" + _ = os.Remove(jsonPath) + + cmd := exec.Command("../bin/ftsb_redisearch", + "--input", "../testdata/minimal.csv", + "--duration", "3s", + "--reporting-period", "1ms", // thousands of progress lines -> overflows the pipe buffer + "--json-out-file", jsonPath, + ) + cmd.Stdout = pw + cmd.Stderr = pw + if err := cmd.Start(); err != nil { + t.Fatalf("start: %v", err) + } + pw.Close() // the child holds its own fd; drop the parent's copy + + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + select { + case <-done: + // completed despite the never-drained consumer + case <-time.After(45 * time.Second): + _ = cmd.Process.Kill() + t.Fatal("ftsb hung with a stalled output consumer (never exited)") + } + + if _, statErr := os.Stat(jsonPath); statErr != nil { + t.Fatalf("result file not written under a stalled consumer: %v", statErr) + } + _ = os.Remove(jsonPath) +} diff --git a/cmd/ftsb_redisearch/main.go b/cmd/ftsb_redisearch/main.go index d5407df..f76ee4b 100644 --- a/cmd/ftsb_redisearch/main.go +++ b/cmd/ftsb_redisearch/main.go @@ -106,21 +106,26 @@ func main() { git_dirty_str = "-dirty" } - // Setup log file if specified + // Route console logging through a non-blocking writer so a stalled output + // consumer (wedged terminal, run-remote/SSH stream stall, full CI buffer) + // can never wedge the benchmark and prevent the result from being written + // (issue #121). The log file, if any, is written directly -- a regular file + // does not stall, and keeping it off the drop path leaves it complete. + // On a healthy consumer each Write is delivered synchronously (so the summary + // and any pre-os.Exit fatal message reach stderr); on a stalled consumer a + // Write waits at most writeTimeout and then drops, so the run still completes + // and writes its result (issue #121). + console := newNonBlockingWriter(os.Stderr, 1024, 2*time.Second) if logFile != "" { f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) if err != nil { log.Fatalf("Failed to open log file %s: %v", logFile, err) } defer f.Close() - - // Create a multi-writer that writes to both stdout and the log file - multiWriter := io.MultiWriter(os.Stdout, f) - - // Redirect log output to both stdout and file - log.SetOutput(multiWriter) - + log.SetOutput(io.MultiWriter(console, f)) log.Printf("Logging to file: %s\n", logFile) + } else { + log.SetOutput(console) } log.Printf("ftsb (git_sha1:%s%s)\n", git_sha, git_dirty_str) diff --git a/cmd/ftsb_redisearch/nonblocking_writer.go b/cmd/ftsb_redisearch/nonblocking_writer.go new file mode 100644 index 0000000..fad8111 --- /dev/null +++ b/cmd/ftsb_redisearch/nonblocking_writer.go @@ -0,0 +1,85 @@ +package main + +import ( + "io" + "time" +) + +// nonBlockingWriter forwards whole log lines to an underlying writer on a +// dedicated drain goroutine, so a stalled consumer can never wedge the caller. +// +// ftsb's progress reporter logs unboundedly (one line per --reporting-period) +// and summary() logs before it writes the --json-out-file result. If stdout/ +// stderr is a pipe whose consumer stops draining (a wedged terminal, an SSH / +// run-remote stream stall, a full CI log buffer), a plain blocking log.Printf +// wedges the whole run: the process never exits and the result is never written +// (issue #121). os.Stderr is in blocking mode, so its write can't be bounded +// with a deadline; instead the blocking write happens on the drain goroutine and +// Write only waits for delivery up to writeTimeout. +// +// On a HEALTHY consumer the drain goroutine writes each line immediately, so +// Write returns only after the line has actually reached the underlying writer. +// That matters for messages logged right before the process exits -- notably the +// final summary and any log.Fatal message, which is emitted via log.Output and +// then os.Exit(1), bypassing any deferred flush. Because Write is synchronous on +// a healthy consumer, those lines are delivered before the exit. +// +// On a STALLED consumer Write waits at most writeTimeout and then returns +// (dropping delivery of that line) so the benchmark still completes and writes +// its result; once the small buffer fills, further lines are dropped without +// waiting at all. +type nonBlockingWriter struct { + ch chan writeReq + writeTimeout time.Duration +} + +type writeReq struct { + b []byte + done chan struct{} // closed by the drain goroutine once b has been written +} + +// newNonBlockingWriter starts a drain goroutine that forwards buffered lines to +// w. bufLines is how many pending lines are tolerated before a stalled consumer +// causes new lines to be dropped; writeTimeout bounds how long a single Write +// waits for its line to be delivered before giving up. +func newNonBlockingWriter(w io.Writer, bufLines int, writeTimeout time.Duration) *nonBlockingWriter { + nb := &nonBlockingWriter{ + ch: make(chan writeReq, bufLines), + writeTimeout: writeTimeout, + } + go func() { + for req := range nb.ch { + // A stalled consumer blocks here; that's fine -- only this detached + // goroutine waits, never the benchmark. Write errors are ignored, + // matching the standard log package's best-effort semantics. + _, _ = w.Write(req.b) + close(req.done) + } + }() + return nb +} + +// Write copies the line (the log package reuses its formatting buffer), hands it +// to the drain goroutine, and waits up to writeTimeout for it to be written. It +// never blocks longer than writeTimeout, and never loses a partial line -- the +// log package calls Write once per fully-formatted line, so a dropped line is +// dropped whole. +func (nb *nonBlockingWriter) Write(p []byte) (int, error) { + b := make([]byte, len(p)) + copy(b, p) + req := writeReq{b: b, done: make(chan struct{})} + select { + case nb.ch <- req: + // Delivered to the drain goroutine; wait (bounded) for it to be written + // so healthy-consumer output -- including a pre-os.Exit fatal line -- is + // actually flushed before we return. + select { + case <-req.done: + case <-time.After(nb.writeTimeout): + } + default: + // Buffer full: the consumer is stalled and we're already behind. Drop + // this line rather than wait, so the benchmark keeps making progress. + } + return len(p), nil +} diff --git a/cmd/ftsb_redisearch/nonblocking_writer_test.go b/cmd/ftsb_redisearch/nonblocking_writer_test.go new file mode 100644 index 0000000..c061db3 --- /dev/null +++ b/cmd/ftsb_redisearch/nonblocking_writer_test.go @@ -0,0 +1,90 @@ +package main + +import ( + "bytes" + "sync" + "sync/atomic" + "testing" + "time" +) + +// blockingWriter blocks every Write until released -- models a stalled console. +type blockingWriter struct{ release chan struct{} } + +func (b *blockingWriter) Write(p []byte) (int, error) { <-b.release; return len(p), nil } + +// The whole point of #121: Write must never block the caller for longer than the +// write timeout, even when the underlying consumer is wedged. +func TestNonBlockingWriterDoesNotBlockOnStall(t *testing.T) { + bw := &blockingWriter{release: make(chan struct{})} + nb := newNonBlockingWriter(bw, 4, 50*time.Millisecond) + + done := make(chan struct{}) + go func() { + for i := 0; i < 100000; i++ { // far more than the buffer -> excess is dropped + _, _ = nb.Write([]byte("progress line\n")) + } + close(done) + }() + + select { + case <-done: + case <-time.After(3 * time.Second): + close(bw.release) + t.Fatal("Write blocked while the underlying writer was stalled") + } + close(bw.release) // let the drain goroutine exit +} + +// countingWriter records how many lines were delivered and their bytes. +type countingWriter struct { + mu sync.Mutex + n int64 + buf bytes.Buffer +} + +func (c *countingWriter) Write(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + atomic.AddInt64(&c.n, 1) + c.buf.Write(p) + return len(p), nil +} +func (c *countingWriter) contains(s string) bool { + c.mu.Lock() + defer c.mu.Unlock() + return bytes.Contains(c.buf.Bytes(), []byte(s)) +} + +// The regression this replaces (#122 CI failure): a message must be on the +// underlying writer by the time Write RETURNS on a healthy consumer, so a line +// logged immediately before os.Exit (log.Fatal, the final summary) is delivered +// even though os.Exit bypasses any deferred flush. Synchronous delivery makes +// this deterministic -- no polling. +func TestNonBlockingWriterDeliversBeforeReturnOnHealthy(t *testing.T) { + cw := &countingWriter{} + nb := newNonBlockingWriter(cw, 16, 2*time.Second) + + _, _ = nb.Write([]byte("Fatal error with X\n")) + + if got := atomic.LoadInt64(&cw.n); got != 1 { + t.Fatalf("delivered %d lines by the time Write returned, want 1", got) + } + if !cw.contains("Fatal error with X") { + t.Fatal("healthy-consumer line was not delivered before Write returned") + } +} + +// When the consumer keeps up, nothing is dropped and every line is delivered by +// the time the write loop finishes -- again deterministic, no polling. +func TestNonBlockingWriterDeliversAllWhenDrained(t *testing.T) { + cw := &countingWriter{} + nb := newNonBlockingWriter(cw, 1024, 2*time.Second) + const lines = 500 + for i := 0; i < lines; i++ { + _, _ = nb.Write([]byte("x")) + } + if got := atomic.LoadInt64(&cw.n); got != lines { + t.Fatalf("delivered %d lines, want %d (nothing should drop when the consumer keeps up)", got, lines) + } +}