From 46c6d162a0837e162c522387e0a4961e8b5a8d77 Mon Sep 17 00:00:00 2001 From: Luna Date: Thu, 4 Jun 2026 23:52:09 -0700 Subject: [PATCH] Harden subprocess I/O against pipe-backpressure deadlock (roadmap A2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The executor drained child stdout/stderr via GLib.IOChannel watches on the main loop, so a stalled or wedged GUI stopped draining the pipes and the child would block on a full (~64KB) pipe forever — the cascade that froze a batch when the editor/terminal died. Replace the main-loop watches with per-stream background reader threads that drain pipes with blocking reads into thread-safe buffers; a GLib timer on the main thread consumes the buffers and does all parsing/UI (threads never touch GObject/GTK). Now the child can never block on a pipe we own, regardless of main-loop state. Also: stdout can go straight to a file when requested, stdin is fed from a file in a thread, the retained end-of-job log is bounded, and wait_end is guarded to run once. Tests (tests/test_executor_io.py): line parsing incl. CR progress lines, 2MB output with no loss/deadlock, completion with the main loop never running (the wedged-GUI case), cancel, stdout-to-file, and exit-code propagation. Co-Authored-By: Claude Opus 4.8 --- src/devedeng/executor.py | 297 ++++++++++++++++++++++++-------------- tests/test_executor_io.py | 139 ++++++++++++++++++ 2 files changed, 328 insertions(+), 108 deletions(-) create mode 100644 tests/test_executor_io.py diff --git a/src/devedeng/executor.py b/src/devedeng/executor.py index 82cf873..6c8af71 100644 --- a/src/devedeng/executor.py +++ b/src/devedeng/executor.py @@ -20,8 +20,16 @@ from gi.repository import GLib, GObject import subprocess import os import signal +import threading import devedeng.configuration_data +# how often (ms) the main loop drains the reader threads' buffers +_IO_POLL_MS = 150 +# cap on retained stdout/stderr text kept for the final log dump (bytes of +# text); the child's output is still streamed to process_*/the UI in full, this +# only bounds memory for the end-of-job log of very chatty processes +_LOG_RETAIN_CHARS = 200000 + class executor(GObject.GObject): """ This class encapsulates everything needed for launching processes """ @@ -33,9 +41,6 @@ class executor(GObject.GObject): GObject.GObject.__init__(self) self.config = devedeng.configuration_data.configuration.get_config() - self.channel_stdin = None - self.channel_stdout = None - self.channel_stderr = None self.text = "" self.stdout_data = "" self.stderr_data = "" @@ -51,6 +56,20 @@ class executor(GObject.GObject): self.pulse_text = None self.handle = None + # --- background pipe draining state (see launch_process) ------------- + # Reader threads drain the child's stdout/stderr into these buffers so + # the child can NEVER block on a full pipe, even if the GTK main loop + # stalls. A GLib timer on the main thread consumes the buffers and does + # all parsing/UI work (threads never touch GObject/GTK). + self._io_lock = threading.Lock() + self._io_buffers = {} # key -> bytearray of undrained bytes + self._io_eof = {} # key -> True once that stream hits EOF + self._io_line_buf = {} # key -> str carry of a partial last line + self._io_threads = [] + self._io_keys = [] # which streams are being drained + self._io_poll_id = None + self._io_finished = False + def add_dependency(self, dep): self.add_dependency2(dep) @@ -154,38 +173,17 @@ class executor(GObject.GObject): if command is not None: self.config.append_log(self.launch_command) - try: - if (self.stdin_file is not None): - self.handle = subprocess.Popen( - command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - self.channel_stdin = GLib.IOChannel(self.handle.stdin.fileno()) - self.channel_stdin.add_watch( - GLib.IO_OUT | GLib.IO_HUP, self.read_stdin_from_file) - self.file_in = open(self.stdin_file, "rb") - else: + if not redirect_output: + # synchronous path (used by the analyzers): communicate() drains + # both pipes itself, so there is no deadlock risk here + try: self.handle = subprocess.Popen( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - except Exception as error_launch: - self.handle = None - self.stderr_data += str(error_launch) - self.wait_end() - return - - if (redirect_output): - self.stdout_buf = "" - self.stderr_buf = "" - self.channel_stdout = GLib.IOChannel(self.handle.stdout.fileno()) - self.channel_stderr = GLib.IOChannel(self.handle.stderr.fileno()) - if (self.stdout_file is not None): - self.channel_stdout.add_watch( - GLib.IO_IN | GLib.IO_HUP, self.read_stdout_to_file) - self.file_out = open(self.stdout_file, "wb") - else: - self.channel_stdout.add_watch( - GLib.IO_IN | GLib.IO_HUP, self.read_stdout) - self.channel_stderr.add_watch( - GLib.IO_IN | GLib.IO_HUP, self.read_stderr) - else: + except Exception as error_launch: + self.handle = None + self.stderr_data += str(error_launch) + self.wait_end() + return (stdout_r, stderr_r) = self.handle.communicate() self.handle = None self.config.append_log(self.launch_command) @@ -199,6 +197,159 @@ class executor(GObject.GObject): self.config.append_log(stderr_r.decode("latin-1")) return (stdout_r, stderr_r) + # asynchronous path: a process whose output we stream to the UI. + # stdout may be redirected straight to a file (the child writes it, so + # it never blocks); stderr (and stdout when parsed) are drained by + # background threads so a stalled main loop can't deadlock the child. + stdout_target = subprocess.PIPE + self._out_file = None + if self.stdout_file is not None: + self._out_file = open(self.stdout_file, "wb") + stdout_target = self._out_file + try: + if self.stdin_file is not None: + self.handle = subprocess.Popen( + command, stdin=subprocess.PIPE, + stdout=stdout_target, stderr=subprocess.PIPE) + else: + self.handle = subprocess.Popen( + command, stdout=stdout_target, stderr=subprocess.PIPE) + except Exception as error_launch: + self.handle = None + if self._out_file is not None: + try: + self._out_file.close() + except OSError: + pass + self.stderr_data += str(error_launch) + self.wait_end() + return + + # feed stdin from a file in a thread (writing can block if the child + # stops reading; blocking a worker thread is harmless) + if self.stdin_file is not None: + t = threading.Thread(target=self._stdin_writer, daemon=True) + self._io_threads.append(t) + t.start() + + # start a draining thread for each pipe we own + if self.stdout_file is None: + self._start_reader("stdout", self.handle.stdout) + self._start_reader("stderr", self.handle.stderr) + + # consume the drained buffers on the main loop + self._io_poll_id = GLib.timeout_add(_IO_POLL_MS, self._io_poll) + + def _start_reader(self, key, fileobj): + self._io_keys.append(key) + self._io_buffers[key] = bytearray() + self._io_eof[key] = False + self._io_line_buf[key] = "" + t = threading.Thread(target=self._reader, args=(key, fileobj), + daemon=True) + self._io_threads.append(t) + t.start() + + def _reader(self, key, fileobj): + """ Runs in a background thread: drains a pipe with blocking reads into + a buffer until EOF. Never touches GObject/GTK. """ + fd = fileobj.fileno() + try: + while True: + chunk = os.read(fd, 65536) # returns b"" at EOF + if not chunk: + break + with self._io_lock: + self._io_buffers[key].extend(chunk) + except OSError: + pass + finally: + with self._io_lock: + self._io_eof[key] = True + + def _stdin_writer(self): + try: + with open(self.stdin_file, "rb") as f: + while True: + chunk = f.read(65536) + if not chunk: + break + self.handle.stdin.write(chunk) + self.handle.stdin.flush() + except (OSError, ValueError): + pass + finally: + try: + self.handle.stdin.close() + except (OSError, ValueError): + pass + + def _io_poll(self): + """ Main-thread timer: pull whatever the reader threads have buffered, + split into lines, and dispatch to process_stdout/process_stderr. + Finishes the job once every stream has hit EOF. """ + self._io_drain() + with self._io_lock: + all_eof = all(self._io_eof.get(k, True) for k in self._io_keys) + if all_eof: + self._io_drain(final=True) + self._io_poll_id = None + self._finish_io() + return False # remove the timer + return True + + def _io_drain(self, final=False): + for key in self._io_keys: + with self._io_lock: + if not self._io_buffers[key]: + data_bytes = b"" + else: + data_bytes = bytes(self._io_buffers[key]) + self._io_buffers[key] = bytearray() + if not data_bytes and not final: + continue + try: + text = data_bytes.decode("utf-8") + except UnicodeDecodeError: + text = data_bytes.decode("latin-1") + self._dispatch_lines(key, text, final) + + def _dispatch_lines(self, key, text, final): + line_data = self._io_line_buf[key] + text + # accumulate a bounded copy for the end-of-job log + if key == "stdout": + self.stdout_data = (self.stdout_data + line_data)[-_LOG_RETAIN_CHARS:] + else: + self.stderr_data = (self.stderr_data + line_data)[-_LOG_RETAIN_CHARS:] + parts = line_data.replace("\r", "\n").split("\n") + if final: + complete = [p for p in parts if p != ""] + self._io_line_buf[key] = "" + elif len(parts) == 1: + complete = [] + self._io_line_buf[key] = parts[0] + else: + complete = parts[:-1] + self._io_line_buf[key] = parts[-1] + if complete: + if key == "stdout": + self.process_stdout(complete) + else: + self.process_stderr(complete) + + def _finish_io(self): + # reader threads have signalled EOF; join them briefly then finish once + for t in self._io_threads: + t.join(timeout=1.0) + self._io_threads = [] + if self._out_file is not None: + try: + self._out_file.close() + except OSError: + pass + self._out_file = None + self.wait_end() + def set_pulse_mode(self, pulse_mode): if pulse_mode == self.pulse_mode: @@ -221,81 +372,6 @@ class executor(GObject.GObject): self.progress_bar[1].pulse() return True - def read_stdout_to_file(self, source, condition): - - if (condition != GLib.IO_IN): - self.channel_stdout = None - if ((self.channel_stderr is None) and (self.channel_stdin is None)): - self.wait_end() - return False - else: - line_data = self.handle.stdout.read1(4096) - self.file_out.write(line_data) - return True - - def read_stdin_from_file(self, source, condition): - - line_data = self.file_in.read1(4096) - if (len(line_data) == 0): - self.channel_stdin = None - self.handle.stdin.close() - if ((self.channel_stderr is None) and (self.channel_stdout is None)): - self.wait_end() - return False - else: - self.handle.stdin.write(line_data) - return True - - def read_stdout(self, source, condition): - - if (condition != GLib.IO_IN): - self.channel_stdout = None - if ((self.channel_stderr is None) and (self.channel_stdin is None)): - self.wait_end() - return False - else: - read_data = self.handle.stdout.read1(4096) - try: - line_data = self.stdout_buf + (read_data.decode("utf-8")) - except: - line_data = self.stdout_buf + (read_data.decode("latin-1")) - self.stdout_data += line_data - data = (line_data).replace("\r", "\n").split("\n") - if (len(data) == 1): - final_data = [] - self.stdout_buf = data[0] - else: - final_data = data[:-1] - self.stdout_buf = data[-1] - if (len(final_data) != 0): - self.process_stdout(final_data) - return True - - def read_stderr(self, source, condition): - - if (condition != GLib.IO_IN): - self.channel_stderr = None - if (((self.channel_stdout is None) or (self.stdout_file is not None)) and ((self.channel_stdin is None) or (self.stdin_file is not None))): - self.wait_end() - return False - else: - read_data = self.handle.stderr.read1(4096) - try: - line_data = self.stderr_buf + (read_data.decode("utf-8")) - except: - line_data = self.stderr_buf + (read_data.decode("latin-1")) - self.stderr_data += line_data - data = (line_data).replace("\r", "\n").split("\n") - if (len(data) == 1): - final_data = [] - self.stderr_buf = data[0] - else: - final_data = data[:-1] - self.stderr_buf = data[-1] - if (len(final_data) != 0): - self.process_stderr(final_data) - return True - def cancel(self): """ Called to kill this process. """ @@ -307,6 +383,11 @@ class executor(GObject.GObject): def wait_end(self): + # guard against being entered twice (e.g. launch error + io finish) + if self._io_finished: + return + self._io_finished = True + if self.handle is not None: retval = self.handle.wait() self.handle = None diff --git a/tests/test_executor_io.py b/tests/test_executor_io.py new file mode 100644 index 0000000..3821b26 --- /dev/null +++ b/tests/test_executor_io.py @@ -0,0 +1,139 @@ +"""Tests for the hardened subprocess I/O in executor.py. + +The executor drains child stdout/stderr with background threads (not the GLib +main loop), so a stalled/wedged GUI can never deadlock a child on a full pipe. +""" + +import time +import os +import tempfile + +import gi +gi.require_version("Gtk", "3.0") +from gi.repository import GLib + +import devedeng.executor as ex + + +class CaptureExecutor(ex.executor): + """Executor subclass that records the lines handed to process_*.""" + + def __init__(self): + super().__init__() + self.out_lines = [] + self.err_lines = [] + + def process_stdout(self, data): + self.out_lines += data + + def process_stderr(self, data): + self.err_lines += data + + +def _run_to_end(executor, timeout_s=20): + """Drive a GLib main loop until the executor emits 'ended' (or times out). + Returns the process return value, or raises on timeout (a hang).""" + loop = GLib.MainLoop() + result = {} + + def on_ended(_obj, ret): + result["ret"] = ret + loop.quit() + + executor.connect("ended", on_ended) + executor.launch_process(executor.command_var) + GLib.timeout_add(int(timeout_s * 1000), + lambda: (result.setdefault("timeout", True), loop.quit())[1]) + loop.run() + assert not result.get("timeout"), "executor hung (no 'ended' within timeout)" + return result["ret"] + + +def test_line_parsing_stdout_and_stderr_with_cr(): + e = CaptureExecutor() + e.command_var = ["python3", "-c", + "import sys\n" + "sys.stdout.write('o1\\no2\\n'); sys.stdout.flush()\n" + "sys.stderr.write('p 1\\rp 2\\rdone\\n'); sys.stderr.flush()\n"] + ret = _run_to_end(e) + assert ret == 0 + assert e.out_lines == ["o1", "o2"] + # carriage-return progress lines are split like newlines + assert e.err_lines == ["p 1", "p 2", "done"] + + +def test_large_output_no_deadlock_no_loss(): + # ~2 MB of stderr, far beyond the ~64 KB pipe buffer + e = CaptureExecutor() + e.command_var = ["python3", "-c", + "import sys\n" + "for i in range(20000): sys.stderr.write('e%d\\n'%i)\n" + "for i in range(10000): sys.stdout.write('o%d\\n'%i)\n"] + ret = _run_to_end(e) + assert ret == 0 + assert len(e.err_lines) == 20000 # nothing lost + assert len(e.out_lines) == 10000 + + +def test_completes_with_no_main_loop_running(): + # the wedged-GUI scenario: never service the GLib loop; the reader threads + # must still drain the pipe so the child can finish instead of deadlocking + e = CaptureExecutor() + e.command_var = ["python3", "-c", + "import sys\n" + "for i in range(20000): sys.stderr.write('x'*100+'\\n')\n"] + e.launch_process(e.command_var) + deadline = time.time() + 15 + while time.time() < deadline: + if e.handle is None or e.handle.poll() is not None: + break + time.sleep(0.05) + rc = "reaped" if e.handle is None else e.handle.poll() + assert rc is not None, "child deadlocked with no draining main loop" + + +def test_cancel_kills_and_ends(): + e = CaptureExecutor() + e.command_var = ["python3", "-c", "import time; time.sleep(60)"] + loop = GLib.MainLoop() + result = {} + e.connect("ended", lambda o, r: (result.update(ret=r), loop.quit())) + e.launch_process(e.command_var) + # cancel shortly after start + GLib.timeout_add(300, lambda: (e.cancel(), False)[1]) + GLib.timeout_add(15000, lambda: (result.setdefault("timeout", True), + loop.quit())[1]) + loop.run() + assert not result.get("timeout"), "cancel did not end the process" + # a killed process reports success (retval forced to 0 by design) + assert result["ret"] == 0 + assert e.killed is True + + +def test_stdout_redirected_to_file(): + tmp = tempfile.mktemp(suffix=".out") + e = CaptureExecutor() + e.stdout_file = tmp + e.command_var = ["python3", "-c", + "import sys\n" + "sys.stdout.write('FILE CONTENT\\n')\n" + "sys.stderr.write('status line\\n')\n"] + try: + ret = _run_to_end(e) + assert ret == 0 + with open(tmp) as f: + assert "FILE CONTENT" in f.read() + # stderr is still parsed while stdout goes to the file + assert "status line" in e.err_lines + # stdout was not also dispatched as lines + assert e.out_lines == [] + finally: + if os.path.exists(tmp): + os.remove(tmp) + + +def test_nonzero_exit_code_propagates(): + e = CaptureExecutor() + e.command_var = ["python3", "-c", "import sys; sys.exit(3)"] + ret = _run_to_end(e) + assert ret == 3