https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> From a main flow, I want to run multi...
# ask-marvin
b
@Marvin From a main flow, I want to run multiple subflows concurrently, with a concurrency limit. What is the best approach? The subflows should be run in the same container/pod as the main flow.
m
✅ thought for 143.2 seconds
If you want multiple subflows concurrently and they must stay in the same container/pod as the main flow, the best approach in Prefect 3.x is: Call subflows directly as nested flows, and make them 
async
so you can run them concurrently with 
asyncio.gather()
or an 
anyio
task group. That keeps everything in the same Python process/container while still creating child flow runs in Prefect. Recommended pattern
Copy code
python
import asyncio
from prefect import flow
from prefect.concurrency.asyncio import concurrency

@flow
async def subflow(item: int):
    # Optional: limit access to some shared resource
    async with concurrency("subflow-slots", occupy=1):
        await asyncio.sleep(2)
        return item * 10

@flow
async def main_flow():
    results = await asyncio.gather(
        *(subflow(i) for i in range(10))
    )
    return results

if __name__ == "__main__":
    asyncio.run(main_flow())
A few important notes: * 
subflow(...)
called directly from a flow is a nested flow. * Nested flows run in the same container/pod as the parent flow. * Using 
asyncio.gather()
lets those nested async subflows run concurrently. * The 
concurrency(...)
context manager can enforce a shared limit. If you want to cap concurrency at, say, 3, create a global concurrency limit first:
Copy code
bash
prefect gcl create subflow-slots --limit 3
I verified the CLI exists in Prefect 3.x via 
prefect gcl --help
/ 
prefect gcl create --help
. So with 10 subflows started concurrently, only 3 will be inside the guarded section at a time. Why this is the right approach Direct nested subflow calls - stay in the same pod/container - create child flow runs visible in the UI - are simplest if you want parent/child structure Deployment-based subflow execution like 
run_deployment(...)
- does not guarantee same container/pod - usually creates a separate flow run to be picked up by a worker - is better when you want isolation or independent scheduling, but not for your requirement What not to do * Don’t use deployment-based execution if “same pod” is required. * Don’t expect 
.submit()
on flows — that is for tasks, not flows. * Don’t rely on task-runner concurrency alone for this specific pattern; task runners control task concurrency, not nested flow scheduling. Best practice summary If your goal is: - same pod/container - concurrent subflows - bounded concurrency then use: 1. 
@flow async def ...
for parent and child flows 2. 
asyncio.gather(...)
to launch subflows concurrently 3. 
prefect.concurrency.asyncio.concurrency(...)
to enforce the limit If you want, I can also give you: * a full example with a Semaphore-only approach * a version using 
anyio.create_task_group()
* or guidance on whether this should actually be modeled as tasks instead of subflows for better performance.
b
@Marvin the subflows are not async, they contains lots of blocking Pandas operations.
m
✅ thought for 17.7 seconds
In that case, I would not use subflows for concurrency. If the work is mostly blocking Pandas code, the best pattern is usually: Use a single flow, turn the blocking units into tasks, and run those tasks concurrently with a 
ThreadPoolTaskRunner
, plus a task-level concurrency limit if needed. Why: * 
asyncio
won’t help much with blocking Pandas work. * Nested sync subflows called directly will generally block the parent unless you add your own thread/process orchestration. * Prefect’s built-in concurrency model is much stronger for tasks than for flows. Recommended approach
Copy code
python
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import wait

@task
def process_partition(partition_id: int):
    # blocking pandas work here
    # df = ...
    # transformed = ...
    return partition_id * 10

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3))
def main_flow():
    futures = [process_partition.submit(i) for i in range(10)]
    wait(futures)
    return [f.result() for f in futures]

if __name__ == "__main__":
    main_flow()
That gives you: * concurrency in the same container/pod * a hard cap via 
max_workers=3
* proper Prefect orchestration primitives If you really need “subflow-like” grouping If those units are currently subflows because you want structure/logging/retries, you have two options: 1. Refactor the heavy Pandas unit into tasks and keep the parent as a flow 2. Keep a thin subflow wrapper, but do the concurrent work as tasks inside it Example:
Copy code
python
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import wait

@task
def pandas_work(item):
    return item * 10

@flow
def subflow(item):
    # sequential wrapper around task(s)
    return pandas_work(item)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3))
def main_flow():
    futures = [pandas_work.submit(i) for i in range(10)]
    wait(futures)
    return [f.result() for f in futures]
But if your real need is concurrent execution, submit tasks, not subflows. When threads are enough vs processes For Pandas: * If operations release the GIL or are I/O-heavy, threads can work well. * If the workload is truly CPU-bound Python-heavy logic, threads may not scale much. In practice, many Pandas/numpy operations do okay with threads, so I’d start with 
ThreadPoolTaskRunner
. If you absolutely must keep them as subflows Then Prefect doesn’t have a built-in “submit subflows concurrently in the same pod” primitive like tasks have. You’d be in “manual Python concurrency” territory, e.g. 
concurrent.futures.ThreadPoolExecutor
calling sync subflows directly. That can work, but it’s not the cleanest/recommended Prefect pattern. Conceptually:
Copy code
python
from concurrent.futures import ThreadPoolExecutor
from prefect import flow

