Test using pytest-run-parallel and related fixups in the tests

ci/tools/run-tests

@@ -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
 
@@ -20,13 +20,20 @@ fi
 
 test_module=${1}
 
+TEST_FT_GROUP=()
+PYTEST_PARALLEL_ARGS=()
+if python -c 'import sys; assert not sys._is_gil_enabled()' 2> /dev/null; then
+  TEST_FT_GROUP=(--group test-ft)
+  PYTEST_PARALLEL_ARGS=(--parallel-threads=4)
+fi
+
 # For standard modes, install pathfinder up front (it is a direct dependency
 # of bindings, and a transitive dependency of core).  Nightly modes install
 # all wheels together in a single pip call further below.
 if [[ "${test_module}" != nightly-* ]]; then
   pushd ./cuda_pathfinder
   echo "Installing pathfinder wheel"
-  pip install ./*.whl --group test
+  pip install ./*.whl --group test "${TEST_FT_GROUP[@]}"
   popd
 fi
 
@@ -36,7 +43,7 @@ if [[ "${test_module}" == "pathfinder" ]]; then
     "LD:${CUDA_PATHFINDER_TEST_LOAD_NVIDIA_DYNAMIC_LIB_STRICTNESS} " \
     "FH:${CUDA_PATHFINDER_TEST_FIND_NVIDIA_HEADERS_STRICTNESS} " \
     "BC:${CUDA_PATHFINDER_TEST_FIND_NVIDIA_BITCODE_LIB_STRICTNESS}"
-  pytest -ra -s -v tests/ |& tee /tmp/pathfinder_test_log.txt
+  pytest -ra -s -v "${PYTEST_PARALLEL_ARGS[@]}" tests/ |& tee /tmp/pathfinder_test_log.txt
   # Report the number of "INFO test_" lines (including zero)
   # to support quick validations based on GHA log archives.
   line_count=$(awk '/^INFO test_/ {count++} END {print count+0}' /tmp/pathfinder_test_log.txt)
@@ -46,26 +53,21 @@ elif [[ "${test_module}" == "bindings" ]]; then
   echo "Installing bindings wheel"
   pushd ./cuda_bindings
   if [[ "${LOCAL_CTK}" == 1 ]]; then
-    pip install "${CUDA_BINDINGS_ARTIFACTS_DIR}"/*.whl --group test
+    pip install "${CUDA_BINDINGS_ARTIFACTS_DIR}"/*.whl --group test "${TEST_FT_GROUP[@]}"
   else
-    pip install $(ls "${CUDA_BINDINGS_ARTIFACTS_DIR}"/*.whl)[all] --group test
+    pip install $(ls "${CUDA_BINDINGS_ARTIFACTS_DIR}"/*.whl)[all] --group test "${TEST_FT_GROUP[@]}"
   fi
   echo "Running bindings tests"
-  ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize tests/
+  ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/
   if [[ "${SKIP_CYTHON_TEST}" == 0 ]]; then
-    ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize tests/cython
+    ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/cython
   fi
   popd
 elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then
   # Shared setup for core and nightly modes.
   TEST_CUDA_MAJOR="$(cut -d '.' -f 1 <<< ${CUDA_VER})"
   TEST_CUDA_MAJOR_MINOR="$(cut -d '.' -f 1-2 <<< "${CUDA_VER}")"
 
-  FREE_THREADING=""
-  if python -c 'import sys; assert not sys._is_gil_enabled()' 2> /dev/null; then
-    FREE_THREADING+="-ft"
-  fi
-
   # Resolve bindings based on BINDINGS_SOURCE (set by env-vars):
   #   main/backport → local wheel from artifacts dir
   #   published     → install from PyPI by version
@@ -101,15 +103,15 @@ elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then
     echo "Installing core wheel"
     # Constrain cuda-toolkit to the requested CTK version to avoid
     # pip pulling in a newer nvidia-cuda-runtime that conflicts with it.
-    pip install "${CORE_WHL[@]}" --group "test-cu${TEST_CUDA_MAJOR}${FREE_THREADING}" "cuda-toolkit==${TEST_CUDA_MAJOR_MINOR}.*"
+    pip install "${CORE_WHL[@]}" --group "test-cu${TEST_CUDA_MAJOR}" "${TEST_FT_GROUP[@]}" "cuda-toolkit==${TEST_CUDA_MAJOR_MINOR}.*"
     echo "Installed packages before core tests:"
     pip list
     echo "Running core tests"
-    ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize tests/
+    ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/
     # Currently our CI always installs the latest bindings (from either major version).
     # This is not compatible with the test requirements.
     if [[ "${SKIP_CYTHON_TEST}" == 0 ]]; then
-      ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize tests/cython
+      ${SANITIZER_CMD} pytest -rxXs -v --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/cython
     fi
     popd
   elif [[ "${test_module}" == "nightly-cuda-core" ]]; then
@@ -123,7 +125,7 @@ elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then
     released_ver=$(pip show cuda-core | awk '/^Version:/{print $2}')
     if [[ -n "${GITHUB_ENV:-}" ]]; then
       echo "CUDA_CORE_RELEASED_VER=${released_ver}" >> "${GITHUB_ENV}"
-      echo "CUDA_CORE_TEST_GROUP=test-cu${TEST_CUDA_MAJOR}${FREE_THREADING}" >> "${GITHUB_ENV}"
+      echo "CUDA_CORE_TEST_GROUP=test-cu${TEST_CUDA_MAJOR}" >> "${GITHUB_ENV}"
     fi
     echo "Installed packages before released cuda-core tests:"
     pip list
@@ -137,7 +139,8 @@ elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then
       "${PATHFINDER_WHL[@]}"
       "${BINDINGS_ARGS[@]}"
       "${CORE_WHL[@]}"
-      --group "test-cu${TEST_CUDA_MAJOR}${FREE_THREADING}"
+      --group "test-cu${TEST_CUDA_MAJOR}"
+      "${TEST_FT_GROUP[@]}"
     )
 
     if [[ "${test_module}" == "nightly-pytorch" ]]; then

cuda_bindings/pyproject.toml

@@ -55,6 +55,9 @@ test = [
     "pytest-randomly==4.1.0",
     "pyglet==2.1.14",
 ]
+test-ft = [
+    "pytest-run-parallel==0.10.0",
+]
 
 [project.urls]
 Repository = "https://github.com/NVIDIA/cuda-python"

cuda_bindings/tests/test_graphics_apis.py

@@ -12,6 +12,8 @@
 
 from cuda.bindings import runtime as cudart
 
+pytestmark = pytest.mark.thread_unsafe(reason="pyglet/OpenGL context is process-global")
+
 
 def _configure_pyglet_headless(pyglet):
     """On headless Linux: enable EGL mode or skip if EGL is absent."""

cuda_core/cuda/core/_stream.pyx

@@ -596,7 +596,6 @@ cdef cydriver.CUstream _handle_from_stream_protocol(obj) except*:
         info = cuda_stream_attr()
     else:
         info = cuda_stream_attr
-        warnings.simplefilter("once", DeprecationWarning)
         warnings.warn(
             "Implementing __cuda_stream__ as an attribute is deprecated; it must be implemented as a method",
             stacklevel=3,

cuda_core/pyproject.toml

@@ -77,12 +77,9 @@ test = [
 ]
 # TODO: drop the Windows 3.15 guard once ml-dtypes publishes cp315 Windows wheels
 ml-dtypes = ["ml-dtypes>=0.5.4,<0.6.0; sys_platform != 'win32' or python_version < '3.15'"]
-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
-# free threaded build, cupy doesn't support free-threaded builds yet, so avoid installing it for now
-# TODO: cupy should support free threaded builds
-test-cu12-ft = [ {include-group = "ml-dtypes" }, {include-group = "test" }, "cuda-toolkit[cudart]==12.*"]
-test-cu13-ft = [ {include-group = "ml-dtypes" }, {include-group = "test" }, "cuda-toolkit[cudart]==13.*"]
+test-ft = ["pytest-run-parallel==0.10.0"]
+test-cu12 = [ {include-group = "ml-dtypes" }, {include-group = "test" }, "cupy-cuda12x; python_version < '3.15'", "cuda-toolkit[cudart]==12.*"]  # runtime headers needed by CuPy
+test-cu13 = [ {include-group = "ml-dtypes" }, {include-group = "test" }, "cupy-cuda13x; python_version < '3.15'", "cuda-toolkit[cudart]==13.*"]  # runtime headers needed by CuPy
 
 [tool.uv]
 conflicts = [
@@ -93,8 +90,6 @@ conflicts = [
     [
         { group = "test-cu12" },
         { group = "test-cu13" },
-        { group = "test-cu12-ft" },
-        { group = "test-cu13-ft" },
     ],
 ]
 

cuda_core/tests/AGENTS.md

@@ -60,7 +60,7 @@ with pytest.raises(RuntimeError, match="IPC is not available"):
 
 ## Release resources at test boundaries
 
-The `_init_cuda_context` fixture in `conftest.py` runs `gc.collect()` followed
+The `init_cuda` fixture in `conftest.py` runs `gc.collect()` followed
 by `cuCtxSynchronize()` before popping the context. Tests should not rely on
 that as a substitute for cleaning up explicitly: prefer context managers for
 resources whose lifetime fits a single scope, and keep pool lifetimes inside

cuda_core/tests/conftest.py

@@ -8,7 +8,6 @@
 import os
 import pathlib
 import sys
-from contextlib import contextmanager
 
 import pytest
 
@@ -72,44 +71,16 @@ def pytest_terminal_summary(terminalreporter):
     oom_diagnostics.report_terminal_summary(terminalreporter)
 
 
-@contextmanager
-def _init_cuda_context():
-    # TODO: rename this to e.g. init_context
-    device = Device(0)
-    device.set_current()
-
-    # Set option to avoid spin-waiting on synchronization.
-    if int(os.environ.get("CUDA_CORE_TEST_BLOCKING_SYNC", 0)) != 0:
-        handle_return(
-            driver.cuDevicePrimaryCtxSetFlags(device.device_id, driver.CUctx_flags.CU_CTX_SCHED_BLOCKING_SYNC)
-        )
-
-    try:
-        yield device
-    finally:
-        # Force any pool/allocation whose only remaining reference was a local
-        # in this test's frame to actually get destroyed now, then drain the
-        # context so the stream-ordered frees that destruction enqueues retire
-        # before the next test runs. Without this, a memory pool's VA
-        # reservation is not returned until both have happened, and per-test
-        # leftovers accumulate across the run -- which is how full-suite runs
-        # can exhaust address space and hit CUDA_ERROR_OUT_OF_MEMORY on a
-        # device with plenty of free physical memory (issue #2381). gc.collect()
-        # must run first: cuCtxSynchronize alone cannot drain frees that were
-        # never enqueued because their owning object had not been collected yet.
-        gc.collect()
-        driver.cuCtxSynchronize()
-        _ = _device_unset_current()
-
-
 def _wrap_worker_cuda_test(func):
     if getattr(func, "_cuda_core_worker_cuda_wrapped", False):
         return func
 
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
         kwargs = dict(kwargs)  # copy before mutating
-        with _init_cuda_context() as device:
+        device = Device(0)
+        device.set_current()
+        try:
             if "init_cuda" in kwargs:
                 kwargs["init_cuda"] = device
             if "mempool_device_x2" in kwargs:
@@ -129,6 +100,12 @@ def wrapper(*args, **kwargs):
                 groups, _ = device.resources.sm.split(SMResourceOptions(count=None))
                 kwargs["green_ctx"] = device.create_context(ContextOptions(resources=[groups[0]]))
             return func(*args, **kwargs)
+        finally:
+            # Unlike the `init_cuda` fixture we do not synchronize here
+            # to avoid doing so while other workers are still running.
+            # (E.g. for stream capture). The fixture cleanup is still run
+            # even with pytest-run-parallel after worker join.
+            _ = _device_unset_current()
 
     wrapper._cuda_core_worker_cuda_wrapped = True
     return wrapper
@@ -186,8 +163,33 @@ def session_setup():
 
 @pytest.fixture
 def init_cuda():
-    with _init_cuda_context() as device:
+    # TODO: rename this to e.g. init_context
+    device = Device(0)
+    device.set_current()
+
+    # Set option to avoid spin-waiting on synchronization.
+    if int(os.environ.get("CUDA_CORE_TEST_BLOCKING_SYNC", 0)) != 0:
+        handle_return(
+            driver.cuDevicePrimaryCtxSetFlags(device.device_id, driver.CUctx_flags.CU_CTX_SCHED_BLOCKING_SYNC)
+        )
+
+    try:
         yield device
+    finally:
+        # Force any pool/allocation whose only remaining reference was a local
+        # in this test's frame to actually get destroyed now, then drain the
+        # context so the stream-ordered frees that destruction enqueues retire
+        # before the next test runs. Without this, a memory pool's VA
+        # reservation is not returned until both have happened, and per-test
+        # leftovers accumulate across the run -- which is how full-suite runs
+        # can exhaust address space and hit CUDA_ERROR_OUT_OF_MEMORY on a
+        # device with plenty of free physical memory (issue #2381). gc.collect()
+        # must run first: cuCtxSynchronize alone cannot drain frees that were
+        # never enqueued because their owning object had not been collected yet.
+        # With pytest-run-parallel this runs after worker join.
+        gc.collect()
+        driver.cuCtxSynchronize()
+        _ = _device_unset_current()
 
 
 def _device_unset_current() -> bool:

cuda_core/tests/graph/test_graph_node_update.py

@@ -14,7 +14,7 @@
 import pytest
 from helpers.graph_kernels import compile_common_kernels
 
-from cuda.core import LaunchConfig, LegacyPinnedMemoryResource
+from cuda.core import Device, LaunchConfig, LegacyPinnedMemoryResource
 from cuda.core._utils._weak_handles import weak_handle
 from cuda.core._utils.cuda_utils import CUDAError, driver, handle_return
 from cuda.core._utils.version import driver_version
@@ -619,7 +619,10 @@ def assert_exec_uses(graph, expected):
 def definition_update_case(request, init_cuda):
     if driver_version() < (12, 2, 0):
         pytest.skip("individual graph node updates require CUDA 12.2+")
-    return request.param(init_cuda)
+    factory = request.param
+    # pytest-run-parallel shares this fixture object across workers. Build the
+    # case at call time on Device() so each worker gets its own graph/node.
+    return lambda: factory(Device())
 
 
 @pytest.mark.agent_authored(model="gpt-5.6")
@@ -745,7 +748,7 @@ def test_memcpy_update_between_host_and_device(init_cuda, device_operand):
 def test_definition_node_update_changes_future_instantiations(
     definition_update_case,
 ):
-    case = definition_update_case
+    case = definition_update_case()
     assert case.original != case.replacement
     old_graph = case.graph_def.instantiate()
 
@@ -762,7 +765,7 @@ def test_definition_node_update_changes_future_instantiations(
 def test_destroyed_definition_node_rejects_update(
     definition_update_case,
 ):
-    case = definition_update_case
+    case = definition_update_case()
     case.node.destroy()
 
     assert not case.node.is_valid
@@ -777,7 +780,7 @@ def test_destroyed_definition_node_rejects_update(
 def test_failed_definition_node_update_preserves_state(
     definition_update_case,
 ):
-    case = definition_update_case
+    case = definition_update_case()
 
     assert case.invalid_update is not None
     assert case.invalid_exception is not None
@@ -793,17 +796,18 @@ def test_failed_definition_node_update_preserves_state(
 def test_definition_node_update_rejects_wrong_type(
     definition_update_case,
 ):
-    if definition_update_case.invalid_argument_update is None:
+    case = definition_update_case()
+    if case.invalid_argument_update is None:
         pytest.skip("update method has no typed positional argument")
     with pytest.raises(TypeError):
-        definition_update_case.invalid_argument_update()
+        case.invalid_argument_update()
 
 
 @pytest.mark.agent_authored(model="gpt-5.6")
 def test_executable_node_update_changes_existing_exec(
     definition_update_case,
 ):
-    case = definition_update_case
+    case = definition_update_case()
     graph = case.graph_def.instantiate()
 
     _update_executable_case(graph, case)
@@ -974,6 +978,7 @@ def test_rejected_executable_update_rolls_back_owners(init_cuda):
     assert ctypes.c_int.from_address(int(active.handle)).value == 1
 
 
+@pytest.mark.thread_unsafe(reason="deferred cleanup on main thread which would wait")
 @pytest.mark.agent_authored(model="gpt-5.6")
 def test_whole_update_replaces_executable_attachment_accumulator(init_cuda):
     if driver_version() < (12, 2, 0):
@@ -1132,6 +1137,7 @@ def second():
     _wait_until(lambda: first_weak() is None and second_weak() is None)
 
 
+@pytest.mark.thread_unsafe(reason="deferred cleanup on main thread which would wait")
 @pytest.mark.agent_authored(model="claude-opus-5")
 def test_child_graph_update_transfers_source_owners_to_executable(init_cuda):
     if driver_version() < (12, 2, 0):

cuda_core/tests/memory/test_copy_batch.py

@@ -212,6 +212,7 @@ def test_batch_matches_sequential_d2d(self, copy_batch_device, device_bufs, copy
 class TestCopyBatchStreamSemantics:
     """Where the batch sits in stream order, and what it cannot be part of."""
 
+    @pytest.mark.thread_unsafe(reason="shared copy_stream and buffers must not interleave")
     @pytest.mark.agent_authored(model="Claude Opus 5")
     def test_ordered_between_prior_and_later_stream_work(self, device_bufs, copy_stream):
         """The batch must observe prior stream work and precede later work.