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`.