From cbbbeea481e8f1c3a48b1115214d8b41b6e68733 Mon Sep 17 00:00:00 2001 From: Ryan Williams Date: Wed, 15 Jul 2026 00:35:21 +0000 Subject: [PATCH 1/2] `join_pipelines`: drain `base_cmd` stdout in a background thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When `base_cmd`'s output exceeds the pipe buffer (~64 KB on Linux) it blocks writing to its captured `stdout=PIPE`, which stops it from draining the FIFOs the pipelines write to, which blocks the `p.wait()` calls → deadlock. Drain the captured stdout in a daemon thread so the writer never blocks. Reproduces on any `git-diff-x '' ` where the piped output per side runs to hundreds of KB (e.g. `gdxc 'jq .' `). Regression test in `TestJoinPipelinesLargeOutput`: `seq 1 30000` vs `seq 30001 60000` (~340 KB diff output); guarded by `SIGALRM` so a broken fix fails loud instead of hanging. --- dffs/utils.py | 18 +++++++++++++++--- tests/test_join_pipelines.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/dffs/utils.py b/dffs/utils.py index ac7e784..f784197 100644 --- a/dffs/utils.py +++ b/dffs/utils.py @@ -5,6 +5,7 @@ from os.path import relpath from subprocess import Popen, PIPE from sys import stdout +from threading import Thread from utz import err, named_pipes, pipeline, process @@ -87,8 +88,18 @@ def join_pipelines( pipe1, pipe2, ] - # Capture stdout so we can suppress it if a pipeline fails + # Capture stdout so we can suppress it if a pipeline fails. + # Drain in a background thread — otherwise, when `base_cmd`'s output + # exceeds the pipe buffer (~64 KB on Linux) it blocks writing, which + # blocks it from reading the FIFOs, which blocks the pipelines' + # writes, which blocks the `p.wait()` calls below → deadlock. proc = Popen(join_cmd, stdout=PIPE) + stdout_buf: list[bytes] = [] + stdout_thread = Thread( + target=lambda: stdout_buf.append(proc.stdout.read() if proc.stdout else b''), + daemon=True, + ) + stdout_thread.start() # Track pipeline processes and their commands pipeline_groups = [] # List of (cmds, procs) tuples @@ -149,9 +160,10 @@ def join_pipelines( stderr_output = stderr_output.decode('utf-8', errors='replace') err(stderr_output.rstrip()) - # Wait for base_cmd and capture its output + # Wait for base_cmd; the drain thread already has its output. proc.wait() - base_stdout = proc.stdout.read() if proc.stdout else b'' + stdout_thread.join() + base_stdout = stdout_buf[0] if stdout_buf else b'' # If any pipeline failed, suppress base_cmd output and return error code if pipeline_failed: diff --git a/tests/test_join_pipelines.py b/tests/test_join_pipelines.py index f60f5af..956cb08 100644 --- a/tests/test_join_pipelines.py +++ b/tests/test_join_pipelines.py @@ -1,4 +1,6 @@ """Tests for join_pipelines function.""" +import signal + import pytest from dffs.utils import join_pipelines @@ -190,3 +192,35 @@ def test_shell_false_multi_stage(self): shell=False, ) assert returncode == 0 + + +class TestJoinPipelinesLargeOutput: + """Regression: previously, `join_pipelines` captured `base_cmd`'s stdout + via `stdout=PIPE` and only read it after `p.wait()` returned for each + pipeline. When `base_cmd`'s output exceeded the pipe buffer (~64 KB on + Linux), the writer blocked, which stopped it from draining the FIFOs the + pipelines wrote to, which blocked `p.wait()` → deadlock. Fixed by draining + `base_cmd`'s stdout in a background thread.""" + + def test_large_diff_output_does_not_deadlock(self): + """Diff of two ≥100 KB-per-side streams that fully differ produces + several hundred KB of diff output — well over the pipe buffer. Must + complete promptly; a broken fix hangs indefinitely.""" + def _timeout(signum, frame): + raise TimeoutError("join_pipelines deadlocked on large output") + prev = signal.signal(signal.SIGALRM, _timeout) + signal.alarm(10) # fix completes in <1 s; 10 s is generous + try: + # `seq 1 30000` ≈ 169 KB per side; differing content ⇒ full diff + # (~340 KB output) — comfortably over the 64 KB pipe buffer. + returncode = join_pipelines( + base_cmd=['diff'], + cmds1=['seq 1 30000'], + cmds2=['seq 30001 60000'], + shell=True, + ) + finally: + signal.alarm(0) + signal.signal(signal.SIGALRM, prev) + # Files differ ⇒ diff exits 1. + assert returncode == 1 From 02a197ace820fb3168f04b75610037a6919b0bc6 Mon Sep 17 00:00:00 2001 From: Ryan Williams Date: Wed, 15 Jul 2026 10:59:19 -0400 Subject: [PATCH 2/2] `join_pipelines`: surface drain-thread errors; assert full output in test Follow-ups on the deadlock fix: - `dffs/utils.py`: the stdout-drain thread previously swallowed any exception from `proc.stdout.read()`, silently returning empty output with a success code. Capture the exception and re-raise it in the main thread after `join()` instead. - `tests/test_join_pipelines.py`: assert the complete diff output (hunk header, all 30000 `<` lines, `---`, all 30000 `>` lines), verifying the drain captured every byte rather than only that it didn't deadlock. Capture via `monkeypatch` on `dffs.utils.stdout` since the module's import-time `from sys import stdout` binding defeats `capsys`/`capfd`. Co-Authored-By: Claude Opus 4.8 (1M context) --- dffs/utils.py | 18 ++++++++++++------ tests/test_join_pipelines.py | 23 +++++++++++++++++++++-- 2 files changed, 33 insertions(+), 8 deletions(-) diff --git a/dffs/utils.py b/dffs/utils.py index f784197..b38a05b 100644 --- a/dffs/utils.py +++ b/dffs/utils.py @@ -94,11 +94,15 @@ def join_pipelines( # blocks it from reading the FIFOs, which blocks the pipelines' # writes, which blocks the `p.wait()` calls below → deadlock. proc = Popen(join_cmd, stdout=PIPE) - stdout_buf: list[bytes] = [] - stdout_thread = Thread( - target=lambda: stdout_buf.append(proc.stdout.read() if proc.stdout else b''), - daemon=True, - ) + drain: dict[str, bytes | BaseException] = {} + + def drain_stdout() -> None: + try: + drain['out'] = proc.stdout.read() if proc.stdout else b'' + except BaseException as e: # surface, don't silently truncate + drain['err'] = e + + stdout_thread = Thread(target=drain_stdout, daemon=True) stdout_thread.start() # Track pipeline processes and their commands @@ -163,7 +167,9 @@ def join_pipelines( # Wait for base_cmd; the drain thread already has its output. proc.wait() stdout_thread.join() - base_stdout = stdout_buf[0] if stdout_buf else b'' + if 'err' in drain: + raise drain['err'] + base_stdout = drain.get('out', b'') # If any pipeline failed, suppress base_cmd output and return error code if pipeline_failed: diff --git a/tests/test_join_pipelines.py b/tests/test_join_pipelines.py index 956cb08..ad6c723 100644 --- a/tests/test_join_pipelines.py +++ b/tests/test_join_pipelines.py @@ -1,5 +1,6 @@ """Tests for join_pipelines function.""" import signal +from io import StringIO import pytest from dffs.utils import join_pipelines @@ -202,10 +203,17 @@ class TestJoinPipelinesLargeOutput: pipelines wrote to, which blocked `p.wait()` → deadlock. Fixed by draining `base_cmd`'s stdout in a background thread.""" - def test_large_diff_output_does_not_deadlock(self): + def test_large_diff_output_does_not_deadlock(self, monkeypatch): """Diff of two ≥100 KB-per-side streams that fully differ produces several hundred KB of diff output — well over the pipe buffer. Must - complete promptly; a broken fix hangs indefinitely.""" + complete promptly (a broken fix hangs indefinitely) *and* emit the + full output (a truncating drain would silently drop lines).""" + # `join_pipelines` writes to the module-level `stdout` it bound at + # import (`from sys import stdout`), so capture that object directly — + # `capsys`/`capfd` don't see writes through this stale reference. + buf = StringIO() + monkeypatch.setattr('dffs.utils.stdout', buf) + def _timeout(signum, frame): raise TimeoutError("join_pipelines deadlocked on large output") prev = signal.signal(signal.SIGALRM, _timeout) @@ -224,3 +232,14 @@ def _timeout(signum, frame): signal.signal(signal.SIGALRM, prev) # Files differ ⇒ diff exits 1. assert returncode == 1 + # Disjoint sequences ⇒ diff replaces every line of side 1 with every + # line of side 2: one `c` hunk header, then all 30000 `<` lines, a + # `---` separator, then all 30000 `>` lines. Asserting the full lists + # verifies the drain thread captured every byte, not a truncated head. + out_lines = buf.getvalue().rstrip("\n").split("\n") + assert out_lines[0] == "1,30000c1,30000" + assert [l[2:] for l in out_lines if l.startswith("< ")] == \ + [str(i) for i in range(1, 30001)] + assert out_lines[30001] == "---" + assert [l[2:] for l in out_lines if l.startswith("> ")] == \ + [str(i) for i in range(30001, 60001)]