From dfcd25997e0d27e249f973249fafdb1cc9908156 Mon Sep 17 00:00:00 2001 From: Gleb Chesnokov Date: Tue, 20 Oct 2026 14:39:29 +0300 Subject: [PATCH] scripts: Keep analyzer stdin flowing during progress updates A progress timeout can occur before the child consumes all supplied input. On Python 3.12, retrying communicate() without input stops scheduling writes to stdin. The child waits for EOF while the driver keeps waiting for output, which can hang the analyzer script tests until the CI job timeout. Run communicate() once in an I/O worker while the main thread waits for its result and updates progress. On interruption, stop the owned process group before joining the worker so blocked pipes cannot prevent cleanup. Make the regression wait for two progress updates before reading stdin to force the pending-input case. Bound the entire child with an alarm so a regression reports a failure instead of hanging the suite. Validation: reproduce both check modes failing on Python 3.12.14 before the fix; all 75 lint-test fixtures pass on chesnokov-vm2 afterwards. All 11 process/log fixtures also pass locally on Python 3.14.8. Fixes: c8648f941072 ("scripts: Share analyzer process and raw-log handling") --- scripts/lint_common.py | 34 +++++++++++++++++-------------- scripts/tests/test_lint_common.py | 7 +++++-- 2 files changed, 24 insertions(+), 17 deletions(-) diff --git a/scripts/lint_common.py b/scripts/lint_common.py index b8a5b8ac4..7d08c01a0 100644 --- a/scripts/lint_common.py +++ b/scripts/lint_common.py @@ -1,6 +1,7 @@ """Process and report helpers for the analyzer drivers.""" import contextlib +from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout import os import re import shlex @@ -106,22 +107,25 @@ def run(command, *, input=None, check=False, progress=None, **kwargs): process = start(command, **kwargs) indicator = Progress(progress, quiet=kwargs.get("stdout") is not None and kwargs.get("stderr") is not None) - try: - if indicator.animated: - indicator.update() - while True: - try: - stdout, stderr = process.communicate( - input, timeout=indicator.interval if progress else None) - break - except subprocess.TimeoutExpired: - # communicate retains pending input and output across retries. - input = None + # Python 3.12 can stop writing pending stdin after a communicate timeout. + # Keep one I/O operation running while the main thread updates progress. + with ThreadPoolExecutor(max_workers=1) as executor: + try: + if indicator.animated: indicator.update() - except BaseException: - indicator.clear() - stop(process) - raise + communication = executor.submit(process.communicate, input) + while True: + try: + stdout, stderr = communication.result( + timeout=indicator.interval if progress else None) + break + except FutureTimeout: + indicator.update() + except BaseException: + indicator.clear() + # Close the process group's pipes before joining the I/O worker. + stop(process) + raise result = subprocess.CompletedProcess(command, exit_status(process.returncode), stdout, stderr) indicator.finish(result.returncode) if check: diff --git a/scripts/tests/test_lint_common.py b/scripts/tests/test_lint_common.py index 334b0dae7..4b91e520c 100644 --- a/scripts/tests/test_lint_common.py +++ b/scripts/tests/test_lint_common.py @@ -53,14 +53,17 @@ class ProcessTest(unittest.TestCase): return size script = """ -import os, sys, time +import os, signal, sys, time from pathlib import Path -payload = sys.stdin.buffer.read() +# Bound a blocked stdin read too, not only the progress handshake. +signal.alarm(5) deadline = time.monotonic() + 5 while not Path(sys.argv[1]).exists(): if time.monotonic() > deadline: sys.exit(99) time.sleep(0.01) +# Read only after progress has advanced, forcing timeouts with pending input. +payload = sys.stdin.buffer.read() os.write(1, payload) os.write(2, b'diagnostic\\x00\\xff') sys.exit(7)