bpo-39098 Fix OSError: handle is closed in ProcessPoolExecutor on shutdown(wait=False)

When a ProcessPoolExecutor was created, a job added and then shutdown with
wait=False, an OSError: handle is closed error was raised by the ThreadWakeup
class. The threadwakeup should not be closed on shutdown if wait is False,
but in the shutdown_worker method.

Author: rad-pat

Lib/concurrent/futures/process.py

@@ -80,18 +80,23 @@
 
 class _ThreadWakeup:
     def __init__(self):
+        self._closed = False
         self._reader, self._writer = mp.Pipe(duplex=False)
 
     def close(self):
-        self._writer.close()
-        self._reader.close()
+        if not self._closed:
+            self._closed = True
+            self._writer.close()
+            self._reader.close()
 
     def wakeup(self):
-        self._writer.send_bytes(b"")
+        if not self._closed:
+            self._writer.send_bytes(b"")
 
     def clear(self):
-        while self._reader.poll():
-            self._reader.recv_bytes()
+        if not self._closed:
+            while self._reader.poll():
+                self._reader.recv_bytes()
 
 
 def _python_exit():
@@ -339,6 +344,7 @@ def shutdown_worker():
 
         # Release the queue's resources as soon as possible.
         call_queue.close()
+        thread_wakeup.close()
         # If .join() is not called on the created processes then
         # some ctx.Queue methods may deadlock on Mac OS X.
         for p in processes.values():
@@ -439,6 +445,7 @@ def shutdown_worker():
                 # this thread if there are no pending work items.
                 if not pending_work_items:
                     shutdown_worker()
+                    thread_wakeup = None
                     return
             except Full:
                 # This is not a problem: we will eventually be woken up (in
@@ -672,15 +679,16 @@ def shutdown(self, wait=True):
         # objects that use file descriptors.
         self._queue_management_thread = None
         if self._call_queue is not None:
-            self._call_queue.close()
             if wait:
+                self._call_queue.close()
                 self._call_queue.join_thread()
             self._call_queue = None
         self._result_queue = None
         self._processes = None
 
         if self._queue_management_thread_wakeup:
-            self._queue_management_thread_wakeup.close()
+            if wait:
+                self._queue_management_thread_wakeup.close()
             self._queue_management_thread_wakeup = None
 
     shutdown.__doc__ = _base.Executor.shutdown.__doc__

Lib/test/test_concurrent_futures.py

@@ -380,6 +380,14 @@ def test_del_shutdown(self):
         for t in threads:
             t.join()
 
+    def test_shutdown_no_wait(self):
+        executor = futures.ThreadPoolExecutor(max_workers=5)
+        executor.map(abs, range(-5, 5))
+        threads = executor._threads
+        executor.shutdown(wait=False)
+        for t in threads:
+            t.join()
+
     def test_thread_names_assigned(self):
         executor = futures.ThreadPoolExecutor(
             max_workers=5, thread_name_prefix='SpecialPool')
@@ -444,6 +452,21 @@ def test_del_shutdown(self):
             p.join()
         call_queue.join_thread()
 
+    def test_shutdown_no_wait(self):
+        executor = futures.ProcessPoolExecutor(max_workers=5)
+        list(executor.map(abs, range(-5, 5)))
+        processes = executor._processes
+        call_queue = executor._call_queue
+        queue_management_thread = executor._queue_management_thread
+        executor.shutdown(wait=False)
+
+        # Make sure that all the executor resources were properly cleaned by
+        # the shutdown process
+        queue_management_thread.join()
+        for p in processes.values():
+            p.join()
+        call_queue.join_thread()
+
 
 create_executor_tests(ProcessPoolShutdownTest,
                       executor_mixins=(ProcessPoolForkMixin,