@Marvin Is there correct way to run multiple concurrent subflows with asyncio.gather() and run_deployment()? We have multiple subflows that execute a series of SQL commands on different tables concurrently, but I'm noticing that some commands are taking longer than usual, seemingly as if they're happening sequentially

✅ thought for 376.3 seconds

Short answer: yes, you can orchestrate many subflow runs concurrently, but the concurrency you get depends on both how you “wait” in the parent and how many workers/pool slots are available to actually execute the child runs. If it looks sequential, it’s usually because the runs are queued behind a single worker (or a concurrency limit) or they’re blocked at the database level. A few concrete points and patterns: What run_deployment() does -

prefect.deployments.run_deployment()

creates a child flow run (a separate flow run executed by a worker). It returns quickly with a FlowRun object; it does not block until completion. - To wait for completion, call

prefect.flow_runs.wait_for_flow_run(run.id)

. If you wait for each one in a loop, you’ll wait sequentially; if you wait for all concurrently, you’ll get parallel waiting. - Docs: Run deployments from flows and API ref for waiting: wait_for_flow_run Recommended patterns 1) Fire-and-forget (don’t wait in the parent)

from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent(tables: list[str]):
    for t in tables:
        run_deployment(
            name="child/table-refresh",
            parameters={"table": t},
            as_subflow=True,  # default True; keeps parent-child linkage
        )
    # parent exits without waiting; children run concurrently on available workers

2) Wait for many child runs concurrently (sync parent)

from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from concurrent.futures import ThreadPoolExecutor

@flow
def parent(tables: list[str]):
    runs = [
        run_deployment(name="child/table-refresh", parameters={"table": t})
        for t in tables
    ]
    with ThreadPoolExecutor(max_workers=len(runs)) as ex:
        list(ex.map(lambda r: wait_for_flow_run(r.id), runs))

3) Wait for many child runs concurrently (async parent) -

wait_for_flow_run

is sync-compatible; in an async flow, run it in a worker thread:

from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
import anyio

@flow
async def parent(tables: list[str]):
    runs = [
        run_deployment(name="child/table-refresh", parameters={"table": t})
        for t in tables
    ]

    async def wait_on(r):
        # run the sync wait in a thread so we can wait on many concurrently
        return await anyio.to_thread.run_sync(wait_for_flow_run, r.id, cancellable=True)

    async with anyio.create_task_group() as tg:
        for r in runs:
            tg.start_soon(wait_on, r)

Note on asyncio.gather(): wrapping

run_deployment()

itself in

gather()

doesn’t add concurrency because

run_deployment()

just submits runs and returns. Use concurrency around the waits (as above), not around the submissions. Why things may look sequential - Only one worker (or a worker/pool with an effective concurrency of 1): Your child flow runs will be picked and executed one after another. Start more workers or ensure your infrastructure can run multiple flow runs concurrently. See workers: Workers - Pool/queue limits: Check for work pool concurrency limits or automations/tag-based concurrency that would put runs into an “AwaitingConcurrencySlot” state in the UI. - Same DB, locking or pool exhaustion: Even with fully parallel flow runs, SQL operations may serialize due to: - A small DB connection pool (e.g.,

pool_size=1

,

max_overflow=0

) causing requests to queue - Long transactions holding row/table locks - DDL or heavyweight operations that acquire broader locks - If this is the bottleneck, increase pool size, shorten transactions, or split conflicting operations.

yea valid point by marvin.

run_deployment

blocks until the remote process is done by default, so if some concurrency is gating them then that'll manifest as blocking in the caller of

run_deployment