Parallel tasks

Run independent task coroutines concurrently with parallel(), and fan out the same task over many inputs with .map().

parallel from flux.tasks runs any number of independent task coroutines at the same time and returns their results in a list, preserving the order of the inputs. Reach for it when tasks do not depend on each other’s output and you want them to overlap in time rather than run sequentially.

Basic usage

from flux import ExecutionContext
from flux.task import task
from flux.tasks import parallel
from flux.workflow import workflow


@task
async def say_hi(name: str) -> str:
    return f"Hi, {name}"


@task
async def say_hello(name: str) -> str:
    return f"Hello, {name}"


@task
async def say_hola(name: str) -> str:
    return f"Hola, {name}"


@workflow
async def greet_workflow(ctx: ExecutionContext[str]):
    results = await parallel(
        say_hi(ctx.input),
        say_hello(ctx.input),
        say_hola(ctx.input),
    )
    return results


if __name__ == "__main__":
    ctx = greet_workflow.run("Joe")
    print(ctx.output)
    # ['Hi, Joe', 'Hello, Joe', 'Hola, Joe']

The three tasks start simultaneously. parallel collects results in the same order as the coroutines were passed, so you can destructure directly:

hi, hello, hola = await parallel(
    say_hi(ctx.input),
    say_hello(ctx.input),
    say_hola(ctx.input),
)

Concurrency semantics

parallel wraps each coroutine in an asyncio.Task and runs them via asyncio.gather. All tasks start before any one of them is awaited — there is no serialization between them. The effective concurrency is limited by your event loop and any I/O or external service constraints, not by the number of tasks passed to parallel.

Durability guarantee

The fan-out is a durable event. When the workflow is replayed (after a crash or a worker restart), Flux checks the execution log. If a previous parallel call already completed, the recorded results are returned immediately and no tasks are re-executed. If the workflow was interrupted mid-fan-out, the already-completed coroutines are skipped and only the remaining ones run.

Error semantics

parallel uses asyncio.gather without return_exceptions=True. This means:

from flux.errors import ExecutionError
from flux import ExecutionContext
from flux.task import task
from flux.tasks import parallel
from flux.workflow import workflow


@task
async def fetch_a() -> str:
    return "result-a"


@task
async def fetch_b() -> str:
    raise RuntimeError("service unavailable")


@workflow
async def resilient_parallel(ctx: ExecutionContext):
    try:
        a, b = await parallel(fetch_a(), fetch_b())
        return {"a": a, "b": b}
    except ExecutionError:
        # One or more tasks failed; handle or re-raise
        return {"error": "fan-out failed"}

Fan-out over a dynamic list

When you have a variable number of inputs and want to apply the same task to each, build the coroutines in a list comprehension and unpack:

from flux import ExecutionContext
from flux.task import task
from flux.tasks import parallel
from flux.workflow import workflow


@task
async def score_document(doc_id: str) -> float:
    # Replace with real scoring logic
    return float(len(doc_id))


@workflow
async def score_all(ctx: ExecutionContext[list[str]]):
    scores = await parallel(*[score_document(doc_id) for doc_id in ctx.input])
    return dict(zip(ctx.input, scores))

For this specific pattern — the same task applied to every element of a list — Flux also provides .map() directly on the task:

@workflow
async def score_all_map(ctx: ExecutionContext[list[str]]):
    scores = await score_document.map(ctx.input)
    return dict(zip(ctx.input, scores))

.map() is equivalent to parallel(*[task(x) for x in inputs]). Both preserve input order in the output. Prefer .map() when the list is homogeneous and all items go through a single task; prefer parallel(...) when combining heterogeneous task coroutines.

Nested fan-outs

parallel accepts any awaitable coroutine returned by a task — including other parallel calls. This lets you express two-dimensional fan-outs:

from flux import ExecutionContext
from flux.task import task
from flux.tasks import parallel
from flux.workflow import workflow


@task
async def fetch_metric(service: str, metric: str) -> float:
    # Replace with a call to your monitoring system
    return float(hash(f"{service}:{metric}") % 100)


@workflow
async def monitor_services(ctx: ExecutionContext[list[str]]):
    services = ctx.input
    cpu_readings, mem_readings = await parallel(
        parallel(*[fetch_metric(svc, "cpu") for svc in services]),
        parallel(*[fetch_metric(svc, "memory") for svc in services]),
    )
    return {"cpu": cpu_readings, "memory": mem_readings}

The outer parallel runs the two inner fan-outs concurrently; within each inner parallel, the per-service metric calls also run concurrently. All results are checkpointed before the outer parallel resolves.

What’s next