ProcessPoolExecutor should not share one BrokenProcessPool exception among all failed futures (#101267)

Author: daniel-shields

Lib/concurrent/futures/process.py

@@ -471,17 +471,15 @@ def terminate_broken(self, cause):
             executor._shutdown_thread = True
             executor = None
 
-        # All pending tasks are to be marked failed with the following
-        # BrokenProcessPool error
-        bpe = BrokenProcessPool("A process in the process pool was "
-                                "terminated abruptly while the future was "
-                                "running or pending.")
-        if cause is not None:
-            bpe.__cause__ = _RemoteTraceback(
-                f"\n'''\n{''.join(cause)}'''")
-
         # Mark pending tasks as failed.
         for work_id, work_item in self.pending_work_items.items():
+            bpe = BrokenProcessPool(
+                "A process in the process pool was "
+                "terminated abruptly while the future was "
+                "running or pending.")
+            if cause is not None:
+                bpe.__cause__ = _RemoteTraceback(
+                    f"\n'''\n{''.join(cause)}'''")
             work_item.future.set_exception(bpe)
             # Delete references to object. See issue16284
             del work_item
@@ -762,7 +760,8 @@ def _spawn_process(self):
     def submit(self, fn, /, *args, **kwargs):
         with self._shutdown_lock:
             if self._broken:
-                raise BrokenProcessPool(self._broken)
+                raise BrokenProcessPool(
+                    'cannot schedule new futures after process pool has broken')
             if self._shutdown_thread:
                 raise RuntimeError('cannot schedule new futures after shutdown')
             if _global_shutdown:

Lib/test/test_concurrent_futures.py

@@ -15,6 +15,7 @@
 import os
 import queue
 import sys
+import traceback
 import threading
 import time
 import unittest
@@ -980,6 +981,27 @@ def test_killed_child(self):
         # Submitting other jobs fails as well.
         self.assertRaises(BrokenProcessPool, self.executor.submit, pow, 2, 8)
 
+    def test_broken_process_pool_traceback(self):
+        # When a child process is abruptly terminated, the whole pool gets
+        # "broken", and a BrokenProcessPool exception should be created
+        # for each future instead of sharing one exception among all futures.
+        futures = [self.executor.submit(time.sleep, 3) for _ in range(3)]
+        # Get one of the processes, and terminate (kill) it.
+        p = next(iter(self.executor._processes.values()))
+        p.terminate()
+        for fut in futures:
+            count = None
+            try:
+                fut.result()
+            except BrokenProcessPool as e:
+                count = sum(
+                    1
+                    for frame_summary in traceback.extract_tb(e.__traceback__)
+                    if frame_summary.filename == __file__
+                )
+            # This code file should appear exactly once in the traceback.
+            self.assertEqual(count, 1)
+
     def test_map_chunksize(self):
         def bad_map():
             list(self.executor.map(pow, range(40), range(40), chunksize=-1))

Misc/ACKS

@@ -1649,6 +1649,7 @@ Charlie Shepherd
 Bruce Sherwood
 Gregory Shevchenko
 Hai Shi
+Daniel Shields
 Alexander Shigin
 Pete Shinners
 Michael Shiplett