gh-74028: concurrent.futures.Executor.map: avoid reference cycles when an exception is raised (#131701)

Co-authored-by: Muse Code (Meta Muse Spark)
Co-authored-by: Thomas Grainger <tagrain@gmail.com>
Co-authored-by: Łukasz Langa <lukasz@langa.pl>

Author: 3 people

Lib/concurrent/futures/_base.py

@@ -723,7 +723,11 @@ def __iter__(self):
     def __next__(self):
         value, exc = next(self.gen)
         if exc is not None:
-            raise exc
+            try:
+                raise exc
+            finally:
+                # avoid a ref cycle where exc is captured in its traceback
+                exc = None
         return value
 
     def close(self):

Lib/concurrent/futures/interpreter.py

@@ -88,7 +88,11 @@ def run(self, task):
             if exc is None:
                 # The exception must have been not shareable.
                 raise  # re-raise
-            raise exc from wrapper
+            try:
+                raise exc from wrapper
+            finally:
+                # avoid a ref cycle where exc is captured in its traceback
+                exc = None
 
 
 class BrokenInterpreterPool(_thread.BrokenThreadPool):

Lib/test/test_concurrent_futures/executor.py

@@ -1,6 +1,8 @@
+import gc
 import itertools
 import threading
 import time
+import traceback
 import weakref
 from concurrent import futures
 from operator import add
@@ -80,7 +82,44 @@ def test_map_exception(self):
         i = self.executor.map(divmod, [5, 5, 5, 5], [2, 3, 0, 5])
         self.assertEqual(next(i), (2, 1))
         self.assertEqual(next(i), (1, 2))
-        self.assertRaises(ZeroDivisionError, next, i)
+
+        exception = None
+        try:
+            i.__next__()
+        except Exception as e:
+            exception = e
+        self.assertTrue(
+            isinstance(exception, ZeroDivisionError),
+            msg="should raise a ZeroDivisionError",
+        )
+
+        # Wait for the completing thread to release its transient reference
+        # to the exception; do not keep the referrers list alive in the loop.
+        for _ in support.sleeping_retry(support.SHORT_TIMEOUT, error=False):
+            if not gc.get_referrers(exception):
+                break
+
+        self.assertFalse(
+            gc.get_referrers(exception),
+            msg="the exception should not have any referrers",
+        )
+
+        self.assertFalse(
+            [
+                (var, val)
+                # go through the frames of the exception's traceback
+                for frame, _ in traceback.walk_tb(exception.__traceback__)
+                # skipping the current frame
+                if frame is not exception.__traceback__.tb_frame
+                # go through the locals captured in that frame
+                for var, val in frame.f_locals.items()
+                # check if one of them is an exception
+                if isinstance(val, Exception)
+                # check if it is captured in its own traceback
+                and frame is val.__traceback__.tb_frame
+            ],
+            msg=f"the exception's traceback should not contain an exception captured in its own traceback",
+        )
         self.assertEqual(next(i), (1, 0))
         self.assertRaises(StopIteration, next, i)
         self.assertRaises(StopIteration, next, i)
@@ -212,7 +251,7 @@ def test_map_buffersize_when_buffer_is_full(self):
         self.assertEqual(
             next(ints),
             buffersize,
-            msg="should have fetched only `buffersize` elements from `ints`.",
+            msg="should have fetched only `buffersize` elements from `ints`",
         )
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()