Initial implementation of IPC-enabled events. Changes the type of max_size memory resource attribute to size_t from int32. Various updates and additions to test helpers.

Author: Andy-Jost

cuda_core/cuda/core/experimental/_device.pyx

@@ -28,6 +28,7 @@ from cuda.core.experimental._utils.cuda_utils import (
 from cuda.core.experimental._stream cimport default_stream
 
 
+
 # TODO: I prefer to type these as "cdef object" and avoid accessing them from within Python,
 # but it seems it is very convenient to expose them for testing purposes...
 _tls = threading.local()
@@ -1273,7 +1274,7 @@ class Device:
         """
         self._check_context_initialized()
         ctx = self._get_current_context()
-        return Event._init(self._id, ctx, options)
+        return Event._init(self._id, ctx, options, True)
 
     def allocate(self, size, stream: Optional[Stream] = None) -> Buffer:
         """Allocate device memory from a specified stream.

cuda_core/cuda/core/experimental/_event.pxd

@@ -11,6 +11,8 @@ cdef class Event:
         cydriver.CUevent _handle
         bint _timing_disabled
         bint _busy_waited
+        bint _ipc_enabled
+        object _ipc_descriptor
         int _device_id
         object _ctx_handle
 

cuda_core/cuda/core/experimental/_event.pyx

@@ -4,7 +4,9 @@
 
 from __future__ import annotations
 
+cimport cpython
 from libc.stdint cimport uintptr_t
+from libc.string cimport memcpy
 
 from cuda.bindings cimport cydriver
 
@@ -14,12 +16,14 @@ from cuda.core.experimental._utils.cuda_utils cimport (
 )
 
 from dataclasses import dataclass
+import multiprocessing
 from typing import TYPE_CHECKING, Optional
 
 from cuda.core.experimental._context import Context
 from cuda.core.experimental._utils.cuda_utils import (
     CUDAError,
     driver,
+    handle_return,
 )
 if TYPE_CHECKING:
     import cuda.bindings
@@ -40,15 +44,15 @@ cdef class EventOptions:
         has actually been completed.
         Otherwise, the CPU thread will busy-wait until the event has
         been completed. (Default to False)
-    support_ipc : bool, optional
+    ipc_enabled : bool, optional
         Event will be suitable for interprocess use.
         Note that enable_timing must be False. (Default to False)
 
     """
 
     enable_timing: Optional[bool] = False
     busy_waited_sync: Optional[bool] = False
-    support_ipc: Optional[bool] = False
+    ipc_enabled: Optional[bool] = False
 
 
 cdef class Event:
@@ -86,24 +90,35 @@ cdef class Event:
         raise RuntimeError("Event objects cannot be instantiated directly. Please use Stream APIs (record).")
 
     @classmethod
-    def _init(cls, device_id: int, ctx_handle: Context, options=None):
+    def _init(cls, device_id: int, ctx_handle: Context, options=None, is_free=False):
         cdef Event self = Event.__new__(cls)
         cdef EventOptions opts = check_or_create_options(EventOptions, options, "Event options")
         cdef unsigned int flags = 0x0
         self._timing_disabled = False
         self._busy_waited = False
+        self._ipc_enabled = False
+        self._ipc_descriptor = None
         if not opts.enable_timing:
             flags |= cydriver.CUevent_flags.CU_EVENT_DISABLE_TIMING
             self._timing_disabled = True
         if opts.busy_waited_sync:
             flags |= cydriver.CUevent_flags.CU_EVENT_BLOCKING_SYNC
             self._busy_waited = True
-        if opts.support_ipc:
-            raise NotImplementedError("WIP: https://github.com/NVIDIA/cuda-python/issues/103")
+        if opts.ipc_enabled:
+            if is_free:
+                raise TypeError(
+                    "IPC-enabled events must be bound; use Stream.record for creation."
+                )
+            flags |= cydriver.CUevent_flags.CU_EVENT_INTERPROCESS
+            self._ipc_enabled = True
+            if not self._timing_disabled:
+                raise TypeError("IPC-enabled events cannot use timing.")
         with nogil:
             HANDLE_RETURN(cydriver.cuEventCreate(&self._handle, flags))
         self._device_id = device_id
         self._ctx_handle = ctx_handle
+        if opts.ipc_enabled:
+            self.get_ipc_descriptor()
         return self
 
     cpdef close(self):
@@ -151,6 +166,40 @@ cdef class Event:
                 raise CUDAError(err)
             raise RuntimeError(explanation)
 
+    def get_ipc_descriptor(self) -> IPCEventDescriptor:
+        """Export an event allocated for sharing between processes."""
+        if self._ipc_descriptor is not None:
+            return self._ipc_descriptor
+        if not self.is_ipc_enabled:
+            raise RuntimeError("Event is not IPC-enabled")
+        cdef cydriver.CUipcEventHandle data
+        with nogil:
+            HANDLE_RETURN(cydriver.cuIpcGetEventHandle(&data, <cydriver.CUevent>(self._handle)))
+        cdef bytes data_b = cpython.PyBytes_FromStringAndSize(<char*>(data.reserved), sizeof(data.reserved))
+        self._ipc_descriptor = IPCEventDescriptor._init(data_b, self._busy_waited)
+        return self._ipc_descriptor
+
+    @classmethod
+    def from_ipc_descriptor(cls, ipc_descriptor: IPCEventDescriptor) -> Event:
+        """Import an event that was exported from another process."""
+        cdef cydriver.CUipcEventHandle data
+        memcpy(data.reserved, <const void*><const char*>(ipc_descriptor._reserved), sizeof(data.reserved))
+        cdef Event self = Event.__new__(cls)
+        with nogil:
+            HANDLE_RETURN(cydriver.cuIpcOpenEventHandle(&self._handle, data))
+        self._timing_disabled = True
+        self._busy_waited = ipc_descriptor._busy_waited
+        self._ipc_enabled = True
+        self._ipc_descriptor = ipc_descriptor
+        self._device_id = -1  # ??
+        self._ctx_handle = None  # ??
+        return self
+
+    @property
+    def is_ipc_enabled(self) -> bool:
+        """Return True if the event can be shared across process boundaries, otherwise False."""
+        return self._ipc_enabled
+
     @property
     def is_timing_disabled(self) -> bool:
         """Return True if the event does not record timing data, otherwise False."""
@@ -161,11 +210,6 @@ cdef class Event:
         """Return True if the event synchronization would keep the CPU busy-waiting, otherwise False."""
         return self._busy_waited
 
-    @property
-    def is_ipc_supported(self) -> bool:
-        """Return True if this event can be used as an interprocess event, otherwise False."""
-        raise NotImplementedError("WIP: https://github.com/NVIDIA/cuda-python/issues/103")
-
     def sync(self):
         """Synchronize until the event completes.
 
@@ -212,12 +256,43 @@ cdef class Event:
         context is set current after a event is created.
 
         """
-
-        from cuda.core.experimental._device import Device  # avoid circular import
-
-        return Device(self._device_id)
+        if self._device_id >= 0:
+            from ._device import Device  # avoid circular import
+            return Device(self._device_id)
 
     @property
     def context(self) -> Context:
         """Return the :obj:`~_context.Context` associated with this event."""
-        return Context._from_ctx(self._ctx_handle, self._device_id)
+        if self._ctx_handle is not None and self._device_id >= 0:
+            return Context._from_ctx(self._ctx_handle, self._device_id)
+
+
+cdef class IPCEventDescriptor:
+    """Serializable object describing an event that can be shared between processes."""
+
+    cdef:
+        bytes _reserved
+        bint _busy_waited
+
+    def __init__(self, *arg, **kwargs):
+        raise RuntimeError("IPCEventDescriptor objects cannot be instantiated directly. Please use Event APIs.")
+
+    @classmethod
+    def _init(cls, reserved: bytes, busy_waited: bint):
+        cdef IPCEventDescriptor self = IPCEventDescriptor.__new__(cls)
+        self._reserved = reserved
+        self._busy_waited = busy_waited
+        return self
+
+    def __eq__(self, IPCEventDescriptor rhs):
+        # No need to check self._busy_waited.
+        return self._reserved == rhs._reserved
+
+    def __reduce__(self):
+        return self._init, (self._reserved, self._busy_waited)
+
+
+def _reduce_event(event):
+    return event.from_ipc_descriptor, (event.get_ipc_descriptor(),)
+
+multiprocessing.reduction.register(Event, _reduce_event)

cuda_core/cuda/core/experimental/_memory.pyx

@@ -231,11 +231,11 @@ cdef class Buffer(_cyBuffer, MemoryResourceAttributes):
         if stream is None:
             # Note: match this behavior to DeviceMemoryResource.allocate()
             stream = default_stream()
-        cdef cydriver.CUmemPoolPtrExportData share_data
-        memcpy(share_data.reserved, <const void*><const char*>(ipc_buffer._reserved), sizeof(share_data.reserved))
+        cdef cydriver.CUmemPoolPtrExportData data
+        memcpy(data.reserved, <const void*><const char*>(ipc_buffer._reserved), sizeof(data.reserved))
         cdef cydriver.CUdeviceptr ptr
         with nogil:
-            HANDLE_RETURN(cydriver.cuMemPoolImportPointer(&ptr, mr._mempool_handle, &share_data))
+            HANDLE_RETURN(cydriver.cuMemPoolImportPointer(&ptr, mr._mempool_handle, &data))
         return Buffer._init(<intptr_t>ptr, ipc_buffer.size, mr, stream)
 
     def copy_to(self, dst: Buffer = None, *, stream: Stream) -> Buffer:
@@ -516,7 +516,7 @@ cdef class DeviceMemoryResourceOptions:
         (Default to 0)
     """
     ipc_enabled : cython.bint = False
-    max_size : cython.int = 0
+    max_size : cython.size_t = 0
 
 
 # TODO: cythonize this?

cuda_core/cuda/core/experimental/_stream.pyx

@@ -262,7 +262,13 @@ cdef class Stream:
         # and CU_EVENT_RECORD_EXTERNAL, can be set in EventOptions.
         if event is None:
             self._get_device_and_context()
-            event = Event._init(<int>(self._device_id), <uintptr_t>(self._ctx_handle), options)
+            event = Event._init(<int>(self._device_id), <uintptr_t>(self._ctx_handle), options, False)
+        elif event.is_ipc_enabled:
+            raise TypeError(
+                "IPC-enabled events should not be re-recorded, instead create a "
+                "new event by supplying options."
+            )
+
         cdef cydriver.CUevent e = (<cyEvent?>(event))._handle
         with nogil:
             HANDLE_RETURN(cydriver.cuEventRecord(e, self._handle))

cuda_core/tests/helpers.py cuda_core/tests/helpers/__init__.pycuda_core/tests/helpers.py renamed to cuda_core/tests/helpers/__init__.py

@@ -22,7 +22,7 @@
     import cuda_python_test_helpers
 except ImportError:
     # Import shared platform helpers for tests across repos
-    sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "cuda_python_test_helpers"))
+    sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[3] / "cuda_python_test_helpers"))
     import cuda_python_test_helpers
 
 

