Language

unilab.training.cuda_mps_cli¶

User-owned CUDA MPS daemon management for the UniLab CLI.

This module owns command-level lifecycle operations for explicitly requested MPS control daemons. It intentionally treats daemon ownership as an external, security-sensitive host service: every stop is gated by a UniLab daemon record containing the recorded process identity, and records are scoped by user and host. It never mutates the calling shell’s CUDA environment.

The topology model distinguishes single-GPU, single-task multi-GPU, and task-per-GPU placements. Only single-GPU execution is currently accepted; the other shapes fail closed rather than silently choosing a future topology.

Functions

`daemon_process_is_live`(daemon, *[, proc_stat])	Check both PID and Linux process start-time to avoid PID reuse.
`daemon_status`(daemon, *[, proc_stat])	Return record identity plus PID/start-time liveness evidence.
`discover_gpus`([run_command])	Discover physical host GPUs through the public `nvidia-smi` CLI.
`doctor`(selector, *[, root, ...])	Validate topology and daemon state without creating or changing anything.
`environment_for_daemon`(daemon)	Return launcher-facing CUDA environment without mutating this process.
`host_status`([root, cuda_visible_devices, ...])	Build read-only host capability and managed-daemon status.
`list_daemon_records`([root, uid, host])	List current-user/current-host daemon records without mutating them.
`main`([argv])	Run the CUDA MPS management CLI and return a process exit code.
`read_daemon_record`(root, name, *[, uid, host])	Load and scope-check a UniLab daemon record.
`resolve_topology`(selector, *[, ...])	Resolve selector syntax to canonical GPU UUIDs and a topology mode.
`select_environment_for_gpu`(gpu_uuid, *[, ...])	Select the sole live user-owned daemon environment for one GPU.
`start_daemon`(selector, *[, name, pipe_dir, ...])	Start an explicitly requested, user-owned daemon and record its identity.
`stop_all_daemons`(*[, root, run_command, ...])	Stop every current-user/current-host daemon recorded by UniLab.
`stop_daemon`(name, *[, root, run_command, ...])	Stop only a daemon whose UniLab record and process identity still match.
`visible_gpus`([cuda_visible_devices, run_command])	Return canonical GPU identities visible under the current host mask.

Classes

`ControlPipeStatus`	A discovered control pipe and whether a UniLab record owns it.
`DaemonRecord`	A UniLab-owned daemon identity; absence means no stop authority.
`GpuIdentity`	Canonical GPU identity suitable for persisted topology records.
`HostStatus`	Read-only host, control-pipe, and managed-daemon status.
`TopologyPlan`	A resolved selector before any host service is created.

Exceptions

`CudaMpsError`	A user-actionable, fail-closed CUDA MPS CLI error.
`StaleCudaMpsDaemonError`	A recorded daemon identity is no longer live; its record was cleaned.

exception unilab.training.cuda_mps_cli.CudaMpsError[source]¶

Bases: RuntimeError

A user-actionable, fail-closed CUDA MPS CLI error.

exception unilab.training.cuda_mps_cli.StaleCudaMpsDaemonError[source]¶

Bases: CudaMpsError

A recorded daemon identity is no longer live; its record was cleaned.

class unilab.training.cuda_mps_cli.GpuIdentity[source]¶

Bases: object

Canonical GPU identity suitable for persisted topology records.

Parameters:

index (int)
uuid (str)

index: int¶

uuid: str¶

to_dict()[source]¶

Return type:: dict[str, str | int]

__init__(index, uuid)¶

Parameters:

index (int)
uuid (str)

class unilab.training.cuda_mps_cli.DaemonRecord[source]¶

Bases: object

A UniLab-owned daemon identity; absence means no stop authority.

Parameters:

name (str)
uid (int)
host (str)
topology_mode (str)
gpu_uuids (tuple[str, ...])
pipe_directory (str)
log_directory (str)
pid (int)
process_start_ticks (int)
created_at (float)

name: str¶

uid: int¶

host: str¶

topology_mode: str¶

gpu_uuids: tuple[str, ...]¶

pipe_directory: str¶

log_directory: str¶

pid: int¶

process_start_ticks: int¶

created_at: float¶

to_dict()[source]¶

Return type:: dict[str, Any]

__init__(name, uid, host, topology_mode, gpu_uuids, pipe_directory, log_directory, pid, process_start_ticks, created_at)¶

