FIX deadlock in test_shutdown_with_sys_exit_at_pickle (#438)

Author: ogrisel

CHANGES.md

@@ -1,5 +1,7 @@
 ### 3.5.0 - in development
 
+- Fix a random deadlock caused by a race condition at executor shutdown that
+  was observed on Linux and Windows. (#438)
 
 - Fix detection of the number of physical cores in
   `cpu_count(only_physical_cores=True)` on some Linux systems and recent

loky/process_executor.py

@@ -305,9 +305,11 @@ def __init__(
         pending_work_items=None,
         running_work_items=None,
         thread_wakeup=None,
+        shutdown_lock=None,
         reducers=None,
     ):
         self.thread_wakeup = thread_wakeup
+        self.shutdown_lock = shutdown_lock
         self.pending_work_items = pending_work_items
         self.running_work_items = running_work_items
         super().__init__(max_size, reducers=reducers, ctx=ctx)
@@ -336,7 +338,8 @@ def _on_queue_feeder_error(self, e, obj):
             if work_item is not None:
                 work_item.future.set_exception(raised_error)
                 del work_item
-            self.thread_wakeup.wakeup()
+            with self.shutdown_lock:
+                self.thread_wakeup.wakeup()
         else:
             super()._on_queue_feeder_error(e, obj)
 
@@ -1139,6 +1142,7 @@ def _setup_queues(self, job_reducers, result_reducers, queue_size=None):
             pending_work_items=self._pending_work_items,
             running_work_items=self._running_work_items,
             thread_wakeup=self._executor_manager_thread_wakeup,
+            shutdown_lock=self._shutdown_lock,
             reducers=job_reducers,
             ctx=self._context,
         )

tests/_executor_mixin.py

@@ -4,6 +4,7 @@
 import pytest
 import threading
 from time import sleep
+import random
 
 from loky import TimeoutError, get_reusable_executor
 from loky.backend import get_context
@@ -29,6 +30,10 @@ def initializer_event(event):
     global _test_event
     _test_event = event
 
+    # Inject some randomness in the initialization to reveal race conditions.
+    if random.random() < 0.2:
+        sleep(random.random() * 0.1)  # 0-100ms
+
 
 def _direct_children_with_cmdline(p):
     """Helper to fetch cmdline from children process list"""

tests/_test_process_executor.py

@@ -107,13 +107,21 @@ def test_run_after_shutdown(self):
 
     def test_shutdown_with_pickle_error(self):
         self.executor.shutdown()
-        with self.executor_type(max_workers=4) as e:
-            e.submit(id, ErrorAtPickle())
+        # Iterate a few times to catch deadlocks/race conditions in the
+        # executor shutdown.
+        for _ in range(5):
+            with self.executor_type(max_workers=4) as e:
+                with pytest.raises(PicklingError):
+                    e.submit(id, ErrorAtPickle()).result()
 
     def test_shutdown_with_sys_exit_at_pickle(self):
         self.executor.shutdown()
-        with self.executor_type(max_workers=4) as e:
-            e.submit(id, ExitAtPickle())
+        # Iterate a few times to catch deadlocks/race conditions in the
+        # executor shutdown.
+        for _ in range(5):
+            with self.executor_type(max_workers=4) as e:
+                with pytest.raises(PicklingError):
+                    e.submit(id, ExitAtPickle()).result()
 
     def test_interpreter_shutdown(self):
         # Free resources to avoid random timeout in CI