diff --git a/dffs/utils.py b/dffs/utils.py index ac7e784..b38a05b 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,22 @@ 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) + 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 pipeline_groups = [] # List of (cmds, procs) tuples @@ -149,9 +164,12 @@ 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() + 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 f60f5af..ad6c723 100644 --- a/tests/test_join_pipelines.py +++ b/tests/test_join_pipelines.py @@ -1,4 +1,7 @@ """Tests for join_pipelines function.""" +import signal +from io import StringIO + import pytest from dffs.utils import join_pipelines @@ -190,3 +193,53 @@ 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, 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) *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) + 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 + # 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)]