Limit coroutines using a pool instead of chunks

[sisp]

I've changed the implementation of the function _run_coros_in_chunks() by using a coroutine pool instead of running coroutines in chunks.

The goal of this function is to limit the number of simultaneous coroutines, but chunking may lead to inferior throughput when coroutines have different execution times, as the next chunk will run only once all coroutines in the previous chunk have completed. In contrast, a coroutine pool will run the next coroutine in the queue as soon as any of the running coroutines completes.

[sisp]

The check CI / gcsfs-pytest is failing with this error:

FAILED gcsfs/gcsfs/tests/test_core.py::test_sign - ImportError: cannot import name 'storage' from 'google.cloud' (unknown location)

But I doubt it's related to this PR.

[martindurant]

With this approach, I agree that a limited number of IO tasks are concurrently running/waiting, but is there any potential downside of simply having a very large number of tasks created and waiting on this semaphore? The previous incantation would only have the given number of top-level tasks at a time.

[sisp]

Good question. I'm not aware of any downsides but can't prove there aren't any. I was also considering to use asyncio.as_completed and pass an iterable of asyncio.Tasks, but it also materializes this iterator as a set internally.

[martindurant]

Let's leave this question open a little while, to see if any knowledgable person comments.

[sisp]

FWIW, I've used this implementation to limit the simultaneous execution of 70,000 coroutines and it worked fine.

[sisp]

You might have a point: https://stackoverflow.com/a/62404509 I wonder whether we could use the shown queue-based approach instead.

[sisp]

I've pushed an implementation based on asyncio.wait inspired by https://death.andgravity.com/limit-concurrency#asyncio-wait. I think this looks good and tests are passing.

[Skylion007]

I actually think the linked solution may be more efficient. Does this solution run into the issue that the next batch does not start until the current batch finishes? That means with large batch size, future workers will be stalled waiting on the slowest worker?

Or am I misunderstanding the current batch async method.

[sisp]

Does this solution run into the issue that the next batch does not start until the current batch finishes? That means with large batch size, future workers will be stalled waiting on the slowest worker?

Correct, the slowest coroutine in a batch determines the runtime of the whole batch and the next batch won't start before all coroutines in the current batch have finished.

[Skylion007]

I think some of the solutions I linked would have alleviated that issue so it could have eagerly continued the next batch.

[sisp]

The solution that @martindurant merged is essentially https://death.andgravity.com/limit-concurrency#asyncio-wait, which you linked.