Context recap (#64192)

Let's consider that we have an input iterable and N = len(iterable).

Current multiprocessing.Pool.imap and multiprocessing.Pool.imap_unordered are O(N) in space (unecessarily expensive on large iterables, completely impossible to use on infinite iterables):
The call results: Iterator = pool.imap(func, iterable) iterates over all the elements of the iterable, submitting N tasks to the pool (results are collected into a list of size N). Following calls to next(results) take the oldest result from the list (FIFO) (waiting for it if not available yet) and return it.

Proposal: add an optional buffersize param

With this proposal, the call results: Iterator = pool.imap(func, iterable, buffersize=b) will iterate only over the first b elements of iterable (acquiring buffersize semaphore while iterating), submitting b tasks to worker threads and then will return the results iterator.

Calls to next(results) will release buffersize semaphore (allowing task_handler thread to get the next input element from iterable to submit a new task to a worker thread) and then return the result.

buffersize semaphores from iterators not exhausted yet are also being released on pool termination to avoid deadlocks.

Benefits:

• The space complexity becomes O(b)
• When using a buffersize the client code takes back the control over the speed of iteration over the input iterable: after an initial spike of b calls to func to fill the buffer, the iteration over input iterable will follow the rate of the iteration over the results (controlled by the client), which is critical when func involves talking to services that you don't want to overload.

Feature history

buffersize support has been recently merged into concurrent.futures.Executor.map implementation (#125663) and many code/test/doc parts are based on ones from there to ensure consistency between modules.
I want to thank authors of that PR for the references.

Links:

• Docs preview

• Issue: Pool.imap doesn't work as advertised #64192