tests: layered defense against IPC child-process hangs (#2004) (#2124)

* tests: layered defense against IPC child-process hangs (#2004)

Replace the per-test kill-on-timeout approach with three layered fixes that
defend the IPC test suite against compute-sanitizer + CUDA driver / toolkit
combinations that wedge child processes during IPC teardown.

CI layer: ci/tools/setup-sanitizer now passes
--target-processes=application-only so spawned multiprocessing.Process
children run without the sanitizer attached. This eliminates the root cause
of the hang reported in issue #2004 (compute-sanitizer's IPC teardown
analysis getting stuck under Python 3.12 + CUDA 12.9.1). The parent pytest
process is still fully sanitized.

Fixture layer: ipc_device and ipc_mempool_device_x2 fixtures now wrap their
yield in track_child_processes(), a context manager that patches
multiprocessing.process.BaseProcess.__init__ to record every Process
instance constructed during the test and kills any survivor at teardown.
This protects ipc_memory_resource.mr.close() from blocking on IPC handles
held by a stuck child, regardless of the original failure mode.

Outer-guard layer: pytest-timeout is added to the test dep group, and
tests/memory_ipc/conftest.py applies pytest.mark.timeout(300) to every test
in the directory. This is the final fallback so no IPC test can wedge the
GHA runner for hours if a new failure mode defeats the earlier layers.

The hardcoded CHILD_TIMEOUT_SEC = 30 in every IPC test module is replaced
with a call to helpers.child_processes.child_timeout_sec(), which returns
30 by default and 120 under compute-sanitizer (detected via the existing
under_compute_sanitizer() helper). The unused CHILD_TIMEOUT_SEC declaration
in test_workerpool.py is removed.

* tests: reframe IPC hang docs as test-side bug rather than CS bug

Reword the inline comment in ci/tools/setup-sanitizer and the docstrings in
helpers/child_processes.py and memory_ipc/conftest.py to make clear that the
underlying problem is insufficient guards in the IPC tests when child
processes spawn slowly (>30s under compute-sanitizer). The sanitizer change
and the new helpers are mitigations / safety nets; the durable fix is making
the tests handle slow children correctly.

* 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.

* tests: add memory_ipc/__init__.py to avoid conftest module collision

Without __init__.py in tests/memory_ipc/, pytest imports
tests/memory_ipc/conftest.py under module name `conftest`, which collides
with the top-level tests/conftest.py. Tests under tests/ that use
`from conftest import X` then resolve to the wrong file at import time,
yielding ImportErrors like:

    ImportError: cannot import name 'skipif_need_cuda_headers' from
    'conftest' (.../tests/memory_ipc/conftest.py)

The fix mirrors tests/system/ (which already has both __init__.py and
conftest.py): adding tests/memory_ipc/__init__.py makes memory_ipc a
proper package, so its conftest is imported as memory_ipc.conftest and
the top-level conftest stays under name `conftest`.

* tests: scale memory_ipc outer timeout with child_timeout_sec

Replace the fixed 300 s pytest.mark.timeout with 3 * child_timeout_sec().
This is the minimum that lets the worst-case test in the suite (currently
TestIpcReexport, with three sequential CHILD_TIMEOUT_SEC waits in the
failure path) reach its own asserts before the outer guard fires.

Resulting budgets:
- Without compute-sanitizer: 90 s (down from 300 s).
- Under compute-sanitizer:  360 s (up from 300 s -- the previous value
  would have killed TestIpcReexport mid-second-join before its
  "process C did not signal completion within timeout" assertion could
  produce a clean failure message).

The meta-test test_outer_timeout_marker_is_applied now computes the
expected timeout from the same formula so the two stay in sync.

* tests: tighten memory_ipc outer timeout to CHILD_TIMEOUT_SEC + 30

The previous 3 * CHILD_TIMEOUT_SEC scaling assumed worst-case wall-clock
where every sequential join/wait hits its full timeout. In practice the
children run concurrently, so expected wall-clock is ~CHILD_TIMEOUT_SEC
regardless of how many joins the test chains -- once a child is done its
join returns immediately. Exceeding CHILD_TIMEOUT_SEC + slack already
means something is genuinely stuck, in which case the outer guard firing
is the right outcome; the autouse track_child_processes() context
manager still cleans up survivors, and the per-test diagnostic message
would not be more informative than "test exceeded its budget".

New budgets:
- Without compute-sanitizer:  60 s (was 90 s).
- Under compute-sanitizer:   150 s (was 360 s).

The meta-test computes its expected value from the same formula.

Author: Andy-Jost

ci/tools/setup-sanitizer

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 #
 # SPDX-License-Identifier: Apache-2.0
 
@@ -12,7 +12,15 @@ set -euo pipefail
 if [[ "${SETUP_SANITIZER}" == 1 ]]; then
   COMPUTE_SANITIZER="${CUDA_HOME}/bin/compute-sanitizer"
   COMPUTE_SANITIZER_VERSION=$(${COMPUTE_SANITIZER} --version | grep -Eo "[0-9]{4}\.[0-9]\.[0-9]" | sed -e 's/\.//g')
-  SANITIZER_CMD="${COMPUTE_SANITIZER} --target-processes=all --launch-timeout=0 --tool=memcheck --error-exitcode=1 --report-api-errors=no"
+  # --target-processes=application-only: attach the sanitizer to the parent
+  # pytest process only. Spawned multiprocessing.Process children run without
+  # the sanitizer. This aims to mitigate a class of CI hangs where child
+  # processes take an extreme amount of time to spawn (>30 seconds). Test bugs
+  # triggered by that specific condition are typically uncovered only in CI,
+  # where they become emergencies and are difficult to debug. The parent
+  # process is still fully sanitized, which is where most of the interesting
+  # host-side IPC plumbing runs anyway.
+  SANITIZER_CMD="${COMPUTE_SANITIZER} --target-processes=application-only --launch-timeout=0 --tool=memcheck --error-exitcode=1 --report-api-errors=no"
   if [[ "$COMPUTE_SANITIZER_VERSION" -ge 202111 ]]; then
     SANITIZER_CMD="${SANITIZER_CMD} --padding=32"
   fi

cuda_core/pyproject.toml

@@ -58,7 +58,7 @@ cu12 = ["cuda-bindings[all]==12.*"]
 cu13 = ["cuda-bindings[all]==13.*"]
 
 [dependency-groups]
-test = ["cython>=3.2,<3.3", "setuptools", "pytest>=6.2.4", "pytest-benchmark", "pytest-randomly", "pytest-repeat", "pytest-rerunfailures", "cloudpickle", "psutil", "cffi"]
+test = ["cython>=3.2,<3.3", "setuptools", "pytest>=6.2.4", "pytest-benchmark", "pytest-randomly", "pytest-repeat", "pytest-rerunfailures", "pytest-timeout", "cloudpickle", "psutil", "cffi"]
 ml-dtypes = ["ml-dtypes>=0.5.4,<0.6.0"]
 test-cu12 = [ {include-group = "ml-dtypes" }, {include-group = "test" }, "cupy-cuda12x; python_version < '3.14'", "cuda-toolkit[cudart]==12.*"]  # runtime headers needed by CuPy
 test-cu13 = [ {include-group = "ml-dtypes" }, {include-group = "test" }, "cupy-cuda13x; python_version < '3.14'", "cuda-toolkit[cudart]==13.*"]  # runtime headers needed by CuPy

cuda_core/tests/conftest.py

@@ -213,14 +213,24 @@ def pop_all_contexts():
 
 @pytest.fixture
 def ipc_device():
-    """Obtains a device suitable for IPC-enabled mempool tests, or skips."""
+    """Obtains a device suitable for IPC-enabled mempool tests, or skips.
+
+    The fixture also tracks every ``multiprocessing.Process`` spawned during
+    the test and kills any survivors at teardown. This prevents a stuck child
+    (e.g., compute-sanitizer wedged during IPC teardown -- see issue #2004)
+    from blocking ``ipc_memory_resource``'s ``mr.close()`` for hours.
+    """
+    from helpers.child_processes import track_child_processes
+
     device = Device(0)
     device.set_current()
 
     if not device.properties.memory_pools_supported:
         pytest.skip("Device does not support mempool operations")
 
-    return _require_ipc_mempool_devices((device,))[0]
+    device = _require_ipc_mempool_devices((device,))[0]
+    with track_child_processes():
+        yield device
 
 
 @pytest.fixture(
@@ -291,8 +301,16 @@ def mempool_device_x3():
 
 @pytest.fixture
 def ipc_mempool_device_x2(mempool_device_x2):
-    """Fixture that provides two IPC-capable mempool devices, or skips."""
-    return _require_ipc_mempool_devices(mempool_device_x2)
+    """Fixture that provides two IPC-capable mempool devices, or skips.
+
+    Also tracks/kills any leftover ``multiprocessing.Process`` children at
+    teardown for the same reasons documented on :func:`ipc_device`.
+    """
+    from helpers.child_processes import track_child_processes
+
+    devices = _require_ipc_mempool_devices(mempool_device_x2)
+    with track_child_processes():
+        yield devices
 
 
 @pytest.fixture(

cuda_core/tests/helpers/child_processes.py

@@ -0,0 +1,103 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Helpers for tests that spawn ``multiprocessing.Process`` children.
+
+These exist primarily to defend IPC tests against a class of CI hang where a
+child process spawns too slowly and the parent does not implement proper guards
+for that (see issue #2004). Without intervention, a zombie child holds an IPC
+memory handle and blocks the parent's ``mr.close()`` in fixture teardown,
+leading to deadlock and wedging the test runner for hours.
+"""
+
+import contextlib
+import multiprocessing.process
+import weakref
+
+from cuda_python_test_helpers import under_compute_sanitizer
+
+CHILD_TIMEOUT_SEC_DEFAULT = 30
+CHILD_TIMEOUT_SEC_SANITIZER = 120
+
+
+def child_timeout_sec() -> int:
+    """Return the per-process join/wait timeout for IPC-style tests.
+
+    Compute-sanitizer significantly slows process startup and CUDA context
+    teardown, so we use a larger budget when it is active.
+    """
+    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
+    alive at exit.
+
+    Patches ``multiprocessing.process.BaseProcess.__init__`` to record every
+    ``Process`` instance constructed inside the ``with`` block. This covers
+    the delegating ``mp.Process`` class as well as direct ``SpawnProcess`` /
+    ``ForkProcess`` instances (including those created by ``mp.Pool``), since
+    all of them inherit from ``BaseProcess``. On exit, any tracked process
+    that is still alive is killed and joined.
+
+    This protects fixture teardown (e.g. ``ipc_memory_resource``'s
+    ``mr.close()``) from blocking on IPC handles held by a stuck child --
+    see issue #2004.
+    """
+    tracked = weakref.WeakSet()
+    base = multiprocessing.process.BaseProcess
+    original_init = base.__init__
+
+    def tracking_init(self, *args, **kwargs):
+        original_init(self, *args, **kwargs)
+        tracked.add(self)
+
+    base.__init__ = tracking_init
+    try:
+        yield
+    finally:
+        base.__init__ = original_init
+        for proc in list(tracked):
+            # is_alive() / kill() raise ValueError if the Process was never
+            # started or has already been closed; nothing to clean up in that
+            # case.
+            with contextlib.suppress(ValueError):
+                if proc.is_alive():
+                    proc.kill()
+                    proc.join()

cuda_core/tests/memory_ipc/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0

cuda_core/tests/memory_ipc/conftest.py

@@ -0,0 +1,41 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Per-directory conftest for memory IPC tests.
+
+Applies an outer-guard ``pytest.mark.timeout`` to every test in this directory.
+Individual tests still drive their own per-process waits using
+``child_timeout_sec()`` from ``helpers.child_processes``; this marker is the
+final fallback so that no IPC test can wedge the CI runner for hours if
+deadlock occurs.
+"""
+
+import pathlib
+
+import pytest
+from helpers.child_processes import child_timeout_sec
+
+_HERE = pathlib.Path(__file__).parent.resolve()
+
+
+def _outer_timeout_sec() -> int:
+    # IPC tests spawn children that run concurrently, so expected wall-clock
+    # is ~CHILD_TIMEOUT_SEC regardless of how many subsequent join/wait
+    # timeouts the test chains together (each subsequent join returns
+    # immediately once its child is already done). Exceeding that already
+    # means something is genuinely stuck, at which point the outer guard
+    # firing is the right outcome -- the per-test asserts wouldn't add
+    # useful diagnostic value over "test exceeded its budget", and the
+    # autouse track_child_processes() context manager still cleans up.
+    return child_timeout_sec() + 30
+
+
+def pytest_collection_modifyitems(config, items):
+    marker = pytest.mark.timeout(_outer_timeout_sec())
+    for item in items:
+        try:
+            item_path = pathlib.Path(str(item.fspath)).resolve()
+        except OSError:
+            continue
+        if _HERE in item_path.parents:
+            item.add_marker(marker)

cuda_core/tests/memory_ipc/test_errors.py

@@ -6,15 +6,31 @@
 import re
 
 import pytest
+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
 
-CHILD_TIMEOUT_SEC = 30
+CHILD_TIMEOUT_SEC = child_timeout_sec()
 NBYTES = 64
 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.
+    """
+    expected = child_timeout_sec() + 30
+    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 == (expected,), 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)."""
@@ -43,6 +59,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,13 +5,14 @@
 
 import pytest
 from helpers.buffers import compare_equal_buffers, make_scratch_buffer
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 from helpers.latch import LatchKernel
 from helpers.logging import TimestampedLogger
 
 from cuda.core import Device, EventOptions
 
 ENABLE_LOGGING = False  # Set True for test debugging and development
-CHILD_TIMEOUT_SEC = 30
+CHILD_TIMEOUT_SEC = child_timeout_sec()
 NBYTES = 64
 
 
@@ -67,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")
 
@@ -162,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,11 +13,12 @@
 import multiprocessing as mp
 
 import pytest
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 from helpers.logging import TimestampedLogger
 
 from cuda.core import Buffer, Device
 
-CHILD_TIMEOUT_SEC = 30
+CHILD_TIMEOUT_SEC = child_timeout_sec()
 NBYTES = 64
 POOL_SIZE = 2097152
 
@@ -84,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,8 +13,9 @@
 else:
     HAVE_PSUTIL = True
 import pytest
+from helpers.child_processes import child_timeout_sec, kill_subprocesses
 
-CHILD_TIMEOUT_SEC = 30
+CHILD_TIMEOUT_SEC = child_timeout_sec()
 NBYTES = 64
 
 USING_FDS = platform.system() == "Linux"
@@ -38,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
 
 
@@ -54,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
 
 
@@ -137,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