What's a best practice look like for setting up hundreds of concurrent REST API calls using prefect tasks? We want to use some kind of concurrency limit to avoid overwhelming the API. However, the concurrency limits in prefect make tasks wait 30 seconds before checking for an available slot. This results in a HUGE slowdown, assuming each API call takes significantly less than 30 seconds a pop (less than 1 second each is normal).

I’d use an AnyIO capacity limiter https://anyio.readthedocs.io/en/stable/synchronization.html#capacity-limiters

It looks like that could work. Is there any solution that avoids using asyncio? We'd like to avoid migrating our pipelines to asyncio if at all possible. It creates a significant technical hurdle.

How are you going to be running them concurrently?

If I wasn't in prefect, I'd use:

from concurrent.futures import ThreadPoolExecutor

values = fetch_values()
with ThreadPoolExecutor(max_workers=5) as executor:
    executor.map(make_api_call, values)

in prefect I'm working with the default concurrent task runner:

@flow
def run_pipeline():
    values = fetch_values()
    make_api_call.map(values)

It sounds like https://github.com/PrefectHQ/prefect/pull/7013 would be what you’re looking for

That does look like exactly what I'm interested in!

Yeah I do think we can improve on this in the future but as a workaround im glad you found something.

You could use a real

threading.Semaphore

too

That is probably wise...time to climb a little further down the threading rabbithole...

Let me know if you have questions. I’m very deep in the concurrency rabbithole these days.

This is a little more elegant:

import time
from functools import wraps
from threading import Semaphore
from typing import Callable

from prefect import flow, task


def limit_concurrency(
    max_workers: int,
) -> Callable[[Callable], Callable]:

    semaphore = Semaphore(max_workers)

    def pseudo_decorator(func: Callable):
        @wraps(func)
        def limited_concurrent_func(*args, **kwargs):
            semaphore.acquire()
            result = func(*args, **kwargs)
            semaphore.release()
            return result

        return limited_concurrent_func

    return pseudo_decorator


@task
@limit_concurrency(max_workers=2)
def concurrent_task(value):
    print(f"starting {value}")
    time.sleep(1)
    print(f"ending {value}")


@flow
def my_flow():
    values = range(30)
    concurrent_task.map(values)


if __name__ == "__main__":
    my_flow()

If your function fails you won’t release the sempahore

ah word good call

You’ll want to use try/finally or use it as a context manager

Got it. Thanks so much! Really appreciate your feedback

You’re welcome 🙂

I'm seeing this implementation use 100% CPU on one core for something simple like running 10 threads with this kind of function:

while True:
   response = requests.get(some_url)
   if response['success']:
       return
   else:
      sleep(30)

It's not using barely any memory but the CPU is maxing out. Any idea why that would happen?

Are you releasing the semaphore while sleeping?

no, not releasing the semaphore while sleeping

strace

might help

Hi y'all. Great thread here. I've been testing out this solution and I am noticing that it creates as many threads as there are elements in my

values

object. My expectation is that this would create as many workers and threads as specified by the

max_workers

variable. For my test case I have tens of thousands of elements I need to operate on and when using this solution Prefect creates tens of thousands of tasks before working on any of them. What am I missing? What am I not understanding about prefect? Thanks!

That's correct and the normal behavior for task.map. the solution here helps with slowing down the execution of the tasks, but all the task instantiation happens up front.

We submit tasks concurrently using the event loop

To rate limit the creation of tasks, rather than using task.map, I can imagine a concurrency limited function that calls task.submit and awaits task results to release a semaphore, although I think you'd need to break prefect patterns a little more to implement this 🤔

In my case, when I unleashed this on the tens of thousands after testing on a much smaller sample, Prefect created all the tasks and then just hung there and none of the tasks ever completed.

Yeah these are things I’m working on very actively