@ebonnal I believe you got this part slightly wrong, "off-by-one". IIUC, the number of pending futures cannot be larger than buffsize. However, after the initial submission of buffsize tasks before, here in this branch you are appending an EXTRA task to the queue, and now you have buffsize + 1 tasks that have potentially not yielded yet.

Fortunately, looks like the fix is trivial: you simply have to yield first, next append to the queue:

diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py
index d98b1ebdd58..3b9ccf4d651 100644
--- a/Lib/concurrent/futures/_base.py
+++ b/Lib/concurrent/futures/_base.py
@@ -628,17 +628,17 @@ def result_iterator():
                 # reverse to keep finishing order
                 fs.reverse()
                 while fs:
+                    # Careful not to keep a reference to the popped future
+                    if timeout is None:
+                        yield _result_or_cancel(fs.pop())
+                    else:
+                        yield _result_or_cancel(fs.pop(), end_time - time.monotonic())
                     if (
                         buffersize
                         and (executor := executor_weakref())
                         and (args := next(zipped_iterables, None))
                     ):
                         fs.appendleft(executor.submit(fn, *args))
-                    # Careful not to keep a reference to the popped future
-                    if timeout is None:
-                        yield _result_or_cancel(fs.pop())
-                    else:
-                        yield _result_or_cancel(fs.pop(), end_time - time.monotonic())
             finally:
                 for future in fs:
                     future.cancel()

Hi @dalcinl,

TL;DR: fyi we have #131467 that is open and tackling this "off-by-one" situation. Would be great to get your review there! 🙏🏻
I have not proposed this variation at first because I think it makes sense as an optional follow up PR given that it integrates slightly less smoothly into existing logic.

You will notice that it is not "just" moving the yield before the enqueue, let's see why on a simple scenario, explaining the 3 behaviors:

Scenario:

it = executor.map(fn, iterable, buffersize=buffersize)
# point A
next(it)
# point B
next(it)
# point C

"enqueue -> wait -> yield" (current, introduced by this PR) behavior

• buffers buffersize tasks
• at # point A there is buffersize buffered tasks
• 1st call to next:
  • enqueue a task, jumping to buffersize+1 tasks in buffer
  • wait for the next result to be available
  • yield the next result, finally going down to buffersize tasks in buffer
• at # point B there is still buffersize buffered tasks
• 2nd call to next: same

pro: buffersize tasks in buffer between two calls to next
con: while waiting for the next result we have buffersize+1 tasks in buffer

"wait -> yield -> enqueue" (your proposal) behavior

• buffers buffersize tasks
• at # point A there is buffersize buffered tasks
• call to next:
  • wait for the next result to be available
  • yield this result, going down to buffersize - 1 tasks in buffer
• at # point B there is buffersize - 1 buffered tasks
• call to next:
  • enqueue a task, jumping back to buffersize tasks in buffer
  • wait for the next result to be available
  • yield next result, going down to buffersize - 1 tasks in buffer
• at # point C there is still buffersize-1 buffered tasks

pro: never exceed buffersize
con: between two calls to next we have only buffersize - 1 tasks in buffer

"wait -> enqueue -> yield" (#131467) behavior

• buffers buffersize tasks
• at # point A there is buffersize buffered tasks
• 1st call to next:
  • wait for next result to be available
  • enqueue a task, jumping to buffersize + 1 tasks in buffer
  • yield already available result without needing to wait, going back to buffersize tasks in buffer
• at # point B there is still buffersize buffered tasks
• 2nd call to next: same

pros:

• buffersize tasks in buffer between two calls to next
• jumps to buffersize + 1 during a call to next but is back instantly to buffersize because the next result is already available when we enqueued the next task (the closest we can get to a yield-and-enqueue-at-the-same-time).

Let me know if it makes sense 🙏🏻