worker¶

The worker harness: a pod of a split task computes units of the ledger in a pool.

simrun-worker (main) claims units of the ledger of its task, computes them in a local process pool with the functions of the task type and writes a report; simrun-merge (merge_main) merges the units once the ledger is complete; simrun-exec (exec_main) runs the command of an unsplit plugin task and writes a report of it. A pod is evicted with SIGTERM and killed after its grace period, and is restarted with the same worker id, which releases the claims of the incarnation before.

MEMORY_SOURCE_ENV `module-attribute` ¶

MEMORY_SOURCE_ENV = 'SIMRUN_MEMORY_SOURCE'

Where the harness reads its memory: cgroup (the default, a pod) or process (its process tree in /proc, the local backend, whose processes share the cgroup of the machine or of the session).

ConfigurationError ¶

Bases: Exception

The environment or the settings of a worker are wrong.

UnitError ¶

UnitError(index)

Bases: RuntimeError

A unit failed in the function of the task type; it was released.

Create the error of the unit index.

Terminated ¶

Bases: Exception

The worker stopped on SIGTERM before its units were done.

WorkerReport `dataclass` ¶

WorkerReport(
    worker,
    node,
    processes,
    units_done=list(),
    seconds=0.0,
    peak_memory_bytes=None,
    memory_limit_bytes=None,
    per_worker_gb=0.0,
    oom_kills=0,
    unit_seconds=0.0,
    failed=False,
    started=None,
    prepare_seconds=0.0,
    cpu_seconds=0.0,
    throttled_seconds=None,
)

What a worker did, written as task_dir/workers/<worker>.json.

Attributes:

Name	Type	Description
`worker`	`str`	The worker id.
`node`	`str`	The server the worker ran on.
`processes`	`int`	The number of processes of the pool.
`units_done`	`list[int]`	The units the worker completed.
`seconds`	`float`	The duration of the worker.
`peak_memory_bytes`	`int \| None`	The largest sample of the anonymous memory of the cgroup (`MemorySampler`), without the page cache.
`memory_limit_bytes`	`int \| None`	The memory limit of the cgroup.
`per_worker_gb`	`float`	The memory per process in GiB the worker was sized with.
`oom_kills`	`int`	The processes of its cgroup the out of memory killer killed while it ran (`memory.events` `oom_kill`), the evidence that an exit 137 was a kill for memory.
`unit_seconds`	`float`	The seconds of the units it completed, each measured in the process which computed it (the setup of a process in its first unit), summed: `unit_seconds / len(units_done)` is the time of a unit in one process, without the waits of the worker (for the units of other workers, a coordinator) and the processes idle at its end.
`failed`	`bool`	Whether the worker failed: a unit failed (exit 3), or the command of `simrun-exec` exited with an error; a stop or a kill is no failure.
`started`	`float \| None`	When the harness started, in seconds since the epoch, before it installed the sources (`HARNESS_STARTED`); None in a report of simrun 0.3.0 and before.
`prepare_seconds`	`float`	The seconds from the start of the harness to the start of its work: the install of the sources, the restart, the imports and the staging of the inputs.
`cpu_seconds`	`float`	The user and system seconds of the harness, of the children it reaped (the command of `simrun-exec`) and of the units it completed, each measured in the process of the pool which computed it.
`throttled_seconds`	`float \| None`	The seconds the processes of the cgroup of the worker waited for its CPU quota during the work, None without a cgroup of its own (the local backend).

from_json `classmethod` ¶

from_json(text)

Read a report; a field this version does not know is ignored.

Parameters:

Name	Type	Description	Default
`text`	`str`	The JSON of `write_report`, possibly of a newer harness.	required

Returns:

Type	Description
`WorkerReport`	The report.

MemorySampler ¶

MemorySampler(source='cgroup')

The maximum of the anonymous memory of the worker over its samples.

The anonymous memory is what the processes allocated: of the cgroup (memory.stat anon), or of the process tree of the harness (process, see process_tree_anon). memory.peak also counts the page cache of staging, installing and writing outputs, which the kernel reclaims at the limit instead of killing.

Attributes:

