motoko 0.1.4

Versatile Workflow Manager

Motoko

Motoko is a workflow manager built on top of BlackDynamite, a Python-based tool for managing job submissions suited for HPC facilities in an automated and highly parallelized fashion.

Full documentation can be found on readthedocs

Installation

Install Motoko and dependencies:

pip install motoko

For installation in development mode:

pip install -e motoko_dir/

To run the local test workflow, do:

pytest

BlackDynamite Studies

More concretely, Motoko manages workflows of interconnected BlackDynamite studies. A BlackDynamite study is a parametrized calculation that is possibly run many times with different inputs.

BlackDynamite parameterizes a calculation using jobs and runs:

• A job is a calculation with a specific input.
• A run is an execution of a job. It carries metadata, the information needed to launch and track the calculation, and output quantities produced by the executable.

Several runs can belong to the same job, for example when the same input is repeated after a failure or evaluated with different software settings.

Key Concepts

Task Manager

A wrapper around one BlackDynamite study. A task is one pair of jobs and runs. The Motoko task manager creates such job/run pairs and can also select subsets of runs using constraints (e.g., selects only finished runs).

Workflow

The top-level object. It owns all task managers, and coordinates their creation and execution.

Orchestrator

A Python module attached to a Workflow. It registers asynchronous routines that the workflow executes following a dependency graph, or events triggering actions when a certain condition becomes true.

Workflow Directory Layout

A workflow directory contains motoko.yaml, an orchestrator Python file, and one subdirectory per study:

workflow/
  motoko.yaml
  orchestrator.py
  study1/
    bd.yaml
    launch.sh
    doIt.py
  study2/
    bd.yaml
    launch.sh
    doIt.py

Each study directory contains three files: bd.yaml, containing information about the parametric space; doIt.py, containing the code that runs the calculations; and launch.sh, which launches doIt.py with a given job submission manager (e.g. Slurm).

Once the workflow has been started, Motoko saves the workflow state in workflow/.wf/. BlackDynamite saves each study's state in workflow/study*/.bd/. Directories containing the produced output of each run are stored in BD-study*-runs/.

motoko.yaml

motoko.yaml is the workflow input file. It declares the task-manager directories and the Python function that will orchestrate them.

Minimal example:

task_managers:
  task1:
  task2:

orchestrator: orchestrator.main

Detailed example:

task_managers:
  prepare:
  solve:
    host: zeo://cluster_name:8010
  analyze:

aliases:
  post: analyze

orchestrator: orchestrator.main

generator: slurmCoat
slurm_options:
  - nodes=1

Fields:

• task_managers: Required mapping. Each key is both the task-manager name and the relative directory name containing that task's BlackDynamite study. Empty values are allowed. A task manager may define host to override the default local ZEO host.
• orchestrator: Required string in module.function form. Motoko loads module.py from the workflow directory and calls function(workflow, **params).
• aliases: Optional mapping from alternative workflow attribute names to task-manager names. For example, post: analyze allows workflow.post to access the analyze task manager.
• generator: Optional BlackDynamite launcher generator (bash, slurm, PBS).
• *_options: Optional launcher-specific option lists. Motoko derives the key from the generator name with generator.replace("Coat", "_options") and appends the listed values to the launcher command.

Python Orchestrator API

The orchestrator module usually exposes two functions:

from motoko.workflow import event


def populate_arg_parser(parser):
    parser.add_argument("--inputs", "-i", type=float, required=True, nargs=2)


async def main(workflow, **params):
    # orchestrate the workflow here

populate_arg_parser(parser) is optional but useful for workflow-specific CLI arguments. Motoko calls it when launching the orchestrator function from the command line (see below).

main(workflow, **params) registers routines and actions. It should be async. It normally ends by setting workflow.finished = True when some condition is met.

Asynchronous routines

Motoko orchestrators are asynchronous. This allows an orchestrator to express a sequence of dependent routines directly in main.

Use this pattern when the workflow follows a dependency graph that can be written as a sequence of awaited task submissions and selections:

async def run_mult(workflow, inputs):
    return await workflow.mult.createTask(x=inputs)


async def run_add(workflow, mult_runs):
    created = []
    for run, job in mult_runs:
        created.extend(await workflow.add.createTask(x=run.y))
    return created


async def main(workflow, **params):
    mult_runs = await run_mult(workflow, params["inputs"])
    add_runs = await run_add(workflow, mult_runs)
    await workflow.norm.createTask(
        mult_ids=[run.id for run, job in add_runs],
    )

Figure: asynchronous routines called directly from main.

flowchart LR
    inputs[CLI inputs] --> mult[run_mult]
    mult --> add[run_add]
    add --> norm[create norm task]

TaskManager.createTask(...) returns an awaitable RunList. Awaiting it commits the transaction, then waits until all created runs reach the FINISHED state and returns the created (run, job) pairs:

created = await workflow.add.createTask(x=1.5)
for run, job in created:
    print(run.id, job.id)

TaskManager.select(...) also returns an awaitable selection. Awaiting a selection polls until at least one run matches the supplied BlackDynamite constraints:

finished = await workflow.mult.select("state = FINISHED")

Registering Actions

Use actions for event-driven workflows. An action registered with add_action(...) is triggered each time its event condition is met. In that case, Motoko polls the condition and calls the action function when the condition fires:

workflow.add_action(event_name, task="__all__", event=..., f=...)

