gh-103061: multiprocessing.Pool: avoid hang when result deserialization fails

Lib/multiprocessing/pool.py

@@ -158,6 +158,19 @@ def __init__(self, /, *args, notifier=None, **kwds):
         self.notifier = notifier
         super().__init__(*args, **kwds)
 
+        self._cache_failed = False
+        self._cache_failed_reason = None
+
+    def _disable_cache(self, exec):
+        self._cache_failed = True
+        self._cache_failed_reason = exec
+
+    def __setitem__(self, key, value):
+        if self._cache_failed:
+            raise RuntimeError("Pool cache is disabled due to previous error") \
+                from self._cache_failed_reason
+        super().__setitem__(key, value)
+
     def __delitem__(self, item):
         super().__delitem__(item)
 
@@ -572,13 +585,30 @@ def _handle_tasks(taskqueue, put, outqueue, pool, cache):
 
     @staticmethod
     def _handle_results(outqueue, get, cache):
+        def _handle_results_failure(cache, e):
+            exc = RuntimeError("Result handler failed to get result from worker and " +
+                               "unable to recover. " +
+                               "This is likely due to a worker process return or raise " +
+                               "an unpicklable object.")
+            exc.__cause__ = e
+            cache._disable_cache(exc)
+            _cache = cache.copy()
+            for value in _cache.values():
+                if isinstance(value, ApplyResult):
+                    chunk_number_left = getattr(value, '_number_left', 1)
+                    for _ in range(chunk_number_left):
+                        value._set(None, (False, exc))
+                elif isinstance(value, IMapIterator):
+                    value._set_length(value._index + 1)
+                    value._set(value._index, (False, exc))
+
         thread = threading.current_thread()
 
         while 1:
             try:
                 task = get()
-            except (OSError, EOFError):
-                util.debug('result handler got EOFError/OSError -- exiting')
+            except Exception as e:
+                _handle_results_failure(cache, e)
                 return
 
             if thread._state != RUN:
@@ -600,8 +630,8 @@ def _handle_results(outqueue, get, cache):
         while cache and thread._state != TERMINATE:
             try:
                 task = get()
-            except (OSError, EOFError):
-                util.debug('result handler got EOFError/OSError -- exiting')
+            except Exception as e:
+                _handle_results_failure(cache, e)
                 return
 
             if task is None:

Misc/NEWS.d/next/Library/2026-03-05-11-13-07.gh-issue-103061.P_ldzn.rst

@@ -0,0 +1 @@
+Fix :code:`multiprocessing.Pool` hang when a unpicklable object is returned