cuda.core: add LaunchConfig.programmatic_stream_serialization for programmatic dependent launch

cuda_core/cuda/core/_launch_config.pxd

@@ -15,6 +15,7 @@ cdef class LaunchConfig:
         public tuple block
         public int shmem_size
         public bint is_cooperative
+        public bint programmatic_stream_serialization
 
         vector[cydriver.CUlaunchAttribute] _attrs
         object __weakref__

cuda_core/cuda/core/_launch_config.pyi

@@ -35,9 +35,13 @@ class LaunchConfig:
         (Default to size 0)
     is_cooperative : bool, optional
         Whether this config can be used to launch a cooperative kernel.
+    programmatic_stream_serialization : bool, optional
+        Whether to allow programmatic stream serialization (PDL). When True,
+        the kernel may overlap with a previous kernel in the same stream that
+        signals completion via programmatic means.
     """
 
-    def __init__(self, grid: int | tuple[int, ...] | None=None, cluster: int | tuple[int, ...] | None=None, block: int | tuple[int, ...] | None=None, shmem_size: int | None=None, is_cooperative: bool=False) -> None:
+    def __init__(self, grid: int | tuple[int, ...] | None=None, cluster: int | tuple[int, ...] | None=None, block: int | tuple[int, ...] | None=None, shmem_size: int | None=None, is_cooperative: bool=False, programmatic_stream_serialization: bool=False) -> None:
         """Initialize LaunchConfig with validation.
 
         Parameters
@@ -52,6 +56,8 @@ class LaunchConfig:
             Dynamic shared memory size in bytes (default: 0)
         is_cooperative : bool, optional
             Whether to launch as cooperative kernel (default: False)
+        programmatic_stream_serialization : bool, optional
+            Whether to allow programmatic stream serialization / PDL (default: False)
         """
 
     def _identity(self) -> tuple[Any, ...]:
@@ -65,7 +71,7 @@ class LaunchConfig:
 
     def __hash__(self) -> int:
         ...
-_LAUNCH_CONFIG_ATTRS = ('grid', 'cluster', 'block', 'shmem_size', 'is_cooperative')
+_LAUNCH_CONFIG_ATTRS = ('grid', 'cluster', 'block', 'shmem_size', 'is_cooperative', 'programmatic_stream_serialization')
 __all__ = ['LaunchConfig']
 
 def _to_native_launch_config(config: LaunchConfig) -> object:

cuda_core/cuda/core/_launch_config.pyx

@@ -13,7 +13,14 @@ from cuda.core._utils.cuda_utils import (
     driver,
 )
 
-_LAUNCH_CONFIG_ATTRS = ('grid', 'cluster', 'block', 'shmem_size', 'is_cooperative')
+_LAUNCH_CONFIG_ATTRS = (
+    'grid',
+    'cluster',
+    'block',
+    'shmem_size',
+    'is_cooperative',
+    'programmatic_stream_serialization',
+)
 
 __all__ = ['LaunchConfig']
 
@@ -48,6 +55,10 @@ cdef class LaunchConfig:
         (Default to size 0)
     is_cooperative : bool, optional
         Whether this config can be used to launch a cooperative kernel.
+    programmatic_stream_serialization : bool, optional
+        Whether to allow programmatic stream serialization (PDL). When True,
+        the kernel may overlap with a previous kernel in the same stream that
+        signals completion via programmatic means.
     """
 
     # TODO: expand LaunchConfig to include other attributes
@@ -60,6 +71,7 @@ cdef class LaunchConfig:
         block: int | tuple[int, ...] | None = None,
         shmem_size: int | None = None,
         is_cooperative: bool = False,
+        programmatic_stream_serialization: bool = False,
     ) -> None:
         """Initialize LaunchConfig with validation.
 
@@ -75,6 +87,8 @@ cdef class LaunchConfig:
             Dynamic shared memory size in bytes (default: 0)
         is_cooperative : bool, optional
             Whether to launch as cooperative kernel (default: False)
+        programmatic_stream_serialization : bool, optional
+            Whether to allow programmatic stream serialization / PDL (default: False)
         """
         # Convert and validate grid and block dimensions
         self.grid = cast_to_3_tuple("LaunchConfig.grid", grid)
