https://github.com/python/cpython/commit/72a88d71c042be97921057935b5dfe3a9bc16daa commit: 72a88d71c042be97921057935b5dfe3a9bc16daa branch: main author: Serhiy Storchaka <[email protected]> committer: serhiy-storchaka <[email protected]> date: 2026-08-29T20:53:36+03:00 summary:
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 <[email protected]> Co-authored-by: Thomas Grainger <[email protected]> files: A Misc/NEWS.d/next/Library/2026-08-14-10-49-45.gh-issue-83371.Km4Rt7.rst M Lib/multiprocessing/pool.py M Lib/test/_test_multiprocessing.py diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index f50bcbe4451bea4..ef9460ac0aa6817 100644 --- a/Lib/multiprocessing/pool.py +++ b/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,11 +859,16 @@ 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 @@ -849,11 +876,16 @@ def _set(self, i, success_result): 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()` diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index e5f618f5f2e84f4..46ed8843fcd0519 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/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', ) diff --git a/Misc/NEWS.d/next/Library/2026-08-14-10-49-45.gh-issue-83371.Km4Rt7.rst b/Misc/NEWS.d/next/Library/2026-08-14-10-49-45.gh-issue-83371.Km4Rt7.rst new file mode 100644 index 000000000000000..ca30859c4a49c5c --- /dev/null +++ b/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`. _______________________________________________ Python-checkins mailing list -- [email protected] To unsubscribe send an email to [email protected] https://mail.python.org/mailman3//lists/python-checkins.python.org Member address: [email protected]
