gh-83371: handle exceptions from user-supplied callbacks in process pools

User-supplied callbacks are called from an internal pool management thread. At
present any exceptions they raise are not caught and so propagate out and kill
the thread. This then causes problems for subsequent pool operations, including
joining the pool hanging.

As a QoL improvement, catch and handle any such exceptions using the system
exception hook. Thus by default details of the exception will be printed to
stderr, but the pool's integrity will remain intact.

Author: duaneg

Lib/multiprocessing/pool.py

@@ -17,6 +17,7 @@
 import itertools
 import os
 import queue
+import sys
 import threading
 import time
 import traceback
@@ -776,13 +777,20 @@ def get(self, timeout=None):
     def _set(self, i, obj):
         self._success, self._value = obj
         if self._callback and self._success:
-            self._callback(self._value)
+            self._handle_exceptions(self._callback, self._value)
         if self._error_callback and not self._success:
-            self._error_callback(self._value)
+            self._handle_exceptions(self._error_callback, self._value)
         self._event.set()
         del self._cache[self._job]
         self._pool = None
 
+    @staticmethod
+    def _handle_exceptions(callback, args):
+        try:
+            return callback(args)
+        except Exception as e:
+            sys.excepthook(*sys.exc_info())
+
     __class_getitem__ = classmethod(types.GenericAlias)
 
 AsyncResult = ApplyResult       # create alias -- see #17805
@@ -813,7 +821,7 @@ def _set(self, i, success_result):
             self._value[i*self._chunksize:(i+1)*self._chunksize] = result
             if self._number_left == 0:
                 if self._callback:
-                    self._callback(self._value)
+                    self._handle_exceptions(self._callback, self._value)
                 del self._cache[self._job]
                 self._event.set()
                 self._pool = None
@@ -825,7 +833,7 @@ def _set(self, i, success_result):
             if self._number_left == 0:
                 # only consider the result ready once all jobs are done
                 if self._error_callback:
-                    self._error_callback(self._value)
+                    self._handle_exceptions(self._error_callback, self._value)
                 del self._cache[self._job]
                 self._event.set()
                 self._pool = None