@@ -101,6 +115,7 @@ cdef class LaunchConfig:
             self.shmem_size = shmem_size
 
         self.is_cooperative = is_cooperative
+        self.programmatic_stream_serialization = programmatic_stream_serialization
 
         if self.is_cooperative and not Device().properties.cooperative_launch:
             raise CUDAError("cooperative kernels are not supported on this device")
@@ -149,6 +164,11 @@ cdef class LaunchConfig:
             attr.value.cooperative = 1
             self._attrs.push_back(attr)
 
+        if self.programmatic_stream_serialization:
+            attr.id = cydriver.CUlaunchAttributeID.CU_LAUNCH_ATTRIBUTE_PROGRAMMATIC_STREAM_SERIALIZATION
+            attr.value.programmaticStreamSerializationAllowed = 1
+            self._attrs.push_back(attr)
+
         drv_cfg.numAttrs = self._attrs.size()
         drv_cfg.attrs = self._attrs.data()
 
@@ -204,6 +224,12 @@ cpdef object _to_native_launch_config(LaunchConfig config):
         attr.value.cooperative = 1
         attrs.append(attr)
 
+    if config.programmatic_stream_serialization:
+        attr = driver.CUlaunchAttribute()
+        attr.id = driver.CUlaunchAttributeID.CU_LAUNCH_ATTRIBUTE_PROGRAMMATIC_STREAM_SERIALIZATION
+        attr.value.programmaticStreamSerializationAllowed = 1
+        attrs.append(attr)
+
     drv_cfg.numAttrs = len(attrs)
     drv_cfg.attrs = attrs
 

cuda_core/tests/test_launcher.py

@@ -183,6 +183,117 @@ class _FakeDev:
     assert attr.value.cooperative == 1, f"Expected cooperative=1, got {attr.value.cooperative}"
 
 
