Harden subprocess I/O against pipe-backpressure deadlock (roadmap A2)

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 <noreply@anthropic.com>

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

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