Add configurable GPU sharing for pipeline workers

docs/cli.md

@@ -121,10 +121,12 @@ standalone raw-comparison and Alard--Lupton review servers are available as
 command-specific contracts. See [xScan](components/xscan.md).
 
 `infer-real-bogus --executor dragon|mpi` distributes complete minibatches;
-`run-pipeline --executor dragon|mpi` distributes complete image pairs through
-xPois, xFit and xScan. Optional round flags retain the workers across passes.
-These commands require a shared input/output filesystem and the matching
-runtime launcher.
+`run-pipeline --executor dragon|mpi|processes|threads` distributes complete
+image pairs through xPois, xFit and xScan. Optional round flags retain the
+workers across passes. Dragon and MPI require a shared input/output
+filesystem and the matching runtime launcher. The local `processes` and
+`threads` pipeline executors share one visible GPU; see
+[GPU sharing](components/xscan.md#share-a-gpu-between-image-pairs).
 
 `infer-real-bogus` consumes packaged datasets. Direct FITS candidate scoring
 uses the CUDA Python API `cuphoton.xscan.fits_inference.predict_fits`.

docs/components/xfit.md

@@ -355,6 +355,12 @@ IDs and input order are restored in the merged artifacts.
 The task count must be at least the MPI rank count. Dragon uses the smaller
 of the requested worker count and task count.
 
+With Dragon, `--workers-per-gpu` allows multiple workers per GPU and
+`--mps-pipe-directory` requires each worker to connect to an existing MPS v2
+service. `--max-workers` caps the total worker count. See
+[GPU sharing](xscan.md#share-a-gpu-between-image-pairs) for service setup and
+connection requirements.
+
 The same executor commands accept a FITS candidate manifest. Planning checks
 headers, candidate bounds and source hashes without decoding image pixels.
 Each bound worker reads the candidate region and retains its stamps during

docs/components/xscan.md

@@ -561,8 +561,8 @@ scientific evidence; Dragon saves each result in
 `items/<item_id>/summary.json` and checks it again in the coordinator.
 The evidence decoder returns the 22 named arrays for comparison.
 
-The `run-pipeline` command accepts the same descriptors with either Dragon or
-MPI. Save the configuration and items above as a manifest:
+The `run-pipeline` command accepts the same descriptors with Dragon, MPI,
+or local GPU workers. Save the configuration and items above as a manifest:
 
 ```python
 import json
@@ -616,6 +616,111 @@ decoding and reader stream completion; `input_h2d` measures NPY upload calls.
 Timings are host elapsed times, so distinguish model initialization,
 first-item compilation and warmed context reuse when comparing runs.
 
+### Share a GPU between image pairs
+
+`--workers-per-gpu N` allows independent image pairs to use the same GPU.
+The default is one worker per GPU. Each worker retains its own model and
+pipeline context, so additional workers consume additional device and host
+memory. Start with a small count and measure throughput and memory on the
+intended images; increasing the count can also slow a saturated GPU.
+
+For a local run, expose exactly one GPU before starting Python and select
+`--executor processes` or `--executor threads`:
+
+```bash
+CUDA_VISIBLE_DEVICES=0 cuphoton xscan run-pipeline \
+  --executor processes --workers-per-gpu 4 \
+  --manifest pipeline.json --output-dir runs --name processes-4 \
+  --worker-timeout-sec 600 --warmup-rounds 1 --measure-rounds 3
+
+CUDA_VISIBLE_DEVICES=0 cuphoton xscan run-pipeline \
+  --executor threads --workers-per-gpu 4 \
+  --manifest pipeline.json --output-dir runs --name threads-4 \
+  --worker-timeout-sec 600 --warmup-rounds 1 --measure-rounds 3
+```
+
+These options call the same Python entry point:
+
+```python
+from pathlib import Path
+from cuphoton.xscan.pipeline_executor import run_pipeline_manifest
+
+result = run_pipeline_manifest(
+    executor="processes",
+    manifest_path=Path("pipeline.json"),
+    output_root=Path("runs"),
+    workers_per_gpu=4,
+    worker_timeout_sec=600,
+)
+```
+
+Process mode creates persistent spawned processes. Thread mode keeps workers
+in one supervised process, with a separate model, CuPy memory pool and pair
+of CUDA streams per thread. Initialization and first use are serialized;
+use a warmup round to exclude that cost from measured rounds. Outputs from
+every planned round are retained. The worker count is capped by the number
+of items. A timeout or worker failure fails the run and stops its owned
+processes; consult the terminal `summary.json` before using timing results.
+
+Thread mode accepts prepared NPY inputs. It works with ordinary Python and
+can also use a compatible free-threaded interpreter. To require the GIL to
+remain disabled, add `--require-free-threaded` (Python:
+`require_free_threaded=True`) to the threads command. This checks the running
+interpreter; it does not disable the GIL. Extensions may re-enable it on
+import, and forcing it off does not establish extension thread safety. See
+[Python's free-threading guide](https://docs.python.org/3/howto/free-threading-python.html).
+Use process mode for FITS inputs. Free-threaded execution requires validation
+of the installed scientific libraries; it is not a general compatibility
+claim for native extensions.
+
+With the shared Dragon executor, the same count applies to every discovered
+GPU. `--max-workers` caps the total number of processes. Dragon uses distinct
+GPUs before assigning additional workers to each one:
+
+```bash
+dragon cuphoton xscan run-pipeline --executor dragon \
+  --workers-per-gpu 4 --max-workers 16 \
+  --manifest pipeline.json --output-dir runs --name dragon-shared
+```
+
+The shared Dragon Python API accepts `workers_per_gpu` as well. Physical GPU
+checks require workers assigned to the same GPU to agree on its identity,
+and still reject collisions between different requested GPUs. MPI retains
+one rank per physical GPU and rejects the sharing options.
+
+### Select an existing MPS service
+
+Process workers can run under CUDA's ordinary process scheduling or an
+externally managed Multi-Process Service (MPS). MPS allows CUDA work from
+different processes to overlap on the GPU. To require a connection to an
+existing MPS v2 service, add an absolute pipe directory:
+
+```bash
+CUDA_VISIBLE_DEVICES=GPU-your-allocated-device-uuid \
+  cuphoton xscan run-pipeline --executor processes --workers-per-gpu 4 \
+  --mps-pipe-directory /tmp/my-job-mps \
+  --manifest pipeline.json --output-dir runs --name mps-4
+```
+
+The Python keyword is `mps_pipe_directory`. It is also accepted by the shared
+Dragon executor. Every worker must appear in that service's client list
+after CUDA initialization, or startup fails. The service and clients must
+use a common PID namespace for this check. On multiple hosts use the same
+absolute node-local directory on each host, such as a job-specific `/tmp`
+path. Dragon resolves its assigned physical device to a UUID before CUDA
+initialization and checks the reported UUID, avoiding MPS ordinal remapping.
+This requires `nvidia-smi` on each worker host. Record the runtime's
+configuration when comparing runs. Without this option, process workers
+inherit the environment's CUDA
+and MPS settings without asserting an MPS connection.
+
+cuPhoton leaves service lifetime, device visibility and resource limits to
+the launcher or administrator. It does not start or stop MPS or change GPU
+compute mode. The explicit connection check uses the
+[MPS v2 control interface](https://docs.nvidia.com/deploy/mps/latest/mpsv2-interface.html).
+Thread mode rejects `--mps-pipe-directory`; its threads share a process and
+CUDA context without needing an interprocess service.
+
 ## Review workflow
 
 ```bash

docs/dragon.md

@@ -31,6 +31,14 @@ pipeline keeps intermediate arrays inside each GPU worker while it
 processes an image pair. Each GPU handles complete items; these paths do not
 split one image across GPUs.
 
+The shared executor accepts `--workers-per-gpu` to place several independent
+workers on each GPU; its default is one. `--max-workers` remains a total
+process cap. See [GPU sharing](components/xscan.md#share-a-gpu-between-image-pairs)
+for CLI and Python examples, local process/thread modes, and external MPS
+connection checks. These sharing and MPS options apply to `xfit fit-dipoles`,
+`xscan infer-real-bogus`, and `xscan run-pipeline` with `--executor dragon`.
+The standalone XPOIS batch API retains its existing placement behavior.
+
 ### Launch descriptors
 
 Dragon workers load immutable, hash-checked launch descriptors from the shared

src/cuphoton/core/cli/executor.py

@@ -6,13 +6,16 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
+
 from ..benchmark import BenchmarkOptions
 from .command import CommandError
 from .invariants import (
     FloatInvariant,
     NonNegativeIntegerInvariant,
     PositiveIntegerInvariant,
     SetInvariant,
+    StringInvariant,
 )
 
 
@@ -26,6 +29,8 @@ class ExecutorOptions:
     rank_setup_timeout_sec: float | None = None
     warmup_rounds: int | None = None
     measure_rounds: int | None = None
+    workers_per_gpu: int | None = None
+    mps_pipe_directory: str | None = None
 
     class ExecutorArg(SetInvariant):
         _arg = "--executor"
@@ -42,10 +47,26 @@ class MaxWorkersArg(PositiveIntegerInvariant):
 
     class WorkerTimeoutSecArg(FloatInvariant):
         _arg = "--worker-timeout-sec"
-        _help = "Dragon worker lifetime across all rounds. [default: 3600]"
+        _help = "Dragon or local GPU worker lifetime. [default: 3600]"
         _default = None
         _min = 0.001
 
+    class WorkersPerGpuArg(PositiveIntegerInvariant):
+        _arg = "--workers-per-gpu"
+        _help = (
+            "Maximum workers sharing each GPU with Dragon or a local "
+            "pipeline executor. [default: 1]"
+        )
+        _default = None
+
+    class MpsPipeDirectoryArg(StringInvariant):
+        _arg = "--mps-pipe-directory"
+        _help = (
+            "Existing MPS v2 pipe directory for Dragon or process workers; "
+            "require every worker to connect to that service."
+        )
+        _default = None
+
     class ResultTimeoutSecArg(FloatInvariant):
         _arg = "--result-timeout-sec"
         _help = "Dragon result and artifact grace period. [default: 60]"
@@ -77,24 +98,34 @@ class MeasureRoundsArg(PositiveIntegerInvariant):
     def executor_options(self) -> dict:
         """Reject ignored flags before loading a numerical runtime."""
 
-        dragon = {
+        dragon_only = {
             "max_workers": self.max_workers,
-            "worker_timeout_sec": self.worker_timeout_sec,
             "result_timeout_sec": self.result_timeout_sec,
         }
+        sharing = {
+            "workers_per_gpu": self.workers_per_gpu,
+            "worker_timeout_sec": self.worker_timeout_sec,
+        }
+        mps = {"mps_pipe_directory": self.mps_pipe_directory}
+        dragon = {**dragon_only, **sharing, **mps}
         mpi = {"rank_setup_timeout_sec": self.rank_setup_timeout_sec}
         rounds = {
             "warmup_rounds": self.warmup_rounds,
             "measure_rounds": self.measure_rounds,
         }
+        invalid: Mapping[str, str | int | float | None]
         if self.executor == "local":
             invalid = {**dragon, **mpi, **rounds}
         elif self.executor == "dragon":
             invalid = mpi
         elif self.executor == "mpi":
             invalid = dragon
+        elif self.executor == "processes":
+            invalid = {**dragon_only, **mpi}
+        elif self.executor == "threads":
+            invalid = {**dragon_only, **mpi, **mps}
         else:
-            raise CommandError("executor must be local, dragon, or mpi")
+            raise CommandError(f"unsupported executor: {self.executor}")
         supplied = [
             "--" + name.replace("_", "-")
             for name, value in invalid.items()
@@ -107,11 +138,16 @@ def executor_options(self) -> dict:
             )
         if self.executor == "local":
             return {}
-        options: dict[str, int | float | BenchmarkOptions | None] = {
+        by_executor: dict[str, Mapping[str, str | int | float | None]] = {
+            "dragon": dragon,
+            "mpi": mpi,
+            "processes": {**sharing, **mps},
+            "threads": sharing,
+        }
+        runtime_options = by_executor[self.executor]
+        options: dict[str, str | int | float | BenchmarkOptions | None] = {
             name: value
-            for name, value in (
-                dragon if self.executor == "dragon" else mpi
-            ).items()
+            for name, value in runtime_options.items()
             if value is not None
         }
         options["benchmark"] = (