+def test_to_native_launch_config_pdl():
+    """LaunchConfig(programmatic_stream_serialization=True) maps to the PDL launch attribute."""
+    from cuda.bindings import driver
+    from cuda.core._launch_config import _to_native_launch_config
+
+    config = LaunchConfig(grid=2, block=4, programmatic_stream_serialization=True)
+    native = _to_native_launch_config(config)
+    assert native.gridDimX == 2
+    assert native.blockDimX == 4
+    assert native.numAttrs == 1
+    attr = native.attrs[0]
+    assert attr.id == driver.CUlaunchAttributeID.CU_LAUNCH_ATTRIBUTE_PROGRAMMATIC_STREAM_SERIALIZATION, (
+        f"Expected CU_LAUNCH_ATTRIBUTE_PROGRAMMATIC_STREAM_SERIALIZATION, got {attr.id}"
+    )
+    assert attr.value.programmaticStreamSerializationAllowed == 1, (
+        f"Expected programmaticStreamSerializationAllowed=1, got {attr.value.programmaticStreamSerializationAllowed}"
+    )
+
+
+@skipif_need_cuda_headers
+def test_pdl_primary_secondary_overlap_same_stream():
+    """Primary + secondary PDL launch on one stream can overlap on Hopper+.
+
+    Secondary is launched with ``programmatic_stream_serialization=True``. After
+    the primary triggers completion, it spins until it observes a flag written by
+    the secondary's independent preamble — proving both grids were resident at
+    once. Without PDL, the secondary cannot start until the primary exits.
+
+    Note concurrency is opportunistic, so a missing overlap execution is reported as
+    an expected failure.
+    """
+    dev = Device()
+    if dev.compute_capability < (9, 0):
+        pytest.skip("Programmatic Dependent Launch requires compute capability >= 9.0")
+    dev.set_current()
+    stream = dev.create_stream(options={"nonblocking": True})
+
+    # clock64 budgets are in GPU cycles; keep the post-trigger window long enough
+    # for the secondary to boot, but short enough for a unit test.
+    code = r"""
+    #include <cuda_device_runtime_api.h>
+
+    extern "C" __global__ void primary_kernel(int* secondary_started, int* overlapped) {
+        cudaTriggerProgrammaticLaunchCompletion();
+
+        const long long deadline = clock64() + 100000000LL;  // ~50ms @ ~2GHz
+        if (threadIdx.x == 0 && blockIdx.x == 0) {
+            while (clock64() < deadline) {
+                if (atomicAdd(secondary_started, 0) != 0) {
+                    atomicExch(overlapped, 1);
+                    return;
+                }
+                __nanosleep(1000);
+            }
+        }
+    }
+
+    extern "C" __global__ void secondary_kernel(int* secondary_started) {
+        if (threadIdx.x == 0 && blockIdx.x == 0) {
+            atomicExch(secondary_started, 1);
+        }
+    }
+    """
+
+    arch = "".join(f"{i}" for i in dev.compute_capability)
+    pro_opts = ProgramOptions(std="c++17", arch=f"sm_{arch}", include_path=helpers.CUDA_INCLUDE_PATH)
+    prog = Program(code, code_type="c++", options=pro_opts)
+    mod = prog.compile("cubin")
+    primary = mod.get_kernel("primary_kernel")
+    secondary = mod.get_kernel("secondary_kernel")
+
+    mr = LegacyPinnedMemoryResource()
+    secondary_started = np.from_dlpack(mr.allocate(4)).view(np.int32)
+    overlapped = np.from_dlpack(mr.allocate(4)).view(np.int32)
+
+    primary_cfg = LaunchConfig(grid=1, block=1)
+    secondary_cfg = LaunchConfig(grid=1, block=1, programmatic_stream_serialization=True)
+    secondary_serial_cfg = LaunchConfig(grid=1, block=1)
+
+    def _run(secondary_launch_cfg: LaunchConfig) -> int:
+        secondary_started[0] = 0
+        overlapped[0] = 0
+        launch(stream, primary_cfg, primary, secondary_started.ctypes.data, overlapped.ctypes.data)
+        launch(stream, secondary_launch_cfg, secondary, secondary_started.ctypes.data)
+        stream.sync()
+        return int(overlapped[0])
+
+    # Without the PDL attribute, same-stream kernels stay serialized.
+    assert _run(secondary_serial_cfg) == 0, "Expected no overlap when programmatic_stream_serialization is False"
+
+    # PDL overlap is opportunistic; retry a few times on a quiet GPU.
+    saw_overlap = False
+    for _ in range(5):
+        if _run(secondary_cfg) == 1:
+            saw_overlap = True
+            break
+
+    if not saw_overlap:
+        # Overlap is never guaranteed by the driver, so a miss is reported as an
+        # expected failure rather than turning a busy GPU into a red CI run.
+        pytest.xfail(
+            "PDL (Programmatic Dependent Launch) overlap was not observed. "
+            "If this keeps xfailing in CI, manually re-check on a quiet Hopper+ GPU."
+        )
+
+    print(
+        f"PDL (Programmatic Dependent Launch) overlap verified on {dev.name} compute capability {dev.compute_capability}",
+        flush=True,
+    )
+
+
 def test_launch_config_cluster_accepts_hopper_cc(monkeypatch):
     """LaunchConfig accepts ``cluster`` when the device reports compute
     capability >= 9.0. Device is mocked so the cluster-cast branch runs on any

cuda_core/tests/test_object_protocols.py

@@ -685,7 +685,8 @@ def sample_switch_node_alt(sample_graphdef):
     (
         "sample_launch_config",
         r"LaunchConfig\(grid=\(\d+, \d+, \d+\), cluster=.+, block=\(\d+, \d+, \d+\), "
-        r"shmem_size=\d+, is_cooperative=(?:True|False)\)",
+        r"shmem_size=\d+, is_cooperative=(?:True|False), "
+        r"programmatic_stream_serialization=(?:True|False)\)",
     ),
     ("sample_kernel", r"<Kernel handle=0x[0-9a-f]+>"),
     # ObjectCode variations (by code_type)