Name	Type	Description
`source`		Where the memory is read.
`peak`	`int \| None`	The largest sample in bytes, None before one was taken.

Create a sampler without samples.

sample ¶

sample()

Read the anonymous memory and keep it when it is the largest.

harness_started ¶

harness_started()

When the harness started, set in HARNESS_STARTED by its first call.

The variable survives the restart after the install of the sources, so the start is the one before the install.

Returns:

Type	Description
`float`	Seconds since the epoch.

cpu_seconds ¶

cpu_seconds()

The user and system seconds of this process and of its ended children.

processes ¶

processes(cores, memory_bytes, base_gb, per_worker_gb)

The number of processes the cores and the memory allow.

Parameters:

Name	Type	Description	Default
`cores`	`int`	The number of cores.	required
`memory_bytes`	`int \| None`	The memory limit in bytes, `None` without a limit.	required
`base_gb`	`float`	The memory of the task besides its processes in GiB.	required
`per_worker_gb`	`float`	The memory per process in GiB.	required

Returns:

Type	Description
`int`	`min(cores, floor((memory - base) / per_worker))`, at least 1.

Raises:

Type	Description
`ValueError`	Not even one process fits into the memory.

write_report ¶

write_report(report, task_dir, attempt=0)

Write the report as task_dir/workers/<worker>-<attempt>.json.

The file is written as a temporary file and renamed.

Parameters:

Name	Type	Description	Default
`report`	`WorkerReport`	The report.	required
`task_dir`	`Path`	The task directory.	required
`attempt`	`int`	The attempt of the pod.	`0`

Returns:

Type	Description
`Path`	The path of the report.

memory_source ¶

memory_source(env)

Where the harness reads its memory, SIMRUN_MEMORY_SOURCE, cgroup by default.

Raises:

Type	Description
`ConfigurationError`	The value is neither `cgroup` nor `process`.

process_tree_anon ¶

process_tree_anon(pid, proc=PROC)

The anonymous memory of a process and all of its descendants in bytes.

The sum over the processes of RssAnon (VmRSS without it): pages shared after a fork count once per process, so it is an upper bound.

Parameters:

Name	Type	Description	Default
`pid`	`int`	The root of the tree.	required
`proc`	`Path`	The mount of procfs.	`PROC`

Returns:

Type	Description
`int \| None`	The bytes, None when the process is gone.

run_worker ¶

run_worker(
    task_type,
    settings,
    paths,
    worker,
    *,
    cores,
    memory_bytes,
    base_gb,
    per_worker_gb,
    grace_s=GRACE_S,
    attempt=0,
    stop=None,
    heartbeat_s=HEARTBEAT_S,
    coordinator=False,
    stale_s=None,
    poll_s=COORDINATOR_POLL_S,
    memory_source="cgroup",
)

Compute units of the ledger of a task until none is pending.

The worker touches its heartbeat in the ledger every heartbeat_s from its start to its end. A coordinator does not return when nothing is pending: until the ledger is complete it waits, releases the claims of workers whose heartbeat is older than stale_s every poll_s and computes them.

Releases the claims of an earlier incarnation of worker, creates the ledger task_dir/units, keeps processes(...) units in flight in a pool of processes which run the setup of the task type once, completes a unit after its function returned and appends a unit_done line to task_dir/progress.jsonl. On SIGTERM (or when stop is set, also before the first claim) it stops claiming, waits for the running units up to grace_s and releases the rest. The report task_dir/workers/<worker>-<attempt>.json is written in every case.

Parameters:

Name	Type	Description	Default
`task_type`	`TaskType`	The task type, splittable.	required
`settings`	`BaseModel`	The settings of the task.	required
`paths`	`TaskPaths`	The locations of the task.	required
`worker`	`str`	The worker id, the same for every incarnation of a pod.	required
`cores`	`int`	The number of cores.	required
`memory_bytes`	`int \| None`	The memory limit in bytes, `None` without a limit.	required
`base_gb`	`float`	The memory of the task besides its processes in GiB.	required
`per_worker_gb`	`float`	The memory per process in GiB.	required
`grace_s`	`float`	Seconds the running units get after SIGTERM.	`GRACE_S`
`attempt`	`int`	The attempt of the pod, which names its report.	`0`
`stop`	`Event \| None`	Set on SIGTERM by a caller which handles it already, e.g. `main`.	`None`
`heartbeat_s`	`float`	Seconds between two heartbeats.	`HEARTBEAT_S`
`coordinator`	`bool`	Whether the worker continues until the ledger is complete.	`False`
`stale_s`	`float \| None`	Seconds after which the heartbeat of another worker is stale, `grace_s + 60` by default.	`None`
`poll_s`	`float`	Seconds between two releases of stale claims by a coordinator.	`COORDINATOR_POLL_S`
`memory_source`	`MemorySource`	Where the memory of the report is read, see `MemorySampler`.	`'cgroup'`

Returns:

Type	Description
`WorkerReport`	The report.

Raises:

Type	Description
`ValueError`	The task type is not splittable, the memory is too small for one process, the worker id is invalid or the ledger has another size.
`UnitError`	A unit failed; it was released.
`BrokenProcessPool`	A process of the pool died, e.g. killed for its memory.
`Terminated`	The worker stopped on SIGTERM.

run_merge ¶

run_merge(task_type, settings, paths)

Merge the units of a task once its ledger is complete.

Parameters:

Name	Type	Description	Default
`task_type`	`TaskType`	The task type, splittable.	required
`settings`	`BaseModel`	The settings of the task.	required
`paths`	`TaskPaths`	The locations of the task.	required

Raises:

Type	Description
`ValueError`	The task type is not splittable.
`RuntimeError`	Units are not done; the message names them.

paths_from_env ¶

paths_from_env(env)

The locations of a task from the environment of its pod.

The task directory is <store>/runs/<run>/tasks/<task> (SIMRUN_TASK_DIR), the inputs are port=path pairs (SIMRUN_INPUTS, a relative path is relative to the store) and the scratch directory is SIMRUN_SCRATCH (default /scratch).

Parameters:

Name	Type	Description	Default
`env`	`Mapping[str, str]`	The environment.	required

Returns:

Type	Description
`TaskPaths`	The locations.

Raises:

Type	Description
`ConfigurationError`	`SIMRUN_TASK_DIR` is not set or a pair is malformed.

shadow_tree ¶

shadow_tree(source, target)

A writable tree of a read-only one: its directories, links to its files.

A build backend may write into the tree it builds (a version file, metadata, a build directory); it writes into the writable directories of the shadow, and never into the source. A link (to a file or a directory) of the source is a link to it in the shadow.

Parameters:

Name	Type	Description	Default
`source`	`Path`	The read-only tree.	required
`target`	`Path`	The shadow to create, replaced when it exists.	required

Returns:

Type	Description
`Path`	`target`.

install_editable ¶

install_editable(snapshots, python=executable, root=None)

Install snapshots of sources as editable packages, without dependencies.

The image has the dependencies and the build backends of the sources (see simrun.environment), so the snapshots are built without isolation and without an index. A snapshot is mounted read-only: each is installed from a writable shadow tree (shadow_tree) below root, whose files are links to the snapshot.

Parameters:

Name	Type	Description	Default
`snapshots`	`list[str]`	The paths of the snapshots; nothing is installed without one.	required
`python`	`str`	The interpreter of the environment to install into.	`executable`
`root`	`Path \| None`	The directory of the shadow trees, `SOURCES_DIR` below the temporary directory by default.	`None`

Raises:

Type	Description
`CalledProcessError`	The installation failed.
`OSError`	A shadow tree cannot be created.

install_sources ¶

install_sources(env, stop)

Install the snapshots of SIMRUN_EDITABLE, then restart in a fresh interpreter.

A source can be a package this process imported already, simrun itself when a run has the simrun repository as a source: its editable install removes the package directory the imported modules came from, so a later import in this interpreter fails. The command therefore starts again (restart) with SOURCES_INSTALLED set, and the new interpreter imports the installed sources. A SIGTERM during the install (stop) skips the restart, so the caller exits.