cuda_core/tests/helpers/latch.py

@@ -0,0 +1,69 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from cuda.core.experimental import (
+    LaunchConfig,
+    LegacyPinnedMemoryResource,
+    Program,
+    ProgramOptions,
+    launch,
+)
+import helpers
+import ctypes
+
+class LatchKernel:
+    """
+    Manages a kernel that blocks progress until released.
+    """
+
+    def __init__(self, device):
+        code = """
+               #include <cuda/atomic>
+
+               extern "C"
+               __global__ void latch(int* val) {
+                   cuda::atomic_ref<int, cuda::thread_scope_system> signal{*val};
+                   while (true) {
+                       if (signal.load(cuda::memory_order_relaxed)) {
+                           break;
+                       }
+                   }
+               }
+               """
+        program_options = ProgramOptions(
+            std="c++17",
+            arch=f"sm_{''.join(f'{i}' for i in device.compute_capability)}",
+            include_path=helpers.CCCL_INCLUDE_PATHS,
+        )
+        prog = Program(code, code_type="c++", options=program_options)
+        mod = prog.compile(target_type="cubin")
+        self.kernel = mod.get_kernel("latch")
+
+        mr = LegacyPinnedMemoryResource()
+        self.buffer = mr.allocate(4)
+        self.busy_wait_flag[0] = 0
+
+    def launch(self, stream):
+        config = LaunchConfig(grid=1, block=1)
+        launch(stream, config, self.kernel, self.busy_wait_flag_address)
+
+    def release(self):
+        self.busy_wait_flag[0] = 1
+
+    @property
+    def busy_wait_flag_address(self):
+        return int(self.buffer.handle)
+
+    @property
+    def busy_wait_flag(self):
+        return ctypes.cast(self.busy_wait_flag_address, ctypes.POINTER(ctypes.c_int32))
+
+    def close(self):
+       buffer = getattr(self, 'buffer', None)
+       if buffer is not None:
+           buffer.close()
+
+    def __del__(self):
+        self.close()
+
+