tests: Harden subprocess management

tests/integration/subinterpreters_program.py

@@ -3,6 +3,7 @@
 import threading
 import time
 
+from subinterpreters_shim import read_n
 from subinterpreters_shim import run_in_new_interpreter
 
 NUM_INTERPRETERS = 3
@@ -28,9 +29,7 @@ def start_interpreter_async(code):
 for _ in range(NUM_INTERPRETERS):
     start_interpreter_async(CODE)
 
-data = b""
-while len(data) < NUM_INTERPRETERS:
-    data += os.read(r_fd, NUM_INTERPRETERS - len(data))
+read_n(r_fd, NUM_INTERPRETERS, timeout=30.0)
 os.close(r_fd)
 os.close(w_fd)
 

tests/integration/subinterpreters_shim.py

@@ -1,20 +1,65 @@
+import os
+import select
+import sys
+import time
+import traceback
+
+
+def abort(reason):
+    """Terminate this process immediately, with a diagnostic on stderr."""
+    sys.stderr.write(f"subinterpreter test program failed: {reason}\n")
+    sys.stderr.flush()
+    os._exit(1)
+
+
+def read_n(read_fd, count, *, timeout):
+    """Read exactly *count* bytes from *read_fd*."""
+    deadline = time.monotonic() + timeout
+    data = b""
+    while len(data) < count:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            abort(f"only {len(data)} of {count} bytes arrived within {timeout} seconds")
+        if not select.select([read_fd], [], [], remaining)[0]:
+            continue
+        chunk = os.read(read_fd, count - len(data))
+        if not chunk:
+            abort(f"EOF with {count - len(data)} bytes missing")
+        data += chunk
+    return data
+
+
 try:
     from concurrent import interpreters  # type: ignore
 
     def run_in_new_interpreter(code):
-        interpreters.create().exec(code)
+        try:
+            interpreters.create().exec(code)
+        except BaseException:
+            traceback.print_exc(file=sys.stderr)
+            abort("Terminating due to the above exception")
 
 except ImportError:
     try:
         import _interpreters  # type: ignore
 
         def run_in_new_interpreter(code):
-            _interpreters.exec(_interpreters.create(), code)
+            excinfo = _interpreters.exec(_interpreters.create(), code)
+            if excinfo is not None:
+                abort(
+                    getattr(excinfo, "errdisplay", None)
+                    or getattr(excinfo, "formatted", None)
+                    or repr(excinfo)
+                )
 
     except ImportError:
         import _xxsubinterpreters  # type: ignore
 
         def run_in_new_interpreter(code):
-            _xxsubinterpreters.run_string(
-                _xxsubinterpreters.create(isolated=False), code
-            )
+            try:
+                _xxsubinterpreters.run_string(
+                    _xxsubinterpreters.create(isolated=False), code
+                )
+            except BaseException:
+                traceback.print_exc(file=sys.stderr)
+                abort("Terminating due to the above exception")

tests/integration/subinterpreters_two_chains_program.py

@@ -3,6 +3,7 @@
 import threading
 import time
 
+from subinterpreters_shim import read_n
 from subinterpreters_shim import run_in_new_interpreter
 
 r_fd, w_fd = os.pipe()
@@ -28,9 +29,7 @@ def launch_chain():
 t1.start()
 t2.start()
 
-data = b""
-while len(data) < 2:
-    data += os.read(r_fd, 2 - len(data))
+read_n(r_fd, 2, timeout=30.0)
 os.close(r_fd)
 os.close(w_fd)
 

tests/integration/subinterpreters_with_threads_program.py

@@ -3,6 +3,7 @@
 import threading
 import time
 
+from subinterpreters_shim import read_n
 from subinterpreters_shim import run_in_new_interpreter
 
 NUM_INTERPRETERS = 2
@@ -48,9 +49,7 @@ def worker():
 
 TOTAL_EXPECTED = NUM_INTERPRETERS * (NUM_THREADS_PER_SUBINTERPRETER + 1)
 
-data = b""
-while len(data) < TOTAL_EXPECTED:
-    data += os.read(r_fd, TOTAL_EXPECTED - len(data))
+read_n(r_fd, TOTAL_EXPECTED, timeout=30.0)
 os.close(r_fd)
 os.close(w_fd)
 

tests/utils.py

@@ -3,6 +3,7 @@
 import itertools
 import os
 import pathlib
+import select
 import shutil
 import subprocess
 import sys
@@ -42,6 +43,70 @@
 Interpreter = collections.namedtuple("Interpreter", "version path has_symbols")
 
 
+def _format_child_output(output) -> str:
+    if isinstance(output, bytes):
+        return output.decode(errors="replace")
+    return output or "<empty>"
+
+
+def _raise_child_startup_error(process: "subprocess.Popen", error: str) -> None:
+    exit_code = process.poll()
+    if exit_code is None:
+        process.terminate()
+        try:
+            stdout, stderr = process.communicate(timeout=1)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            stdout, stderr = process.communicate(timeout=TIMEOUT)
+        status = "still running and was terminated by the test"
+    else:
+        stdout, stderr = process.communicate(timeout=TIMEOUT)
+        status = f"exited with status {exit_code}"
+
+    raise AssertionError(
+        f"Child process {process.pid} {error}; it {status}.\n"
+        f"Command: {process.args!r}\n"
+        f"stdout:\n{_format_child_output(stdout)}\n"
+        f"stderr:\n{_format_child_output(stderr)}"
+    )
+
+
+def _wait_for_child_ready(
+    process: "subprocess.Popen", fifo: pathlib.Path, timeout: float = TIMEOUT
+) -> None:
+    deadline = time.monotonic() + timeout
+    response = bytearray()
+
+    read_fd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK)
+    # Ensure the FIFO has been opened for writing before we try to read it.
+    write_fd = os.open(fifo, os.O_WRONLY | os.O_NONBLOCK)
+    try:
+        while True:
+            if process.poll() is not None:
+                _raise_child_startup_error(process, "exited before reporting readiness")
+
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                _raise_child_startup_error(
+                    process, f"did not report readiness within {timeout:g} seconds"
+                )
+
+            readable, _, _ = select.select([read_fd], [], [], min(remaining, 0.1))
+            if not readable:
+                continue
+
+            response.extend(os.read(read_fd, 4096))
+            if response == b"ready":
+                return
+            if response and not b"ready".startswith(response):
+                _raise_child_startup_error(
+                    process, f"reported unexpected readiness value {bytes(response)!r}"
+                )
+    finally:
+        os.close(write_fd)
+        os.close(read_fd)
+
+
 def find_all_available_pythons() -> Iterable[Interpreter]:  # pragma: no cover
     versions: List[Tuple[Tuple[int, int], str]]
     test_version = os.getenv("PYTHON_TEST_VERSION")
@@ -98,12 +163,9 @@ def spawn_child_process(
         stderr=subprocess.PIPE,
         text=True,
     ) as process:
-        with open(fifo, "r") as fifo_file:
-            response = fifo_file.read()
-
-        assert response == "ready"
-        time.sleep(0.1)
         try:
+            _wait_for_child_ready(process, fifo)
+            time.sleep(0.1)
             yield process
         finally:
             os.remove(fifo)
@@ -126,10 +188,7 @@ def generate_core_file(
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
     ) as process:
-        with open(fifo, "r") as fifo_file:
-            response = fifo_file.read()
-
-        assert response == "ready"
+        _wait_for_child_ready(process, fifo)
         subprocess.run(
             ["gcore", str(process.pid)],
             check=True,