Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions src/execnet/gateway_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
36 changes: 28 additions & 8 deletions src/execnet/multi.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Comment on lines +363 to +367
reply.get() # propagate worker exceptions, if any
workerpool.waitall(timeout=wait_timeout)


default_group = Group()
Expand Down
38 changes: 38 additions & 0 deletions testing/test_multi.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Comment on lines +294 to +318