gh-158497: Fix traceback loss and a reference cycle in Executor.map()

_process_chunk() returned a failed call's bare exception, so
ProcessPoolExecutor.map() lost the _RemoteTraceback cause that
_process_worker() attaches. Wrap it in _ExceptionWithTraceback.

_MapResultIterator.__next__() kept the exception it raised in a local,
so the exception referenced its own raising frame through its
traceback. Clear the local after raising, as Future.__get_result()
does.

Author: fedonman

Lib/concurrent/futures/_base.py

@@ -723,7 +723,10 @@ def __iter__(self):
     def __next__(self):
         value, exc = next(self.gen)
         if exc is not None:
-            raise exc
+            try:
+                raise exc
+            finally:
+                exc = None
         return value
 
     def close(self):

Lib/concurrent/futures/process.py

@@ -206,7 +206,7 @@ def _process_chunk(fn, chunk):
         try:
             result = (fn(*args), None)
         except BaseException as exc:
-            result = (None, exc)
+            result = (None, _ExceptionWithTraceback(exc, exc.__traceback__))
         results.append(result)
     return results
 

Lib/test/test_concurrent_futures/executor.py

@@ -1,6 +1,8 @@
+import gc
 import itertools
 import threading
 import time
+import types
 import weakref
 from concurrent import futures
 from operator import add
@@ -93,6 +95,22 @@ def test_map_exception(self):
         self.assertRaises(StopIteration, next, i)
         self.assertRaises(StopIteration, next, i)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    @support.cpython_only
+    def test_map_exception_refcycle(self):
+        # The iterator's frame that re-raises the exception must not keep
+        # a reference to it, or the exception and its traceback stay alive
+        # until the next garbage collection.
+        i = self.executor.map(raiser, [ValueError])
+        try:
+            next(i)
+        except ValueError as e:
+            exc = e
+        code = futures._base._MapResultIterator.__next__.__code__
+        frames = [r for r in gc.get_referrers(exc)
+                  if isinstance(r, types.FrameType) and r.f_code is code]
+        self.assertEqual(frames, [])
+
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_timeout_from_callable(self):
         # A TimeoutError from the callable is not the map() timeout, whether

Lib/test/test_concurrent_futures/test_process_pool.py

@@ -14,7 +14,7 @@
 from test.support import hashlib_helper, threading_helper, warnings_helper
 from test.test_importlib.metadata.fixtures import parameterize
 
-from .executor import ExecutorTest, mul
+from .executor import ExecutorTest, mul, raiser
 from .util import (
     ProcessPoolForkMixin, ProcessPoolForkserverMixin, ProcessPoolSpawnMixin,
     create_executor_tests, setup_module)
@@ -141,6 +141,16 @@ def test_traceback(self):
         self.assertIn('raise RuntimeError(123) # some comment',
                       f1.getvalue())
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_map_traceback(self):
+        # The traceback from the child process is also kept for map().
+        i = self.executor.map(raiser, [RuntimeError])
+        with self.assertRaises(RuntimeError) as cm:
+            next(i)
+        cause = cm.exception.__cause__
+        self.assertIs(type(cause), futures.process._RemoteTraceback)
+        self.assertIn('raise exception(msg)', cause.tb)
+
     def test_traceback_when_child_process_terminates_abruptly(self):
         # gh-139462 enhancement - BrokenProcessPool exceptions
         # should describe which process terminated.