gh-83371: Fix deadlock when a Pool callback raises an exception (GH-155777)

The exception killed the thread which handles results, so that the pool
hung forever.  It is now the result of the job and is raised by
AsyncResult.get(), with the original error as its context.

Co-authored-by: Sindri Guðmundsson <sindrigudmundsson@gmail.com>
Co-authored-by: Thomas Grainger <tagrain@gmail.com>

Author: 3 people

Lib/multiprocessing/pool.py

@@ -763,6 +763,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()`
 #
@@ -800,13 +810,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)
 
@@ -837,23 +859,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

@@ -3474,12 +3474,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`.