Hi all, in prefect

2.14.3

I’ve noticed that the global concurrency is blocking tasks longer than needed. I have 1 flow submitting multiple tasks like so:

from prefect import task, flow
from prefect.concurrency.asyncio import concurrency

@task(...)
async def my_task(...):
    print(f"Checking for available concurrency limit slot from {tags}")
    async with concurrency(tags):
        print(f"Acquired concurrency limit slot from {tags}")
        ...
    print(f"Releasing concurrency limit slot from {tags}")

@flow(...)
async def my_flow(task):
    for task in tasks:
        await task.submit(wait_for=...)

The first set of tasks run as expected as there are slots available in the global concurrency limit. Occasionally when they finish and release their slots the other tasks do not acquire these slots, even though they’re free. How long does

concurrency

wait between attempts to acquire a slot?

The

concurrency

manager tries to predict when a new slot will be free, based on a moving average of the previous slot acquire/release times. So if the tasks you have are fairly variable in time then it could be that when a fast task completes it might take a while before the manager tries to grab a new slot.

Thanks for the explanation @Chris Pickett. Is there a way to change this to a set polling frequency (e.g. something like

concurrency(tags, polling=10)

)?

That’s not possible at the moment, but it seems like a decent idea in general, we’d want a lower bounds on the polling frequency, but that’s easily doable. Another option to unblock you, though I’m not sure it meets your requirements, is to use it as a

rate_limit

, that wouldn’t change the polling frequency, but it would ensure that slots are released at a regular pace, and the predictive acquisition works better because of that. But if you actually need to make sure that only 1 of these tasks is running at a time then this doesn’t solve your problem.

wonder if it is related to this https://prefect-community.slack.com/archives/C048SVCEFF0/p1700639636500989