# How StepRunner Executes Experiment Steps in Topological Order in Marin

> Discover how StepRunner executes experiment steps in topological order within Marin. Learn about DAGs, Kahn's algorithm, and efficient dependency management for seamless experiment runs.

- Repository: [The Marin Project/marin](https://github.com/marin-community/marin)
- Tags: internals
- Published: 2026-08-29

---

**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`](https://github.com/marin-community/marin/blob/main/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)](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](https://github.com/marin-community/marin/blob/main/lib/marin/src/marin/execution/step_runner.py#L181-L190) of [`step_runner.py`](https://github.com/marin-community/marin/blob/main/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](https://github.com/marin-community/marin/blob/main/lib/marin/src/marin/execution/step_runner.py#L224-L226)). 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.

```python

# 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])

```

```python

# 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`](https://github.com/marin-community/marin/blob/main/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](https://github.com/marin-community/marin/blob/main/lib/marin/src/marin/execution/step_runner.py#L181-L190) of [`step_runner.py`](https://github.com/marin-community/marin/blob/main/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](https://github.com/marin-community/marin/blob/main/lib/marin/src/marin/execution/step_runner.py#L224-L226)), ensuring that remote task submissions remain consistent across threads. If a step fails, downstream execution is cancelled to maintain workflow integrity.