gh-74028: concurrent.futures.Executor.map: add buffersize param for lazy behavior

[ebonnal]

Context recap (#74028)

concurrent.futures.Executor.map is not lazy:

• it cannot be used to map infinite input iterables
• even for finite input iterables the inconvenience is that it is $O(n)$ memory-wise, collecting all the results into a list
• also because it launches all the futures at once it becomes impossible for client code to limit the rate of the iteration over input iterables, which is a critical feature when the mapped fn calls an external service that you don't want to overload.

Proposal: add buffersize param

Similar to the work of @graingert in #18566 and @Jason-Y-Z in #114975, i.e. use a queue of fixed size to hold the not-yet-yielded future results, and only iterating on input iterables if the queue is not full.

In addition this PR:

• uses the intuitive term "buffer"
• keeps the exact existing list-based behavior when buffersize=None
• integrates concisely into existing logic
• forbids the usage of both timeout and buffersize at the same time
• tests "If the buffer is full, then the iteration over iterables is paused until a result is yielded from the buffer."
• intuitive docstrings

• Issue: Make Executor.map work with infinite/large inputs correctly #74028

---

📚 Documentation preview 📚: https://cpython-previews--125663.org.readthedocs.build/

[cpython-cla-bot]

All commit authors signed the Contributor License Agreement.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[Zheaoli]

Thanks for the PR.

First, I think this is a big behavior change for Executor. I think we need to discuss it in the https://discuss.python.org/ first.

In my personal opinion, I think this is not a good choice to add the buffersize argument to the api. For now, the API design is based on the original map API. I think this argument will bring more inconsistent into the codebase. And BTW, even if we need the buffersize argument, I think it's not reasonable to forbids the usage of both timeout and buffersize at the same time

The returned iterator raises a TimeoutError if next() is called and the result isn’t available after timeout seconds from the original call to Executor.map().

[ebonnal]

Hi @Zheaoli, thank you for your comment!

First, I think this is a big behavior change for Executor.

You mean big alternative behavior, right? (the default behavior when ommitting buffersize remaining unchanged)

I think we need to discuss it in the https://discuss.python.org/ first.

Fair, I will start a thread there and ping you.

For now, the API design is based on the original map API. I think this argument will bring more inconsistent into the codebase.

I'm not sure to get it, could you detail that point? 🙏🏻

I think it's not reasonable to forbids the usage of both timeout and buffersize at the same time

You are completely right, makes more sense! I have fixed that (commit)

[Zheaoli]

I'm not sure to get it, could you detail that point? 🙏🏻

For me, the basic map API's behavior is when we put an infinite iterator, the result would be infinite and only stop when the iterator has been stoped. I think we need to keep the same behavior between map and executor.map

[ebonnal]

Hi @Zheaoli

For me, the basic map API's behavior is when we put an infinite iterator, the result would be infinite and only stop when the iterator has been stoped. I think we need to keep the same behavior between map and executor.map

There may be a misunderstanding here, the goal of this PR is precisely to make Executor.map closer to the builtin map behavior, i.e. make it lazier. (map and current executor.map do not have the same behavior)

I will recap the behaviors so that everybody is on the same page:

built-in map

infinite_iterator = itertools.count(0)

# a `map` instance is created and the func and iterable are just stored as attributes
mapped_iterator = map(str, infinite_iterator)

# retrieves the first element of its input iterator, applies
# the transformation and returns the result
assert next(mapped_iterator) == "0" 

# the next element in the input iterator is the 2nd
assert next(infinite_iterator) == 1

# one can next infinitely
assert next(mapped_iterator) == "2"
assert next(mapped_iterator) == "3" 
assert next(mapped_iterator) == "4" 
assert next(mapped_iterator) == "5" 
...

Executor.map without buffersize (= current Executor.map)

infinite_iterator = itertools.count(0)

# this line runs FOREVER, trying to iterate over input iterator until exhaustion
mapped_iterator = executor.map(str, infinite_iterator)

⏫ this line will run forever because it collects the entire input iterable eagerly, in order to build the entire future results list fs = [self.submit(fn, *args) for args in zip(*iterables)] which requires infinite time and memory.

Executor.map with buffersize

infinite_iterator = itertools.count(0)

# retrieves the first 2 elements (=buffersize) and submits 2 tasks for them
mapped_iterator = executor.map(str, infinite_iterator, buffersize=2)

# retrieves the 3rd element of input iterator and submits a task for it,
# then wait for the oldest future in the buffer to complete and returns the result
assert next(mapped_iterator) == "0" 

# the next element of the input iterator is the 4th
assert next(infinite_iterator) == 3

# one can next infinitely while only a buffer of finite not-yet-yielded future results is kept in memory
assert next(mapped_iterator) == "1" 
assert next(mapped_iterator) == "2" 
assert next(mapped_iterator) == "4"
assert next(mapped_iterator) == "5" 
...

note

I used the example of an infinite input iterator because this is an example where current Executor.map is just unusable at all. But even for finite input iterables, if a developer writes mapped_iterator = executor.map(fn, iterable), they often don’t want the iterable to be eagerly exhausted right away, but rather to be iterated at the same rate as mapped_iterator. This PR's proposal is to allow them to do so by setting a buffersize.

[ebonnal]

hey @rruuaanng, fyi I have applied your requested changes regarding the integration of unit tests into existing class 🙏🏻

[rruuaanng]

Why does it have to be None?

[ebonnal]

it has to be None OR to be greater than 0, would the addition of either make it clearer ? -> ValueError("buffersize must be either None or >= 1.")

[rruuaanng]

Isn't buffersize empty? Can you introduce it? (Forgive me for not understanding it).

[ebonnal]

absolutely np, thank you for taking the time to review my proposal. To be sure to understand the question well, what do you mean by "Isn't buffersize empty?"