Pipelines: internals¶

How the milk package is built: its layout and layering, the objects behind the commands, how the state is computed, the task contract, the errors, and the normative terms.

Audience: Internals · See also: Pipelines · Plans and tasks · Configuration · Glossary

1. Package layout and layering¶

milk/
  config/      what a pipeline is: the model, {var} substitution, confresolve. Pure data.
  provision/   confdirs and rootdirs: clone, consolidate, init, drift, resync. Files only.
  operate/     the live pipeline: Pipeline, ComputeUnit, tasks, plans, state, rundir.
  show/        milk-show: displays (matplotlib) of pipeline results.
  errors.py    every exception of the package
  env.py       the only reader of the MILK_* environment variables

The CLIs only parse arguments and call the API; provision and operate both build on config and never import each other.

flowchart TD
    PI["milk-pipeinit"] --> PR["milk.provision<br/>files only"]
    PP["milk-pipes"] --> OP["milk.operate<br/>live system, reads only the rootdir"]
    MS["milk-show"] --> SH["milk.show"]
    PR --> CF["milk.config<br/>pure data"]
    OP --> CF
    PR -. "never imports" .- OP

Provisioning is file operations and never touches SHM, FPS, tmux or processes; its only contact with the live system is the "is a run process active?" callable that resync receives (Drift and resync). Operating reads only the rootdir.

2. Configuration: PipelineConfig¶

confresolve is the only way to obtain a configuration (Configuration: resolving). It merges the layers, expands the substitutions and validates, every time it is called, with no side effects. The result is a frozen PipelineConfig (a pydantic model): long_name, name, loop_number, kind, compute_units (a dict of ComputeUnitSpec: exec, fps), dataloads, hooks, systemlinkage, plans, teardown_kill_tmux. fps_name(cu) is the one place where FPS names are made.

3. Operating: Pipeline and ComputeUnit¶

from milk.operate.pipeline import Pipeline

pp = Pipeline("basic")                 # <cwd>/basic-rootdir
pp = Pipeline("/elsewhere/basic-rootdir")
pp.config                              # the resolved PipelineConfig
pp.root_folder, pp.work_folder         # the rootdir, and rootdir/work (the FPS work folder)
pp.compute_units                       # {'delay0': ComputeUnit, 'delay1': ComputeUnit}
pp.compute_units["delay0"].fps         # the pyMilk FPS, or None while it does not exist
pp.any_run_active()
pp.link("loopback"); pp.linked(); pp.unlink("all")

A Pipeline is a view on one rootdir: it resolves the configuration once, at construction, and derives everything else from the live system when asked. compute_units builds fresh ComputeUnit views on each access; a ComputeUnit wraps one fpsexec executable and its FPS (fpsinit, confstart, confstop, runstart, runstop run the executable's commands, in tmux when asked). You seldom call these yourself: tasks do.

4. State¶

from milk.operate.state import state

st = state(pp)            # PipelineState(state=<State.CONFIGURED>, reasons=())
st.state.value            # 'cold' | 'inconsistent' | 'deployed' | 'configured' | 'running' | 'mixed'

The state is never stored. It is computed from the rundir claim ($MILK_SHM_DIR/pipe_rundirs/ <name>/: boot id, rootdir, loop number) and from the existence and the conf/run status of each CompU's FPS. cold means nothing is deployed; inconsistent means the rundir and the FPSs disagree (a stale rundir from a previous boot, an FPS destroyed by hand, a failed deploy), and only teardown is accepted from there. Operations that need a given state raise a StateError (ColdSystemError, InconsistentStateError) that names the command to run. The diagrams are in Operating: lifecycle.

5. Errors and environment¶

Every exception of the package lives in milk.errors, under MilkError: EnvError, ConfigError, ProvisionError, TaskError (TaskCannotRunError, TaskDidNotSucceedError), ComputeUnitError, LinkageError, StateError (ColdSystemError, InconsistentStateError) and BestEffortError (the aggregate of a best-effort plan, with .failures). An error carries context when it has some (pipeline, compu, task, path, cmd, stderr), appended to its message one per line. Library code never prints or exits; pyMilk exceptions propagate as they are. How the CLIs print errors is in Operating: errors.

milk.env reads MILK_ROOT, MILK_INSTALLDIR and MILK_SHM_DIR when a function is called, not at import, and raises EnvError naming what is missing. import milk and confresolve work without a milk environment; anything that touches SHM or runs an executable needs one.

6. Task contract¶

A task subclasses milk.operate.tasks.Task and implements three methods. pipeline.task_do calls them in this order:

flowchart TD
    S1{"success()"} -- yes --> D1["done: no-op"]
    S1 -- no --> C{"can()"}
    C -- no --> E1["TaskCannotRunError(failure_reason)"]
    C -- yes --> F["forward()"]
    F --> S2{"success()"}
    S2 -- no --> E2["TaskDidNotSucceedError(failure_reason)"]
    S2 -- yes --> D2["done"]

The no-op on an already reached goal is what makes plans idempotent.

• success() checks whether the goal already holds, and must not have side effects.
• can() checks the task's preconditions. Before returning False, it sets self.failure_reason.
• forward() does the work. For asynchronous effects (tmux, processes), it waits with wait_until(self.success, self.TIMEOUT) (from milk.wait); TIMEOUT defaults to 10 s.
• self.compute_units is the step's CompU subset, as {name: ComputeUnit}. self._existing() is the part of that subset whose FPS exists, as (ComputeUnit, FPS) pairs.
• Tasks act on the live pipeline (SHM, FPS, tmux, processes) and never modify rootdir/conf/.

Direct subclasses of Task in milk/operate/tasks.py are registered under their class name automatically. A task defined anywhere else needs TASKS.register("MyTask", MyTask); it can then be named in a Step (Plans and tasks).

7. Normative glossary¶

Published copy of the normative terms of the pipelines design (design-v7 §0). Specs and docs use these words only in these senses. Short definitions for users are in the Glossary.