Can Subflows run concurrently?

[EDigioacchinoTBL]

I've patterned a subflow for my use-case which batches out according to some CDC rules generating the tasks to be operated on. I'd like to reuse this component for ETL, and would also like to have multiple subflows operate concurrently from a common parent.

The documentation covers Task concurrency at length, but not Subflows.

[zzstoatzz]

hi @EDigioacchinoTBL - to run flows concurrently, just write async python. some people wrap flows in tasks to get .submit/.map but in my opinion there's no strong reason to do that and it just creates extra wrapping

import asyncio

from prefect import flow


@flow
async def single_flow():
    print("doing some stuff")


@flow
async def a_different_flow():
    print("doing some different stuff")


@flow
async def parent_flow():
    await asyncio.gather(*[single_flow(), single_flow(), a_different_flow()])


if __name__ == "__main__":
    asyncio.run(parent_flow())

[EDigioacchinoTBL]

If I am deploying through main, using flow.serve(), how can I wrap that with async?

[br413]

@EDigioacchinoTBL — you don't need to wrap flow.serve() itself in async. serve() is a long-running registration/polling process that lives in if __name__ == "__main__"; the async concurrency pattern belongs inside the flow function that the deployment runs.

Pattern: async orchestrator flow + serve()

Put asyncio.gather() (or run_deployment) in your parent flow body, then serve that parent:

import asyncio
from prefect import flow
from prefect.deployments import run_deployment


@flow
async def etl_batch(batch_id: str):
    # your reusable CDC/batching subflow logic
    ...


@flow
async def orchestrator():
    # Option A: call other @flow functions directly (in-process subflows)
    await asyncio.gather(
        etl_batch("batch-1"),
        etl_batch("batch-2"),
        etl_batch("batch-3"),
    )

    # Option B: trigger separate *deployments* concurrently
    await asyncio.gather(
        run_deployment("etl-batch/etl-batch", parameters={"batch_id": "batch-1"}),
        run_deployment("etl-batch/etl-batch", parameters={"batch_id": "batch-2"}),
    )


if __name__ == "__main__":
    # serve() blocks here — that's expected. Workers/pollers run your async flow body.
    orchestrator.serve(name="orchestrator")

When a scheduled or ad-hoc run starts, Prefect executes orchestrator() with the async event loop, so await asyncio.gather(...) works as usual.

Serving multiple deployments from one main

If you need several deployments (parent + children) from one entrypoint, use prefect.serve:

from prefect import serve

if __name__ == "__main__":
    serve(
        orchestrator.to_deployment(name="orchestrator"),
        etl_batch.to_deployment(name="etl-batch"),
    )

Each deployment still runs its own async/sync flow function; the orchestrator deployment is where you coordinate concurrency.

run_deployment tips for concurrent child runs

• Use timeout=0 on run_deployment if you want fire-and-forget child runs instead of waiting for completion.
• Child runs triggered from inside a flow appear as subflows in the UI by default (as_subflow=True). Set as_subflow=False if you want them shown as independent runs.

Docs:

• Trigger ad-hoc deployment runs (async run_deployment section)
• Serve flows in local processes

Hope that clarifies the serve() vs async distinction — happy to adapt this to your exact CDC batching layout if you share how your subflows are split across deployments.