From 945640933df57b62d1fddf0485a99c25e5e4e246 Mon Sep 17 00:00:00 2001 From: "Ralf W. Grosse-Kunstleve" Date: Mon, 19 Oct 2026 08:40:56 -0700 Subject: [PATCH] Consolidate multiprocessing child-process test helpers Share the child runner, picklable no-op target, and cached spawn warmup between the VMM IPC and file-descriptor leak tests. Kill children when joining is interrupted and close Process handles before returning. Document helper reuse and cover successful execution, failed exits, timeouts, interrupted joins, warmup retries, and descriptor stability. Follow up on #3146. Signed-off-by: Ralf W. Grosse-Kunstleve --- cuda_core/tests/AGENTS.md | 9 ++ cuda_core/tests/helpers/child_processes.py | 32 +++++ cuda_core/tests/memory_ipc/test_leaks.py | 34 +---- cuda_core/tests/memory_ipc/test_vmm_ipc.py | 30 +---- cuda_core/tests/test_child_processes.py | 150 +++++++++++++++++++++ 5 files changed, 202 insertions(+), 53 deletions(-) create mode 100644 cuda_core/tests/test_child_processes.py diff --git a/cuda_core/tests/AGENTS.md b/cuda_core/tests/AGENTS.md index 4cd50de44a2..0ea219877f9 100644 --- a/cuda_core/tests/AGENTS.md +++ b/cuda_core/tests/AGENTS.md @@ -96,6 +96,15 @@ Follow these rules when adding or moving shared test code: `tests/helpers/` instead. - Import helpers explicitly from the test root, for example: `from helpers.memory import create_managed_memory_resource_or_skip`. +- Reuse `run_child` and `noop` from `helpers.child_processes` for synchronous + child-process tests. The runner applies the sanitizer-aware timeout, kills + surviving children before asserting success, and closes process handles. + Before measuring file-descriptor counts around spawned processes, call + `warmup_multiprocessing` from the same module to start the persistent + resource-tracker pipe outside the baseline. Keep explicit process management + when the test needs to communicate with a running child or inspect its lifetime; + use `kill_subprocesses` before assertions and `track_child_processes` for + fixture cleanup so a stuck child cannot block IPC resource teardown. - Search `tests/helpers/` and `cuda_python_test_helpers/` for prior art before adding a new helper; consolidate duplicates across diff --git a/cuda_core/tests/helpers/child_processes.py b/cuda_core/tests/helpers/child_processes.py index ed9300c77bd..b13137200b0 100644 --- a/cuda_core/tests/helpers/child_processes.py +++ b/cuda_core/tests/helpers/child_processes.py @@ -11,6 +11,8 @@ """ import contextlib +import functools +import multiprocessing as mp import multiprocessing.process import weakref @@ -64,6 +66,36 @@ def kill_subprocesses(*processes): return killed +def run_child(target, *args): + """Run ``target(*args)`` in a child, killing it on timeout or interrupted join. + + Fail unless the child exits successfully. Close the Process object before + returning so its handles cannot affect file-descriptor leak checks. + """ + with contextlib.closing(mp.Process(target=target, args=args)) as process: + try: + process.start() + process.join(timeout=child_timeout_sec()) + finally: + survivors = kill_subprocesses(process) + assert not survivors, "child did not exit within timeout" + assert process.exitcode == 0, f"child exited with {process.exitcode}" + + +def noop(*args): + """A picklable child target that accepts and discards any arguments.""" + + +@functools.cache +def warmup_multiprocessing(): + """Start spawn's persistent resource-tracker pipe before counting open FDs. + + Cache only a successful warmup; the pipe outlives individual Process objects. + """ + assert mp.get_start_method() == "spawn" + run_child(noop) + + @contextlib.contextmanager def track_child_processes(): """Context manager that kills any ``multiprocessing.Process`` children still diff --git a/cuda_core/tests/memory_ipc/test_leaks.py b/cuda_core/tests/memory_ipc/test_leaks.py index 8bb29922f3c..50240f57731 100644 --- a/cuda_core/tests/memory_ipc/test_leaks.py +++ b/cuda_core/tests/memory_ipc/test_leaks.py @@ -13,7 +13,7 @@ else: HAVE_PSUTIL = True import pytest -from helpers.child_processes import child_timeout_sec, kill_subprocesses +from helpers.child_processes import child_timeout_sec, kill_subprocesses, noop, run_child, warmup_multiprocessing CHILD_TIMEOUT_SEC = child_timeout_sec() NBYTES = 64 @@ -40,16 +40,7 @@ def test_alloc_handle(ipc_memory_resource): def exec_success(obj, number=1): """Successfully run a child process.""" for _ in range(number): - process = mp.Process(target=child_main, args=(obj,)) - process.start() - process.join(timeout=CHILD_TIMEOUT_SEC) - survivors = kill_subprocesses(process) - assert not survivors, "child did not exit within timeout" - assert process.exitcode == 0 - - -def child_main(obj, *args): - pass + run_child(noop, obj) def exec_launch_failure(obj, number=1): @@ -78,7 +69,7 @@ def exec_reduce_failure(obj, number=1): for _ in range(number): fails_to_reduce = Irreducible() with contextlib.suppress(RuntimeError): - mp.Process(target=child_main, args=(obj, fails_to_reduce)).start() + mp.Process(target=noop, args=(obj, fails_to_reduce)).start() class Irreducible: @@ -137,7 +128,7 @@ def __init__(self): self.process = psutil.Process() def __enter__(self): - prime() + warmup_multiprocessing() gc.collect() self.initial_fds = self.process.num_fds() return self @@ -150,20 +141,3 @@ def __exit__(self, exc_type, exc_val, exc_tb): final_fds = self.process.num_fds() assert final_fds == self.initial_fds, f"file descriptors leaked: {self.initial_fds} -> {final_fds}" return False - - -prime_was_run = False - - -def prime(): - """Multiprocessing consumes a file descriptor on first launch.""" - assert mp.get_start_method() == "spawn" - global prime_was_run - if not prime_was_run: - process = mp.Process() - process.start() - process.join(timeout=CHILD_TIMEOUT_SEC) - survivors = kill_subprocesses(process) - assert not survivors, "child did not exit within timeout" - assert process.exitcode == 0 - prime_was_run = True diff --git a/cuda_core/tests/memory_ipc/test_vmm_ipc.py b/cuda_core/tests/memory_ipc/test_vmm_ipc.py index 373f98f7fb8..0aba6ad4bea 100644 --- a/cuda_core/tests/memory_ipc/test_vmm_ipc.py +++ b/cuda_core/tests/memory_ipc/test_vmm_ipc.py @@ -22,7 +22,7 @@ import threading import pytest -from helpers.child_processes import child_timeout_sec, kill_subprocesses +from helpers.child_processes import child_timeout_sec, kill_subprocesses, noop, run_child, warmup_multiprocessing from helpers.contexts import assert_no_cuda_warning from cuda.bindings import driver @@ -149,20 +149,6 @@ class _CustomResource(VirtualMemoryResource): """A subclass, to check that pickling preserves the type.""" -def _run_child(target, *args): - """Run ``target(*args)`` in a spawned child and fail if it does not exit cleanly.""" - process = mp.Process(target=target, args=args) - process.start() - process.join(timeout=CHILD_TIMEOUT_SEC) - survivors = kill_subprocesses(process) - assert not survivors, "child did not exit within timeout" - assert process.exitcode == 0, f"child exited with {process.exitcode}" - - -def _noop(*args): - pass - - def _close_argument(buffer): buffer.close() @@ -260,7 +246,7 @@ def test_main(self, vmm_ipc_device): assert _open_fds() == fds_before + 3 fds = [int(handle) for handle in desc._handles] - _run_child(self.child_main, device.device_id, desc, gran, queue) + run_child(self.child_main, device.device_id, desc, gran, queue) result = queue.get(timeout=CHILD_TIMEOUT_SEC) assert result["chunks"] == [bytes([value]) * 16 for value in values] assert result["released"] is True @@ -311,7 +297,7 @@ def test_main(self, vmm_ipc_device_x2): peers_possible = dev1.can_access_peer(dev0) queue = mp.Queue() - _run_child(self.child_main, dev0.device_id, dev1.device_id, desc, peers_possible, queue) + run_child(self.child_main, dev0.device_id, dev1.device_id, desc, peers_possible, queue) result = queue.get(timeout=CHILD_TIMEOUT_SEC) assert f"device {dev0.device_id}" in result["wrong_device"] assert result["device_id"] == dev1.device_id @@ -495,7 +481,7 @@ def test_export_and_pickling_raise_before_any_driver_call(self, vmm_ipc_device): with parent_conn, child_conn, pytest.raises(RuntimeError, match=_REEXPORT_MESSAGE): parent_conn.send(imported) with pytest.raises(RuntimeError, match=_REEXPORT_MESSAGE): - mp.Process(target=_noop, args=(imported,)).start() + mp.Process(target=noop, args=(imported,)).start() # The original descriptor still serves new importers, and the data is intact. _fill(buf, 0x3C) alias = Buffer.from_ipc_descriptor(mr, desc, stream=stream) @@ -589,9 +575,7 @@ def test_exporter_holds_none_without_a_descriptor(self, vmm_ipc_device): @pytest.mark.thread_unsafe(reason="counts process-global file descriptors") def test_process_argument_releases_with_the_process_object(self, vmm_ipc_device): """A buffer passed to a spawned Process imports there; its descriptor goes when the Process object does.""" - # The first spawn opens a process-wide resource-tracker pipe, which - # outlives Process objects. Start it before taking the descriptor baseline. - _run_child(_noop) + warmup_multiprocessing() device = vmm_ipc_device mr = _resource(device) with mr.allocate(1) as buf: @@ -926,7 +910,7 @@ def test_round_trip(self, vmm_ipc_device): assert alias.is_mapped alias.close() queue = mp.Queue() - _run_child(self.child_main, device.device_id, desc, queue) + run_child(self.child_main, device.device_id, desc, queue) assert queue.get(timeout=CHILD_TIMEOUT_SEC) == {"size": 0, "is_mapped": True} empty.close() @@ -948,7 +932,7 @@ def test_imported_buffer_feeds_a_consumer(self, vmm_ipc_device): buf = mr.allocate(1) _fill(buf, 0x7E) queue = mp.Queue() - _run_child(self.child_main, device.device_id, buf, queue) + run_child(self.child_main, device.device_id, buf, queue) result = queue.get(timeout=CHILD_TIMEOUT_SEC) assert result["dl_device"] == (int(DLDeviceType.kDLCUDA), device.device_id) assert result["shape"] == (buf.size,) diff --git a/cuda_core/tests/test_child_processes.py b/cuda_core/tests/test_child_processes.py new file mode 100644 index 00000000000..827f88d15d7 --- /dev/null +++ b/cuda_core/tests/test_child_processes.py @@ -0,0 +1,150 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Behavioral checks for the shared multiprocessing test helpers.""" + +import contextlib +import gc +import multiprocessing as mp +import os +import pathlib +import threading + +import pytest +from helpers import child_processes + +pytestmark = pytest.mark.thread_unsafe( + reason="changes multiprocessing globals and counts process-wide file descriptors" +) + + +def _write_arguments(path, *args): + path.write_text(repr(args)) + + +def _exit_failure(): + raise SystemExit(7) + + +def _wait_forever(ready): + ready.set() + threading.Event().wait() + + +@pytest.fixture(autouse=True) +def _spawn_context(): + previous = mp.get_start_method(allow_none=True) + mp.set_start_method("spawn", force=True) + child_processes.warmup_multiprocessing.cache_clear() + try: + yield + finally: + child_processes.warmup_multiprocessing.cache_clear() + mp.set_start_method(previous, force=True) + + +@pytest.fixture +def processes(monkeypatch): + original_process = mp.Process + tracked = [] + + def make_process(*args, **kwargs): + process = original_process(*args, **kwargs) + tracked.append(process) + return process + + monkeypatch.setattr(mp, "Process", make_process) + try: + yield tracked + finally: + child_processes.kill_subprocesses(*tracked) + for process in tracked: + with contextlib.suppress(ValueError): + process.close() + + +def _assert_closed(processes): + assert processes + for process in processes: + with pytest.raises(ValueError): + process.is_alive() + + +def _timeout_after_start(monkeypatch, ready): + startup_timeout = child_processes.child_timeout_sec() + + def short_timeout(): + assert ready.wait(timeout=startup_timeout), "child did not reach its target" + return 0.05 + + monkeypatch.setattr(child_processes, "child_timeout_sec", short_timeout) + + +@pytest.mark.agent_authored(model="gpt-6") +def test_run_child_passes_arguments_and_closes_process(tmp_path, processes): + output = tmp_path / "arguments.txt" + child_processes.run_child(_write_arguments, output, 42, "payload", None) + assert output.read_text() == repr((42, "payload", None)) + _assert_closed(processes) + + +@pytest.mark.agent_authored(model="gpt-6") +def test_run_child_reports_failed_exit_and_closes_process(processes): + with pytest.raises(AssertionError): + child_processes.run_child(_exit_failure) + _assert_closed(processes) + + +@pytest.mark.agent_authored(model="gpt-6") +def test_run_child_kills_hung_child(monkeypatch, processes): + ready = mp.Event() + _timeout_after_start(monkeypatch, ready) + with pytest.raises(AssertionError): + child_processes.run_child(_wait_forever, ready) + assert ready.is_set() + _assert_closed(processes) + + +@pytest.mark.agent_authored(model="gpt-6") +def test_run_child_kills_child_when_join_is_interrupted(monkeypatch, processes): + ready = mp.Event() + _timeout_after_start(monkeypatch, ready) + original_join = mp.process.BaseProcess.join + + def interrupted_join(process, timeout=None): + if timeout is not None: + raise InterruptedError("join interrupted") + return original_join(process) + + monkeypatch.setattr(mp.process.BaseProcess, "join", interrupted_join) + with pytest.raises(InterruptedError): + child_processes.run_child(_wait_forever, ready) + _assert_closed(processes) + + +@pytest.mark.agent_authored(model="gpt-6") +def test_warmup_retries_failure_and_caches_success(monkeypatch, processes): + noop = child_processes.noop + monkeypatch.setattr(child_processes, "noop", _exit_failure) + with pytest.raises(AssertionError): + child_processes.warmup_multiprocessing() + + monkeypatch.setattr(child_processes, "noop", noop) + child_processes.warmup_multiprocessing() + + monkeypatch.setattr(child_processes, "noop", _exit_failure) + child_processes.warmup_multiprocessing() + _assert_closed(processes) + + +@pytest.mark.skipif(not pathlib.Path("/proc/self/fd").is_dir(), reason="requires Linux file descriptor accounting") +@pytest.mark.agent_authored(model="gpt-6") +def test_repeated_warmup_and_run_child_leave_fd_count_stable(processes): + child_processes.warmup_multiprocessing() + gc.collect() + initial_fds = len(os.listdir("/proc/self/fd")) + for _ in range(3): + child_processes.warmup_multiprocessing() + child_processes.run_child(child_processes.noop, 42, "discarded", None) + assert len(os.listdir("/proc/self/fd")) == initial_fds + _assert_closed(processes)