fixup! pythongh-105829: Fix concurrent.futures.ProcessPoolExecutor deadlock

This reverts the previous fix and instead opts to remove the locking
completely when clearing the wakeup pipe.

We can do this because clear() and close() are both called from the
same thread and nowhere else.

In this version of this fix, the call to ProcessPoolExecutor.submit
can still block on the wakeup pipe if it happens to fill up. This
should not be an issue as there are already other cases where the
submit call can block and if the wakeup pipe is full it implies
there is already a lot of work items queued up.

Co-authored-by: Antoine Pitrou <[email protected]>

Author: elfstrom

Lib/concurrent/futures/process.py

@@ -69,27 +69,21 @@ class _ThreadWakeup:
     def __init__(self):
         self._closed = False
         self._reader, self._writer = mp.Pipe(duplex=False)
-        os.set_blocking(self._writer.fileno(), False)
 
     def close(self):
+        # Please note that we do not take the shutdown lock when
+        # calling clear() (to avoid deadlocking) so this method can
+        # only be called safely from the same thread as all calls to
+        # clear() even if you hold the shutdown lock. Otherwise we
+        # might try to read from the closed pipe.
         if not self._closed:
             self._closed = True
             self._writer.close()
             self._reader.close()
 
     def wakeup(self):
         if not self._closed:
-            try:
-                self._writer.send_bytes(b"")
-            except BlockingIOError:
-                # Assuming BlockingIOError is only raised when there
-                # is data in the pipe then we can skip the wake-up
-                # here because we are holding the shutdown_lock and
-                # the clear() call is also protected by this
-                # lock. This means the reader will wake up again (or
-                # is already awake) due to the existing data in the
-                # pipe.
-                pass
+            self._writer.send_bytes(b"")
 
     def clear(self):
         if not self._closed:
@@ -437,8 +431,12 @@ def wait_result_broken_or_wakeup(self):
         elif wakeup_reader in ready:
             is_broken = False
 
-        with self.shutdown_lock:
-            self.thread_wakeup.clear()
+        # No need to hold the _shutdown_lock here because:
+        # 1. we're the only thread to use the wakeup reader
+        # 2. we're also the only thread to call thread_wakeup.close()
+        # 3. we want to avoid a possible deadlock when both reader and writer
+        #    would block (gh-105829)
+        self.thread_wakeup.clear()
 
         return result_item, is_broken, cause
 
@@ -717,7 +715,10 @@ def __init__(self, max_workers=None, mp_context=None,
         # as it could result in a deadlock if a worker process dies with the
         # _result_queue write lock still acquired.
         #
-        # _shutdown_lock must be locked to access _ThreadWakeup.
+        # _shutdown_lock must be locked to access _ThreadWakeup.close() and
+        # .wakeup(). Care must also be taken to not call clear or close from
+        # more than one thread since _ThreadWakeup.clear() is not protected by
+        # the _shutdown_lock
         self._executor_manager_thread_wakeup = _ThreadWakeup()
 
         # Create communication channels for the executor