Parameters:

name (str)
uid (int)
host (str)
topology_mode (str)
gpu_uuids (tuple[str, ...])
pipe_directory (str)
log_directory (str)
pid (int)
process_start_ticks (int)
created_at (float)

class unilab.training.cuda_mps_cli.ControlPipeStatus[source]¶

Bases: object

A discovered control pipe and whether a UniLab record owns it.

Parameters:

path (str)
kind (str)
managed_names (tuple[str, ...])

path: str¶

kind: str¶

managed_names: tuple[str, ...]¶

to_dict()[source]¶

Return type:: dict[str, Any]

__init__(path, kind, managed_names)¶

Parameters:

path (str)
kind (str)
managed_names (tuple[str, ...])

class unilab.training.cuda_mps_cli.HostStatus[source]¶

Bases: object

Read-only host, control-pipe, and managed-daemon status.

Parameters:

platform (str)
nvidia_smi_available (bool)
cuda_visible_devices (str | None)
gpus (tuple[GpuIdentity, ...])
control_pipes (tuple[ControlPipeStatus, ...])
managed_daemons (tuple[DaemonRecord, ...])
diagnostic (str | None)
compatible_topology_modes (tuple[str, ...])

platform: str¶

nvidia_smi_available: bool¶

cuda_visible_devices: str | None¶

gpus: tuple[GpuIdentity, ...]¶

control_pipes: tuple[ControlPipeStatus, ...]¶

managed_daemons: tuple[DaemonRecord, ...]¶

diagnostic: str | None = None¶

compatible_topology_modes: tuple[str, ...] = ('single_gpu', 'single_task_multi_gpu', 'task_per_gpu')¶

to_dict()[source]¶

Return type:: dict[str, Any]

__init__(platform, nvidia_smi_available, cuda_visible_devices, gpus, control_pipes, managed_daemons, diagnostic=None, compatible_topology_modes=('single_gpu', 'single_task_multi_gpu', 'task_per_gpu'))¶

Parameters:

platform (str)
nvidia_smi_available (bool)
cuda_visible_devices (str | None)
gpus (tuple[GpuIdentity, ...])
control_pipes (tuple[ControlPipeStatus, ...])
managed_daemons (tuple[DaemonRecord, ...])
diagnostic (str | None)
compatible_topology_modes (tuple[str, ...])

class unilab.training.cuda_mps_cli.TopologyPlan[source]¶

Bases: object

A resolved selector before any host service is created.

Parameters:

selector (str)
topology_mode (str)
gpus (tuple[GpuIdentity, ...])

selector: str¶

topology_mode: str¶

gpus: tuple[GpuIdentity, ...]¶

to_dict()[source]¶

Return type:: dict[str, Any]

__init__(selector, topology_mode, gpus)¶

Parameters:

selector (str)
topology_mode (str)
gpus (tuple[GpuIdentity, ...])

unilab.training.cuda_mps_cli.discover_gpus(run_command=<function run>)[source]¶

Discover physical host GPUs through the public nvidia-smi CLI.

Parameters:: run_command (Callable[..., CompletedProcess[str]])
Return type:: tuple[GpuIdentity, ...]

unilab.training.cuda_mps_cli.visible_gpus(cuda_visible_devices=None, run_command=<function run>)[source]¶

Return canonical GPU identities visible under the current host mask.

Parameters:

cuda_visible_devices (str | None)
run_command (Callable[..., CompletedProcess[str]])

Return type:

tuple[GpuIdentity, ...]

unilab.training.cuda_mps_cli.read_daemon_record(root, name, *, uid=None, host=None)[source]¶

Load and scope-check a UniLab daemon record.

name=None selects the sole current-user/current-host live daemon. The common single-GPU workflow therefore needs no daemon label.

Parameters:

root (Path | None)
name (str | None)
uid (int | None)
host (str | None)

Return type:

DaemonRecord

unilab.training.cuda_mps_cli.list_daemon_records(root=None, *, uid=None, host=None)[source]¶

List current-user/current-host daemon records without mutating them.

Parameters:

root (Path | None)
uid (int | None)
host (str | None)

Return type:

tuple[DaemonRecord, ...]

unilab.training.cuda_mps_cli.daemon_process_is_live(daemon, *, proc_stat=<function _proc_stat>)[source]¶

