Make test more consistent

Author: tfzee

Lib/test/test_concurrent_futures/test_deadlock.py

@@ -332,30 +332,39 @@ def clear(self):
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_shutdown_should_not_deadlock_if_result_pipe_full(self):
-        # Exiting workers put their pid on the result queue while shutdown()
-        # joins them. If nothing drains the queue, a full pipe blocks the
-        # workers forever. Fill the pipe once the executor manager thread
-        # stops reading results, right before the workers are told to exit.
+        # Workers must not write to the result queue on shutdown: the
+        # executor manager thread does not read it while it joins them.
+        # Fill the pipe once the executor manager thread stops reading
+        # results, right before the workers are told to exit.
         self.executor.shutdown(wait=True)
 
         data = b"a" * support.PIPE_MAX_SIZE
-        fillers = []
-        shutdown_workers = futures.process._ExecutorManagerThread.shutdown_workers
-        def mock_shutdown_workers(self):
+        join_executor_internals = (
+            futures.process._ExecutorManagerThread._join_executor_internals)
+        def mock_join_executor_internals(self, broken=False):
             # The put() blocks holding the result queue's write lock until
-            # the pipe is drained, so every exiting worker blocks behind it.
+            # the pipe is drained, so any worker writing to the result queue
+            # on exit blocks behind it.
             filler = threading.Thread(target=self.result_queue.put,
                                       args=(data,))
             filler.start()
-            fillers.append(filler)
-            shutdown_workers(self)
+            wlock = self.result_queue._wlock
+            if wlock is not None:
+                # Wait for the filler to hold the write lock.
+                while wlock.acquire(block=False):
+                    wlock.release()
+                    time.sleep(0.001)
+            join_executor_internals(self, broken)
+            # Unblock the filler.
+            self.result_queue.get()
+            filler.join()
 
         executor = self.executor_type(max_workers=2,
                                       mp_context=self.get_context())
         self.executor = executor  # Allow clean up in fail_on_deadlock
         with unittest.mock.patch.object(futures.process._ExecutorManagerThread,
-                                        'shutdown_workers',
-                                        mock_shutdown_workers):
+                                        '_join_executor_internals',
+                                        mock_join_executor_internals):
             self.assertEqual(list(executor.map(int, range(10))),
                              list(range(10)))
             shutdown = threading.Thread(target=executor.shutdown)
@@ -364,9 +373,6 @@ def mock_shutdown_workers(self):
             if shutdown.is_alive():
                 self._fail_on_deadlock(executor)
 
-        self.assertEqual(len(fillers), 1)
-        fillers[0].join()
-
 
 create_executor_tests(globals(), ExecutorDeadlockTest,
                       executor_mixins=(ProcessPoolForkMixin,