https://github.com/python/cpython/commit/75d96f8ade89a9aab9ff9f986c12c9ddaf105424
commit: 75d96f8ade89a9aab9ff9f986c12c9ddaf105424
branch: main
author: Neil Schemenauer <[email protected]>
committer: nascheme <[email protected]>
date: 2026-10-27T21:15:21-07:00
summary:
gh-158975: Replace fixed sleeps in multiprocessing tests (gh-158977)
Replace fixed sleeps used for synchronization in multiprocessing tests
with bounded waits to avoid failures on slow or overloaded systems.
Ensure pool cleanup runs even when assertions fail in the imap buffer
tests.
files:
A Misc/NEWS.d/next/Tests/2026-10-07-12-00-58.gh-issue-158975.yR8q2L.rst
M Lib/test/_test_multiprocessing.py
diff --git a/Lib/test/_test_multiprocessing.py
b/Lib/test/_test_multiprocessing.py
index d267c9d3dfa8a98..050d146f921fe95 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -1222,8 +1222,10 @@ def test_put(self):
queue.put(5, False, None)
queue.put_nowait(6)
- # the values may be in buffer but not yet in pipe so sleep a bit
- time.sleep(DELTA)
+ # The values may be in the buffer but not yet in the pipe.
+ for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+ if not queue_empty(queue):
+ break
self.assertEqual(queue_empty(queue), False)
self.assertEqual(queue_full(queue, MAXSIZE), True)
@@ -1344,8 +1346,10 @@ def test_fork(self):
for i in range(10):
queue.put(i)
- # wait to make sure thread starts before we fork a new process
- time.sleep(DELTA)
+ # Wait for the feeder thread to write to the pipe before forking.
+ for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+ if not queue_empty(queue):
+ break
# fork process
p = self.Process(target=self._test_fork, args=(queue,))
@@ -1568,7 +1572,7 @@ def test_repr_lock(self):
args=(lock, l),
name=tname)
t.start()
- time.sleep(0.1)
+ threading_helper.join_thread(t)
self.assertEqual(f'<Lock(owner=MainProcess|{tname})>', l[0])
lock.release()
@@ -1576,7 +1580,7 @@ def test_repr_lock(self):
args=(lock,),
name=tname)
t.start()
- time.sleep(0.1)
+ threading_helper.join_thread(t)
self.assertEqual('<Lock(owner=SomeOtherThread)>', repr(lock))
lock.release()
@@ -3177,25 +3181,28 @@ def produce_args():
last_produced_task_arg.value = arg
yield arg
- method = getattr(p, method_name)
- it = method(functools.partial(sqr, wait=0.2), produce_args())
+ def wait_for_produced_args(threshold):
+ for _ in support.sleeping_retry(support.SHORT_TIMEOUT,
error=False):
+ if last_produced_task_arg.value > threshold:
+ break
+ self.assertGreater(last_produced_task_arg.value, threshold)
- time.sleep(0.2)
- # `iterable` could've been advanced only `processes` times,
- # but in fact it advances further (`> processes`) because of
- # not waiting for workers or user code to catch up.
- self.assertGreater(last_produced_task_arg.value, processes)
+ try:
+ method = getattr(p, method_name)
+ it = method(functools.partial(sqr, wait=0.2), produce_args())
- next(it)
- time.sleep(0.2)
- self.assertGreater(last_produced_task_arg.value, processes + 1)
+ # Without a buffer limit, the iterable advances without waiting
+ # for workers or user code to catch up.
+ wait_for_produced_args(processes)
- next(it)
- time.sleep(0.2)
- self.assertGreater(last_produced_task_arg.value, processes + 2)
+ next(it)
+ wait_for_produced_args(processes + 1)
- p.terminate()
- p.join()
+ next(it)
+ wait_for_produced_args(processes + 2)
+ finally:
+ p.terminate()
+ p.join()
@unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes')
@warnings_helper.ignore_fork_in_thread_deprecation_warnings()
@@ -3215,23 +3222,29 @@ def produce_args():
last_produced_task_arg.value = arg
yield arg
- method = getattr(p, method_name)
- it = method(functools.partial(sqr, wait=0.2), produce_args(),
- buffersize=processes)
+ def wait_for_produced_args(expected):
+ # Consuming a result allows the task handler to advance the
+ # iterable, but does not wait for it to do so.
+ for _ in support.sleeping_retry(support.SHORT_TIMEOUT,
error=False):
+ if last_produced_task_arg.value >= expected:
+ break
+ self.assertEqual(last_produced_task_arg.value, expected)
- time.sleep(0.2)
- self.assertEqual(last_produced_task_arg.value, processes)
+ try:
+ method = getattr(p, method_name)
+ it = method(functools.partial(sqr, wait=0.2), produce_args(),
+ buffersize=processes)
- next(it)
- time.sleep(0.2)
- self.assertEqual(last_produced_task_arg.value, processes + 1)
+ wait_for_produced_args(processes)
- next(it)
- time.sleep(0.2)
- self.assertEqual(last_produced_task_arg.value, processes + 2)
+ next(it)
+ wait_for_produced_args(processes + 1)
- p.terminate()
- p.join()
+ next(it)
+ wait_for_produced_args(processes + 2)
+ finally:
+ p.terminate()
+ p.join()
@warnings_helper.ignore_fork_in_thread_deprecation_warnings()
@support.subTests('method_name', ("imap", "imap_unordered"))
@@ -3437,8 +3450,11 @@ def test_release_task_refs(self):
self.pool.map(identity, objs)
del objs
- time.sleep(DELTA) # let threaded cleanup code run
- support.gc_collect() # For PyPy or other GCs.
+ for _ in support.sleeping_retry(support.SHORT_TIMEOUT, error=False):
+ support.gc_collect() # For PyPy or other GCs.
+ if (all(wr() is None for wr in refs)
+ and CountedObject.n_instances == 0):
+ break
self.assertEqual(set(wr() for wr in refs), {None})
# With a process pool, copies of the objects are returned, check
# they were released too.
@@ -4032,7 +4048,7 @@ def test_connection(self):
self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1)
conn.send(None)
- time.sleep(.1)
+ self.assertTrue(conn.poll(support.SHORT_TIMEOUT))
self.assertEqual(poll(TIMEOUT1), True)
self.assertTimingAlmostEqual(poll.elapsed, 0)
@@ -4333,11 +4349,10 @@ def test_issue14725(self):
p = self.Process(target=self._test, args=(l.address,))
p.daemon = True
p.start()
- time.sleep(1)
- # On Windows the client process should by now have connected,
- # written data and closed the pipe handle by now. This causes
- # ConnectNamdedPipe() to fail with ERROR_NO_DATA. See Issue
- # 14725.
+ join_process(p)
+ # On Windows the client process has now connected, written data
+ # and closed the pipe handle. This causes ConnectNamedPipe()
+ # to fail with ERROR_NO_DATA. See Issue 14725.
conn = l.accept()
self.assertEqual(conn.recv(), 'hello')
conn.close()
diff --git
a/Misc/NEWS.d/next/Tests/2026-10-07-12-00-58.gh-issue-158975.yR8q2L.rst
b/Misc/NEWS.d/next/Tests/2026-10-07-12-00-58.gh-issue-158975.yR8q2L.rst
new file mode 100644
index 000000000000000..450af29aea272b7
--- /dev/null
+++ b/Misc/NEWS.d/next/Tests/2026-10-07-12-00-58.gh-issue-158975.yR8q2L.rst
@@ -0,0 +1,3 @@
+Replace fixed sleeps used for synchronization in multiprocessing tests with
+bounded waits to avoid failures on slow or overloaded systems. Ensure pool
+cleanup runs even when assertions fail in the imap buffer tests.
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
