Run vs Submit

What you’ll learn¶

Execute steps with run() (blocking) and submit() (non-blocking)
Wire downstream steps before upstream ones finish using StepFuture
Choose between run() and submit() for different situations

Prerequisites: Your First Pipeline. Estimated time: 10 minutes GPU required: No.

from __future__ import annotations

import time

from artisan.operations.examples import (
    DataGenerator,
    DataTransformer,
    MetricCalculator,
)
from artisan.orchestration import Backend, PipelineManager
from artisan.utils import tutorial_setup

env = tutorial_setup("run_vs_submit")
DELTA_ROOT = env.delta_root
STAGING_ROOT = env.staging_root
WORKING_ROOT = env.working_root

PipelineManager has two ways to execute a step: run() and submit(). They accept the same arguments and produce the same results — the difference is when they return.

	`run()`	`submit()`
Blocks?	Yes — waits for the step to finish	No — returns immediately
Returns	`StepResult`	`StepFuture`
Get outputs	`result.output(role)`	`future.output(role)`
Relationship	`run()` = `submit().result()`	Underlying primitive

Under the hood, run() calls submit() and waits for the result. Everything else — caching, provenance, input wiring — is identical.

Blocking execution with `run()`¶

run() blocks until the step finishes and returns a StepResult directly. This is the default pattern you’ve already used in Your First Pipeline.

pipeline = PipelineManager.create(
    name="blocking",
    delta_root=DELTA_ROOT,
    staging_root=STAGING_ROOT,
    working_root=WORKING_ROOT,
)
output = pipeline.output

t0 = time.perf_counter()
step0 = pipeline.run(
    operation=DataGenerator,
    name="generate",
    params={"count": 3, "seed": 42},
    backend=Backend.LOCAL,
)
elapsed = time.perf_counter() - t0

print(f"Type:         {type(step0).__name__}")
print(f"Step name:    {step0.step_name}")
print(f"Success:      {step0.success}")
print(f"Output roles: {step0.output_roles}")
print(f"Wall time:    {elapsed:.3f}s (blocked until done)")

To wire steps together, use pipeline.output(step_name, role) to create an OutputReference from a completed step. Above, we saved this as output = pipeline.output for brevity. Since run() blocks, the step is guaranteed to be complete by the time you reference its outputs.

pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 0.5, "variants": 1, "seed": 100},
    backend=Backend.LOCAL,
)

pipeline.run(
    operation=MetricCalculator,
    name="metrics",
    inputs={"dataset": output("transform", "dataset")},
    backend=Backend.LOCAL,
)

summary = pipeline.finalize()
print(
    f"\nPipeline: {summary['total_steps']} steps, success={summary['overall_success']}"
)

With run(), each step finishes before the next one starts. The pipeline executes strictly top-to-bottom. This is the right choice when your steps are fast or when you want to inspect intermediate results before continuing.

Non-blocking execution with `submit()`¶

submit() returns a StepFuture immediately -- the step runs in a background thread. This lets you define your entire pipeline upfront without waiting for each step to finish.

We create a fresh tutorial environment here so the submit() examples use a separate delta root from the run() section above.

env2 = tutorial_setup("run_vs_submit_async")

pipeline = PipelineManager.create(
    name="non_blocking",
    delta_root=env2.delta_root,
    staging_root=env2.staging_root,
    working_root=env2.working_root,
)
output = pipeline.output

t0 = time.perf_counter()
future0 = pipeline.submit(
    operation=DataGenerator,
    name="generate",
    params={"count": 3, "seed": 42},
    backend=Backend.LOCAL,
)
elapsed = time.perf_counter() - t0

print(f"Type:    {type(future0).__name__}")
print(f"Status:  {future0.status}")
print(f"Done:    {future0.done}")
print(f"Elapsed: {elapsed:.4f}s (returned without waiting)")

Wiring outputs before completion¶

The key advantage of submit(): you can call future.output(role) to get an OutputReference before the step finishes. This reference is a lazy placeholder — the framework resolves it to concrete artifact IDs when the downstream step actually needs the data.

ref = future0.output("datasets")
print(f"OutputReference: step={ref.source_step}, role={ref.role}")
print("(created instantly — step may still be running)")

This means you can wire your entire pipeline in one block. The framework handles execution ordering — a downstream step automatically waits for its upstream dependencies before dispatching.

pipeline.submit(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("generate", "datasets")},
    params={"scale_factor": 0.5, "variants": 1, "seed": 100},
    backend=Backend.LOCAL,
)

future2 = pipeline.submit(
    operation=MetricCalculator,
    name="metrics",
    inputs={"dataset": output("transform", "dataset")},
    backend=Backend.LOCAL,
)

# finalize() waits for all steps to complete
summary = pipeline.finalize()
print(
    f"\nPipeline: {summary['total_steps']} steps, success={summary['overall_success']}"
)

Getting results from futures¶

If you need to inspect results after the pipeline runs, call future.result() to get the underlying StepResult. After finalize(), all futures are already complete, so this returns immediately.

result = future2.result()
print(f"Step name:       {result.step_name}")
print(f"Succeeded count: {result.succeeded_count}")
print(f"Output roles:    {result.output_roles}")

When to use which¶

Situation	Use
Interactive exploration, debugging	`run()` — see each step’s output before continuing
Production pipelines, SLURM submission	`submit()` — define the whole DAG upfront
Need to inspect intermediate results mid-pipeline	`run()` for that step
Steps with no data dependencies (parallel branches)	`submit()` — they can execute concurrently

You can freely mix run() and submit() in the same pipeline.

Summary¶

run() blocks and returns a StepResult -- use for interactive work and debugging
submit() returns a StepFuture immediately -- use for production pipelines and SLURM workflows where you define the full DAG upfront
Both methods accept the same arguments and produce the same results; run() is equivalent to submit().result()
You can mix run() and submit() freely in the same pipeline

Next steps¶

Resume and Caching -- Resume pipelines and skip duplicate work
SLURM Execution -- Run steps on HPC clusters with Backend.SLURM
Pipeline Patterns -- Sources, branching, filtering, and more
Execution Flow -- How steps are dispatched, batched, and committed