gh-83371: Fix deadlock when a Pool callback raises an exception

Lib/multiprocessing/pool.py

@@ -754,6 +754,16 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.terminate()
 
+def _chain_context(exc, context):
+    'Set context as the context of exc, avoiding a cycle.'
+    seen = {id(context)}
+    while exc is not None and id(exc) not in seen:
+        seen.add(id(exc))
+        if exc.__context__ is None:
+            exc.__context__ = context
+            return
+        exc = exc.__context__
+
 #
 # Class whose instances are returned by `Pool.apply_async()`
 #
@@ -791,13 +801,25 @@ 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)
-        if self._error_callback and not self._success:
-            self._error_callback(self._value)
-        self._event.set()
-        del self._cache[self._job]
-        self._pool = None
+        try:
+            if self._success:
+                if self._callback:
+                    self._callback(self._value)
+            else:
+                if self._error_callback:
+                    self._error_callback(self._value)
+        except BaseException as exc:
+            # A failed callback becomes the result of the job.  If it
+            # propagated, it would kill the result handler thread.
+            if not self._success:
+                # do not lose the original error
+                _chain_context(exc, self._value)
+            self._success = False
+            self._value = exc
+        finally:
+            self._event.set()
+            del self._cache[self._job]
+            self._pool = None
 
     __class_getitem__ = classmethod(types.GenericAlias)
 
@@ -828,23 +850,33 @@ def _set(self, i, success_result):
         if success and self._success:
             self._value[i*self._chunksize:(i+1)*self._chunksize] = result
             if self._number_left == 0:
-                if self._callback:
-                    self._callback(self._value)
-                del self._cache[self._job]
-                self._event.set()
-                self._pool = None
+                try:
+                    if self._callback:
+                        self._callback(self._value)
+                except BaseException as exc:
+                    self._success = False
+                    self._value = exc
+                finally:
+                    del self._cache[self._job]
+                    self._event.set()
+                    self._pool = None
         else:
             if not success and self._success:
                 # only store first exception
                 self._success = False
                 self._value = 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)
-                del self._cache[self._job]
-                self._event.set()
-                self._pool = None
+                try:
+                    if self._error_callback:
+                        self._error_callback(self._value)
+                except BaseException as exc:
+                    _chain_context(exc, self._value)
+                    self._value = exc
+                finally:
+                    del self._cache[self._job]
+                    self._event.set()
+                    self._pool = None
 
 #
 # Class whose instances are returned by `Pool.imap()`

Lib/test/_test_multiprocessing.py

@@ -3452,12 +3452,114 @@ def test_resource_warning(self):
             pool = None
             support.gc_collect()
 
+class CallbackError(Exception): pass
+
+class CallbackBaseException(BaseException): pass
+
 def raising():
     raise KeyError("key")
 
+def raising_map(x):
+    raise KeyError("key")
+
+def reraise(exc):
+    raise exc
+
+def raise_with_context(exc):
+    try:
+        raise ZeroDivisionError
+    except ZeroDivisionError:
+        raise CallbackError('callback failed')
+
 def unpickleable_result():
     return lambda: 42
 
+class _TestPoolCallbackErrors(BaseTestCase):
+    ALLOWED_TYPES = ('processes', )
+
+    @staticmethod
+    def _raise(value):
+        raise CallbackError('callback failed')
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_apply_async_callback_raises(self):
+        with multiprocessing.Pool(1) as p:
+            res = p.apply_async(sqr, (7,), callback=self._raise)
+            with self.assertRaises(CallbackError):
+                res.get(support.SHORT_TIMEOUT)
+            # the pool is still usable
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+            self.assertTrue(p._result_handler.is_alive())
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_apply_async_callback_raises_base_exception(self):
+        def raise_base(value):
+            raise CallbackBaseException
+        with multiprocessing.Pool(1) as p:
+            res = p.apply_async(sqr, (7,), callback=raise_base)
+            with self.assertRaises(CallbackBaseException):
+                res.get(support.SHORT_TIMEOUT)
+            # the pool did not hang
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_apply_async_error_callback_raises(self):
+        with multiprocessing.Pool(1) as p:
+            res = p.apply_async(raising, error_callback=self._raise)
+            with self.assertRaises(CallbackError) as cm:
+                res.get(support.SHORT_TIMEOUT)
+            # the original error is not lost
+            self.assertIsInstance(cm.exception.__context__, KeyError)
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_apply_async_error_callback_reraises(self):
+        with multiprocessing.Pool(1) as p:
+            res = p.apply_async(raising, error_callback=reraise)
+            with self.assertRaises(KeyError) as cm:
+                res.get(support.SHORT_TIMEOUT)
+            # the error is not its own context
+            self.assertIsNone(cm.exception.__context__)
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_map_async_error_callback_reraises(self):
+        with multiprocessing.Pool(1) as p:
+            res = p.map_async(raising_map, [0], error_callback=reraise)
+            with self.assertRaises(KeyError) as cm:
+                res.get(support.SHORT_TIMEOUT)
+            self.assertIsNone(cm.exception.__context__)
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_apply_async_error_callback_raises_with_context(self):
+        # the original error is kept at the end of the context chain
+        with multiprocessing.Pool(1) as p:
+            res = p.apply_async(raising, error_callback=raise_with_context)
+            with self.assertRaises(CallbackError) as cm:
+                res.get(support.SHORT_TIMEOUT)
+            context = cm.exception.__context__
+            self.assertIsInstance(context, ZeroDivisionError)
+            self.assertIsInstance(context.__context__, KeyError)
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_map_async_callback_raises(self):
+        with multiprocessing.Pool(1) as p:
+            res = p.map_async(sqr, list(range(3)), callback=self._raise)
+            with self.assertRaises(CallbackError):
+                res.get(support.SHORT_TIMEOUT)
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+    def test_map_async_error_callback_raises(self):
+        with multiprocessing.Pool(1) as p:
+            res = p.map_async(raising_map, [0], error_callback=self._raise)
+            with self.assertRaises(CallbackError) as cm:
+                res.get(support.SHORT_TIMEOUT)
+            self.assertIsInstance(cm.exception.__context__, KeyError)
+            self.assertEqual(p.apply(sqr, (3,)), 9)
+
 class _TestPoolWorkerErrors(BaseTestCase):
     ALLOWED_TYPES = ('processes', )
 

Misc/NEWS.d/next/Library/2026-08-14-10-49-45.gh-issue-83371.Km4Rt7.rst

@@ -0,0 +1,6 @@
+Fix a deadlock in :class:`multiprocessing.pool.Pool` when *callback* or
+*error_callback* raises an exception.
+It killed the thread which handles results, so that the pool hung forever.
+The exception is now the result of the job,
+as an error raised while iterating the input,
+and is raised by :meth:`!AsyncResult.get`.