From ce11819cb20bd8d3318d0d1cec1141723ee8ff0b Mon Sep 17 00:00:00 2001 From: Yue Sun Date: Fri, 31 Jul 2026 16:20:44 +0800 Subject: [PATCH] Spool native warnings without pipes during EP compilation (#1266) ## Summary - replace the pipe used by native warning suppression with a file-backed temporary spool - filter and replay preserved native diagnostics only after restoring stderr - retain fail-open behavior, Windows handle restoration, warning filtering, and bounded memory usage ## Root cause The warning filter introduced in #1246 redirected native stderr to a pipe. VitisAI can hang inside `ort.InferenceSession` when its compiler sees that pipe handle. The reader was draining correctly, so this was not the full-buffer deadlock fixed by #1223; changing the reader to defer replay still hung, which isolated the pipe handle itself as the trigger. A temporary file preserves warning filtering without pipe semantics or a finite producer buffer. With an empty VAIP cache and no VitisAI-specific bypass, `facebook/convnext-tiny-224` completed on VitisAI NPU in 144.1 seconds. Disabling warning filtering entirely completed the same workload in 145.1 seconds. ## Validation - `uv run --no-sync pytest tests/unit/utils/test_native_stderr.py tests/unit/commands/test_perf_cli.py -q --basetemp temp/pytest_tmp/native-warning-file-backed-final` (143 passed, 1 platform skip) - `uvx ruff check src/winml/modelkit/utils/native_stderr.py tests/unit/utils/test_native_stderr.py src/winml/modelkit/commands/perf.py tests/unit/commands/test_perf_cli.py` - cold-cache VitisAI NPU perf with warning filtering enabled: PASS in 144.1s --- src/winml/modelkit/utils/native_stderr.py | 79 ++++++++------------- tests/unit/utils/test_native_stderr.py | 84 ++++++----------------- 2 files changed, 50 insertions(+), 113 deletions(-) diff --git a/src/winml/modelkit/utils/native_stderr.py b/src/winml/modelkit/utils/native_stderr.py index 598e03963..af723e1ef 100644 --- a/src/winml/modelkit/utils/native_stderr.py +++ b/src/winml/modelkit/utils/native_stderr.py @@ -10,7 +10,7 @@ * ``suppress_native_stderr`` - discard to devnull (startup noise) * ``capture_native_stderr`` - capture via pipe and re-log (compilation output) -* ``suppress_native_warnings`` - hide warning lines, replay everything else +* ``suppress_native_warnings`` - spool to disk, hide warnings, replay other lines The first two are no-ops on non-Windows. ``suppress_native_warnings`` works via fd 2 on all platforms and also keeps the Win32 stderr handle in sync on Windows. @@ -22,8 +22,9 @@ import os import re import sys +import tempfile import threading -from contextlib import contextmanager +from contextlib import ExitStack, contextmanager from pathlib import Path from typing import TYPE_CHECKING, cast @@ -198,19 +199,25 @@ def suppress_native_warnings( ``[E:...]``. This process-wide fd redirect is opt-in; CLI entry points pass ``enabled=True`` only around native-heavy work. ``-v`` / ``-vv`` or ``WINMLCLI_SHOW_ALL_WARNINGS=1`` leaves stderr untouched. + + Output is spooled to a temporary file and replayed only after fd 2 is + restored. A pipe is deliberately not used here: some native execution- + provider compilers change behavior or hang when stderr is a pipe, even when + that pipe is drained concurrently. A file also bounds Python memory usage + without imposing a finite producer buffer. """ if not enabled or _show_native_warnings_requested(): yield return - read_fd: int | None = None - write_fd: int | None = None old_fd: int | None = None + capture_stack = ExitStack() try: - read_fd, write_fd = os.pipe() + capture = capture_stack.enter_context( + tempfile.TemporaryFile(mode="w+b") # noqa: SIM115 - owned by ExitStack + ) except OSError: - _close_fd(read_fd) - _close_fd(write_fd) + capture_stack.close() logger.debug( "Native warning suppression setup failed; leaving stderr unchanged", exc_info=True, @@ -223,8 +230,7 @@ def suppress_native_warnings( try: old_fd = os.dup(2) except OSError: - _close_fd(read_fd) - _close_fd(write_fd) + capture_stack.close() logger.debug( "Native warning suppression setup failed; leaving stderr unchanged", exc_info=True, @@ -232,17 +238,10 @@ def suppress_native_warnings( redirect_failed = True if not redirect_failed: assert old_fd is not None - reader = threading.Thread( - target=_drain_filtered_native_stderr, - args=(read_fd, old_fd, preserve_unclassified), - name="suppress-native-warnings", - daemon=True, - ) try: - os.dup2(write_fd, 2) + os.dup2(capture.fileno(), 2) except OSError: - _close_fd(read_fd) - _close_fd(write_fd) + capture_stack.close() _close_fd(old_fd) logger.debug( "Native warning suppression redirect failed; leaving stderr unchanged", @@ -250,22 +249,26 @@ def suppress_native_warnings( ) redirect_failed = True if not redirect_failed: - _close_fd(write_fd) _set_win32_std_handle_to_current_fd(2) - reader.start() try: yield finally: _restore_redirected_fd(2, old_fd) _set_win32_std_handle_to_current_fd(2) _refresh_click_windows_console_stream(2) - reader.join(timeout=_NATIVE_READER_JOIN_TIMEOUT_SECONDS) - if reader.is_alive(): + try: + capture.flush() + capture.seek(0) + for line in capture: + if _should_preserve_native_line(line, preserve_unclassified): + _write_all(old_fd, line) + except OSError: logger.debug( - "Native warning suppression reader did not finish after stderr restore" + "Could not replay filtered native stderr", + exc_info=True, ) - else: - _close_fd(old_fd) + capture_stack.close() + _close_fd(old_fd) if redirect_failed: yield @@ -307,32 +310,6 @@ def _restore_redirected_fd(fd: int, old_fd: int | None) -> None: ) -def _drain_filtered_native_stderr( - read_fd: int, - target_fd: int, - preserve_unclassified: bool, -) -> None: - pending = b"" - try: - while chunk := os.read(read_fd, 4096): - pending += chunk - while b"\n" in pending: - line, pending = pending.split(b"\n", 1) - _write_non_warning_line(target_fd, line + b"\n", preserve_unclassified) - if pending: - _write_non_warning_line(target_fd, pending, preserve_unclassified) - except OSError: - # fd redirection is best-effort cleanup; restore path handles usability. - pass - finally: - os.close(read_fd) - - -def _write_non_warning_line(fd: int, line: bytes, preserve_unclassified: bool) -> None: - if _should_preserve_native_line(line, preserve_unclassified): - _write_all(fd, line) - - def _should_preserve_native_line(line: bytes, preserve_unclassified: bool) -> bool: match = _NATIVE_SEVERITY_TOKEN_RE.search(line) if match is not None: diff --git a/tests/unit/utils/test_native_stderr.py b/tests/unit/utils/test_native_stderr.py index 22cb2a56d..c1905cc70 100644 --- a/tests/unit/utils/test_native_stderr.py +++ b/tests/unit/utils/test_native_stderr.py @@ -243,6 +243,23 @@ def test_filters_native_warning_lines_and_preserves_errors(self, monkeypatch, ca assert "useful error" in stderr assert "plain diagnostic" in stderr + def test_uses_file_backed_capture_instead_of_pipe(self, monkeypatch): + monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False) + logging.getLogger().setLevel(logging.WARNING) + pipe_called = False + + def fail_pipe() -> tuple[int, int]: + nonlocal pipe_called + pipe_called = True + raise AssertionError("warning suppression must not create a pipe") + + monkeypatch.setattr(native_stderr_module.os, "pipe", fail_pipe) + + with native_stderr_module.suppress_native_warnings(enabled=True): + os.write(2, b"2026 [W:custom-native:, file.cc:1 WarningFunc] hidden\n") + + assert pipe_called is False + def test_filters_native_prefix_info_without_dropping_python_stderr(self, monkeypatch, capfd): monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False) logging.getLogger().setLevel(logging.WARNING) @@ -299,14 +316,14 @@ def test_show_all_warnings_env_leaves_native_warnings_visible(self, monkeypatch, assert "env warning" in capfd.readouterr().err - def test_pipe_setup_failure_does_not_abort_wrapped_code(self, monkeypatch): + def test_capture_setup_failure_does_not_abort_wrapped_code(self, monkeypatch): monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False) logging.getLogger().setLevel(logging.WARNING) - def fail_pipe() -> tuple[int, int]: + def fail_capture(*args: object, **kwargs: object): raise OSError(1, "Incorrect function") - monkeypatch.setattr(native_stderr_module.os, "pipe", fail_pipe) + monkeypatch.setattr(native_stderr_module.tempfile, "TemporaryFile", fail_capture) ran = False with native_stderr_module.suppress_native_warnings(enabled=True): @@ -314,37 +331,21 @@ def fail_pipe() -> tuple[int, int]: assert ran - def test_restore_failure_closes_redirected_fd_and_bounds_reader_join(self, monkeypatch): + def test_restore_failure_closes_redirected_fd(self, monkeypatch): monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False) logging.getLogger().setLevel(logging.WARNING) closed: list[int] = [] - join_timeouts: list[float | None] = [] dup2_calls = 0 - class FakeThread: - def __init__(self, *args: object, **kwargs: object) -> None: - pass - - def start(self) -> None: - pass - - def join(self, timeout: float | None = None) -> None: - join_timeouts.append(timeout) - - def is_alive(self) -> bool: - return False - def fake_dup2(src: int, dst: int) -> None: nonlocal dup2_calls dup2_calls += 1 if dup2_calls == 2: raise OSError(1, "restore failed") - monkeypatch.setattr(native_stderr_module.os, "pipe", lambda: (10, 11)) monkeypatch.setattr(native_stderr_module.os, "dup", lambda fd: 12) monkeypatch.setattr(native_stderr_module.os, "dup2", fake_dup2) monkeypatch.setattr(native_stderr_module.os, "close", lambda fd: closed.append(fd)) - monkeypatch.setattr(native_stderr_module.threading, "Thread", FakeThread) monkeypatch.setattr( native_stderr_module, "_set_win32_std_handle_to_current_fd", @@ -360,48 +361,7 @@ def fake_dup2(src: int, dst: int) -> None: pass assert 2 in closed - assert join_timeouts == [native_stderr_module._NATIVE_READER_JOIN_TIMEOUT_SECONDS] - - def test_reader_owned_old_fd_stays_open_when_reader_outlives_join(self, monkeypatch): - monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False) - logging.getLogger().setLevel(logging.WARNING) - closed: list[int] = [] - join_timeouts: list[float | None] = [] - - class FakeThread: - def __init__(self, *args: object, **kwargs: object) -> None: - pass - - def start(self) -> None: - pass - - def join(self, timeout: float | None = None) -> None: - join_timeouts.append(timeout) - - def is_alive(self) -> bool: - return True - - monkeypatch.setattr(native_stderr_module.os, "pipe", lambda: (10, 11)) - monkeypatch.setattr(native_stderr_module.os, "dup", lambda fd: 12) - monkeypatch.setattr(native_stderr_module.os, "dup2", lambda src, dst: None) - monkeypatch.setattr(native_stderr_module.os, "close", lambda fd: closed.append(fd)) - monkeypatch.setattr(native_stderr_module.threading, "Thread", FakeThread) - monkeypatch.setattr( - native_stderr_module, - "_set_win32_std_handle_to_current_fd", - lambda fd: None, - ) - monkeypatch.setattr( - native_stderr_module, - "_refresh_click_windows_console_stream", - lambda fd, handle=None: None, - ) - - with native_stderr_module.suppress_native_warnings(enabled=True): - pass - - assert join_timeouts == [native_stderr_module._NATIVE_READER_JOIN_TIMEOUT_SECONDS] - assert 12 not in closed + assert 12 in closed @pytest.mark.skipif(sys.platform != "win32", reason="Win32 only") def test_win32_std_handle_sync_failure_does_not_abort_wrapped_code(self, monkeypatch):