From f45ed22a1d080c6341b97780d35b2b1f47d24a1b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Sat, 22 Aug 2026 15:29:45 -0700 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=AA=20test(fork):=20fork=20once=20the?= =?UTF-8?q?=20event=20loop=20has=20closed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The canceled-acquire test forked from inside the coroutine, right after awaiting shutdown_default_executor(). Thread.join() returns once the interpreter releases the thread state, while the OS thread is still unwinding, and NetBSD deadlocks a child forked from a process that still carries one, so the fork wedged and the 10s subprocess timeout killed it (#707). Move the descriptor swap and the fork to module scope, after asyncio.run has closed the loop, leave the worker time to go, and report a thread that is still counted rather than forking on it. Raise the subprocess timeout to 30s so a slow builder is not the failure mode either. --- tests/test_fork_coordination.py | 80 ++++++++++++++++++--------------- 1 file changed, 43 insertions(+), 37 deletions(-) diff --git a/tests/test_fork_coordination.py b/tests/test_fork_coordination.py index a6f56af4..0dba8c1d 100644 --- a/tests/test_fork_coordination.py +++ b/tests/test_fork_coordination.py @@ -780,13 +780,14 @@ def test_cancelled_async_acquire_unregisters_reused_descriptor(tmp_path: Path) - import os import sys import threading +import time import warnings from filelock import AsyncFileLock warnings.filterwarnings("ignore", message=".*multi-threaded, use of fork.*", category=DeprecationWarning) -async def main() -> int: +async def main() -> tuple[int, tuple[int, int]]: callback_started = asyncio.Event() finish_callback = threading.Event() loop = asyncio.get_running_loop() @@ -812,50 +813,55 @@ def block_after_acquire(fd: int) -> None: except asyncio.CancelledError: pass else: - return 1 + raise SystemExit(1) if lock.is_locked or lock.lock_counter != 0: - return 2 - - path_stat = os.stat(sys.argv[1]) - if (path_stat.st_dev, path_stat.st_ino) != identity: - return 3 - - occupant = os.open(os.devnull, os.O_RDONLY) - if occupant != descriptor: - os.dup2(occupant, descriptor) - source = os.open(sys.argv[1], os.O_RDWR) - if source == descriptor: - return 4 - if (source_stat := os.fstat(source)).st_dev != identity[0] or source_stat.st_ino != identity[1]: - return 5 - os.dup2(source, descriptor) - os.close(source) - if occupant != descriptor: - os.close(occupant) - - # The canceled acquire ran the flock backend in the default executor, leaving an idle worker thread behind. Join it - # before forking so the child is single-threaded: NetBSD deadlocks a child forked from a multi-threaded process. - await loop.shutdown_default_executor() + raise SystemExit(2) + return descriptor, identity + +descriptor, identity = asyncio.run(main()) + +path_stat = os.stat(sys.argv[1]) +if (path_stat.st_dev, path_stat.st_ino) != identity: + raise SystemExit(3) + +occupant = os.open(os.devnull, os.O_RDONLY) +if occupant != descriptor: + os.dup2(occupant, descriptor) +source = os.open(sys.argv[1], os.O_RDWR) +if source == descriptor: + raise SystemExit(4) +if (source_stat := os.fstat(source)).st_dev != identity[0] or source_stat.st_ino != identity[1]: + raise SystemExit(5) +os.dup2(source, descriptor) +os.close(source) +if occupant != descriptor: + os.close(occupant) + +# The canceled acquire ran the flock backend in the default executor. asyncio.run() joins that worker on the way out, +# but Thread.join() returns while the OS thread is still unwinding, and NetBSD deadlocks a child forked from a process +# that still carries one (#707). So fork from module scope with the loop closed, leave the worker time to go, and +# report a thread that is still counted rather than wedging the fork on it. +time.sleep(0.5) +if threading.active_count() != 1: + raise SystemExit(6) - child_pid = os.fork() - if child_pid == 0: - try: - replacement_stat = os.fstat(descriptor) - except OSError: - os._exit(1) - os._exit(0 if (replacement_stat.st_dev, replacement_stat.st_ino) == identity else 2) - _, status = os.waitpid(child_pid, 0) - os.close(descriptor) - return os.waitstatus_to_exitcode(status) - -raise SystemExit(asyncio.run(main())) +child_pid = os.fork() +if child_pid == 0: + try: + replacement_stat = os.fstat(descriptor) + except OSError: + os._exit(1) + os._exit(0 if (replacement_stat.st_dev, replacement_stat.st_ino) == identity else 2) +_, status = os.waitpid(child_pid, 0) +os.close(descriptor) +raise SystemExit(os.waitstatus_to_exitcode(status)) """ result = subprocess.run( [sys.executable, "-c", script, str(tmp_path / "canceled.lock")], check=False, capture_output=True, text=True, - timeout=10, + timeout=30, ) assert (result.returncode, result.stderr) == (0, "")