Short answer: yes — there are a few issues in that pattern that can leave runs “stuck” in RUNNING. The big ones: - Mixing asyncio primitives with Prefect futures - Awaiting flows/tasks incorrectly - Using

.submit()

on the result of a task call (Prefect 1.x style) - A sync flow using

await

Key problems in your snippet -

@flow def my_flow():

uses

await

— if you’re going to

await

, the flow must be

@flow async def ...

. -

summarize(loaded).submit()

— in Prefect 3.x you submit the task itself, not the return value. Use

summarize.submit(...)

. -

as_completed

— you imported

asyncio.as_completed

but you’re not creating asyncio Tasks (

asyncio.create_task(...)

), you’re collecting bare coroutines; that can hang. If you’re submitting Prefect work, prefer

from prefect.futures import as_completed

. -

asyncio.Semaphore

— this only limits concurrency inside a single event loop/process. It won’t coordinate across Prefect workers and is a common source of surprising “RUNNING” states. Prefer either: - Limit concurrency via the parent flow’s task runner (max_workers), or - Use Global Concurrency Limits for cross-worker limits. A solid Prefect-native rewrite (no asyncio semaphore) - Runs subflows concurrently - Caps concurrent subflows to 10 via the parent flow’s task runner - Streams results as they complete using Prefect’s

as_completed

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import as_completed

@task
def extract(x): ...
@task
def transform(x): ...
@task
def load(x): ...
@task
def summarize(items): ...
@task
def get_all_chunks() -> list: ...
@task
def record_all_summaries(summaries): ...

@flow
def subflow(chunk: list):
    # Each .map returns Prefect futures; chaining .map with futures is supported
    extracted = extract.map(chunk)
    transformed = transform.map(extracted)
    loaded = load.map(transformed)

    # If summarize needs the loaded values as a single list:
    loaded_values = [f.result() for f in loaded]
    return summarize.submit(loaded_values).result()

@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def my_flow():
    all_chunks = get_all_chunks()  # returns a Python list
    # Launch subflows concurrently; max 10 active from task_runner
    subflow_futures = [
        subflow.submit(all_chunks[i:i+10])
        for i in range(0, len(all_chunks), 10)
    ]

    summaries = []
    for fut in as_completed(subflow_futures):
        summaries.append(fut.result())

    record_all_summaries.submit(summaries).wait()

If you really want asyncio-style orchestration - Make your flows async - Create asyncio tasks explicitly - Use

asyncio.as_completed

on those tasks (not Prefect futures) - Avoid

asyncio.Semaphore

for cross-worker limits (use Global Concurrency Limits instead)

import asyncio
from prefect import flow

@flow
async def subflow(chunk):  # make it async if you want to await it
    # you can still use task.map(...) inside; those run on the subflow's task runner
    ...

@flow
async def my_flow():
    all_chunks = await get_all_chunks()  # must be awaitable (async task/func)
    sem = asyncio.Semaphore(10)

    async def run_chunk(c):
        async with sem:
            return await subflow(c)

    tasks = [
        asyncio.create_task(run_chunk(all_chunks[i:i+10]))
        for i in range(0, len(all_chunks), 10)
    ]

    summaries = []
    for t in asyncio.as_completed(tasks):
        summaries.append(await t)

    await record_all_summaries(summaries)

When subflows get “stuck” in RUNNING, common causes - Returning unresolved Prefect futures from a flow and never resolving them in the caller. - Using

asyncio.as_completed

on bare coroutines (not scheduled tasks). - Blocking the event loop inside async flows (e.g.,

time.sleep

instead of

await asyncio.sleep

). - Local semaphores that never release across processes/workers.

- Not actually awaiting/collecting subflow results, keeping the parent flow open. References - Run work concurrently: docs - Futures API (wait, as_completed, result): docs - Subflows (flows within flows): docs - Global Concurrency Limits: docs If you can share: - Whether `my_flow`/`subflow` are actually async in your code - Whether

get_all_chunks

and

record_all_summaries

are Prefect tasks or plain async funcs - Whether you’re running on a single worker or distributed …I can tailor the fix exactly to your setup.

Great context — that helps. Direct answers to your questions - If you do

await subflow(...)

inside an async flow, you get the subflow’s return value (a plain Python value), not a PrefectFuture. It blocks the parent at that await point. - To get a PrefectFuture for a subflow (so you can fan out and wait later), call

subflow.submit(...)

. Those should be driven with Prefect’s futures utilities (e.g.,

prefect.futures.as_completed

,

wait

), not

asyncio.as_completed

. - Don’t mix

asyncio.as_completed

with PrefectFutures, and don’t mix

prefect.futures.as_completed

with asyncio Tasks. Pick one model (all-async or all-Prefect-futures) and stick to it. Since you’re on a single worker - You don’t need

asyncio.Semaphore

for cross-worker coordination. If you want “max 10 subflows at once,” use the parent flow’s task runner max_workers, or keep the semaphore if you choose the pure-async pattern — but avoid mixing with Prefect futures. Two clean patterns you can choose from 1) Prefect-futures pattern (recommended when using .submit/.map) - Subflows run concurrently as tasks - Limit to 10 with the task runner - Stream results with Prefect’s

as_completed

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import as_completed

@flow
def subflow(chunk):
    extracted = extract.map(chunk)
    transformed = transform.map(extracted)
    loaded = load.map(transformed)

    # If summarize reduces the entire chunk:
    values = [f.result() for f in loaded]
    return summarize.submit(values).result()

