How StepRunner Executes Experiment Steps in Topological Order in Marin

StepRunner analyzes StepSpec dependencies to build a directed acyclic graph (DAG), applies Kahn’s algorithm to produce a topological ordering, and dispatches steps to a thread pool while ensuring every dependency completes before its downstream step begins.

Marin is an open-source framework for orchestrating complex machine learning experiments. When an experiment defines a chain of dependent operations, StepRunner executes experiment steps in topological order to guarantee that data flows correctly from one stage to the next. This orchestration logic lives in lib/marin/src/marin/execution/step_runner.py, where the runner transforms high-level step definitions into an executable task graph.

Constructing the Execution DAG

StepRunner begins by accepting a list of StepSpec objects, typically produced by lowering experiment handles via the lower() utility in [lib/marin/src/marin/execution/lazy.py](https://github.com/marin-community/marin/blob/main/lib/marin/src/marin/execution/lazy.py). Each StepSpec declares its downstream dependencies through a deps attribute that points to other steps.

Inside the private method _collect_steps, the runner recursively walks these dependencies to discover every reachable node. This traversal builds a complete directed graph where vertices represent step IDs and edges represent dependency relationships. By exhaustively collecting all transitive dependencies, StepRunner ensures that no step is overlooked when the workflow begins.

Topological Sorting with Kahn’s Algorithm

Once the DAG is fully materialized, StepRunner must produce a linear execution order that respects all dependency constraints. It delegates this work to the _toposort function (lines 181‑190 of step_runner.py).

The implementation follows Kahn’s algorithm:

1. Initialize a queue with all nodes that have zero incoming edges (no dependencies).
2. Yield each node from the queue, marking it as visited.
3. Remove the yielded node’s outgoing edges; any neighbor that now has zero incoming edges is added to the queue.
4. Repeat until the queue is empty.

This process guarantees that a step is never executed until every step it depends on has already finished. The algorithm also detects cycles: if any nodes remain unvisited after the queue empties, the graph contains a circular dependency and the runner raises an error.

Concurrent Execution and Context Propagation

After computing the topological order, StepRunner iterates over the sorted steps and submits them to an underlying executor. The max_concurrent parameter controls the size of the thread pool, allowing independent branches of the DAG to run in parallel while respecting resource limits.

Before launching a worker thread, the runner captures the current Fray client—the controller used for remote task submission—and injects it into the worker’s context (lines 224‑226). This ensures that steps executed on worker threads use the same Fray client as the main process, maintaining consistent state across distributed components.

If any step raises an exception, StepRunner immediately propagates the error and halts scheduling of downstream steps, preserving the integrity of the DAG.

Practical Usage Examples

The following examples demonstrate how to define a dependency graph and run it through StepRunner.

# Example: building a simple three-step DAG and running it

from marin.execution.step_runner import StepRunner
from marin.execution.lazy import lower  # converts handle graph to StepSpec graph

# Define three dummy steps (the real ones would be `marin.experiments.*` handles)

step_a = lower(build_a())               # no deps

step_b = lower(build_b(deps=[step_a]))  # depends on a

step_c = lower(build_c(deps=[step_b]))  # depends on b

# Run the DAG – StepRunner will execute a → b → c

StepRunner(max_concurrent=2).run([step_c])

# Example: explicitly controlling concurrency

from marin.execution.step_runner import StepRunner

# `steps` is a list of StepSpec objects discovered from an experiment

StepRunner(max_concurrent=8).run(steps)

Summary

• StepRunner acts as Marin’s DAG scheduler, residing in lib/marin/src/marin/execution/step_runner.py.
• It constructs a complete dependency graph by recursively collecting StepSpec nodes via _collect_steps.
• Kahn’s algorithm (implemented in _toposort, lines 181‑190) produces a valid topological ordering that honors every deps declaration.
• Steps run concurrently up to the max_concurrent limit, with the Fray client propagated to each worker thread (lines 224‑226).
• Execution halts on first failure, preventing downstream steps from running against invalid state.

Frequently Asked Questions

What is StepRunner in Marin?

StepRunner is the core execution engine in Marin that orchestrates experiment workflows. It accepts a collection of step definitions, orders them according to their dependencies, and manages their execution across a thread pool while respecting a user-defined concurrency limit.

How does StepRunner handle step dependencies?

StepRunner interprets the deps attribute on each StepSpec to build a directed acyclic graph. Using the _collect_steps method, it recursively discovers all transitive dependencies, then uses Kahn’s algorithm to ensure a step only runs after every dependency has completed.

What algorithm does StepRunner use for topological sorting?

StepRunner uses Kahn’s algorithm, implemented in the _toposort function at lines 181‑190 of step_runner.py. This algorithm repeatedly selects nodes with zero incoming edges, guaranteeing a valid linear order for any acyclic graph.

How does StepRunner manage concurrency and thread safety?

StepRunner respects the max_concurrent parameter to limit parallel workers. It captures the active Fray client from the main thread and injects it into each worker’s context (lines 224‑226), ensuring that remote task submissions remain consistent across threads. If a step fails, downstream execution is cancelled to maintain workflow integrity.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}