Parameters:

Name	Type	Description	Default
`env`	`Mapping[str, str]`	The environment of the pod.	required
`stop`	`Event`	Set on SIGTERM.	required

Raises:

Type	Description
`CalledProcessError`	The installation failed.
`OSError`	A shadow tree cannot be created.

restart ¶

restart()

Replace the process with a new interpreter of the same command line.

stage ¶

stage(pairs, store, scratch)

Copy inputs to the scratch directory.

Parameters:

Name	Type	Description	Default
`pairs`	`list[tuple[str, str]]`	`(source, destination)`; a relative source is relative to the store, a relative destination to the scratch directory.	required
`store`	`Path`	The root of the run store.	required
`scratch`	`Path`	The scratch directory.	required

cores ¶

cores(env)

The cores of a worker: the quota of its cgroup, else the CPUs of the process.

SIMRUN_CORES caps them, e.g. the cores of the plan of a task on the local backend, which has no cgroup per task.

Raises:

Type	Description
`ConfigurationError`	`SIMRUN_CORES` is not a positive integer.

memory_limit ¶

memory_limit(env)

The memory limit of a worker in bytes: the one of its cgroup.

Without a limit of the cgroup, or with the memory of the process tree (SIMRUN_MEMORY_SOURCE=process, whose cgroup is not the worker's), it is SIMRUN_MEMORY_LIMIT_GB, the memory of the plan of the task, by which the harness sizes its pool; None without either.

Raises:

Type	Description
`ConfigurationError`	`SIMRUN_MEMORY_LIMIT_GB` or `SIMRUN_MEMORY_SOURCE` is invalid.

main ¶

main()

Run simrun-worker in a pod of a split task; exits with its code.

SIGTERM is handled from the start, so a pod evicted while it prepares (as PID 1 without a handler it would ignore SIGTERM) stops before its first claim. The worker of SIMRUN_ROLE=base is the coordinator, which returns once the ledger is complete (stale claims released after SIMRUN_STALE_S); extra (the default) returns when nothing is pending.

Exit codes: 0 done, 3 a unit failed (logged and released), 4 a configuration error, 137 a process of the pool died (retried with more memory), 143 stopped on SIGTERM (retried).

merge_main ¶

merge_main()

Run simrun-merge after the workers of a split task; exits with its code.

SIGTERM is handled from the start: a merge stopped before it began exits 143. A successful merge writes out/.simrun-merged (MERGED_MARKER).

Exit codes: 0 merged, 3 the ledger is incomplete or the merge failed, 4 a configuration error, 143 stopped on SIGTERM before the merge.

exec_main ¶

exec_main()

Run simrun-exec <command...>, the command of an unsplit plugin task.

The harness installs the snapshots (SIMRUN_EDITABLE), stages the inputs, sets the thread variables of BLAS and OpenMP to 1 and runs the command as its child, while its heartbeat thread samples the memory. A SIGTERM is passed on to the command, which is killed after GRACE_S. The report task_dir/workers/<task>-<attempt>.json (the worker is the task id, or SIMRUN_WORKER) is written in every case after the command started, so the controller learns the memory of the task.

Exit codes: 0 done, 3 the command failed, 4 a configuration error (also a command which cannot be started), 137 the command was killed (by the out of memory killer, retried with more memory) or exited 137, 143 stopped on SIGTERM (retried).

worker¶

MEMORY_SOURCE_ENV module-attribute ¶

ConfigurationError ¶

UnitError ¶

Terminated ¶

WorkerReport dataclass ¶

from_json classmethod ¶

MemorySampler ¶

sample ¶

harness_started ¶

cpu_seconds ¶

processes ¶

write_report ¶

memory_source ¶

process_tree_anon ¶

run_worker ¶

run_merge ¶

paths_from_env ¶

shadow_tree ¶

install_editable ¶

install_sources ¶

restart ¶

stage ¶

cores ¶

memory_limit ¶

main ¶

merge_main ¶

exec_main ¶

MEMORY_SOURCE_ENV `module-attribute` ¶

WorkerReport `dataclass` ¶

from_json `classmethod` ¶