Merge pull request #419 from RonnyPfannschmidt/fix/safe-terminate-timeout-and-gevent-fdopen

fix: bound safe_terminate waits and gevent stdio exit

Author: RonnyPfannschmidt

src/execnet/gateway_base.py

@@ -273,10 +273,12 @@ def start(self, func, args=()) -> None:
         gevent.spawn(func, *args)
 
     def fdopen(self, fd, mode, bufsize=1, closefd=True):
-        # XXX
         import gevent.fileobject
 
-        return gevent.fileobject.FileObjectThread(fd, mode, bufsize, closefd=closefd)
+        # Prefer FileObject (FileObjectPosix on Unix). FileObjectThread keeps a
+        # native threadpool alive and can prevent interpreter shutdown, which
+        # hangs tests/scripts that open stdio via init_popen_io and then exit.
+        return gevent.fileobject.FileObject(fd, mode, bufsize, closefd=closefd)
 
     def Lock(self):
         import gevent.lock

src/execnet/multi.py

@@ -17,6 +17,7 @@
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Literal
+from typing import TypeAlias
 from typing import overload
 
 from . import gateway_bootstrap
@@ -328,25 +329,44 @@ def waitclose(self) -> None:
             raise first
 
 
+TermKillFunc: TypeAlias = Callable[[], object]
+TermKillPair: TypeAlias = tuple[TermKillFunc, TermKillFunc]
+
+
 def safe_terminate(
-    execmodel: ExecModel, timeout: float | None, list_of_paired_functions
+    execmodel: ExecModel,
+    timeout: float | None,
+    list_of_paired_functions: Sequence[TermKillPair],
 ) -> None:
+    """Run terminate/kill pairs in parallel with a hard wait bound.
+
+    Each termfunc is given ``timeout``.  If it does not finish, killfunc runs.
+    Waiting for the worker pool is also bounded so a stuck kill cannot hang
+    the caller forever (see issues #43 / #221).
+    """
     workerpool = WorkerPool(execmodel)
 
-    def termkill(termfunc, killfunc) -> None:
+    def termkill(termfunc: TermKillFunc, killfunc: TermKillFunc) -> None:
         termreply = workerpool.spawn(termfunc)
         try:
             termreply.get(timeout=timeout)
         except OSError:
             killfunc()
 
-    replylist = []
-    for termfunc, killfunc in list_of_paired_functions:
-        reply = workerpool.spawn(termkill, termfunc, killfunc)
-        replylist.append(reply)
+    replylist = [
+        workerpool.spawn(termkill, termfunc, killfunc)
+        for termfunc, killfunc in list_of_paired_functions
+    ]
+    # Allow term timeout plus a kill attempt; never block indefinitely.
+    wait_timeout = None if timeout is None else timeout * 2
     for reply in replylist:
-        reply.get()
-    workerpool.waitall(timeout=timeout)
+        try:
+            reply.waitfinish(timeout=wait_timeout)
+        except OSError:
+            # termkill still running (typically stuck in killfunc).
+            continue
+        reply.get()  # propagate worker exceptions, if any
+    workerpool.waitall(timeout=wait_timeout)
 
 
 default_group = Group()

testing/test_multi.py

@@ -278,3 +278,41 @@ def kill() -> None:
     sleep(0.1)
     gc.collect()
     assert threading.active_count() == active
+
+
+@pytest.mark.timeout(5)
+def test_safe_terminate_does_not_hang_when_kill_blocks(
+    execmodel: ExecModel,
+) -> None:
+    """Regression for #43/#221: a stuck kill must not hang terminate forever.
+
+    Before the fix, reply.get() waited without a timeout after termfunc timed
+    out, so a blocking killfunc made Group.terminate() hang indefinitely
+    (seen via pytest-xdist teardown).
+    """
+    kill_started = execmodel.Event()
+    release_kill = execmodel.Event()
+    other_killed: list[int] = []
+
+    def term_slow() -> None:
+        execmodel.sleep(10)
+
+    def kill_hang() -> None:
+        kill_started.set()
+        release_kill.wait()
+
+    def kill_ok() -> None:
+        other_killed.append(1)
+
+    safe_terminate(
+        execmodel,
+        0.2,
+        [
+            (term_slow, kill_hang),
+            (term_slow, kill_ok),
+        ],
+    )
+
+    assert kill_started.is_set()
+    assert other_killed == [1]
+    release_kill.set()