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)