[cuda.compute] Add explicit device binding for cuda.compute algorithms

docs/python/compute/developer_overview.rst

@@ -539,7 +539,10 @@ build targets compute capabilities with no GPU queried, and a deserialized
 wrapper is reconstructed without a device binding — so its device is known only
 at call time. Each call resolves the per-device loaded result from a dictionary
 inside the per-cc build results, where a completed lookup also takes no explicit
-lock.
+lock. ``clone(device_id=...)`` copies per-call descriptors recursively while
+sharing ``_PerCCBuildResults``. An integer selects the bound fast path;
+``None`` clears bindings without device queries or code loading, deferring
+device resolution to each call. The source wrapper is unchanged.
 
 Design requirements
 +++++++++++++++++++

docs/python/compute/index.rst

@@ -410,21 +410,35 @@ The same argument-matching rules described above for the object-based API apply
 to a deserialized algorithm: the dtypes, iterator kinds, and operator you pass
 when invoking it must match those used when it was originally built.
 
+Use ``clone`` on a deserialized or ``make_*`` executor to obtain independent
+mutable host-side argument state while sharing compiled code:
+
+.. code-block:: python
+
+   executor = cuda.compute.deserialize(blob)
+   bound_executor = executor.clone(device_id=0)
+   unbound_executor = bound_executor.clone(device_id=None)
+
+``device_id`` is required and keyword-only. An integer binds the clone to that
+device, loading code once without recompiling; calls then skip device queries
+and build-result lookups. The device must be current during execution. Cloning
+restores the previous CUDA context even on failure.
+
+``None`` clears any binding on the copy, including in nested algorithms, without
+device access or code loading. The unbound copy resolves the current device when
+called. Cloning leaves the source unchanged, and serialization omits bindings.
+
+Use each executor from one thread at a time (see
+:ref:`Free-threaded Python <cuda.compute.free_threading>`). For concurrent use,
+give each thread its own clone; deserializing separately also works but loads
+native build state independently.
+
 Blobs are versioned and self-describing, but they are not a long-term storage
 format: compatibility across ``cuda-cccl`` versions is not guaranteed, and
 loading a blob produced by a different version may be rejected with a clear
 error. Persist the inputs needed to rebuild (or re-serialize) rather than
 relying on old blobs surviving an upgrade.
 
-The same threading rules also apply: like any other reusable algorithm object,
-a deserialized algorithm must be used by one thread at a time unless access is
-externally serialized (see :ref:`Free-threaded Python <cuda.compute.free_threading>`).
-For concurrent use, call :func:`deserialize <cuda.compute.algorithms.deserialize>`
-in each thread — reconstruction performs no recompilation, so per-thread
-deserialization from one shared blob is cheap. (Currently each deserialized
-object loads its native build state independently; a future release may share
-that state behind the scenes.)
-
 .. _cuda.compute.ahead_of_time_compilation:
 
 Ahead-of-Time Compilation

python/cuda_cccl/cuda/compute/_cccl_interop.py

@@ -472,8 +472,8 @@ def resolve_build_result(build_results: dict, bound_result=None):
 
     ``bound_result`` is a default-build wrapper's construction-time binding
     (see cache_build_results): already the loaded result for the wrapper's
-    device, returned without any device query. Deserialized wrappers have no
-    binding and resolve per call.
+    device, returned without any device query. Deserialized wrappers resolve
+    per call unless explicitly bound with ``clone(device_id=...)``.
     """
     if bound_result is not None:
         return bound_result

python/cuda_cccl/cuda/compute/_serialization/serializable.py

@@ -259,6 +259,91 @@ def _after_deserialize(self) -> None:
         override this to rebind it; the default is a no-op.
         """
 
+    def clone(self: _S, *, device_id: None | int) -> _S:
+        """Return an executor with independent mutable host-side argument state, sharing compiled code.
+
+        If ``device_id`` is an integer, bind the executor to that device;
+        if it is ``None``, select the current device at each call.
+
+        Args:
+            device_id: Required keyword-only CUDA ordinal, or ``None`` to clear
+                bindings, including nested ones, without device access or code
+                loading.
+
+        Returns:
+            A new executor with the same call signature; the source is unchanged.
+            Use each executor from one thread at a time.
+
+        Raises:
+            RuntimeError: If no compiled result is compatible with the device,
+                or loading that result fails.
+
+        Binding loads code once without recompiling and restores the previous
+        CUDA context even on failure. The bound device must be current during
+        execution. Bindings are not serialized.
+        """
+        if device_id is None:
+            return self._clone(device_id=None, cc=None)
+
+        from cuda.bindings import driver
+        from cuda.core import Device
+        from cuda.core._utils.cuda_utils import handle_return
+
+        from .._cccl_interop import cc_to_key
+
+        device = Device(device_id)
+        previous_context = handle_return(driver.cuCtxGetCurrent())
+        try:
+            device.set_current()
+            cc = cc_to_key(tuple(device.compute_capability))
+            return self._clone(device.device_id, cc)
+        finally:
+            handle_return(driver.cuCtxSetCurrent(previous_context))
+
+    def _clone(self: _S, device_id: None | int, cc: None | int) -> _S:
+        """Copy per-call descriptors, optionally binding code on the current device."""
+        from .._cccl_interop import key_to_cc
+
+        obj = type(self).__new__(type(self))
+        obj._bound_build_result = None
+        for attr, kind in self.__serialization_schema__:
+            value = getattr(self, attr)
+            if isinstance(kind, _BuildResults):
+                # Share the ownership/clone protocol, not mutable descriptors.
+                # A single cubin can also run on a newer minor of the same SM
+                # major, matching the native serialization compatibility rule.
+                if device_id is not None:
+                    assert cc is not None
+                    build_cc = cc
+                    if len(value) == 1:
+                        (candidate,) = value
+                        if candidate // 10 == cc // 10 and candidate <= cc:
+                            build_cc = candidate
+                    if build_cc not in value:
+                        available = ", ".join(
+                            f"{major}.{minor}"
+                            for major, minor in (key_to_cc(k) for k in sorted(value))
+                        )
+                        major, minor = key_to_cc(cc)
+                        raise RuntimeError(
+                            f"This algorithm was compiled for compute capabilities [{available}], "
+                            f"but device {device_id} has compute capability {major}.{minor}. "
+                            f"Rebuild with compute_capability including {cc}."
+                        )
+                    obj._bound_build_result = value.resolve(build_cc, device_id)
+            elif isinstance(kind, _SubObject) and isinstance(value, Serializable):
+                # For example, select executes a nested three-way partitioner.
+                value = value._clone(device_id, cc)
+            else:
+                # Descriptor codecs omit runtime pointers and reconstruct fresh
+                # Cython objects, without serializing or reloading native code.
+                w = codec.Writer()
+                kind.write(w, value, self)
+                value = kind.read(codec.Reader(w.getvalue()), obj)
+            setattr(obj, attr, value)
+        obj._after_deserialize()
+        return obj
+
     def serialize(self) -> bytes:
         """Serialize this built algorithm to a self-contained serialization blob."""
         w = codec.begin(type(self).__qualname__)