tests: add kill_subprocesses helper and apply to IPC tests

Make every IPC test responsible for its own child-process cleanup, rather
than leaning on the fixture-level tracker as the primary mechanism.

helpers.child_processes.kill_subprocesses(*processes) returns the list of
processes that were still alive when called and kills them. Tests pair this
with an "assert not survivors" check, so a timeout produces a clean failure
message ("timed out waiting on: ['Process-3']") instead of "assert None ==
0", and the held IPC handles are released before any further test code
runs.

Every join+exitcode pattern in tests/memory_ipc/ is converted to the new
shape:

    process.join(timeout=CHILD_TIMEOUT_SEC)
    survivors = kill_subprocesses(process)
    assert not survivors, "child did not exit within timeout"
    assert process.exitcode == 0

For test_send_buffers.py:TestIpcReexport (the test from #2121), the
event.wait return value is also captured and asserted on so a timeout there
is reported explicitly rather than dropped on the floor.

The autouse track_child_processes() context manager in the ipc_device
fixture remains as defense in depth for any future test that forgets the
pattern.

memory_ipc/test_errors.py adds a meta-test test_outer_timeout_marker_is_applied
that verifies tests/memory_ipc/conftest.py is loaded and applies the
pytest.mark.timeout(300) marker. This catches the "nested conftest silently
not picked up" failure mode at test time with a clear error message.

Author: Andy-Jost

cuda_core/tests/helpers/child_processes.py

@@ -29,6 +29,41 @@ def child_timeout_sec() -> int:
     return CHILD_TIMEOUT_SEC_SANITIZER if under_compute_sanitizer() else CHILD_TIMEOUT_SEC_DEFAULT
 
 
+def kill_subprocesses(*processes):
+    """Kill any of the given Process objects that are still alive.
+
+    Returns the list of processes that were killed (i.e. that were still alive
+    when the call was made). Callers should ``assert not survivors`` to convert
+    a non-empty return value into a clean test failure, e.g.::
+
+        proc_a.join(timeout=CHILD_TIMEOUT_SEC)
+        proc_b.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(proc_a, proc_b)
+        assert not survivors, f"timed out waiting on: {[p.name for p in survivors]}"
+        assert proc_a.exitcode == 0
+        assert proc_b.exitcode == 0
+
+    Killing survivors before the subsequent asserts prevents a zombie child
+    from holding IPC handles past the test body and blocking fixture
+    teardown.
+    """
+    killed = []
+    for proc in processes:
+        try:
+            alive = proc.is_alive()
+        except (ValueError, AssertionError):
+            # is_alive() raises if the Process was never started or has
+            # already been closed; nothing to clean up.
+            continue
+        if not alive:
+            continue
+        with contextlib.suppress(ValueError, AssertionError):
+            proc.kill()
+            proc.join()
+        killed.append(proc)
+    return killed
+
+
 @contextlib.contextmanager
 def track_child_processes():
     """Context manager that kills any ``multiprocessing.Process`` children still

cuda_core/tests/memory_ipc/test_errors.py

@@ -6,7 +6,7 @@
 import re
 
 import pytest
-from helpers.child_processes import child_timeout_sec
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 
 from cuda.core import Buffer, Device, DeviceMemoryResource, DeviceMemoryResourceOptions
 from cuda.core._utils.cuda_utils import CUDAError
@@ -16,6 +16,20 @@
 POOL_SIZE = 2097152
 
 
+def test_outer_timeout_marker_is_applied(request):
+    """Verify that memory_ipc/conftest.py applies the outer pytest-timeout marker.
+
+    If this test fails, the per-directory conftest is not being loaded, or its
+    pytest_collection_modifyitems hook is not adding the marker. Without this
+    marker, the only thing protecting the GHA runner from a wedged IPC test is
+    the in-test cleanup -- which we want to keep as defense in depth, not as
+    the sole guard.
+    """
+    marker = request.node.get_closest_marker("timeout")
+    assert marker is not None, "memory_ipc/conftest.py did not apply a timeout marker"
+    assert marker.args == (300,), f"unexpected timeout value: {marker.args!r}"
+
+
 class ChildErrorHarness:
     """Test harness for checking errors in child processes. Subclasses override
     PARENT_ACTION, CHILD_ACTION, and ASSERT (see below for examples)."""
@@ -44,6 +58,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
 
             # Wait for the child process.
             process.join(timeout=CHILD_TIMEOUT_SEC)
+            survivors = kill_subprocesses(process)
+            assert not survivors, "child did not exit within timeout"
             assert process.exitcode == 0
         finally:
             for mr in self._extra_mrs:

cuda_core/tests/memory_ipc/test_event_ipc.py

@@ -5,7 +5,7 @@
 
 import pytest
 from helpers.buffers import compare_equal_buffers, make_scratch_buffer
-from helpers.child_processes import child_timeout_sec
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 from helpers.latch import LatchKernel
 from helpers.logging import TimestampedLogger
 
@@ -68,6 +68,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
         log("releasing stream1")
         latch.release()
         process.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(process)
+        assert not survivors, "child did not exit within timeout"
         assert process.exitcode == 0
         log("done")
 
@@ -163,6 +165,8 @@ def test_main(self, ipc_device, blocking_sync, use_options_cls, use_option_kw):
         assert props[5] is None
 
         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(self, q_in, q_out):

cuda_core/tests/memory_ipc/test_ipc_duplicate_import.py

@@ -13,7 +13,7 @@
 import multiprocessing as mp
 
 import pytest
-from helpers.child_processes import child_timeout_sec
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 from helpers.logging import TimestampedLogger
 
 from cuda.core import Buffer, Device
@@ -85,6 +85,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
 
         log("waiting for child")
         process.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(process)
         log(f"child exit code: {process.exitcode}")
+        assert not survivors, "child did not exit within timeout"
         assert process.exitcode == 0, f"Child process failed with exit code {process.exitcode}"
         log("done")

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
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 
 CHILD_TIMEOUT_SEC = child_timeout_sec()
 NBYTES = 64
@@ -39,6 +39,8 @@ def exec_success(obj, number=1):
         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
 
 
@@ -55,6 +57,8 @@ def exec_launch_failure(obj, number=1):
         process = mp.Process(target=child_main_bad, 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
 
 
@@ -138,5 +142,7 @@ def prime():
         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_memory_ipc.py

@@ -5,7 +5,7 @@
 
 import pytest
 from helpers.buffers import PatternGen
-from helpers.child_processes import child_timeout_sec
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 
 from cuda.core import Buffer, DeviceMemoryResource
 
@@ -40,6 +40,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
 
         # Wait for the child process.
         process.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(process)
+        assert not survivors, "child did not exit within timeout"
         assert process.exitcode == 0
 
         # Verify that the buffer was modified.
@@ -83,6 +85,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
         # Wait for the child processes.
         p1.join(timeout=CHILD_TIMEOUT_SEC)
         p2.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(p1, p2)
+        assert not survivors, f"timed out waiting on: {[p.name for p in survivors]}"
         assert p1.exitcode == 0
         assert p2.exitcode == 0
 
@@ -136,6 +140,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
         # Wait for children.
         p1.join(timeout=CHILD_TIMEOUT_SEC)
         p2.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(p1, p2)
+        assert not survivors, f"timed out waiting on: {[p.name for p in survivors]}"
         assert p1.exitcode == 0
         assert p2.exitcode == 0
 
@@ -186,6 +192,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
         # Wait for children.
         p1.join(timeout=CHILD_TIMEOUT_SEC)
         p2.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(p1, p2)
+        assert not survivors, f"timed out waiting on: {[p.name for p in survivors]}"
         assert p1.exitcode == 0
         assert p2.exitcode == 0
 

cuda_core/tests/memory_ipc/test_peer_access.py

@@ -5,7 +5,7 @@
 
 import pytest
 from helpers.buffers import PatternGen
-from helpers.child_processes import child_timeout_sec
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 
 from cuda.core import Device, DeviceMemoryResource, DeviceMemoryResourceOptions
 from cuda.core._utils.cuda_utils import CUDAError
@@ -36,6 +36,8 @@ def test_main(self, ipc_mempool_device_x2):
         process = mp.Process(target=self.child_main, args=(mr,))
         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
 
         # Verify parent's MR still has peer access set (independent state)
@@ -82,6 +84,8 @@ def test_main(self, ipc_mempool_device_x2, grant_access_in_parent):
         process = mp.Process(target=self.child_main, args=(mr, buffer))
         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
 
         buffer.close()

cuda_core/tests/memory_ipc/test_send_buffers.py

@@ -6,7 +6,7 @@
 
 import pytest
 from helpers.buffers import PatternGen
-from helpers.child_processes import child_timeout_sec
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 
 from cuda.core import Device, DeviceMemoryResource, DeviceMemoryResourceOptions
 
@@ -38,8 +38,11 @@ def test_main(self, ipc_device, nmrs):
             process = mp.Process(target=self.child_main, args=(device, buffers))
             process.start()
 
-            # Wait for the child process.
+            # Wait for the child process, then kill any survivor so subsequent
+            # asserts cannot block on a held IPC handle.
             process.join(timeout=CHILD_TIMEOUT_SEC)
+            survivors = kill_subprocesses(process)
+            assert not survivors, "child did not exit within timeout"
             assert process.exitcode == 0
 
             # Verify that the buffers were modified.
@@ -96,11 +99,16 @@ def test_main(self, ipc_device, ipc_memory_resource):
         proc_b.start()
         proc_c.start()
 
-        # Wait for C to signal completion then clean up.
-        event_c.wait(timeout=CHILD_TIMEOUT_SEC)
+        # Wait for C to signal completion, then let B finish and join both.
+        # Gather all state (event result + joins + survivor kills) before
+        # asserting so cleanup happens regardless of which check fires.
+        completed = event_c.wait(timeout=CHILD_TIMEOUT_SEC)
         event_b.set()  # b can finish now
         proc_b.join(timeout=CHILD_TIMEOUT_SEC)
         proc_c.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(proc_b, proc_c)
+        assert completed, "process C did not signal completion within timeout"
+        assert not survivors, f"timed out waiting on: {[p.name for p in survivors]}"
         assert proc_b.exitcode == 0
         assert proc_c.exitcode == 0
 

cuda_core/tests/memory_ipc/test_serialize.py

@@ -7,7 +7,7 @@
 
 import pytest
 from helpers.buffers import PatternGen
-from helpers.child_processes import child_timeout_sec
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 
 from cuda.core import Buffer, Device, DeviceMemoryResource, PinnedMemoryResource
 
@@ -47,6 +47,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
 
         # Wait for the child process.
         process.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(process)
+        assert not survivors, "child did not exit within timeout"
         assert process.exitcode == 0
 
         # Confirm buffers were modified.
@@ -104,6 +106,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
 
         # Wait for the child process.
         process.join(timeout=CHILD_TIMEOUT_SEC)
+        survivors = kill_subprocesses(process)
+        assert not survivors, "child did not exit within timeout"
         assert process.exitcode == 0
 
         # Confirm buffer was modified.
@@ -152,6 +156,8 @@ def test_main(self, ipc_device, ipc_memory_resource):
         process = mp.Process(target=self.child_main, args=(alloc_handle, mr, buffer_desc, buffer))
         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
 
         pgen.verify_buffer(buffer, seed=True)