mstar.utils.cpus

The CPUs this process may use, and the plan every process of a deployment reads.

os.cpu_count() reports the host’s CPUs. In a container the cgroup CPU quota can be far smaller, and a thread pool sized to the host then runs more threads than the quota lets run, so a fork-join op waits on whichever thread the scheduler parked.

CpuPlan holds the process mask, how many of its CPUs the quota lets run, the CPUs the step_thread_core knob pins (each worker’s step threads and the API server’s result threads, see mstar.utils.threads), and the pool sizes, taken from the CPUs left after those. The API server and every worker build it with deployment_plan from the same inputs (the knob, the mask, the deployment’s devices), so they agree on it without talking.

Placement under auto is by the device’s rank among the host’s GPUs on its NUMA node, as sysfs lists them (every GPU of the host, not only the ones this process was given), so servers on different GPUs pick different CPUs without knowing about each other. A worker takes the core at its device’s rank from the bottom of the node’s allowed CPUs, the API server the core at its first device’s rank from the top, so the cores a GPU picks from must hold two per GPU or auto pins nothing. Servers on one GPU can’t be told apart that way, so the first holds a lock on it (claim_gpus) and the next pins nothing.

Functions

cgroup_cpu_quota([root, proc_cgroup])	The tightest cgroup v2 cpu.max quota over this process's cgroup and its ancestors, in CPUs; None when no level sets one (or cgroup v2 is absent).
claim_gpus(addresses)	Hold a lock on each GPU for the life of this process; if one can't be had, hold none of them and say why.
deployment_plan(setting, addresses, *[, ...])	The plan for a deployment whose workers' devices sit at PCI addresses, in the conductor's worker order (None for a device sysfs doesn't place, such as a CPU worker; only auto reads them).
device_addresses(devices)	The PCI address of each worker's device (cuda:1); None for a CPU worker, or where the platform doesn't report one.
devices_plan(setting, devices)	deployment_plan for the workers' device strings, in the conductor's worker order: the same call in the API server and in every worker.
format_cpu_list(cpus)	cpus as ranges, the inverse of parse_cpu_list.
host_gpus(like[, root])	The host's GPUs in PCI address order: the functions with the vendor and class of the device at like (0000:40:00.0).
install(new)	Make new this process's plan (None: back to nothing pinned).
parse_cpu_list(text)	The CPUs in a list such as 0-3,8,10-11 (as sysfs and taskset write them); empty for an empty string.
plan()	This process's plan: the one install set, else nothing pinned.
process_mask()	The CPUs this process may run on; every CPU where affinity is unknown.
smt_siblings(cpu)	The hardware threads of cpu's core, cpu included.
usable_cpus()	CPUs this process may run on (its affinity, else os.cpu_count()), capped by the cgroup quota rounded down.

Classes

CpuPlan(mask, usable[, workers, result, why])	See the module docstring.
HostGpu(address, node, local)	One of the host's GPUs as sysfs lists it: its PCI address, NUMA node and the node's CPUs (None where sysfs doesn't say).
StepCpus(cpus, why)	The CPUs one worker's main, GPU and plan threads share (one core's hardware threads in the mask), and how they were chosen, for the log.

Exceptions

CpuPlanError

A step_thread_core value the process mask can't honour.

class mstar.utils.cpus.CpuPlan(mask, usable, workers=(), result=None, why='step_thread_core off')

Bases: object

See the module docstring. workers has one entry per worker of the deployment, in the conductor’s worker order (empty, or all None, when nothing is pinned).

Parameters:

• mask (frozenset[int])

• usable (int)

• workers (tuple[StepCpus | None, ...])

• result (int | None)

• why (str)

property copy_workers: int

Threads a large output read copies over (host_copy.HostCopier): one encode’s share of the CPUs left. Reads run one at a time on the data worker’s thread, each ahead of its own encode.

property encode_workers: int

Outputs the data worker encodes at once. A PNG deflates on png_threads of its own, so the pool times that stays within the CPUs left.

property left: int

what the pools are sized from.

Type:

The CPUs the quota lets run, less the pinned ones

mask: frozenset[int]

property pinned: frozenset[int]

property png_threads: int

Threads one PNG deflates on (image_io.uint8_to_png).

property preprocess_threads: int

The API server’s intra-op threads when neither the config nor the model sets them. A quarter of the CPUs left, at most 4: the process shares them with the engine workers, and its per-request ops are a few MB.

result: int | None = None

usable: int

why: str = 'step_thread_core off'

workers: tuple[StepCpus | None, ...] = ()

exception mstar.utils.cpus.CpuPlanError

Bases: ValueError

A step_thread_core value the process mask can’t honour.

class mstar.utils.cpus.HostGpu(address, node, local)

Bases: NamedTuple

One of the host’s GPUs as sysfs lists it: its PCI address, NUMA node and the node’s CPUs (None where sysfs doesn’t say).

Parameters:

• address (str)

• node (str | None)

• local (frozenset[int] | None)

address: str

Alias for field number 0

local: frozenset[int] | None

Alias for field number 2

node: str | None

Alias for field number 1

class mstar.utils.cpus.StepCpus(cpus, why)

Bases: NamedTuple

The CPUs one worker’s main, GPU and plan threads share (one core’s hardware threads in the mask), and how they were chosen, for the log.

Parameters:

• cpus (frozenset[int])

• why (str)

cpus: frozenset[int]

Alias for field number 0

why: str

Alias for field number 1

mstar.utils.cpus.cgroup_cpu_quota(root=_CGROUP_ROOT, proc_cgroup='/proc/self/cgroup')

The tightest cgroup v2 cpu.max quota over this process’s cgroup and its ancestors, in CPUs; None when no level sets one (or cgroup v2 is absent).

Parameters:

• root (Path)

• proc_cgroup (str)

Return type:

float | None

mstar.utils.cpus.claim_gpus(addresses)

Hold a lock on each GPU for the life of this process; if one can’t be had, hold none of them and say why.

auto depends only on the PCI address and the mask, so two servers on one GPU (two MIG instances report one address) would take the same CPUs. The kernel drops an flock however the process exits, so a dead server leaves nothing to clean up. The file is never unlinked: another process may already hold an fd on this inode, and removing it would let a third lock a different one. The lock only reaches servers that share _GPU_LOCK_DIR.

Parameters:

addresses (Iterable[str | None])

Return type:

str | None

mstar.utils.cpus.deployment_plan(setting, addresses, *, mask=None, usable=None, pci_root=_PCI_DEVICES, siblings=smt_siblings)

The plan for a deployment whose workers’ devices sit at PCI addresses, in the conductor’s worker order (None for a device sysfs doesn’t place, such as a CPU worker; only auto reads them).

off pins nothing. A CPU number is the first worker’s CPU; the next workers take the next cores above it, and the API server the last core of the mask; a number outside the mask, or too near its top for every worker, raises CpuPlanError. auto places by device rank (module docstring), and pins nothing, saying why in CpuPlan.why, when the mask is too small. Either way at least one CPU of the mask stays unpinned.

Parameters:

• setting (str | int)

• addresses (Sequence[str | None])

• mask (Collection[int] | None)

• usable (int | None)

• pci_root (Path)

• siblings (Callable[[int], frozenset[int]])

Return type:

CpuPlan

mstar.utils.cpus.device_addresses(devices)

The PCI address of each worker’s device (cuda:1); None for a CPU worker, or where the platform doesn’t report one.

Parameters:

devices (Sequence[str])

Return type:

list[str | None]

mstar.utils.cpus.devices_plan(setting, devices)

deployment_plan for the workers’ device strings, in the conductor’s worker order: the same call in the API server and in every worker.

Parameters:

• setting (str | int)

• devices (Sequence[str])

Return type:

CpuPlan

mstar.utils.cpus.format_cpu_list(cpus)

cpus as ranges, the inverse of parse_cpu_list.

Parameters:

cpus (Iterable[int])

Return type:

str

mstar.utils.cpus.host_gpus(like, root=_PCI_DEVICES)

The host’s GPUs in PCI address order: the functions with the vendor and class of the device at like (0000:40:00.0). Every process on the host sees the same list, whatever GPUs it was given; empty without sysfs.

Parameters:

• like (str)

• root (Path)

Return type:

list[HostGpu]

mstar.utils.cpus.install(new)

Make new this process’s plan (None: back to nothing pinned).

Parameters:

new (CpuPlan | None)

Return type:

None

mstar.utils.cpus.parse_cpu_list(text)

The CPUs in a list such as 0-3,8,10-11 (as sysfs and taskset write them); empty for an empty string.

Parameters:

text (str)

Return type:

frozenset[int]

mstar.utils.cpus.plan()

This process’s plan: the one install set, else nothing pinned.

Return type:

CpuPlan

mstar.utils.cpus.process_mask()

The CPUs this process may run on; every CPU where affinity is unknown.

Return type:

frozenset[int]

mstar.utils.cpus.smt_siblings(cpu)

The hardware threads of cpu’s core, cpu included.

Parameters:

cpu (int)

Return type:

frozenset[int]

mstar.utils.cpus.usable_cpus()

CPUs this process may run on (its affinity, else os.cpu_count()), capped by the cgroup quota rounded down. At least 1.

Return type:

int