From ff499698aecb60f6cb456de4eaf50d15435728fa Mon Sep 17 00:00:00 2001
From: Meng Xiangzhuo <aumo@foxmail.com>
Date: Sun, 17 Sep 2023 00:25:12 +0800
Subject: [PATCH 1/8] Make concurrent.futures.Executor.map() behave consistent
 with built-in map()

---
 Lib/concurrent/futures/_base.py               | 68 ++++++++++++++++++-
 Lib/concurrent/futures/process.py             | 17 ++++-
 Lib/test/test_concurrent_futures/executor.py  |  3 +
 .../test_thread_pool.py                       |  9 +--
 4 files changed, 87 insertions(+), 10 deletions(-)

diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py
index 6742a07753c9217..03580cfc2dbfb40 100644
--- a/Lib/concurrent/futures/_base.py
+++ b/Lib/concurrent/futures/_base.py
@@ -314,7 +314,9 @@ def wait(fs, timeout=None, return_when=ALL_COMPLETED):
 def _result_or_cancel(fut, timeout=None):
     try:
         try:
-            return fut.result(timeout)
+            return _FutureResult.from_value(fut.result(timeout))
+        except Exception as e:
+            return _FutureResult.from_exception(e)
         finally:
             fut.cancel()
     finally:
@@ -566,6 +568,46 @@ def set_exception(self, exception):
 
     __class_getitem__ = classmethod(types.GenericAlias)
 
+
+class _FutureResult(object):
+    """
+    This is used to record the exception instead of throwing them.
+
+    _FutureResult must contain either the value of future or an exception
+    that was thrown during the computation of future. Use is_exception
+    property to determine which one it is.
+    """
+
+    def __init__(self, exception, value):
+        self._exception = exception
+        self._value = value
+
+    @classmethod
+    def from_exception(cls, exc):
+        return cls(exc, None)
+
+    @classmethod
+    def from_value(cls, value):
+        return cls(None, value)
+
+    @property
+    def exception(self):
+        if not self.is_exception:
+            raise RuntimeError("No exception thrown.")
+        return self._exception
+
+    @property
+    def value(self):
+        if self.is_exception:
+            raise RuntimeError(
+                "Cannot get result value because an exception was thrown.")
+        return self._value
+
+    @property
+    def is_exception(self):
+        return self._exception is not None
+
+
 class Executor(object):
     """This is an abstract base class for concrete asynchronous executors."""
 
@@ -602,6 +644,11 @@ def map(self, fn, *iterables, timeout=None, chunksize=1):
                 before the given timeout.
             Exception: If fn(*args) raises for any values.
         """
+        return _MapResultIterator.from_generator(
+            self._map(fn, *iterables, timeout=timeout)
+        )
+
+    def _map(self, fn, *iterables, timeout=None):
         if timeout is not None:
             end_time = timeout + time.monotonic()
 
@@ -648,6 +695,25 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         return False
 
 
+class _MapResultIterator(object):
+    """The iterator returned by map()."""
+    def __init__(self, gen):
+        self.gen = gen
+
+    @classmethod
+    def from_generator(cls, gen):
+        return cls(gen)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        result = next(self.gen)
+        if result.is_exception:
+            raise result.exception
+        return result.value
+
+
 class BrokenExecutor(RuntimeError):
     """
     Raised when a executor has become non-functional after a severe failure.
diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
index f4b5cd1d869067d..1ad25955247d94c 100644
--- a/Lib/concurrent/futures/process.py
+++ b/Lib/concurrent/futures/process.py
@@ -204,7 +204,14 @@ def _process_chunk(fn, chunk):
     This function is run in a separate process.
 
     """
-    return [fn(*args) for args in chunk]
+    results = []
+    for args in chunk:
+        try:
+            result = _base._FutureResult.from_value(fn(*args))
+        except Exception as e:
+            result = _base._FutureResult.from_exception(e)
+        results.append(result)
+    return results
 
 
 def _sendback_result(result_queue, work_id, result=None, exception=None,
@@ -617,6 +624,9 @@ def _chain_from_iterable_of_lists(iterable):
     careful not to keep references to yielded objects.
     """
     for element in iterable:
