How Marin StepRunner Executes Experiment Steps in Topological Order

Marin's StepRunner constructs a directed acyclic graph from step dependencies, performs a topological sort to determine execution order, and dispatches tasks to the Fray engine while respecting concurrency limits and propagating execution context.

The StepRunner class in the marin-community/marin repository serves as the core DAG scheduler for machine learning experiments. When you invoke StepRunner().run(), the system translates high-level step specifications into a deterministic, dependency-aware execution plan. Understanding how this component enforces topological ordering while managing distributed resources is essential for debugging pipeline failures and optimizing experiment throughput.

Building the Step Dependency Graph

Each StepSpec object may declare a deps attribute listing other steps that must complete before it can run. In lib/marin/src/marin/execution/step_runner.py, the runner's internal _build_graph logic converts these specifications into a directed graph where edges point from dependencies to dependents.

The graph construction validates that no circular dependencies exist, ensuring the experiment structure forms a true directed acyclic graph (DAG). This validation step prevents infinite loops and guarantees that a topological ordering exists before execution begins.

Topological Sorting and Execution Order

Once the graph is built, the StepRunner leverages Python's standard library TopologicalSorter (or an equivalent depth-first search implementation) to compute a linear execution sequence. Steps with zero pending predecessors are marked as ready and eligible for immediate dispatch.

The algorithm guarantees that a step never executes until all its declared dependencies have completed successfully. This deterministic ordering ensures that data artifacts produced by upstream steps are available before downstream consumers attempt to access them.

Concurrency Control and max_concurrent Constraints

After sorting, the runner enters its main execution loop, pulling ready steps from the topological sorter and submitting them to the Fray distributed execution engine (or a local thread pool for single-node runs). The max_concurrent parameter acts as a throttle, preventing the system from launching more than the specified number of simultaneous steps.

This backpressure mechanism protects cluster resources while ensuring the topological order is still strictly honored. Even when running under high concurrency limits, the scheduler only considers steps eligible for dispatch once all their graph ancestors have finished.

Context Propagation and Failure Handling

Before dispatching a step, the runner captures the current Fray client via fray.current_client() and active contextvars around line 220 in step_runner.py. These values are attached to worker threads, ensuring each step inherits the same logging, authentication, and tracing context as the main process.

If a step fails, the runner records the exception and immediately cancels any downstream steps that depend on the failed node. This cascade failure behavior maintains data consistency by preventing partial or stale inputs from propagating through the remainder of the DAG.

Lazy Lowering from High-Level Handles

High-level experiment definitions use "handle" objects that represent lazy computations. Before the StepRunner can schedule these, they must be converted into concrete StepSpec instances. This transformation occurs in lib/marin/src/marin/execution/lazy.py via the lower() function, which traverses the handle graph and generates the corresponding StepSpec DAG.

The output of this lowering phase is what the StepRunner ultimately consumes, bridging the gap between user-friendly experiment APIs and the low-level execution engine.

Practical Implementation Examples

Running a Simple Pipeline

from marin.execution.step_runner import StepRunner
from marin.execution.lazy import lower
from my_experiment import build  # returns a handle graph

# Lower the high-level graph to a StepSpec DAG

steps = [lower(step) for step in build()]

# Execute with default concurrency

StepRunner().run(steps)

Limiting Concurrency with Custom Fray Client

from marin.execution.step_runner import StepRunner
from fray.client import FrayClient
from marin.execution.lazy import lower
from my_experiment import build

client = FrayClient(cluster="production-cluster")
steps = [lower(s) for s in build()]

runner = StepRunner(max_concurrent=4, fray_client=client)
runner.run(steps)

Accessing Step Results

runner = StepRunner()
results = runner.run(steps)  # returns mapping {step_id: result}

print(results[my_step.id])   # access specific step output

Summary

• Dependency Graph Construction: The StepRunner builds a DAG from StepSpec dependencies in step_runner.py, validating that no cycles exist.
• Topological Ordering: Execution order is determined by a topological sort that ensures prerequisites always complete before dependent steps begin.
• Concurrency Management: The max_concurrent parameter limits simultaneous step execution while preserving topological constraints.
• Context Safety: Execution context and Fray client instances are propagated to worker threads around line 220 to maintain consistent runtime state.
• Failure Isolation: Step failures trigger cancellation of downstream dependents, preventing error propagation through the DAG.
• Graph Lowering: High-level handles are transformed into executable StepSpec objects via the lower() function in lazy.py.

Frequently Asked Questions

What is the maximum number of concurrent steps StepRunner supports?

The max_concurrent parameter accepts any positive integer, effectively limited only by available cluster resources and the Fray engine's capacity. By default, the runner typically uses a conservative value (often 8) to prevent overwhelming the underlying execution backend.

How does StepRunner handle circular dependencies in experiment steps?

During the graph construction phase in _build_graph, the StepRunner detects cycles in the dependency declarations and raises a validation error before execution begins. This prevents infinite loops and guarantees that the experiment structure can be topologically sorted.

Can I use StepRunner without the Fray distributed execution engine?

Yes. While designed to integrate with Fray for distributed workloads, the StepRunner can operate in local mode using a standard Python thread pool. When no Fray client is provided or when running in a single-node context, it falls back to local thread-based execution while maintaining the same topological ordering guarantees.

What happens when a step fails during topological execution?

When a step raises an exception, the runner captures the error and immediately cancels all downstream steps that depend on the failed node. This ensures that no step attempts to consume artifacts from a failed or incomplete predecessor, maintaining data integrity across the experiment DAG.

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