@flow
def subflow(x):
    # blocking pandas work
    return x * 10

@flow
def main_flow():
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(subflow, i) for i in range(10)]
        return [f.result() for f in futures]
I’d treat this as a fallback, not the first choice. My recommendation For blocking Pandas work in the same pod: Prefer tasks + 
ThreadPoolTaskRunner(max_workers=N)
over concurrent subflows. Docs and APIs: * Prefect task runners: https://docs.prefect.io/v3/develop/task-runners * Prefect concurrency concepts: https://docs.prefect.io/v3/develop/global-concurrency-limits * Flows: https://docs.prefect.io/v3/develop/write-flows If you want, I can help convert your current “main flow + subflows” design into the closest equivalent task-based pattern while preserving retries, logging, and result handling.
b
@Marvin The subflow can be run standalone, and performs processing on a single file. There is also a flow which iterates through a list of files and runs a subflow for each file. I want to run the subflow concurrently from the main flow. I do not want to change the flow to a task
m
✅ thought for 249.8 seconds
Got it — if you must keep it as a flow and want to call it concurrently from a parent flow in the same pod/container, the answer is: Prefect 3.x does not really support concurrent execution of sync nested flows as a first-class pattern. Sync nested flows are designed to block the parent flow until they finish. The officially supported concurrent nested-flow pattern is async nested flows. So for your exact constraints: - subflow remains a 
@flow
- subflow is sync / blocking Pandas - parent flow wants to launch many of them concurrently - same pod/container - do not convert the subflow to a task there is no clean built-in Prefect primitive that does this the way 
task.submit()
does for tasks. What is supported - 
@flow
called from another flow => nested flow, same process/container, but blocking - async nested flows with 
asyncio.gather(...)
=> concurrent - tasks with 
ThreadPoolTaskRunner
=> concurrent in same pod What is not a recommended/supported pattern Using 
ThreadPoolExecutor
to call sync subflows concurrently, like this:
Copy code
python
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(process_file_flow, f) for f in files]
This is problematic because Prefect uses context variables for flow-run context, and new threads do not automatically inherit that nested-flow context correctly. So this is not a reliable / recommended Prefect pattern. Best practical pattern if you want to preserve a standalone flow entrypoint The usual compromise is: 1. Put the actual file-processing logic in a plain Python function 2. Wrap that function in a standalone 
@flow
for independent execution 3. From the parent flow, run the plain function concurrently as tasks or threads That preserves a standalone flow for one-file runs, while using a concurrency-friendly execution path in the parent. Conceptually:
Copy code
python
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import wait

def process_file_logic(path: str):
    # blocking pandas work
    return path

@flow
def process_file_flow(path: str):
    return process_file_logic(path)

@task
def process_file_task(path: str):
    return process_file_logic(path)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
def process_many_files_flow(files: list[str]):
    futures = [process_file_task.submit(f) for f in files]
    wait(futures)
    return [f.result() for f in futures]
This gives you: - a standalone flow for single-file processing - concurrent same-pod execution from the parent - proper Prefect-supported concurrency I realize you said you do not want to change the flow to a task — this pattern does not remove the standalone flow. It just extracts the shared logic so the parent can use Prefect’s supported concurrency model. If you insist on flow-to-flow concurrency without converting anything You would be in unsupported/manual territory. There are hacky approaches involving manual context hydration across threads, but I would not recommend them for production unless you’re willing to own the complexity and edge cases. So the honest answer is You have three options: 1. Best supported: refactor shared logic into a plain function, keep the single-file flow wrapper, and use tasks for concurrent parent execution 2. If possible: make the subflow async and use 
asyncio.gather(...)
3. Unsupported fallback: manually thread sync subflow calls, with Prefect context workarounds For most users, option 1 is the right answer. Relevant docs: - Flows / nested flows: https://docs.prefect.io/v3/develop/write-flows - Task runners: https://docs.prefect.io/v3/develop/task-runners - Global concurrency limits: https://docs.prefect.io/v3/develop/global-concurrency-limits If you want, I can show you the smallest refactor that preserves your current 
process_file_flow(...)
API while adding a parent flow that runs N files concurrently in the same pod.
b
@Marvin Just to be clear, I mean running the subflow separately in parallel, say 10x files at a time. Currently in using anyio semaphore, with await gather() which calls run_deployment. However, run_deployment starts a new pod/container for every file. I want to keep all the processing in the same pod/container.
m
✅ thought for 23.4 seconds
Yes — that clarification helps. If you want to process 10 files at a time in parallel inside the same pod/container, then do not use 
run_deployment()
. 
run_deployment()
creates a separate flow run that is meant to be picked up by deployment infrastructure / workers, so it will typically launch separate execution environments rather than stay inside the current pod. So your requirement is: - one parent flow running in one pod - process up to 10 files concurrently - each file currently corresponds to a “subflow” - keep everything in the same pod - avoid spawning a new pod per file For that, the honest answer is: Prefect does not have a built-in supported way to execute sync subflows in parallel in the same pod the way tasks can be executed in parallel. So if you stay with flow-per-file and want parallelism in one pod, you’re fighting the model a bit. What 
run_deployment()
is doing - creates independent flow runs - those runs are scheduled/executed by workers - they are isolated from the parent process - therefore you get a new pod/container per file in Kubernetes-style setups That behavior is expected. What would keep execution in the same pod Only work that executes inside the parent process: - direct nested flow calls - tasks - plain Python concurrency But: - direct sync nested flow calls are blocking - concurrent sync nested flows are not a first-class/supported Prefect pattern - tasks are the supported same-process concurrency primitive So, given your constraints, you have two realistic paths: Option 1: Best supported Keep the “single-file” unit as a standalone flow for standalone use, but have the parent flow call shared processing logic concurrently in-process. Example structure:
Copy code
python
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import wait

