Great question! Assuming Prefect 3.x: - Within a single flow run, Prefect uses a task runner to control parallelism. - Default is

ThreadPoolTaskRunner

, which is effectively unlimited by default (it sets

max_workers

to

sys.maxsize

). So any tasks you

.submit(...)

can run in parallel up to your system limits. - To cap it, set a max:

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

    @task
    def do_work(i): ...

    @flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
    def my_flow():
        for i in range(100):
            do_work.submit(i)

Or via env var:

export PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS=8

- For CPU-bound work, use processes instead:

from prefect.task_runners import ProcessPoolTaskRunner

    @flow(task_runner=ProcessPoolTaskRunner())  # defaults to number of CPU cores
    def cpu_flow(): ...

Env var to override:

export PREFECT_TASKS_RUNNER_PROCESS_POOL_MAX_WORKERS=4

- Higher-level (across flow runs) controls: - Limit concurrent flow runs per worker:

prefect worker start --pool my-pool --limit 10

- Global Concurrency Limits to cap a shared resource across flows:

prefect gcl create database-queries --limit 5

Then in code:

from prefect.concurrency.sync import concurrency

    with concurrency("database-queries", occupy=1):
        ...  # work

Helpful docs: - Task runners: Task runners (Prefect 3.x) - Global concurrency limits: Global Concurrency Limits Quick note: tasks run concurrently when you call `task_fn.submit(...)`; calling

task_fn(...)

runs them inline/sequentially. Would you like to control concurrency within a single flow run, or the number of flow runs a worker handles at once?