• event_name: Human-readable name stored in workflow logs.
• task: Task manager to watch. Use "__all__" to evaluate the event against every task manager.
• event: Condition that decides whether the action fires. It may be a BlackDynamite constraint string, a list of constraint strings, a run, job function, a workflow, task_manager function, or a no-argument function.
• f: Callback to run when the event fires. If the event returns a run selection, Motoko passes it as runs=...; otherwise the callback receives workflow=... and the runtime parameters.

Example actions:

@event
async def spawn_init_tasks(workflow, **kwargs):
    await workflow.mult.createTask(x=kwargs["inputs"])


@event
async def spawn_add_tasks(runs=None, workflow=None, **kwargs):
    for run, job in runs:
        created = await workflow.add.createTask(x=run.y)
        run.state = "FORWARDED"
        run.dependencies = [f"add.{r.id}" for r, j in created]

Constraint-based event:

workflow.add_action(
    "mult_finished",
    task="mult",
    event=["runs.id < 3", "state = FINISHED"],
    f=spawn_add_tasks,
)

Python condition event:

def ready_for_norm(workflow, task_manager):
    if len(workflow.mult.select([])) != 2:
        return False
    if workflow.mult.select(["state != FORWARDED"]):
        return False
    return True


workflow.add_action("need_norm", event=ready_for_norm, f=spawn_norm_tasks)

Workflow API Reference

• Workflow(filename): Load motoko.yaml, create task manager objects, and record workflow paths.
• workflow.create(validated=False): Reset .wf/ and initialize all task-manager BlackDynamite studies.
• workflow.start_launcher_daemons(args=None): Start BlackDynamite launcher daemons for all or selected task managers.
• workflow.add_action(...): Register an event condition and callback.
• workflow.add_error_handler(event="state = FAILED", f=...): Register a fail-fast action for failed runs.
• workflow.execute(**params): Run the orchestrator and event polling loop.
• workflow.get_runs(["add.1", "norm.3"]): Resolve dependency references into persistent run objects grouped by task manager.
• workflow.vars: Persistent workflow variable namespace backed by .wf/wf.db.
• workflow.<task_manager>: Attribute access to task managers, for example workflow.mult.

Task Manager API Reference

• TaskManager.createTask(run_params=None, **job_params): Create one or more BlackDynamite jobs/runs. job_params are expanded through the study's BlackDynamite job schema and default job_space. run_params are stored on each run. Returns an awaitable RunList of (run, job) pairs.
• await TaskManager.createTask(...): Wait until all created runs reach FINISHED, then return the created (run, job) pairs.
• TaskManager.select(constraints=None): Return a lazy TaskSelection. Constraints use BlackDynamite syntax, such as "state = FINISHED" or ["runs.id < 3", "state = FINISHED"]. When workflow.run_name is set, Motoko automatically adds a matching run_name constraint.
• await TaskManager.select(...): Poll until the selection becomes non-empty.
• TaskSelection.all(...): Build an awaitable condition that waits until all selected runs satisfy one of the supplied constraint sets.

Launching a Motoko workflow

Motoko workflows can be run using the command line or within a Python script.

Command Line Usage

Create or reset the BlackDynamite studies for a workflow:

motoko create workflow_dir

Start launcher daemons from inside the workflow directory:

cd workflow_dir
motoko launcher

Run the orchestrator in the foreground:

motoko orchestrator start --run_name test --inputs 2.1 3.2

Run the orchestrator detached with zdaemon:

motoko orchestrator start --detach --run_name test --inputs 2.1 3.2

Inspect workflow state:

motoko info
motoko info --verbose
motoko info --bd_study mult

Stop daemons:

motoko orchestrator stop
motoko kill

Clean BlackDynamite runs:

motoko clean
motoko clean --delete

Python interface

The command line interface is a thin wrapper around the Python API. A workflow can also be created and executed directly from Python:

from motoko.workflow import Workflow


workflow = Workflow("motoko.yaml")
workflow.create()
workflow.start_launcher_daemons()
workflow.run_name = "test"
workflow.execute(inputs=[2.1, 3.1])

The Workflow constructor reads motoko.yaml, creates one TaskManager per entry in task_managers, and exposes each manager as an attribute. Tasks can also be created manually outside the orchestrator function:

workflow = Workflow("motoko.yaml")
workflow.run_name = "manual"


async def submit_tasks():
    runs = await workflow.mult.createTask(x=[2.1, 3.1])
    finished = workflow.mult.select("state = FINISHED")
    return runs, finished

In a script, workflow.run_name must be set before creating tasks. Motoko stores it on every created run and uses it to scope later selections, so separate workflow executions do not accidentally interfere with one another's runs.

Motoko loads orchestrator.py from the workflow directory and calls main(workflow, **params). Parameters come from CLI arguments and are also accepted by workflow.execute(**params) when running from Python.

For scripts that only need to add or inspect tasks, use the task managers directly:

workflow = Workflow("motoko.yaml")
workflow.run_name = "inspection"

selected = workflow.norm.select(["state = FINISHED"])
for run, job in selected:
    print(run.id, run.state)

When using the Python API outside the CLI, the caller is responsible for starting and stopping BlackDynamite launcher daemons if tasks should execute automatically:

subprocess.call("motoko kill", shell=True, cwd=workflow_dir)

Troubleshooting

• FATAL: not in a motoko directory (needs motoko.yaml): Several CLI commands expect to run from inside a workflow directory.
• No runs execute: Confirm motoko launcher is running for the relevant task managers.
• Selections return old or unexpected runs: Use a distinct --run_name. Motoko scopes selections by run name when it is set.
• Workflow never finishes: Ensure one action eventually sets workflow.finished = True.
• ZEO cache warnings: Stop daemons with motoko kill, recreate the workflow with motoko create, and rerun launchers.