Short answer: - You can’t “burst” above a ThreadPoolTaskRunner’s max size at runtime and there’s no task priority feature. - In Prefect 3.x, the safest fix for starvation from tasks spawning subtasks is to either increase/remove the thread cap or restructure so orchestration happens at the flow/subflow level (not inside a running task). Details and options: 1) Raise or remove the thread cap - In Prefect 3.x, ThreadPoolTaskRunner defaults to effectively unbounded threads (sys.maxsize) specifically to avoid deadlocks when tasks submit more tasks. - If you’ve set a small

max_workers

, nested

.submit()

+

.result()

inside tasks can starve the pool. - How to set: - In code:

from prefect import flow
    from prefect.task_runners import ThreadPoolTaskRunner

    @flow(task_runner=ThreadPoolTaskRunner(max_workers=128))  # or omit to use default
    def my_flow():
        ...

- Via env var (either name is accepted):

export PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS=128
    # or
    export PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS=128

- Check current value:

from prefect.settings import get_current_settings
    print(get_current_settings().tasks.runner.thread_pool_max_workers)

- Docs/refs: - Task runners concept - ThreadPoolTaskRunner API - Settings reference - Source: ThreadPoolTaskRunner impl (defaults to sys.maxsize) - Test proving recursive/nested submission works when not artificially capped: recursive nested tasks test 2) Don’t block inside a task; let the flow orchestrate - Starvation often comes from a task that submits children and then immediately calls

.result()

while the pool is small. - Prefer: submit in the task (or better, in the flow), return futures, and wait at the flow level so the parent task doesn’t keep a worker occupied while children need threads. Example pattern:

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

@task
def launch_children(n):
    # Submit but do not call .result() here
    return [child.submit(i) for i in range(n)]

@task
def child(i):
    return i * 2

@flow(task_runner=ThreadPoolTaskRunner(max_workers=32))
def my_flow():
    futs = launch_children.submit(50)       # parent task exits quickly
    results = wait(futs.result())           # wait in the flow, not inside the task
    return [f.result() for f in results]

3) Use subflows for orchestration boundaries - Each subflow gets its own task runner and thread pool, so “child” work doesn’t compete with “parent” threads. - This is a clean way to avoid pool contention and gain better observability. - Docs: Organize flows with subflows and tasks 4) No built-in priorities - Prefect’s local task runners don’t support a priority queue. - If you need ordering, control the submission order or split work into separate flows/runners. 5) Consider the right runner for the workload - IO-bound: ThreadPoolTaskRunner (default) - CPU-bound: ProcessPoolTaskRunner (give it enough workers; similar caveat about nesting if you cap too low) If you can share: - Your Prefect version (assuming 3.x) - How you’re constructing the task runner (any

max_workers

set?) - A minimal snippet showing the task that spawns subtasks and where you call

.result()

…I can suggest the smallest change to fix the starvation in your setup.

Yes — this is exactly what Prefect’s tag-based task concurrency limits are for. How it works - Tag your tasks by type (e.g., download/transform/upload). - Create a concurrency limit per tag. The Prefect server will only let that many tasks with the tag run at once, across all workers, while still respecting your flow’s overall thread pool. - There’s no priority, but limits give you “only N of this kind concurrently” control. Example

from prefect import flow, task

@task(tags=["download"])
def download(...): ...

@task(tags=["transform"])
def transform(...): ...

@task(tags=["upload"])
def upload(...): ...

@flow
def pipeline():
    # submit lots of each; server will gate by tag limits
    ...

Set limits - CLI

prefect concurrency-limit create download 4
  prefect concurrency-limit create transform 4
  # omit upload for "unlimited" (it’ll only be bounded by available workers)
  prefect concurrency-limit ls
  prefect concurrency-limit inspect download

- Python (async)

from prefect import get_client

  async with get_client() as client:
      await client.create_concurrency_limit(tag="download", concurrency_limit=4)
      await client.create_concurrency_limit(tag="transform", concurrency_limit=4)

Important caveats - Limits are enforced by the server/orchestrator. They are not applied in pure client-side runs. Use workers (or Cloud) for enforcement. - If a task has multiple tags, it needs an available slot for all of them (AND logic). - “Unlimited” just means no tag limit; execution is still capped by your runner’s

max_workers

. Docs and refs - Task runners: concepts - ThreadPoolTaskRunner API: API - Tag-based concurrency limits: concepts and CLI reference If you share your deployment mode (local vs worker/Cloud) I can help wire this up so the limits are actually enforced in your environment.