gh-146392: Delay Executor.map shutdown error until buffer is drained

Author: ashm-dev

Lib/concurrent/futures/_base.py

@@ -651,21 +651,35 @@ def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
         # Yield must be hidden in closure so that the futures are submitted
         # before the first iterator value is required.
         def result_iterator():
+            submit_exc = None
+            exhausted = object()
+
             try:
                 # reverse to keep finishing order
                 fs.reverse()
                 while fs:
-                    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 buffersize and submit_exc is None:
+                        args = next(zipped_iterables, exhausted)
+                        if args is not exhausted:
+                            executor = executor_weakref()
+                            if executor is None:
+                                submit_exc = RuntimeError(
+                                    "cannot schedule new futures after shutdown"
+                                )
+                            else:
+                                try:
+                                    fs.appendleft(executor.submit(fn, *args))
+                                except RuntimeError as exc:
+                                    submit_exc = exc
+
+                    # buffered result should still be yielded
                     if timeout is None:
                         yield _result_or_cancel(fs.pop())
                     else:
                         yield _result_or_cancel(fs.pop(), end_time - time.monotonic())
+
+                if submit_exc is not None:
+                    raise submit_exc
             finally:
                 for future in fs:
                     future.cancel()

Lib/test/test_concurrent_futures/executor.py

@@ -168,6 +168,33 @@ def test_map_buffersize_when_buffer_is_full(self):
             msg="should have fetched only `buffersize` elements from `ints`.",
         )
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_map_buffersize_shutdown_drains_buffer_then_raises(self):
+        with self.executor_type(max_workers=1) as executor:
+            it = executor.map(str, range(8), buffersize=2)
+
+        self.assertEqual(next(it), "0")
+        self.assertEqual(next(it), "1")
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "cannot schedule new futures after shutdown",
+        ):
+            next(it)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_map_buffersize_shutdown_after_iteration_started_drains_buffer_then_raises(self):
+        with self.executor_type(max_workers=1) as executor:
+            it = executor.map(str, range(8), buffersize=2)
+            self.assertEqual(next(it), "0")
+
+        self.assertEqual(next(it), "1")
+        self.assertEqual(next(it), "2")
+        with self.assertRaisesRegex(
+            RuntimeError,
+            "cannot schedule new futures after shutdown",
+        ):
+            next(it)
+
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_shutdown_race_issue12456(self):
         # Issue #12456: race condition at shutdown where trying to post a

Misc/NEWS.d/next/Library/2026-03-25-01-03-05.gh-issue-146392.zb-lzE.rst

@@ -0,0 +1,2 @@
+Fix Executor.map(buffersize=...) to drain buffered results before raising
+after shutdown. Patch by Shamil Abdulaev.