From 411fe0f34559920ebc7e38ce82065076bb0faff0 Mon Sep 17 00:00:00 2001 From: Harry Callahan Date: Thu, 6 Aug 2026 19:11:54 +0100 Subject: [PATCH 1/2] fix: prevent local async runtime deadlock on long subprocess output lines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LocalRuntimeBackend._log_from_pipe drained subprocess stdout/stderr with `async for line in stream`, which calls StreamReader.readline(). readline() raises LimitOverrunError (surfaced as ValueError) once a single line exceeds the StreamReader limit — 64 KiB by default, as create_subprocess_exec was invoked with no `limit=`. A tool emitting a very long line without a newline (e.g. Jasper's cov-unr UNR report, observed at 116 398 chars) crashed the reader task. With no consumer, asyncio paused the transport, the OS pipe filled, the tool blocked writing to it, and `await process.wait()` hung forever. The legacy launcher wrote tool output straight to the log fd and could not deadlock this way, so the async rewrite was the regression. Read fixed-size chunks via StreamReader.read() instead, which imposes no line-length limit. An incremental UTF-8 decoder (surrogateescape) handles multibyte characters that straddle a chunk boundary. Harden the reader against a recurrence: - Broaden the `except` so an unexpected error is logged via log.exception rather than silently killing the reader task (which would re-deadlock the pipe), while still passing asyncio.CancelledError through untouched. - Pass an explicit `limit=` to create_subprocess_exec. Keep the two concerns as separate constants: SUBPROCESS_STREAM_LIMIT is the StreamReader/readline buffer cap (64 KiB, matching the asyncio default) and SUBPROCESS_READ_CHUNK_SIZE is the read() chunk size used in _log_from_pipe. Signed-off-by: Harry Callahan --- src/dvsim/runtime/local.py | 33 ++++++++++++++++++++++++++++++--- 1 file changed, 30 insertions(+), 3 deletions(-) diff --git a/src/dvsim/runtime/local.py b/src/dvsim/runtime/local.py index e01e82fe..37e4ec1a 100644 --- a/src/dvsim/runtime/local.py +++ b/src/dvsim/runtime/local.py @@ -5,6 +5,7 @@ """Legacy launcher adapter interface for the new async scheduler design.""" import asyncio +import codecs import contextlib import os import pty @@ -51,6 +52,15 @@ class LocalRuntimeBackend(RuntimeBackend): INTERACTIVE_TEE_READ_SIZE = 1024 # Read 1024 bytes to balance efficiency with responsiveness + # StreamReader buffer limit for subprocess pipes, matching the asyncio default of 64 KiB. + # Only relevant to line-oriented reads (readline/readuntil) + SUBPROCESS_STREAM_LIMIT = 64 * 2**10 # 64 KiB + + # Size of each fixed-size read from a subprocess pipe in `_log_from_pipe`. Independent of the + # StreamReader limit above: `StreamReader.read(n)` returns up to n bytes and never raises on a + # long line, so this only trades off syscall frequency against per-read memory. + SUBPROCESS_READ_CHUNK_SIZE = 64 * 2**10 # 64 KiB + def __init__( self, *, @@ -86,13 +96,29 @@ async def _log_from_pipe( """Write piped asyncio subprocess stream contents to a job's log file.""" if stream is None or not handle.log_file: return + # Read fixed-size chunks - iterating lines can cause unexpected errors if the + # subprocess emits very long lines which overflows the StreamReader buffer. + # Uses an incremental decoder to handle multibyte characters that straddle + # a chunk boundary. + decoder = codecs.getincrementaldecoder("utf-8")(errors="surrogateescape") try: - async for line in stream: - decoded = line.decode("utf-8", errors="surrogateescape") - handle.log_file.write(decoded) + while True: + chunk = await stream.read(self.SUBPROCESS_READ_CHUNK_SIZE) + if not chunk: + break + handle.log_file.write(decoder.decode(chunk)) + handle.log_file.flush() + tail = decoder.decode(b"", final=True) + if tail: + handle.log_file.write(tail) handle.log_file.flush() except asyncio.CancelledError: pass + except Exception: # noqa: BLE001 + log.exception( + "Error while streaming subprocess output to log file for job '%s'.", + handle.spec.full_name, + ) async def _monitor_job(self, handle: LocalJobHandle) -> None: """Wait for subprocess completion and emit a completion event.""" @@ -238,6 +264,7 @@ async def _launch_job( # useful to make this behaviour optional on some global `IoPolicy`. stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + limit=self.SUBPROCESS_STREAM_LIMIT, env=env, ) except BlockingIOError: From f47e824502851e211310178a34037cb884fd957c Mon Sep 17 00:00:00 2001 From: Harry Callahan Date: Thu, 6 Aug 2026 19:12:05 +0100 Subject: [PATCH 2/2] test: cover local backend subprocess-output streaming The existing runtime tests only exercised the backend registry and never launched a subprocess, so regressions in _log_from_pipe (empty logs, a NameError, or a re-introduced deadlock) passed CI unnoticed. Add TestLocalBackendStreaming: - streams stdout to the log and passes. The sentinel is emitted via chr() codes so it cannot appear in the "[Executing]" command preamble the monitor writes, i.e. the assertion only holds if real subprocess output was captured. - a single line far larger than the 64 KiB StreamReader limit is logged without deadlocking. Regression test for the original readline() LimitOverrunError bug; @timeout turns a re-regression into a failure rather than a hang. - a multibyte UTF-8 character split across a read-chunk boundary decodes correctly (incremental decoder). - an unexpected log-write error is caught and reported via log.exception rather than propagating (a dead reader would re-deadlock the pipe). - normal cancellation is silent (no error logged). Confirmed these fail on both the original readline deadlock and a bare-name NameError in the read loop. Signed-off-by: Harry Callahan --- tests/test_runtime.py | 160 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 159 insertions(+), 1 deletion(-) diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 07bfbd6d..7f481f12 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -4,20 +4,30 @@ """Test the DVSim Runtime Backends.""" +import asyncio import importlib +import io +import shlex +import sys from collections.abc import Generator +from pathlib import Path +from types import SimpleNamespace import pytest from hamcrest import ( assert_that, calling, + contains_string, equal_to, instance_of, raises, ) from pytest_mock import MockerFixture +from dvsim.job.status import JobStatus +from dvsim.runtime import local as local_module from dvsim.runtime.backend import RuntimeBackend +from dvsim.runtime.data import JobCompletionEvent from dvsim.runtime.legacy import LegacyLauncherAdapter from dvsim.runtime.local import LocalRuntimeBackend from dvsim.runtime.registry import ( @@ -26,7 +36,7 @@ register_backend, register_legacy_launcher_backend, ) -from tests.test_scheduler import MockLauncher +from tests.test_scheduler import MockLauncher, job_spec_factory @pytest.fixture(autouse=True) @@ -99,3 +109,151 @@ def test_lazy_register_launcher(mocker: MockerFixture) -> None: assert_that(backend, instance_of(LegacyLauncherAdapter)) assert_that(backend.name, equal_to("mock")) assert_that(mock_import.call_count, equal_to(1)) + + +async def _run_job_to_completion( + backend: LocalRuntimeBackend, job: object, timeout: float = 20.0 +) -> list[JobCompletionEvent]: + """Submit a single job to a backend and wait for its completion event(s).""" + events: list[JobCompletionEvent] = [] + done = asyncio.Event() + + async def on_complete(batch: object) -> None: + events.extend(batch) + done.set() + + backend.attach_completion_callback(on_complete) + await backend.submit(job) + await asyncio.wait_for(done.wait(), timeout=timeout) + return events + + +def _python_cmd(program: str) -> str: + """Build a shell command that runs `program` with the current test interpreter.""" + return f"{shlex.quote(sys.executable)} -c {shlex.quote(program)}" + + +class TestLocalBackendStreaming: + """Tests for `LocalRuntimeBackend` streaming subprocess output into the job log.""" + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.timeout(30) + async def test_streams_output_to_log(tmp_path: Path) -> None: + """A normal job's stdout is written to its log file and the job passes. + + The sentinel is emitted via `chr()` codes so the literal string never appears in the + command line: `_monitor_job` writes the command into the log as an "[Executing]" preamble, + so asserting on a literal that is also in the command would pass even if no subprocess + output were captured at all. + """ + sentinel = "STREAMED_OUTPUT_OK" + program = f"print(''.join(chr(c) for c in {[ord(c) for c in sentinel]}))" + job = job_spec_factory(tmp_path, cmd=_python_cmd(program)) + backend = LocalRuntimeBackend() + + events = await _run_job_to_completion(backend, job) + + assert_that(len(events), equal_to(1)) + assert_that(events[0].status, equal_to(JobStatus.PASSED)) + assert_that(job.log_path.read_text(), contains_string(sentinel)) + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.timeout(30) + async def test_long_line_does_not_deadlock(tmp_path: Path) -> None: + """A single line far larger than the 64 KiB StreamReader limit is logged, not deadlocked. + + Regression test for the original bug: `readline()` raised `LimitOverrunError` on such a + line, killing the reader task; with nobody draining the pipe the subprocess blocked + writing and `process.wait()` hung forever. The `@timeout` turns a re-regression into a + test failure rather than a hang. + """ + fill = "x" + line_len = 512 * 1024 # 512 KiB on a single line, no embedded newline + program = f"import sys; sys.stdout.write('{fill}' * {line_len})" + job = job_spec_factory(tmp_path, cmd=_python_cmd(program)) + backend = LocalRuntimeBackend() + + events = await _run_job_to_completion(backend, job) + + assert_that(events[0].status, equal_to(JobStatus.PASSED)) + # The entire unbroken run must reach the log intact (the "[Executing]" preamble contains + # only a single `x`, so a run this long can only be the job's own output). + assert_that(job.log_path.read_text(), contains_string(fill * line_len)) + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.timeout(30) + async def test_decodes_multibyte_across_chunk_boundary() -> None: + """A multibyte UTF-8 character split across two chunk reads is decoded correctly. + + `_log_from_pipe` reads fixed-size chunks, so a character whose bytes straddle a chunk + boundary must be held by the incremental decoder rather than corrupted. + """ + chunk = LocalRuntimeBackend.SUBPROCESS_READ_CHUNK_SIZE + # Place the two bytes of 'é' (U+00E9 -> 0xC3 0xA9) either side of the first chunk boundary. + data = b"a" * (chunk - 1) + "é".encode() + b"b" * 10 + reader = asyncio.StreamReader() + reader.feed_data(data) + reader.feed_eof() + + log_file = io.StringIO() + handle = SimpleNamespace(log_file=log_file, spec=SimpleNamespace(full_name="job")) + + await LocalRuntimeBackend()._log_from_pipe(handle, reader) # noqa: SLF001 + + expected = "a" * (chunk - 1) + "é" + "b" * 10 + assert_that(log_file.getvalue(), equal_to(expected)) + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.timeout(30) + async def test_reader_error_is_logged_not_raised(mocker: MockerFixture) -> None: + """An unexpected error while writing the log is caught and logged, not propagated. + + A dead reader task stops draining the pipe and re-introduces the deadlock, so + `_log_from_pipe` must never let an unexpected exception escape - but it must also make the + failure visible via `log.exception` rather than swallow it silently. + """ + spy = mocker.spy(local_module.log, "exception") + + class _RaisingLog: + def write(self, _: str) -> int: + raise OSError("simulated disk-full") + + def flush(self) -> None: + pass + + reader = asyncio.StreamReader() + reader.feed_data(b"some output\n") + reader.feed_eof() + handle = SimpleNamespace(log_file=_RaisingLog(), spec=SimpleNamespace(full_name="job")) + + # Must return normally (no exception bubbling out to kill the caller's task)... + await LocalRuntimeBackend()._log_from_pipe(handle, reader) # noqa: SLF001 + + # ...but must surface the failure via `log.exception` rather than swallow it silently. + assert_that(spy.call_count, equal_to(1)) + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.timeout(30) + async def test_cancellation_is_silent(mocker: MockerFixture) -> None: + """Cancelling the reader (the normal teardown path) is not reported as an error.""" + spy = mocker.spy(local_module.log, "exception") + + reader = asyncio.StreamReader() # never fed EOF: the read blocks until cancelled + log_file = io.StringIO() + handle = SimpleNamespace(log_file=log_file, spec=SimpleNamespace(full_name="job")) + + task = asyncio.create_task( + LocalRuntimeBackend()._log_from_pipe(handle, reader) # noqa: SLF001 + ) + # Let the task start and block on the first read before cancelling it. + for _ in range(3): + await asyncio.sleep(0) + task.cancel() + await task # `CancelledError` is caught internally, so this returns without raising. + + assert_that(spy.call_count, equal_to(0))