How Fan-Out/Fan-In Works in Dapr Workflow

#programming #architecture #microservices #dapr

Introduction

You have a list of items to process, and each item takes a while. Running them one after another wastes time, and running them in parallel on threads loses everything when the process dies halfway. Fan-out/fan-in in Dapr Workflow solves both. You schedule a batch of activities at once, wait for all of them with when_all, and combine the results. Each finished activity is recorded in the workflow's history, so after a crash the workflow only schedules the activities that hadn't finished. This post walks through the Python version and what a crash does to a half-finished batch. If the idea of a workflow history is new to you, read the durable execution concepts first. The code is tested with Dapr 1.19.

One of six workflow patterns

The Dapr docs describe six workflow patterns: task chaining, fan-out/fan-in, async HTTP APIs, monitor, external system interaction, and compensation. Fan-out/fan-in is the one where the size of your data matters most, because every result ends up in the workflow's history. If you build it by hand, you have to answer the questions the docs raise yourself. How do you control the degree of parallelism? How do you know when to start aggregating? What if the number of items changes on every run?

The code

From Python SDK 1.19, workflow support ships inside the core dapr package. Install it with the workflow extra instead of the separate dapr-ext-workflow package, which stops at 1.18:

pip install "dapr[workflow]"

The workflow below gets a list of items, starts one process_item activity per item, waits for all of them, and passes the sum to a final activity:

import dapr.ext.workflow as wf

wfr = wf.WorkflowRuntime()

@wfr.workflow(name="batch")
def batch(ctx: wf.DaprWorkflowContext, size: int):
    items = yield ctx.call_activity(get_items, input=size)
    tasks = [ctx.call_activity(process_item, input=i) for i in items]
    results = yield wf.when_all(tasks)
    yield ctx.call_activity(save_total, input=sum(results))

@wfr.activity(name="get_items")
def get_items(ctx, size: int):
    return list(range(1, size + 1))

@wfr.activity(name="process_item")
def process_item(ctx, item: int):
    return item * 2

@wfr.activity(name="save_total")
def save_total(ctx, total: int):
    print(f"total: {total}")

What happens when it runs

The list comprehension creates the tasks, but nothing runs yet. When the workflow yields on when_all, Dapr schedules them all at once, and they can run in parallel across however many workers you have.

While it waits, the workflow unloads itself from memory. Each result is written to the history as it arrives. When the workflow wakes up, it replays from that history, so finished activities return their recorded result instead of running again.

That's what makes a crash cheap. The docs use the example of 100 tasks where 40 finish before the process crashes. On restart, the workflow schedules only the other 60. I tried a smaller version locally: 20 items, with the app stopped after 8 had finished. After the restart only items 9 to 20 ran, none of them twice, and save_total printed 420.

Four things to watch

Return small results. Every result lands in the history, and each dispatch to your app carries the workflow's full history. That dispatch is capped by the sidecar's --max-body-size, which is 4 MiB by default. If a dispatch would go over 95% of the limit, Dapr stalls the workflow. You don't get an error; the workflow just stops until you raise the limit, purge it, or change the code. Return an ID, not the whole document.
Cap the parallelism if a downstream service can't take the load. Schedule the items in chunks and call when_all per chunk.
Split very large fan-outs into child workflows with ctx.call_child_workflow. Each child keeps its own history, so the parent's stays small. The docs recommend this once a workflow schedules thousands of tasks.
Plan for failures. when_all completes as soon as one task fails, so one bad item fails the whole batch. Pass a wf.RetryPolicy to call_activity for transient errors, or catch the error inside the activity and return a status when the other results should still count.

Summary

Fan-out/fan-in comes down to creating the tasks, yielding on when_all, and adding up the results. Dapr Workflow records each result in the history as it arrives, and after a restart it reschedules only the work that didn't finish. The part you control is how much goes into that history. Keep results small, and when a batch gets large, split it into chunks or child workflows. To see the history fill up item by item, run the sample with the Dapr Dev Dashboard. To practice the pattern hands-on in a browser sandbox, take the free Dapr Workflow course at Dapr University. If you work in .NET, the in-depth guide to the Dapr Workflow patterns shows the same pattern with Task.WhenAll.

DEV Community