gh-128364: Fix flaky test_concurrent_futures.test_wait tests

Use events instead of relying on `time.sleep()`. The tests are also now about
four times faster.

Author: colesbury

Lib/test/test_concurrent_futures/test_wait.py

@@ -1,6 +1,5 @@
 import sys
 import threading
-import time
 import unittest
 from concurrent import futures
 from test import support
@@ -16,24 +15,25 @@
 def mul(x, y):
     return x * y
 
-def sleep_and_raise(t):
-    time.sleep(t)
+def wait_and_raise(e):
+    e.wait()
     raise Exception('this is an exception')
 
 
 class WaitTests:
     def test_20369(self):
         # See https://bugs.python.org/issue20369
-        future = self.executor.submit(time.sleep, 1.5)
+        future = self.executor.submit(mul, 1, 2)
         done, not_done = futures.wait([future, future],
                             return_when=futures.ALL_COMPLETED)
         self.assertEqual({future}, done)
         self.assertEqual(set(), not_done)
 
 
     def test_first_completed(self):
+        event = self.create_event()
         future1 = self.executor.submit(mul, 21, 2)
-        future2 = self.executor.submit(time.sleep, 1.5)
+        future2 = self.executor.submit(event.wait)
 
         done, not_done = futures.wait(
                 [CANCELLED_FUTURE, future1, future2],
@@ -42,8 +42,12 @@ def test_first_completed(self):
         self.assertEqual(set([future1]), done)
         self.assertEqual(set([CANCELLED_FUTURE, future2]), not_done)
 
+        event.set()
+        future2.result()  # wait for job to finish
+
     def test_first_completed_some_already_completed(self):
-        future1 = self.executor.submit(time.sleep, 1.5)
+        event = self.create_event()
+        future1 = self.executor.submit(event.wait)
 
         finished, pending = futures.wait(
                  [CANCELLED_AND_NOTIFIED_FUTURE, SUCCESSFUL_FUTURE, future1],
@@ -54,11 +58,24 @@ def test_first_completed_some_already_completed(self):
                 finished)
         self.assertEqual(set([future1]), pending)
 
-    @support.requires_resource('walltime')
+        event.set()
+        future1.result()  # wait for job to finish
+
     def test_first_exception(self):
+        event1 = self.create_event()
+        event2 = self.create_event()
+
         future1 = self.executor.submit(mul, 2, 21)
-        future2 = self.executor.submit(sleep_and_raise, 1.5)
-        future3 = self.executor.submit(time.sleep, 3)
+        future2 = self.executor.submit(wait_and_raise, event1)
+        future3 = self.executor.submit(event2.wait)
+
+        # Ensure that future1 is completed before future2 finishes
+        def wait_for_future1():
+            future1.result()
+            event1.set()
+
+        t = threading.Thread(target=wait_for_future1)
+        t.start()
 
         finished, pending = futures.wait(
                 [future1, future2, future3],
@@ -67,9 +84,14 @@ def test_first_exception(self):
         self.assertEqual(set([future1, future2]), finished)
         self.assertEqual(set([future3]), pending)
 
+        t.join()
+        event2.set()
+        future3.result()  # wait for job to finish
+
     def test_first_exception_some_already_complete(self):
+        event = self.create_event()
         future1 = self.executor.submit(divmod, 21, 0)
-        future2 = self.executor.submit(time.sleep, 1.5)
+        future2 = self.executor.submit(event.wait)
 
         finished, pending = futures.wait(
                 [SUCCESSFUL_FUTURE,
@@ -83,8 +105,12 @@ def test_first_exception_some_already_complete(self):
                               future1]), finished)
         self.assertEqual(set([CANCELLED_FUTURE, future2]), pending)
 
+        event.set()
+        future2.result()  # wait for job to finish
+
     def test_first_exception_one_already_failed(self):
-        future1 = self.executor.submit(time.sleep, 2)
+        event = self.create_event()
+        future1 = self.executor.submit(event.wait)
 
         finished, pending = futures.wait(
                  [EXCEPTION_FUTURE, future1],
@@ -93,6 +119,9 @@ def test_first_exception_one_already_failed(self):
         self.assertEqual(set([EXCEPTION_FUTURE]), finished)
         self.assertEqual(set([future1]), pending)
 
+        event.set()
+        future1.result()  # wait for job to finish
+
     def test_all_completed(self):
         future1 = self.executor.submit(divmod, 2, 0)
         future2 = self.executor.submit(mul, 2, 21)
@@ -114,9 +143,9 @@ def test_all_completed(self):
 
     def test_timeout(self):
         short_timeout = 0.050
-        long_timeout = short_timeout * 10
 
-        future = self.executor.submit(time.sleep, long_timeout)
+        event = self.create_event()
+        future = self.executor.submit(event.wait)
 
         finished, pending = futures.wait(
                 [CANCELLED_AND_NOTIFIED_FUTURE,
@@ -132,6 +161,10 @@ def test_timeout(self):
                          finished)
         self.assertEqual(set([future]), pending)
 
+        # Set the event to allow the future to complete
+        event.set()
+        future.result()  # wait for job to finish
+
 
 class ThreadPoolWaitTests(ThreadPoolMixin, WaitTests, BaseTestCase):
 

Lib/test/test_concurrent_futures/util.py

@@ -1,5 +1,6 @@
 import multiprocessing
 import sys
+import threading
 import time
 import unittest
 from concurrent import futures
@@ -50,14 +51,19 @@ def setUp(self):
                 max_workers=self.worker_count,
                 mp_context=self.get_context(),
                 **self.executor_kwargs)
+            self.manager = multiprocessing.Manager()
         else:
             self.executor = self.executor_type(
                 max_workers=self.worker_count,
                 **self.executor_kwargs)
+            self.manager = None
 
     def tearDown(self):
         self.executor.shutdown(wait=True)
         self.executor = None
+        if self.manager is not None:
+            self.manager.shutdown()
+            self.manager = None
 
         dt = time.monotonic() - self.t1
         if support.verbose:
@@ -73,11 +79,17 @@ def get_context(self):
 class ThreadPoolMixin(ExecutorMixin):
     executor_type = futures.ThreadPoolExecutor
 
+    def create_event(self):
+        return threading.Event()
+
 
 @support.skip_if_sanitizer("gh-129824: data races in InterpreterPool tests", thread=True)
 class InterpreterPoolMixin(ExecutorMixin):
     executor_type = futures.InterpreterPoolExecutor
 
+    def create_event(self):
+        self.skipTest("InterpreterPoolExecutor doesn't support events")
+
 
 class ProcessPoolForkMixin(ExecutorMixin):
     executor_type = futures.ProcessPoolExecutor
@@ -94,6 +106,9 @@ def get_context(self):
             self.skipTest("TSAN doesn't support threads after fork")
         return super().get_context()
 
+    def create_event(self):
+        return self.manager.Event()
+
 
 class ProcessPoolSpawnMixin(ExecutorMixin):
     executor_type = futures.ProcessPoolExecutor
@@ -106,6 +121,9 @@ def get_context(self):
             self.skipTest("ProcessPoolExecutor unavailable on this system")
         return super().get_context()
 
+    def create_event(self):
+        return self.manager.Event()
+
 
 class ProcessPoolForkserverMixin(ExecutorMixin):
     executor_type = futures.ProcessPoolExecutor
@@ -122,6 +140,9 @@ def get_context(self):
             self.skipTest("TSAN doesn't support threads after fork")
         return super().get_context()
 
+    def create_event(self):
+        return self.manager.Event()
+
 
 def create_executor_tests(remote_globals, mixin, bases=(BaseTestCase,),
                           executor_mixins=(ThreadPoolMixin,