[3.13] gh-101267: ProcessPoolExecutor no longer shares 1 BrokenProcessPool exception...

[3.14] gh-101267: ProcessPoolExecutor no longer shares 1 BrokenProcessPool exception among all failed futures (GH-101268) (GH-151431)

* gh-101267: ProcessPoolExecutor no longer shares 1 BrokenProcessPool exception among all failed futures (GH-101268)
(cherry picked from commit 3c00ebc2bbd902495b163def850bc931420209fc)

(cherry picked from commit 4e8c9c6a5742659ca9f77719178802cce06a3507)

* Drop the abrupt-exit-code reporting from the 3.14 backport

Reporting the exit codes of processes that died without a known cause
is a new feature, not part of the gh-101267 bugfix. Keep only the
bugfix on 3.14: each failed future gets its own BrokenProcessPool
exception instead of one shared instance.

---------
(cherry picked from commit 27ff2c8c53e29e5075cba7d393f593a3a682a2f3)

Co-authored-by: Gregory P. Smith <68491+gpshead@users.noreply.github.com>
Co-authored-by: Daniel Shields <daniel.shields@twosigma.com>
Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>

diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
index 8ff717f5e003bfb44ae6111ad56d9a965385a8c2..ad718bf92fe2ac9f2b99ceafdb4df3a899401623 100644 (file)
--- a/Lib/concurrent/futures/process.py
+++ b/Lib/concurrent/futures/process.py
@@ -469,17 +469,20 @@ class _ExecutorManagerThread(threading.Thread):
             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.")
+        # All pending tasks are to be marked failed with a
+        # BrokenProcessPool error, as separate instances to avoid sharing
+        # a traceback (gh-101267).
+        cause_tb = None
         if cause is not None:
-            bpe.__cause__ = _RemoteTraceback(
-                f"\n'''\n{''.join(cause)}'''")
+            cause_tb = 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_tb is not None:
+                bpe.__cause__ = _RemoteTraceback(cause_tb)
             try:
                 work_item.future.set_exception(bpe)
             except _base.InvalidStateError:

diff --git a/Lib/test/test_concurrent_futures/test_process_pool.py b/Lib/test/test_concurrent_futures/test_process_pool.py
index 8b1bdaa33d8f5c4e3eaaf3053bd407f288c9213f..7cddcd0d9328453867c8da6e4f9c8f9946255837 100644 (file)
--- a/Lib/test/test_concurrent_futures/test_process_pool.py
+++ b/Lib/test/test_concurrent_futures/test_process_pool.py
@@ -2,6 +2,7 @@ import os
 import sys
 import threading
 import time
+import traceback
 import unittest
 from concurrent import futures
 from concurrent.futures.process import BrokenProcessPool
@@ -43,6 +44,31 @@ class ProcessPoolExecutorTest(ExecutorTest):
         # 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.
+        event = self.create_event()
+        futures = [self.executor.submit(event.wait) for _ in range(3)]
+        p = next(iter(self.executor._processes.values()))
+        p.terminate()
+        for fut in futures:
+            # Don't use assertRaises(): it clears the traceback off exc.
+            try:
+                fut.result()
+            except BrokenProcessPool as exc:
+                tb = exc.__traceback__
+            else:
+                self.fail("BrokenProcessPool not raised")
+            count = sum(
+                1
+                for frame_summary in traceback.extract_tb(tb)
+                if frame_summary.filename == __file__
+            )
+            # This code file should appear exactly once in the traceback.
+            # A shared exception would accumulate a frame per result() call.
+            self.assertEqual(count, 1)
+
     def test_map_chunksize(self):
         def bad_map():
             list(self.executor.map(pow, range(40), range(40), chunksize=-1))

diff --git a/Misc/ACKS b/Misc/ACKS
index d7eb6f44300c0441d2701b26f1f419a61c0ce150..c1216b21e3732121c41dd57184d9d0693bb90c47 100644 (file)
--- a/Misc/ACKS
+++ b/Misc/ACKS
@@ -1711,6 +1711,7 @@ Charlie Shepherd
 Bruce Sherwood
 Gregory Shevchenko
 Hai Shi
+Daniel Shields
 Alexander Shigin
 Pete Shinners
 Michael Shiplett

diff --git a/Misc/NEWS.d/next/Library/2023-01-23-21-23-50.gh-issue-101267._f-cFH.rst b/Misc/NEWS.d/next/Library/2023-01-23-21-23-50.gh-issue-101267._f-cFH.rst
new file mode 100644 (file)
index 0000000..901a3fb
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2023-01-23-21-23-50.gh-issue-101267._f-cFH.rst
@@ -0,0 +1,7 @@
+When a worker process terminates unexpectedly,
+:class:`concurrent.futures.ProcessPoolExecutor` now sets a separate
+:exc:`~concurrent.futures.process.BrokenProcessPool` exception on each pending
+future instead of sharing a single instance among them all.  Sharing one
+exception produced malformed tracebacks: each
+:meth:`Future.result() <concurrent.futures.Future.result>` call re-raised the
+same object, appending another copy of the traceback to it.