diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py index e728b8e0a91f744..2d6557cd549389f 100644 --- a/Lib/concurrent/futures/_base.py +++ b/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): diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py index 7f4f225c0ad4fbd..c55bda8d402eff4 100644 --- a/Lib/concurrent/futures/process.py +++ b/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 diff --git a/Lib/test/test_concurrent_futures/executor.py b/Lib/test/test_concurrent_futures/executor.py index ff7bd0db0c2199c..681dc83b637c6e3 100644 --- a/Lib/test/test_concurrent_futures/executor.py +++ b/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 diff --git a/Lib/test/test_concurrent_futures/test_process_pool.py b/Lib/test/test_concurrent_futures/test_process_pool.py index dafbda862c51c24..9ea959f88d80d16 100644 --- a/Lib/test/test_concurrent_futures/test_process_pool.py +++ b/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.