def process_file_logic(path: str):
    # heavy blocking pandas logic
    return path

@flow
def process_file_flow(path: str):
    return process_file_logic(path)

@task
def process_file_task(path: str):
    return process_file_logic(path)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def process_many_files_flow(files: list[str]):
    futures = [process_file_task.submit(path) for path in files]
    wait(futures)
    return [f.result() for f in futures]
This preserves: - 
process_file_flow(path)
as a standalone flow - parallel processing of 10 files at a time - all execution in the same pod/container This is the cleanest answer. Option 2: Manual Python concurrency around plain functions If you absolutely refuse to use tasks for the parent-side parallelism, you can use 
ThreadPoolExecutor
on a plain Python function.
Copy code
python
from concurrent.futures import ThreadPoolExecutor
from prefect import flow

def process_file_logic(path: str):
    # blocking pandas logic
    return path

@flow
def process_file_flow(path: str):
    return process_file_logic(path)

@flow
def process_many_files_flow(files: list[str]):
    with ThreadPoolExecutor(max_workers=10) as executor:
        results = list(executor.map(process_file_logic, files))
    return results
This also keeps everything in the same pod. But note: - now the parallel units are not subflow runs - you lose per-file child-flow observability in Prefect - this is Python concurrency, not Prefect orchestration What I would not recommend Trying to concurrently invoke the sync 
@flow
itself in threads, e.g. 
executor.submit(process_file_flow, path)
That is the problematic part. Prefect flow context is not designed for that as a clean supported nested-flow concurrency mechanism. Bottom line If you need 10 files at a time in one pod, you have to keep the parallel work inside the current process. That means one of: 1. Preferred: parent flow submits tasks that call shared file-processing logic 2. Fallback: parent flow uses Python threads/processes on shared file-processing logic
3. Not suitable: 
run_deployment()
because it creates separate runs/infrastructure 4. Not really supported: concurrent sync nested subflows in one pod So the shortest possible answer is: You cannot get “parallel subflow runs in the same pod” via 
run_deployment()
. If you want same-pod parallelism, move the concurrent unit of work to tasks or plain Python functions inside the parent process. Docs: - Flows: https://docs.prefect.io/v3/develop/write-flows - Task runners: https://docs.prefect.io/v3/develop/task-runners - Deployments / flow runs: https://docs.prefect.io/v3/deploy/index If you want, I can sketch the exact refactor pattern so that: - your existing single-file flow remains unchanged for standalone runs - your multi-file flow gets a minimal change - concurrency is capped at 10 - everything stays in the same pod.
n
hmm marvin is kinda yappy since the model upgrade
😆 1
Copy code
@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def do_things()
  return process.map(inputs).result()
is what i'd recommend. but if you don't feel like making them tasks (which are the things that intentionally have the concurrency primitives) then a semaphore is what i'd use. no run_deployment, just giving flows the stdlib asyncio treatment
b
Cheers @Nate, the subflow is quite a large complex flow which contains many tasks and is deployed as a flow which can be run standalone to process a single file. The parent flow in this case needs to call the subflow for each file. In the past I used a wrapper task which calls the subflow, that way I could just 
.submit()
the task and use the ThreadPoolTaskRunner to limit concurrency. This worked, but is adds an extra layer of DAG bars to drop down in the UI to get to the tasks for a given subflow - not an issue by any means, was just wondering if there was a neater way to do it. Currently I have something like this, but since changing infrastructure a bit I don't want to spawn a new pod for every file, so I need to call the subflow directly. Claude suggested changing 
run_deployment
to 
<http://anyio.to|anyio.to>_thread.run_sync
, which appears to had the desired effect.
Copy code
@flow
def subflow(file_id):
    df = download_file(file_id)
    df = processing_task_1(df)
    df = processing_task_2(df)


@flow
async def parent_flow():
    file_ids = get_file_ids()

    # semaphore limits how many subflows run concurrently
    semaphore = anyio.Semaphore(10)

    async def process_file(file_id) -> dict:
        async with semaphore:
            await run_deployment(...) # runs subflow

    await gather(
            *(lambda id=id: process_file(id) for id in file_ids)
        )
7 Views
Open in Slack
PreviousNext
Powered by