Check both PID and Linux process start-time to avoid PID reuse.

Parameters:

daemon (DaemonRecord)
proc_stat (Callable[[int], tuple[int, int] | None])

Return type:

bool

unilab.training.cuda_mps_cli.resolve_topology(selector, *, cuda_visible_devices=None, run_command=<function run>)[source]¶

Resolve selector syntax to canonical GPU UUIDs and a topology mode.

None selects the single visible GPU. On a single-GPU host this is the default CLI path: users need no selector, daemon name, or custom paths.

Parameters:

selector (str | None)
cuda_visible_devices (str | None)
run_command (Callable[..., CompletedProcess[str]])

Return type:

TopologyPlan

unilab.training.cuda_mps_cli.daemon_status(daemon, *, proc_stat=<function _proc_stat>)[source]¶

Return record identity plus PID/start-time liveness evidence.

Parameters:

daemon (DaemonRecord)
proc_stat (Callable[[int], tuple[int, int] | None])

Return type:

dict[str, Any]

unilab.training.cuda_mps_cli.host_status(root=None, *, cuda_visible_devices=None, run_command=<function run>, uid=None, host=None)[source]¶

Build read-only host capability and managed-daemon status.

Parameters:

root (Path | None)
cuda_visible_devices (str | None)
run_command (Callable[..., CompletedProcess[str]])
uid (int | None)
host (str | None)

Return type:

HostStatus

unilab.training.cuda_mps_cli.start_daemon(selector, *, name=None, pipe_dir=None, log_dir=None, root=None, cuda_visible_devices=None, run_command=<function run>, daemon=True, foreground_runner=None)[source]¶

Start an explicitly requested, user-owned daemon and record its identity.

The caller remains responsible for exporting the resulting environment into a training launcher; this command never changes its parent shell.

Parameters:

selector (str | None)
name (str | None)
pipe_dir (str | None)
log_dir (str | None)
root (Path | None)
cuda_visible_devices (str | None)
run_command (Callable[..., CompletedProcess[str]])
daemon (bool)
foreground_runner (Callable[[Mapping[str, str]], int] | None)

Return type:

DaemonRecord

unilab.training.cuda_mps_cli.environment_for_daemon(daemon)[source]¶

Return launcher-facing CUDA environment without mutating this process.

Parameters:: daemon (DaemonRecord)
Return type:: dict[str, str]

unilab.training.cuda_mps_cli.select_environment_for_gpu(gpu_uuid, *, root=None, proc_stat=<function _proc_stat>)[source]¶

Select the sole live user-owned daemon environment for one GPU.

This is launcher-facing data only. It does not mutate the current process, start a daemon, stop a daemon, or provide stop authority.

Parameters:

gpu_uuid (str)
root (Path | None)
proc_stat (Callable[[int], tuple[int, int] | None])

Return type:

dict[str, str]

unilab.training.cuda_mps_cli.stop_daemon(name, *, root=None, run_command=<function run>, proc_stat=<function _proc_stat>)[source]¶

Stop only a daemon whose UniLab record and process identity still match.

Parameters:

name (str)
root (Path | None)
run_command (Callable[..., CompletedProcess[str]])
proc_stat (Callable[[int], tuple[int, int] | None])

Return type:

DaemonRecord

unilab.training.cuda_mps_cli.stop_all_daemons(*, root=None, run_command=<function run>, proc_stat=<function _proc_stat>)[source]¶

Stop every current-user/current-host daemon recorded by UniLab.

Parameters:

root (Path | None)
run_command (Callable[..., CompletedProcess[str]])
proc_stat (Callable[[int], tuple[int, int] | None])

Return type:

tuple[DaemonRecord, ...]

unilab.training.cuda_mps_cli.doctor(selector, *, root=None, cuda_visible_devices=None, run_command=<function run>, torch_module=None, uid=None, host=None)[source]¶

Validate topology and daemon state without creating or changing anything.

Parameters:

selector (str | None)
root (Path | None)
cuda_visible_devices (str | None)
run_command (Callable[..., CompletedProcess[str]])
torch_module (Any)
uid (int | None)
host (str | None)

Return type:

dict[str, Any]

unilab.training.cuda_mps_cli.main(argv=None)[source]¶

Run the CUDA MPS management CLI and return a process exit code.

Parameters:: argv (Sequence[str] | None)
Return type:: int