+        if element.is_exception:
+            raise element.exception
+        element = element.value
         element.reverse()
         while element:
             yield element.pop()
@@ -830,10 +840,11 @@ def map(self, fn, *iterables, timeout=None, chunksize=1):
         if chunksize < 1:
             raise ValueError("chunksize must be >= 1.")
 
-        results = super().map(partial(_process_chunk, fn),
+        results = super()._map(partial(_process_chunk, fn),
                               _get_chunks(*iterables, chunksize=chunksize),
                               timeout=timeout)
-        return _chain_from_iterable_of_lists(results)
+        return _base._MapResultIterator.from_generator(
+            _chain_from_iterable_of_lists(results))
 
     def shutdown(self, wait=True, *, cancel_futures=False):
         with self._shutdown_lock:
diff --git a/Lib/test/test_concurrent_futures/executor.py b/Lib/test/test_concurrent_futures/executor.py
index 1e7d4344740943d..330145b82d5a784 100644
--- a/Lib/test/test_concurrent_futures/executor.py
+++ b/Lib/test/test_concurrent_futures/executor.py
@@ -52,6 +52,9 @@ def test_map_exception(self):
         self.assertEqual(i.__next__(), (0, 1))
         self.assertEqual(i.__next__(), (0, 1))
         self.assertRaises(ZeroDivisionError, i.__next__)
+        self.assertEqual(i.__next__(), (0, 1))
+        self.assertRaises(StopIteration, i.__next__)
+        self.assertRaises(StopIteration, i.__next__)
 
     @support.requires_resource('walltime')
     def test_map_timeout(self):
diff --git a/Lib/test/test_concurrent_futures/test_thread_pool.py b/Lib/test/test_concurrent_futures/test_thread_pool.py
index 812f989d8f3ad2d..de30c6bf57bf0be 100644
--- a/Lib/test/test_concurrent_futures/test_thread_pool.py
+++ b/Lib/test/test_concurrent_futures/test_thread_pool.py
@@ -79,16 +79,13 @@ def log_n_wait(ident):
             # submit work to saturate the pool
             fut = pool.submit(log_n_wait, ident="first")
             try:
-                with contextlib.closing(
-                    pool.map(log_n_wait, ["second", "third"], timeout=0)
-                ) as gen:
-                    with self.assertRaises(TimeoutError):
-                        next(gen)
+                iterator = pool.map(log_n_wait, ["second"], timeout=0)
+                with self.assertRaises(TimeoutError):
+                    next(iterator)
             finally:
                 stop_event.set()
             fut.result()
         # ident='second' is cancelled as a result of raising a TimeoutError
-        # ident='third' is cancelled because it remained in the collection of futures
         self.assertListEqual(log, ["ident='first' started", "ident='first' stopped"])
 
 

From 2e0970e87f576a2e3f862a1c717b4f9a990a9e89 Mon Sep 17 00:00:00 2001
From: Serhiy Storchaka <storchaka@gmail.com>
Date: Sun, 4 Feb 2024 11:49:52 +0200
Subject: [PATCH 2/8] Fix _chain_from_iterable_of_lists().

---
 Lib/concurrent/futures/process.py | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
index 2f864157a64b1c3..979b8a45d7f90ef 100644
--- a/Lib/concurrent/futures/process.py
+++ b/Lib/concurrent/futures/process.py
@@ -627,9 +627,6 @@ def _chain_from_iterable_of_lists(iterable):
     careful not to keep references to yielded objects.
     """
     for element in iterable:
-        if element.is_exception:
-            raise element.exception
-        element = element.value
         element.reverse()
         while element:
             yield element.pop()

From 1d0f9e4a55ef95b8e15c94bbf664309e3864c03c Mon Sep 17 00:00:00 2001
From: Serhiy Storchaka <storchaka@gmail.com>
Date: Sun, 4 Feb 2024 12:30:37 +0200
Subject: [PATCH 3/8] Remove _FutureResult and simplify code.

---
 Lib/concurrent/futures/_base.py   | 63 +++++--------------------------
 Lib/concurrent/futures/process.py |  9 ++---
 2 files changed, 13 insertions(+), 59 deletions(-)

diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py
index 03580cfc2dbfb40..b515ec307e0c82a 100644
--- a/Lib/concurrent/futures/_base.py
+++ b/Lib/concurrent/futures/_base.py
@@ -314,9 +314,9 @@ def wait(fs, timeout=None, return_when=ALL_COMPLETED):
 def _result_or_cancel(fut, timeout=None):
     try:
         try:
-            return _FutureResult.from_value(fut.result(timeout))
-        except Exception as e:
-            return _FutureResult.from_exception(e)
+            return (fut.result(timeout), None)
+        except BaseException as exc:
+            return (None, exc)
         finally:
             fut.cancel()
     finally:
@@ -569,45 +569,6 @@ def set_exception(self, exception):
     __class_getitem__ = classmethod(types.GenericAlias)
 
 
-class _FutureResult(object):
-    """
-    This is used to record the exception instead of throwing them.
-
-    _FutureResult must contain either the value of future or an exception
-    that was thrown during the computation of future. Use is_exception
-    property to determine which one it is.
-    """
-
-    def __init__(self, exception, value):
-        self._exception = exception
-        self._value = value
-
-    @classmethod
-    def from_exception(cls, exc):
-        return cls(exc, None)
-
-    @classmethod
-    def from_value(cls, value):
-        return cls(None, value)
-
-    @property
-    def exception(self):
-        if not self.is_exception:
-            raise RuntimeError("No exception thrown.")
-        return self._exception
-
-    @property
-    def value(self):
-        if self.is_exception:
-            raise RuntimeError(
-                "Cannot get result value because an exception was thrown.")
-        return self._value
-
-    @property
-    def is_exception(self):
-        return self._exception is not None
-
-
 class Executor(object):
     """This is an abstract base class for concrete asynchronous executors."""
 
@@ -644,9 +605,7 @@ def map(self, fn, *iterables, timeout=None, chunksize=1):
                 before the given timeout.
             Exception: If fn(*args) raises for any values.
         """
-        return _MapResultIterator.from_generator(
-            self._map(fn, *iterables, timeout=timeout)
-        )
+        return _MapResultIterator(self._map(fn, *iterables, timeout=timeout))
 
     def _map(self, fn, *iterables, timeout=None):
         if timeout is not None:
@@ -695,23 +654,19 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         return False
 
 
-class _MapResultIterator(object):
+class _MapResultIterator:
     """The iterator returned by map()."""
     def __init__(self, gen):
         self.gen = gen
 
-    @classmethod
-    def from_generator(cls, gen):
-        return cls(gen)
-
     def __iter__(self):
         return self
 
     def __next__(self):
-        result = next(self.gen)
-        if result.is_exception:
-            raise result.exception
-        return result.value
+        value, exc = next(self.gen)
+        if exc is not None:
+            raise exc
+        return value
 
 
 class BrokenExecutor(RuntimeError):
diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
index 979b8a45d7f90ef..6fbd7b0c15c3cfe 100644
--- a/Lib/concurrent/futures/process.py
+++ b/Lib/concurrent/futures/process.py
@@ -202,9 +202,9 @@ def _process_chunk(fn, chunk):
     results = []
     for args in chunk:
         try:
-            result = _base._FutureResult.from_value(fn(*args))
-        except Exception as e:
-            result = _base._FutureResult.from_exception(e)
+            result = (fn(*args), None)
+        except BaseException as exc:
+            result = (None, exc)
         results.append(result)
     return results
 
@@ -846,8 +846,7 @@ def map(self, fn, *iterables, timeout=None, chunksize=1):
         results = super().map(partial(_process_chunk, fn),
                               itertools.batched(zip(*iterables), chunksize),
                               timeout=timeout)
-        return _base._MapResultIterator.from_generator(
-            _chain_from_iterable_of_lists(results))
+        return _base._MapResultIterator(_chain_from_iterable_of_lists(results))
 
     def shutdown(self, wait=True, *, cancel_futures=False):
         with self._shutdown_lock:

From 2d147a15bd85e1dc3aed4dcef9317a4c03efab2d Mon Sep 17 00:00:00 2001
From: Serhiy Storchaka <storchaka@gmail.com>
Date: Sun, 4 Feb 2024 13:02:58 +0200
Subject: [PATCH 4/8] Add _MapResultIterator.close().

---
 Lib/concurrent/futures/_base.py                      | 3 +++
 Lib/test/test_concurrent_futures/test_thread_pool.py | 9 ++++++---
 2 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py
index b515ec307e0c82a..e66d0309e7e42a2 100644
--- a/Lib/concurrent/futures/_base.py
+++ b/Lib/concurrent/futures/_base.py
@@ -668,6 +668,9 @@ def __next__(self):
             raise exc
         return value
 
+    def close(self):
+        self.gen.close()
+
 
 class BrokenExecutor(RuntimeError):
     """
diff --git a/Lib/test/test_concurrent_futures/test_thread_pool.py b/Lib/test/test_concurrent_futures/test_thread_pool.py
index 081697d2a2d0740..9fa31ce69a63a02 100644
--- a/Lib/test/test_concurrent_futures/test_thread_pool.py
+++ b/Lib/test/test_concurrent_futures/test_thread_pool.py
@@ -79,13 +79,16 @@ def log_n_wait(ident):
             # submit work to saturate the pool
             fut = pool.submit(log_n_wait, ident="first")
             try:
-                iterator = pool.map(log_n_wait, ["second"], timeout=0)
-                with self.assertRaises(TimeoutError):
-                    next(iterator)
+                with contextlib.closing(
+                    pool.map(log_n_wait, ["second", "third"], timeout=1)
+                ) as gen:
+                    with self.assertRaises(TimeoutError):
+                        next(gen)
             finally:
                 stop_event.set()
             fut.result()
         # ident='second' is cancelled as a result of raising a TimeoutError
+        # ident='third' is cancelled because it remained in the collection of futures
         self.assertListEqual(log, ["ident='first' started", "ident='first' stopped"])
 
 

From a2c0f7e1d1e8ae5ebf506ec80e517731278f9a05 Mon Sep 17 00:00:00 2001
From: Serhiy Storchaka <storchaka@gmail.com>
Date: Sun, 4 Feb 2024 13:16:34 +0200
Subject: [PATCH 5/8] Add more tests.

---
 Lib/test/test_concurrent_futures/executor.py  | 29 ++++++++++++++++---
 .../test_thread_pool.py                       |  2 +-
 2 files changed, 26 insertions(+), 5 deletions(-)

diff --git a/Lib/test/test_concurrent_futures/executor.py b/Lib/test/test_concurrent_futures/executor.py
index 330145b82d5a784..1799e70f8f580d1 100644
--- a/Lib/test/test_concurrent_futures/executor.py
+++ b/Lib/test/test_concurrent_futures/executor.py
@@ -48,11 +48,19 @@ def test_map(self):
                 list(map(pow, range(10), range(10))))
 
     def test_map_exception(self):
-        i = self.executor.map(divmod, [1, 1, 1, 1], [2, 3, 0, 5])
-        self.assertEqual(i.__next__(), (0, 1))
-        self.assertEqual(i.__next__(), (0, 1))
+        i = self.executor.map(divmod, [5, 5, 5, 5], [2, 3, 0, 5])
+        self.assertEqual(i.__next__(), (2, 1))
+        self.assertEqual(i.__next__(), (1, 2))
         self.assertRaises(ZeroDivisionError, i.__next__)
-        self.assertEqual(i.__next__(), (0, 1))
+        self.assertEqual(i.__next__(), (1, 0))
+        self.assertRaises(StopIteration, i.__next__)
+        self.assertRaises(StopIteration, i.__next__)
+
+        i = self.executor.map(divmod, [5, 5, 5, 5], [2, 0, 3, 5], chunksize=3)
+        self.assertEqual(i.__next__(), (2, 1))
+        self.assertRaises(ZeroDivisionError, i.__next__)
+        self.assertEqual(i.__next__(), (1, 2))
+        self.assertEqual(i.__next__(), (1, 0))
         self.assertRaises(StopIteration, i.__next__)
         self.assertRaises(StopIteration, i.__next__)
 
@@ -71,6 +79,19 @@ def test_map_timeout(self):
 
         self.assertEqual([None, None], results)
 
+    def test_map_close(self):
+        i = self.executor.map(divmod, [5, 5, 5, 5], [2, 0, 3, 5])
+        self.assertEqual(i.__next__(), (2, 1))
+        i.close()
+        self.assertRaises(StopIteration, i.__next__)
+        self.assertRaises(StopIteration, i.__next__)
+
+        i = self.executor.map(divmod, [5, 5, 5, 5], [2, 0, 3, 5], chunksize=3)
+        self.assertEqual(i.__next__(), (2, 1))
+        i.close()
+        self.assertRaises(StopIteration, i.__next__)
+        self.assertRaises(StopIteration, i.__next__)
+
     def test_shutdown_race_issue12456(self):
         # Issue #12456: race condition at shutdown where trying to post a
         # sentinel in the call queue blocks (the queue is full while processes
diff --git a/Lib/test/test_concurrent_futures/test_thread_pool.py b/Lib/test/test_concurrent_futures/test_thread_pool.py
index 9fa31ce69a63a02..5926a632aa4bec2 100644
--- a/Lib/test/test_concurrent_futures/test_thread_pool.py
+++ b/Lib/test/test_concurrent_futures/test_thread_pool.py
@@ -80,7 +80,7 @@ def log_n_wait(ident):
             fut = pool.submit(log_n_wait, ident="first")
             try:
                 with contextlib.closing(
-                    pool.map(log_n_wait, ["second", "third"], timeout=1)
+                    pool.map(log_n_wait, ["second", "third"], timeout=0)
                 ) as gen:
                     with self.assertRaises(TimeoutError):
                         next(gen)

From 488abe6b6be3a51ad054acb0fe4abe3816b49131 Mon Sep 17 00:00:00 2001
From: Serhiy Storchaka <storchaka@gmail.com>
Date: Sun, 4 Feb 2024 14:27:11 +0200
Subject: [PATCH 6/8] Update docs and cancel all on timeout.

---
 Doc/library/concurrent.futures.rst            | 13 ++++-
 Doc/whatsnew/3.13.rst                         |  5 ++
 Lib/concurrent/futures/_base.py               |  2 +
 Lib/test/test_concurrent_futures/executor.py  | 54 +++++++++----------
 ...-02-04-13-56-48.gh-issue-108518.6NCPk_.rst |  3 ++
 5 files changed, 46 insertions(+), 31 deletions(-)
 create mode 100644 Misc/NEWS.d/next/Library/2024-02-04-13-56-48.gh-issue-108518.6NCPk_.rst

diff --git a/Doc/library/concurrent.futures.rst b/Doc/library/concurrent.futures.rst
index d3c7a40aa9d390d..481615218c9359b 100644
--- a/Doc/library/concurrent.futures.rst
+++ b/Doc/library/concurrent.futures.rst
@@ -51,11 +51,18 @@ Executor Objects
       The returned iterator raises a :exc:`TimeoutError`
       if :meth:`~iterator.__next__` is called and the result isn't available
       after *timeout* seconds from the original call to :meth:`Executor.map`.
-      *timeout* can be an int or a float.  If *timeout* is not specified or
+      *timeout* can be an int or a float.
+      It cancels all future calls of *fn* and closes the iterator.
+      If *timeout* is not specified or
       ``None``, there is no limit to the wait time.
 
       If a *fn* call raises an exception, then that exception will be
       raised when its value is retrieved from the iterator.
+      It does not cancel future calls of *fn*.
+
+      The returned iterator has method :meth:`!close` which cancels all
+      future calls of *fn* and discards the results of already finished calls
+      if they are available.
 
       When using :class:`ProcessPoolExecutor`, this method chops *iterables*
       into a number of chunks which it submits to the pool as separate
@@ -68,6 +75,10 @@ Executor Objects
       .. versionchanged:: 3.5
          Added the *chunksize* argument.
 
+      .. versionchanged:: 3.13
+         The returned iterator no longer automatically closed if a *fn* call
+         raises an exception.
+
    .. method:: shutdown(wait=True, *, cancel_futures=False)
 
       Signal the executor that it should free any resources that it is using
diff --git a/Doc/whatsnew/3.13.rst b/Doc/whatsnew/3.13.rst
index f17c6ec0775beff..258dd7ba35f5bcb 100644
--- a/Doc/whatsnew/3.13.rst
+++ b/Doc/whatsnew/3.13.rst
@@ -144,6 +144,11 @@ Other Language Changes
   (Contributed by Levi Sabah, Zackery Spytz and Hugo van Kemenade in
   :gh:`73965`.)
 
+* The iterator returned by :meth:`concurrent.futures.Executor.map` is no longer
+  automatically closed if a function call raises an exception.
+  Use method :meth:`!close` to explicitly close the iterator.
+  (Contributed by xzmeng and Serhiy Storchaka in :gh:`108518`.)
+
 New Modules
 ===========
 
diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py
index e66d0309e7e42a2..b618f70a0e3e2b0 100644
--- a/Lib/concurrent/futures/_base.py
+++ b/Lib/concurrent/futures/_base.py
@@ -315,6 +315,8 @@ def _result_or_cancel(fut, timeout=None):
     try:
         try:
             return (fut.result(timeout), None)
+        except TimeoutError:
+            raise
         except BaseException as exc:
             return (None, exc)
         finally:
diff --git a/Lib/test/test_concurrent_futures/executor.py b/Lib/test/test_concurrent_futures/executor.py
index 1799e70f8f580d1..f864bf43ac12543 100644
--- a/Lib/test/test_concurrent_futures/executor.py
+++ b/Lib/test/test_concurrent_futures/executor.py
@@ -49,48 +49,42 @@ def test_map(self):
 
     def test_map_exception(self):
         i = self.executor.map(divmod, [5, 5, 5, 5], [2, 3, 0, 5])
-        self.assertEqual(i.__next__(), (2, 1))
-        self.assertEqual(i.__next__(), (1, 2))
-        self.assertRaises(ZeroDivisionError, i.__next__)
-        self.assertEqual(i.__next__(), (1, 0))
-        self.assertRaises(StopIteration, i.__next__)
-        self.assertRaises(StopIteration, i.__next__)
+        self.assertEqual(next(i), (2, 1))
+        self.assertEqual(next(i), (1, 2))
+        self.assertRaises(ZeroDivisionError, next, i)
+        self.assertEqual(next(i), (1, 0))
+        self.assertRaises(StopIteration, next, i)
+        self.assertRaises(StopIteration, next, i)
 
         i = self.executor.map(divmod, [5, 5, 5, 5], [2, 0, 3, 5], chunksize=3)
-        self.assertEqual(i.__next__(), (2, 1))
-        self.assertRaises(ZeroDivisionError, i.__next__)
-        self.assertEqual(i.__next__(), (1, 2))
-        self.assertEqual(i.__next__(), (1, 0))
-        self.assertRaises(StopIteration, i.__next__)
-        self.assertRaises(StopIteration, i.__next__)
+        self.assertEqual(next(i), (2, 1))
+        self.assertRaises(ZeroDivisionError, next, i)
+        self.assertEqual(next(i), (1, 2))
+        self.assertEqual(next(i), (1, 0))
+        self.assertRaises(StopIteration, next, i)
+        self.assertRaises(StopIteration, next, i)
 
     @support.requires_resource('walltime')
     def test_map_timeout(self):
-        results = []
-        try:
-            for i in self.executor.map(time.sleep,
-                                       [0, 0, 6],
-                                       timeout=5):
-                results.append(i)
-        except futures.TimeoutError:
-            pass
-        else:
-            self.fail('expected TimeoutError')
-
-        self.assertEqual([None, None], results)
+        i = self.executor.map(time.sleep, [0, 0, 6, 0], timeout=5)
+        next(i)
+        next(i)
+        self.assertRaises(futures.TimeoutError, next, i)
+        self.assertRaises(StopIteration, next, i)
+        self.assertRaises(StopIteration, next, i)
 
     def test_map_close(self):
         i = self.executor.map(divmod, [5, 5, 5, 5], [2, 0, 3, 5])
-        self.assertEqual(i.__next__(), (2, 1))
+        self.assertEqual(next(i), (2, 1))
         i.close()
-        self.assertRaises(StopIteration, i.__next__)
-        self.assertRaises(StopIteration, i.__next__)
+        self.assertRaises(StopIteration, next, i)
+        self.assertRaises(StopIteration, next, i)
 
         i = self.executor.map(divmod, [5, 5, 5, 5], [2, 0, 3, 5], chunksize=3)
-        self.assertEqual(i.__next__(), (2, 1))
+        self.assertEqual(next(i), (2, 1))
         i.close()
-        self.assertRaises(StopIteration, i.__next__)
-        self.assertRaises(StopIteration, i.__next__)
+        self.assertRaises(StopIteration, next, i)
+        self.assertRaises(StopIteration, next, i)
 
     def test_shutdown_race_issue12456(self):
         # Issue #12456: race condition at shutdown where trying to post a
diff --git a/Misc/NEWS.d/next/Library/2024-02-04-13-56-48.gh-issue-108518.6NCPk_.rst b/Misc/NEWS.d/next/Library/2024-02-04-13-56-48.gh-issue-108518.6NCPk_.rst
new file mode 100644
index 000000000000000..d7cf5ba88fa1868
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2024-02-04-13-56-48.gh-issue-108518.6NCPk_.rst
@@ -0,0 +1,3 @@
+The iterator returned by :meth:`concurrent.futures.Executor.map` is no longer
+automatically closed if a function call raises an exception.
+Use method :meth:`!close` to explicitly close the iterator.

From e021fbdceae97c47639c07cc83f6a2755b2ea0f9 Mon Sep 17 00:00:00 2001
From: Serhiy Storchaka <storchaka@gmail.com>
Date: Tue, 11 Aug 2026 15:40:34 +0300
Subject: [PATCH 7/8] Ignore the fork-in-thread deprecation warning in
 test_map_close

---
 Lib/test/test_concurrent_futures/executor.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/Lib/test/test_concurrent_futures/executor.py b/Lib/test/test_concurrent_futures/executor.py
index 8ced3cd87432b81..5d9f27c83bf9a81 100644
--- a/Lib/test/test_concurrent_futures/executor.py
+++ b/Lib/test/test_concurrent_futures/executor.py
@@ -108,6 +108,7 @@ def test_map_timeout(self):
         self.assertRaises(StopIteration, next, i)
         self.assertRaises(StopIteration, next, i)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_close(self):
         i = self.executor.map(divmod, [5, 5, 5, 5], [2, 0, 3, 5])
         self.assertEqual(next(i), (2, 1))

From 17b229aa015baf5442e5ef9771fb7f77e0796a9d Mon Sep 17 00:00:00 2001
From: Serhiy Storchaka <storchaka@gmail.com>
Date: Tue, 11 Aug 2026 15:48:34 +0300
Subject: [PATCH 8/8] close() is not a new method of the iterator returned by
 map()

---
 Doc/library/concurrent.futures.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Doc/library/concurrent.futures.rst b/Doc/library/concurrent.futures.rst
index 8f84434766c9d23..1f34dc66228e308 100644
--- a/Doc/library/concurrent.futures.rst
+++ b/Doc/library/concurrent.futures.rst
@@ -91,7 +91,7 @@ Executor Objects
 
       .. versionchanged:: next
          The returned iterator is no longer automatically closed if a *fn*
-         call raises an exception.  Added its :meth:`!close` method.
+         call raises an exception.
 
    .. method:: shutdown(wait=True, *, cancel_futures=False)