Task run concurrency limits help prevent too many tasks from running simultaneously. For example, if many tasks across multiple flows are designed to interact with a database that only allows 10 connections. Task run concurrency limits use task tags. You can specify an optional concurrency limit as the maximum number of concurrent task runs in a Running state for tasks with a given tag.
Tag-based task concurrency is different from Global concurrency limits, though they can be used to achieve similar outcomes. Global concurrency limits are a more general way to control concurrency for any Python-based operation, whereas tag-based concurrency limits are specific to Prefect tasks.
If a task has multiple tags, it will run only if all tags have available concurrency. Tags without specified concurrency limits are treated as unlimited. Setting a tag’s concurrency limit to 0 causes immediate abortion of any task runs with that tag, rather than delaying them.

​
Execution behavior

Task tag limits are checked whenever a task run attempts to enter a Running state. If there are no concurrency slots available for any one of your task’s tags, it delays the transition to a Running state and instructs the client to try entering a Running state again in 30 seconds (or the value specified by the PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS setting).

​
Configure task run concurrency limits

Flow run concurrency limits are set at a work pool, work queue, or deployment levelWhile task run concurrency limits are configured through tags (as shown below), flow run concurrency limits are configured through work pools and/or work queues.
You can set concurrency limits on as few or as many tags as you wish. You can set limits through:

​
CLI

You can manage task run concurrency with the Prefect CLI. 
prefect concurrency-limit [command] [arguments]
CommandDescription
createCreate a concurrency limit by specifying a tag and limit.
deleteDelete the concurrency limit set on the specified tag.
inspectView details about a concurrency limit set on the specified tag.
lsView all defined concurrency limits.
For example, to set a concurrency limit of 10 on the ‘small_instance’ tag: 
prefect concurrency-limit create small_instance 10
To delete the concurrency limit on the ‘small_instance’ tag: 
prefect concurrency-limit delete small_instance
To view details about the concurrency limit on the ‘small_instance’ tag: 
prefect concurrency-limit inspect small_instance

​
API

You can manage task run concurrency with the Prefect API.

​
Terraform

You can manage task run concurrency with the Terraform provider for Prefect.

​
Python client

To update your tag concurrency limits programmatically, use PrefectClient.orchestration.create_concurrency_limit. create_concurrency_limit takes two arguments:
  • tag specifies the task tag on which you’re setting a limit.
  • concurrency_limit specifies the maximum number of concurrent task runs for that tag.
For example, to set a concurrency limit of 10 on the ‘small_instance’ tag: 
from prefect import get_client

async with get_client() as client:
    # set a concurrency limit of 10 on the 'small_instance' tag
    limit_id = await client.create_concurrency_limit(
        tag="small_instance", 
        concurrency_limit=10
        )
To remove all concurrency limits on a tag, use PrefectClient.delete_concurrency_limit_by_tag, passing the tag: 
async with get_client() as client:
    # remove a concurrency limit on the 'small_instance' tag
    await client.delete_concurrency_limit_by_tag(tag="small_instance")
If you wish to query for the current limit on a tag, use PrefectClient.read_concurrency_limit_by_tag, passing the tag: To see all of your limits across all of your tags, use PrefectClient.read_concurrency_limits. 
async with get_client() as client:
    # query the concurrency limit on the 'small_instance' tag
    limit = await client.read_concurrency_limit_by_tag(tag="small_instance")