From 91bc1ae732368be0f997f2e83a0e501367714e05 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Fri, 14 Aug 2026 11:00:02 +0300 Subject: [PATCH] gh-83371: Fix deadlock when a Pool callback raises an exception MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- Lib/multiprocessing/pool.py | 66 +++++++++--- Lib/test/_test_multiprocessing.py | 102 ++++++++++++++++++ ...6-08-14-10-49-45.gh-issue-83371.Km4Rt7.rst | 6 ++ 3 files changed, 157 insertions(+), 17 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2026-08-14-10-49-45.gh-issue-83371.Km4Rt7.rst diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index 8fd0f98a02dd3a6..e58958f21255379 100644 --- a/Lib/multiprocessing/pool.py +++ b/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,11 +850,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 @@ -840,11 +867,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 ba1c0de5d283323..2f90f44cf404fe7 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/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', ) 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`.