test(core): consolidate multiprocessing child-process helpers

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
 

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

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

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,)

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)