Orchestrator#

The orchestrator drives a sweep: Orchestrator.run() enqueues one run, Orchestrator.plan() reports what a drain would do without running it, and Orchestrator.get() reads a metric back across the sweep.

Bases: object

Top-most orchestrator instance managing runs and persisting metrics.

Parameters:

root – The directory holding one folder per job.
owned – What a job’s identity is computed over, as module prefixes or as a Modules carrying its own exclusions; guessed per run from the callable’s own package by default.
lock – The lock file fingerprinting the environment; discovered from the current directory by default.
source – A directory inside the repository holding the code being run, whose commit is recorded with every job; the current directory by default.
force – Re-run every job, whatever is already recorded.
isolate – Run each job in a fresh interpreter of its own, which keeps one job’s leftover state from reaching the next. This requires the experiment and its parameters to be picklable, and a sweep script’s top level to be guarded by if __name__ == “__main__”:; see krum.orchestration.execution.
trace – Record which owned functions each job actually entered, and re-check them on a later pass. This closes the gap a static read of the code leaves open, a dependency reached only at runtime being invisible to a job’s key; see krum.orchestration.tracing. On by default: the cost is one callback per distinct function, and the alternative is keeping a stale result. Turn it off for a sweep that must share monitoring with a debugger or a profiler.

decide(pending: PendingRun, force: bool | None = None) → JobDecision[source]#

Decide whether one enqueued run needs executing, and say why.

A job runs when it has no recorded result, when its last attempt failed, when the environment it was recorded in no longer matches, or when it is forced. Otherwise its stored output stands.

Parameters:

pending – The enqueued run to decide on.
force – Override the orchestrator’s own force setting.

Returns:

The decision, carrying the reasons a job has to run.

drain(force: bool | None = None) → RunSummary[source]#

Execute every enqueued run that is not already done.

A run whose folder is marked done is skipped. A run that fails stops the drain, and the queue is abandoned: the results that did complete stay readable, which they would not be if a later get re-ran the failing job. Re-enqueueing retries it, its folder being marked failed rather than done.

Parameters:: force – Override the orchestrator’s own force setting.
Returns:: The summary of this drain.
Raises:: RunFailed – If a run raised, after recording its traceback.

property enqueued: tuple[str, ...]#: The jobs of this sweep, in the order they were enqueued.

get(metric: str) → MetricTable[source]#

Read one metric back across this sweep’s jobs.

Drains the queue first, so that a sweep need not be run explicitly.

Only the jobs enqueued here are read, in that order. A store accumulates a folder per version of an experiment, so reading all of them would mix code versions, two of which can carry the same label and be drawn over each other. Pass krum.orchestration.metrics.collect() the store directly to read everything it holds.

Parameters:

metric – The metric name.

Returns:

A tidy :class:`~krum.orchestration.metrics.MetricTable` of
`[step, value, *params, job_key]`. Call
:meth:`~krum.orchestration.metrics.MetricTable.to_pandas`, or hand
:meth:`~krum.orchestration.metrics.MetricTable.to_dict` to the
dataframe library you prefer, to group or filter it.

Raises:

KeyError – If none of this sweep’s jobs recorded that metric.
RunFailed – If draining the queue hit a failing run.

metrics() → list[str][source]#: Every metric name recorded by a completed job, sorted.

plan(force: bool | None = None) → list[JobDecision][source]#

What a drain would do, without executing anything.

Parameters:: force – Override the orchestrator’s own force setting.
Returns:: One decision per enqueued run, in queue order.

property queued: int#: How many runs are enqueued.

run(callable: RunCallable, **params: Any) → Hash[source]#

Enqueue one run, computing its identity now.

The key is computed eagerly so that an ill-fitting parameter, or a dependency that cannot be hashed, is reported at the call site rather than after a long sweep has already started.

Parameters:

callable – The user-defined function to execute.
**params – The hyper parameter values to execute it on.

Returns:

The run’s key, which names its output folder.

Raises:

TypeError – If the parameters do not fit the callable’s signature.
ValueError – If a parameter name is one a metric frame owns.

runner(owned: Iterable[str] = ()) → Runner[source]#

How this orchestrator executes a job’s body.

Parameters:: owned – Module prefixes whose functions are worth recording, when tracing. A sweep is normally one experiment, so this is taken from the first enqueued run.
Returns:: The runner, configured for isolation and tracing.

property store: JobStore#: The job store this orchestrator reads and writes.

witnesses() → dict[str, Any][source]#

The facts the current environment would stamp on a result.

Computed once per orchestrator: a sweep is one environment, and the lock file should not be re-read for every job.

Runs and decisions#

class krum.orchestration.PendingRun(callable: RunCallable, params: dict[str, Any], key: Hash | None = None)[source]#

Bases: object

Information about an enqueued run.

property bound_params: dict[str, Any]#

The parameters in signature order, defaults included.

This is what the key was computed from, and so what the manifest records: a default left unstated still identifies the job.

property callable: RunCallable#: The user-defined function this run executes.

property key: Hash#: The run’s static identity, computed on first access.

property params: dict[str, Any]#: The hyper parameter values this run was enqueued with.

class krum.orchestration.JobDecision(key: str, action: str, reasons: Iterable[str] = ())[source]#

Bases: object

Whether one enqueued run needs executing, and why.

Separating the decision from the execution is what lets a sweep be inspected before it is started, through Orchestrator.plan().

property action: str#: Either run or skip.

property key: str#: The job key.

property reasons: tuple[str, ...]#: Why the job has to run; empty when it is skipped.

property runs: bool#: Whether this job would be executed.

krum.orchestration.owned_for(callable: Any) → set[str][source]#

Guess the module prefixes a run’s identity should be computed over.

The user’s own package and krum are what a change should invalidate on; everything else is a dependency, and belongs to the environment fingerprint rather than to the key.

Parameters:: callable – The user-defined function a run executes.
Returns:: The owned module prefixes.

Outcomes#

class krum.orchestration.JobOutcome(key: str, status: str, seconds: float | None = None, path: Path | None = None, error: str | None = None, reasons: Iterable[str] = ())[source]#

Bases: object

What became of one enqueued run during a drain.

property error: str | None#: The traceback of a failed job.

property key: str#: The job key.

property path: Path | None#: The job’s folder, once it has one.

property reasons: tuple[str, ...]#: Why this job was executed rather than skipped.

property seconds: float | None#: How long the job took, or None if it was skipped.

property status: str#: One of done, skipped or failed.

class krum.orchestration.RunSummary(outcomes: Iterable[JobOutcome], pending: int = 0)[source]#

Bases: object

A report on one drain of the queue.

count(status: str) → int[source]#: How many runs ended in a given status.

property failed: tuple[JobOutcome, ...]#: The runs that failed.

property outcomes: tuple[JobOutcome, ...]#: Every run’s outcome, in execution order.

property pending: int#: How many runs were left unstarted, after a failure stopped the drain.

report() → str[source]#: Render a line per run, for a human reading the end of a sweep.

exception krum.orchestration.RunFailed(summary: RunSummary, message: str)[source]#

Bases: RuntimeError

Raised when a run fails, stopping the sweep.

Carries the summary of the drain, so the caller sees what ran, what was skipped and what was left unstarted alongside the failure itself.

property summary: RunSummary#: The summary of the drain this failure stopped.