@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def my_flow():
    chunks = get_all_chunks()  # returns a regular Python list if this is a sync task/func

    subflow_futures = [subflow.submit(chunks[i:i+10])
                       for i in range(0, len(chunks), 10)]

    summaries = []
    for f in as_completed(subflow_futures):
        summaries.append(f.result())

    record_all_summaries.submit(summaries).wait()

2) Pure-async pattern (no PrefectFutures; everything is awaited with asyncio) - Make parent and subflow async - Use

asyncio.create_task

+

asyncio.as_completed

- If you keep a semaphore, it’ll work locally on your single worker

import asyncio
from prefect import flow

@flow
async def subflow(chunk):
    extracted = extract.map(chunk)
    transformed = transform.map(extracted)
    loaded = load.map(transformed)
    values = [await l.result() if hasattr(l, "result") else l for l in loaded]  # or [f.result() for f in loaded] if they’re Prefect futures in a sync context
    return await summarize(values)

@flow
async def my_flow():
    chunks = await get_all_chunks()

    sem = asyncio.Semaphore(10)

    async def run_chunk(c):
        async with sem:
            return await subflow(c)

    tasks = [asyncio.create_task(run_chunk(chunks[i:i+10]))
             for i in range(0, len(chunks), 10)]

    summaries = []
    for t in asyncio.as_completed(tasks):
        summaries.append(await t)

    await record_all_summaries(summaries)

Important fix in your original subflow -

summarize(loaded).submit()

is a Prefect 1.x pattern. In 3.x, submit the task itself: - Per-element:

summarize.map(loaded)

if

summarize

operates on single items - Reduce a list:

summarize.submit([f.result() for f in loaded])

Rules of thumb - Need a PrefectFuture? Use

.submit()

, then use `prefect.futures.as_completed`/`wait`. - Staying async? Call the flow/task directly and use

asyncio.create_task

+

asyncio.as_completed

. - Don’t cross the streams:

asyncio.as_completed

expects asyncio Tasks, not PrefectFutures; Prefect’s

as_completed

expects PrefectFutures. Docs for deeper dives - Run work concurrently: docs - Futures API (wait/as_completed/result): docs - Flows and subflows: docs

- Global Concurrency Limits (if you later distribute): docs If you share whether you want to stick with the async style or switch to Prefect futures, I can adapt your exact code (including your async `get_all_chunks`/`record_all_summaries`) into the correct pattern.

Great question — you’re right to reconcile these two ideas. Short answer - Each flow run (including subflows) gets its own task runner. The parent’s task runner does not cap the child’s internal parallelism. - What “wins” at any given time depends on how you invoke the subflow: - Serial subflows: only one flow is active at a time; the active flow’s task runner controls concurrency. - Parallel subflows: you multiply concurrency (parent’s parallelism × each child’s parallelism). How it actually shakes out 1) Serial subflow calls (most common) - When the parent calls a subflow like a normal function, the call is blocking. The parent pauses, the child runs to completion, then the parent resumes. - While the child is running, only the child’s task runner is active — so its

max_workers

is the cap for that period. - You do not need a semaphore to keep total local concurrency bounded in this pattern; just set

max_workers

on whichever flow actually does the parallel work. Example

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
def child_flow(items):
    # Up to 4 child tasks at a time here
    ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))
def parent_flow(batches):
    for batch in batches:
        # This call blocks; only the child's runner is active while it runs
        child_flow(batch)

In this setup, even though the parent has 16 workers, the total concurrency while the child runs is capped at the child’s 4. 2) Parallel subflow calls (where surprises happen) - If you kick off multiple subflows concurrently (e.g., by running them from multiple parent tasks or using your own threading/async fan-out), each subflow gets its own runner. - Then total concurrency ≈ number_of_concurrent_subflows × child_max_workers. - The parent’s

max_workers

limits how many parent tasks can start subflows at once, but does not cap the per-subflow workers. Example pitfall

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
def child_flow(i):
    # Up to 4 tasks in this child
    ...

@task
def launch(i):
    child_flow(i)  # This starts a new flow run with its own 4-worker pool

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3))
def parent_flow():
    # Up to 3 parent tasks concurrently, each running a child with 4 workers
    for i in range(10):
        launch.submit(i)
    # Potential total concurrency ~ 3 * 4 = 12

Controlling concurrency across the whole tree - If you need a single cap across parent and all subflows (especially if subflows may run in parallel), use Prefect’s concurrency limits instead of a local semaphore: - Add a concurrency context in the sections that actually consume the constrained resource (tasks in child and/or parent). - This works across flow boundaries and across runs. Example with Prefect concurrency limits

from prefect import flow, task
from prefect.concurrency.sync import concurrency
from prefect.task_runners import ThreadPoolTaskRunner

@task
def heavy(i):
    with concurrency("local_slots", occupy=1):
        # do work that should be globally limited
        ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
def child_flow(n):
    for i in range(n):
        heavy.submit(i)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3))
def parent_flow():
    # Even if parent and child run in parallel, the "local_slots" limit gates total concurrency
    child_flow(10)

Note: Define/manage the "local_slots" limit in Prefect Cloud/Server so the runtime can coordinate slot acquisition. Key takeaways - Each flow run gets its own task runner; the parent’s runner does not cap the child’s. - If you call subflows serially, only one runner is active at a time, so you can rely on the child’s

max_workers

without a semaphore.

- If you fan out subflows in parallel, you multiply concurrency; use Prefect concurrency limits to enforce a single cap across the whole tree.