@Marvin, can you explain Global Concurrency Limits and Task Run Concurrency Limits and are there any limits or quotas for them?

Global Concurrency Limits and Task Run Concurrency Limits are mechanisms in Prefect to control the execution of workflows at different granularities: Global Concurrency Limits: Global concurrency limits allow you to limit the total number of concurrently running flow runs across your entire Prefect instance. This is particularly useful when you want to manage the overall load on your system or when there are strict limitations on external resources that all your flows may use (like a specific API that has rate limits). To configure global concurrency limits, you can use the

slot_decay_per_second

parameter to control how fast slots become available. For example, if you set a higher

slot_decay_per_second

value, slots will become available quickly after being occupied by a task. Conversely, a lower value results in slower slot availability. The

concurrency

context manager allows you to control this behavior by creating concurrency limits that are initially inactive but are activated as needed. You can read more about this in the Prefect documentation. Task Run Concurrency Limits: Task run concurrency limits are used to restrict the number of task runs that can execute simultaneously based on tags. This feature is useful when tasks across multiple flows interact with an external system with connection limits, such as a database. By setting task concurrency limits, you make sure that no more than the specified limit of task runs with a given tag are in a

Running

state at the same time. If a task run attempts to enter a

Running

state and the limit has been reached, the run may be delayed until a slot is free. If you set the concurrency limit for a tag to 0, any task run with that tag will be aborted instead of delayed. You can read more about task run concurrency limits in the Prefect documentation. Regarding limits or quotas for global concurrency limits and task run concurrency limits, the documentation does not mention any specific limits or quotas imposed by Prefect itself. However, you should be mindful of the resources available in your own infrastructure and external services you are interacting with. If you have further questions or need to explore specific use cases, don't hesitate to dive deeper into the documentation or ask additional questions.

@Marvin , what's the difference between

from prefect.concurrency.sync import concurrency

and

from prefect.concurrency.sync import rate_limit

?

Both

concurrency

and

rate_limit

come from the Prefect concurrency module, but they serve different purposes:

concurrency

: The

concurrency

context manager is used to control the concurrency limit, which is the total number of concurrent operations that can occur at the same time. When you use

concurrency

, you specify a name for your concurrency limit and how many slots you want to occupy. If the concurrency limit is reached (i.e., all slots are occupied), any subsequent operations that try to use the same concurrency context will have to wait until a slot frees up. Here's an example of how to use the

concurrency

context manager:

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def process_data(x, y):
    with concurrency("database", occupy=1):
        return x + y

@flow
def my_flow():
    for x, y in [(1, 2), (2, 3), (3, 4), (4, 5)]:
        process_data.submit(x, y)

rate_limit

: The

rate_limit

function, on the other hand, is designed to control the rate at which operations are performed. This is particularly useful when you need to comply with external rate limits, such as when making API calls to services that have a maximum allowed number of requests per second. The

rate_limit

function requires your concurrency limit to have a slot decay configured so that it can govern the frequency of operations. Here's an example of how to use the

rate_limit

function:

python
from prefect import flow, task
from prefect.concurrency.sync import rate_limit

@task
def make_http_request():
    rate_limit("rate-limited-api")
    print("Making an HTTP request...")

@flow
def my_flow():
    for _ in range(10):
        make_http_request.submit()

Basically,

concurrency

is for controlling the number of concurrent operations, and

rate_limit

is for governing the frequency of operations to avoid hitting rate limits. For more details, you can check the Prefect documentation on Global Concurrency Limits.

@Marvin should I use rate limit and concurrency together? Show example please

It is possible to use both

concurrency

and

rate_limit

together in Prefect if you need to control both the number of concurrent operations as well as the frequency of those operations. For example, you might want to ensure that you do not exceed a certain number of concurrent API calls to a service that has rate limiting. Unfortunately, the documentation does not provide a direct example of using them together. However, here’s a conceptual example to illustrate how you could combine

concurrency

with

rate_limit

in your task:

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency, rate_limit

@task
def make_rated_http_request():
    # Control concurrency using the concurrency limit named "db-slots"
    with concurrency("db-slots", occupy=1):
        # Control the rate of requests using the rate limit named "api-rate-limiter"
        rate_limit("api-rate-limiter")
        print("Making a rate-limited HTTP request...")

@flow
def my_flow():
    for _ in range(20):
        make_rated_http_request.submit()

if __name__ == "__main__":
    my_flow()

In this example: -

concurrency

is used with a limit named

"db-slots"

. Each use of the concurrency context manager occupies one slot in this limit. Other operations that try to use the same limit will wait if no slots are available. -

rate_limit

is used with a limit named

"api-rate-limiter"

. It follows the configured decay slot rate which controls how frequently a slot becomes available, thereby limiting the rate of operations. Remember, you should adjust the details such as number of slots, decay rate, and limit names according to your use case and system constraints. For more information on concurrency and rate limits, you may refer to the Prefect Global Concurrency Limits documentation.