docs/pipeline/execution.md

# Execution Model

"Async" means three different things in this framework. They operate at different levels and solve different problems. Keeping them separate is key to understanding the concurrency model.

---

## Three types of concurrency

| Type | Level | Problem it solves |
|------|-------|-------------------|
| **Async steps** | single step | Don't block the thread during I/O |
| **`async_boundary`** | across samples | Start the next sample before the current one finishes |

| **Branch parallelism** | within one sample | Run independent work simultaneously on the same data |



Each mechanism is independent. They compose freely — you can have async steps inside branches, behind an `async_boundary`, run with multiple workers.



---



## Entry points: `run()` and `run_async()`



=== "Sync"



    ```python

    # For regular (non-async) callers

    results = pipe.run(contexts, workers=4)

    ```



=== "Async"



    ```python

    # For async callers (e.g. inside an async framework)

    results = await pipe.run_async(contexts, workers=4)

    ```



=== "With cancellation"



    ```python

    from pipeline import CancellationToken



    token = CancellationToken()

    results = pipe.run(contexts, workers=4, cancel_token=token)

    # Call token.cancel() from another thread to stop processing

    ```



`run()` is a thin wrapper that calls `asyncio.run(self.run_async(...))`. Both accept the same parameters and return `list[SampleResult]`.



Optional parameters: `on_sample_done` (callback after each sample), `cancel_token` (stop between steps). See [API Reference](api-reference.md) for full signatures.



---



## Workers: sample-level parallelism



The `workers` parameter on `run()` / `run_async()` controls how many samples are processed through foreground steps simultaneously:



```python

import time

from pipeline import Pipeline, StepContext





class SlowStep:

    requires = frozenset()

    provides = frozenset({"result"})



    def __call__(self, ctx: StepContext) -> StepContext:

        time.sleep(0.1)  # Simulate expensive work

        return ctx.replace(

            metadata=MappingProxyType({**ctx.metadata, "result": "done"})

        )





pipe = Pipeline().then(SlowStep())

samples = [StepContext(sample=f"s{i}") for i in range(6)]



# Sequential: 6 × 0.1s ≈ 0.6s

results = pipe.run(samples, workers=1)



# Parallel: 0.1s (all 6 run at once)

results = pipe.run(samples, workers=6)

```



Under the hood, `workers` creates an `asyncio.Semaphore` — at most N samples flow through the foreground steps at any given time.



---



## Async steps — non-blocking I/O



A step that makes network calls (HTTP requests, API calls, subprocess) can be defined as a coroutine to avoid blocking the thread:



=== "Sync step"



    ```python

    class FetchStep:

        requires = frozenset()

        provides = frozenset({"response"})



        def __call__(self, ctx: StepContext) -> StepContext:

            response = requests.get(ctx.sample)  # blocks the thread

            return ctx.replace(

                metadata=MappingProxyType({**ctx.metadata, "response": response})

            )

    ```



=== "Async step"



    ```python

    class FetchStep:

        requires = frozenset()

        provides = frozenset({"response"})



        async def __call__(self, ctx: StepContext) -> StepContext:

            async with aiohttp.ClientSession() as session:

                response = await session.get(ctx.sample)  # yields the thread

            return ctx.replace(

                metadata=MappingProxyType({**ctx.metadata, "response": response})

            )

    ```



The pipeline detects async steps automatically via `asyncio.iscoroutinefunction` and awaits them. Sync steps are wrapped with `asyncio.to_thread()` so they're safe in an async context too.



!!! note

    Async steps are about **not blocking the thread**, not about parallelism. The pipeline is still sequential — it just yields the thread during I/O waits.



---



## Async boundary — fire-and-forget background



**Problem:** Some steps are slow (e.g. LLM calls for analysis). Waiting for them before starting the next sample hurts throughput.

**Solution:** A step declares `async_boundary = True`. Everything from that step onward runs in a background thread. The pipeline loop moves to the next sample immediately.

```python

class SlowScoreStep:

    requires = frozenset({"tokens"})

    provides = frozenset({"score"})

    async_boundary = True    # hand off to background from here

    max_workers = 3          # up to 3 scoring threads in parallel



    def __call__(self, ctx: StepContext) -> StepContext:

        time.sleep(0.5)  # Expensive scoring

        score = len(ctx.metadata["tokens"]) * 10

        return ctx.replace(

            metadata=MappingProxyType({**ctx.metadata, "score": score})

        )

```

```mermaid

graph LR

    A1[Tokenize] --> B1[Uppercase] -->|async_boundary| C1[SlowScore]



    style A1 fill:#6366f1,stroke:#4f46e5,color:#fff

    style B1 fill:#6366f1,stroke:#4f46e5,color:#fff

    style C1 fill:#3b82f6,stroke:#2563eb,color:#fff

```

> **Indigo** = foreground (returns immediately) · **Blue** = background (fire-and-forget)

Multiple samples flow through this simultaneously — sample 2 starts its foreground steps while sample 1's background steps are still running.

### Using the boundary

```python

pipe = Pipeline().then(Tokenize()).then(Uppercase()).then(SlowScoreStep())



# run() returns immediately after foreground steps (Tokenize + Uppercase)

results = pipe.run(samples, workers=4)



# Background scoring continues — results not yet populated

print(pipe.background_stats())

# {'active': 3, 'completed': 1}



# Block until all background work finishes

pipe.wait_for_background(timeout=30.0)



# Now all SampleResult.output fields are fully populated

for r in results:

    print(r.output.metadata["score"])

```

### Background pool model

Each step **class** has a single shared `ThreadPoolExecutor`:

- `SlowScoreStep.max_workers = 3` means one pool of 3 threads for all `SlowScoreStep` instances, regardless of how many pipelines are running
- The pool is created lazily at first use and persists for the process lifetime
- If two users need different concurrency limits for the same step type, they should subclass

!!! warning "Boundary rules"
    - **One boundary per pipeline.** If multiple steps declare `async_boundary = True`, the pipeline raises `PipelineConfigError` at construction time.
    - **No boundary inside Branch children.** A boundary inside a branch child raises `PipelineConfigError`. Branch children always block until joined — detaching mid-branch is incoherent.
    - **Nested pipeline boundary is ignored.** When a pipeline is used as a step inside another pipeline, `async_boundary` is warned and ignored — there is no "next sample" to move to from the outer pipeline's perspective.

---

## `workers` vs `max_workers` — independent pools



These two knobs control different thread pools and do not interact:



| Knob | Pool | Controls |

|------|------|----------|

| `pipe.run(contexts, workers=N)` | foreground pool | How many samples run through pre-boundary steps simultaneously |

| `step.max_workers = K` | background pool (per step class) | How many instances of that step run in the background simultaneously |

A sample leaves the foreground pool when it crosses the `async_boundary` and enters the background step's pool.

**Mental model:** `workers` controls throughput *into* the pipeline; `max_workers` controls throughput *through* each background step.

!!! warning "Rate limits"
    `workers` and `max_workers` are independent pools, but total concurrent outbound calls = foreground calls + background calls. With `workers=4` and `max_workers=3`, up to 7 requests may be in-flight simultaneously. Account for this when configuring per-provider rate limits.


---

## Rule of thumb

| Question | Answer |
|----------|--------|
| Does the step wait on I/O? | `async def __call__` |
| Do I want to process more samples while previous ones are still in background steps? | `async_boundary = True` on the handoff step |
| Can two steps on the same sample run simultaneously? | [`Branch`](branching.md) |
| Do I want N samples going through the pipeline at the same time? | `workers=N` on `run()` |