launcher

Defines launchers for individual test jobs

class Launcher

Bases: ABC

abstractmethod run(job: Job) → int

class SubprocessLauncher(args: list[str] | None = None)

Bases: Launcher

context(job: Job) → Generator[None, None, None]

run(job: Job) → int

class MeasuredProcess(factory: Callable[[], Popen], *, name: str | None = None, sample_children: bool = False)

Bases: object

Wrapper around subprocess.Popen that samples resource usage via psutil.

Notes

• This is not a multiprocessing.Process. It’s intended to measure the actual launched workload PID (e.g., mpiexec/srun) and later can be extended to include children.

• Sampling is explicit: call sample_metrics() from your polling loop.

property pid: int | None

property returncode: int | None

start() → None

poll() → int | None

wait(timeout: float | None = None) → int

terminate() → None

kill() → None

shutdown(signum: int, grace_period: float = 0.05) → None

Best-effort: send signum, wait grace_period, then SIGKILL if still alive.

sample_metrics() → None

Append a sample dict to self.samples. Safe to call frequently.

get_measurements() → dict[str, Any]

Summarize collected samples into min/max/ave by key.

default_job_launcher(case: Job) → Launcher