Add ability to run `asyncio.gather()` with concurrency limit

[shidenko97]

Feature or enhancement

Proposal:

The main purpose of the proposal is to be able to run asyncio.gather() with some limit to not overload resources. For example you want to fetch 1000 pages of website, want to do it concurrently, but with some limit, to not to overload both client and server. For such task you can simply use asyncio.gather() with asyncio.Semaphore() and got needed result. It means with limit of 10 (for example) you will have only 10 concurrent tasks running, while other 990 will be still waiting and after every one task finished - new one will be started.

# implementation example
async def asyncio_gather_with_concurrency_limit(*tasks, limit: int):
    semaphore = asyncio.Semaphore(limit)

    async def semaphored_task(task):
        async with semaphore:
            return await task

    return await asyncio.gather(*(semaphored_task(task) for task in tasks))

# usage example
await asyncio_gather_with_concurrency_limit(*[asyncio.sleep(i) for i in range(100)], limit=10)

Has this already been discussed elsewhere?

No response given

Links to previous discussion of this feature:

No response

[serhiy-storchaka]

I think that it has nothing to do with gather(), it has to do with your tasks. Waiting on the semaphore should be the part of your tasks. You have shown that it is easy to make wrappers.

BTW, not all of your tasks may need to wait on the semaphore. For example you can have a long running task that informs the user how many pages were fetched. You can also guard with the semaphore only the part of the task. For example, limit the number of concurrent page fetches, but do not limit the number of concurrent saves on the disk. Or vice versa, limit the number of concurrent saves in the database with other semaphore.

[shidenko97]

You're right, it could be complicated if task has more than one asynchronous call inside. Of course it can be easily implemented on the client's side, I thought about this implementation as a kind of additional util to gather, to make it easier for users (I believe I'm not the only one who had to implement such feature, so it can be useful common util).

[serhiy-storchaka]

You may be not the one, but this feature looks too niche and too limited. Python provides building blocks, and you can combine them in arbitrary solutions.

[karolinepauls]

I don't know if it's the right place to complain about this but I'll try.

Python's stdlib is known to be "batteries included". You don't have to implement your own Queue, Counter, Decimal, Popen, or choose one from N different 3rd party libraries. This is generally a good thing.

The lack of concurrency limiting in gather makes people implement their own versions - often wrong. And it's hard to blame them for that. Writing the same wrapper many times is boring. This is not batteries included.

A quick web search shows examples like:

• https://gist.github.com/benfasoli/650a57923ab1951e1cb6355f033cbc8b
• https://www.reddit.com/r/Python/comments/xw4f0k/comment/ir6nat3/

Last but not least, semaphores are not an elegant solution. Let's say I have 100k tasks and my concurrency limit is 20. Why burn cycles and memory starting tasks that will block on that same semaphore? Yes, that will only happen once the other tasks are blocked. But coroutine state isn't free.

gather, when given a limit of N, could only spawn N tasks at first and then spawn subsequent tasks, up to the limit of N, when tasks prior to that finish, until the input tasks are exhausted.