Skip to content

gh-128364: Fix flaky test_concurrent_futures.test_wait tests #130742

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Mar 6, 2025
Merged

gh-128364: Fix flaky test_concurrent_futures.test_wait tests #130742

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
eff5d73
gh-128364: Fix flaky `test_concurrent_futures.test_wait` tests
colesbury Mar 1, 2025
f3cc50f
Use self.get_context().Manager()
colesbury Mar 2, 2025
8eeaf63
Use threading_helper.join() and try/finally
colesbury Mar 5, 2025
0a94b54
Merge branch 'main' into gh-128364-flaky-futures-wait-v2
colesbury Mar 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 46 additions & 13 deletions Lib/test/test_concurrent_futures/test_wait.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import sys
import threading
import time
import unittest
from concurrent import futures
from test import support
Expand All @@ -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],
Expand All @@ -42,8 +42,12 @@ def test_first_completed(self):
self.assertEqual(set([future1]), done)
self.assertEqual(set([CANCELLED_FUTURE, future2]), not_done)

event.set()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use maybe try/finally to unblock the task on test failure? Same remark for other tests.

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],
Expand All @@ -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],
Expand All @@ -67,9 +84,14 @@ def test_first_exception(self):
self.assertEqual(set([future1, future2]), finished)
self.assertEqual(set([future3]), pending)

t.join()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use maybe threading_helper.join_thread(t) (from test.support.threading_helper) to not wait forever if something goes wrong.

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,
Expand All @@ -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)
Comment on lines +118 to +119
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if it could be more helpful, but what about using self.addCleanup(event.set) and self.addCleanup(future1.result) here?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Cleanup functions are called too late. They run after tearDown() where we shutdown the executor and manager.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh right..


finished, pending = futures.wait(
[EXCEPTION_FUTURE, future1],
Expand All @@ -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)
Expand All @@ -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,
Expand All @@ -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):

Expand Down
21 changes: 21 additions & 0 deletions Lib/test/test_concurrent_futures/util.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import multiprocessing
import sys
import threading
import time
import unittest
from concurrent import futures
Expand Down Expand Up @@ -50,14 +51,19 @@ def setUp(self):
max_workers=self.worker_count,
mp_context=self.get_context(),
**self.executor_kwargs)
self.manager = self.get_context().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:
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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,
Expand Down
Loading