concurrent.futures.Executor.map: raise after executor shutdown

Author: ebonnal

Lib/concurrent/futures/_base.py

@@ -614,12 +614,10 @@ def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):
             fs = collections.deque(
                 self.submit(fn, *args) for args in islice(zipped_iterables, buffersize)
             )
+            executor = self
         else:
             fs = [self.submit(fn, *args) for args in zipped_iterables]
-
-        # Use a weak reference to ensure that the executor can be garbage
-        # collected independently of the result_iterator closure.
-        executor_weakref = weakref.ref(self)
+            executor = None
 
         # Yield must be hidden in closure so that the futures are submitted
         # before the first iterator value is required.
@@ -628,11 +626,7 @@ def result_iterator():
                 # reverse to keep finishing order
                 fs.reverse()
                 while fs:
-                    if (
-                        buffersize
-                        and (executor := executor_weakref())
-                        and (args := next(zipped_iterables, None))
-                    ):
+                    if buffersize 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:

Lib/test/test_concurrent_futures/executor.py

@@ -139,6 +139,19 @@ def test_map_buffersize_on_multiple_infinite_iterables(self):
         self.assertEqual(next(res, None), 2)
         self.assertEqual(next(res, None), 4)
 
+    def test_map_buffersize_raise_on_shutdown(self):
+        for start_iteration in (False, True):
+            with self.subTest(start_iteration=start_iteration):
+                with self.executor_type(max_workers=1) as executor:
+                    results = executor.map(str, range(4), buffersize=1)
+                    if start_iteration:
+                        next(results)
+                with self.assertRaisesRegex(
+                    RuntimeError,
+                    "cannot schedule new futures after shutdown",
+                ):
+                    next(results)
+
     def test_map_buffersize_on_empty_iterable(self):
         res = self.executor.map(str, [], buffersize=2)
         self.assertIsNone(next(res, None))