diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 7292128fb78870a..0967623dde18c68 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -1220,8 +1220,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) @@ -1342,8 +1344,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,)) @@ -1566,7 +1570,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'', l[0]) lock.release() @@ -1574,7 +1578,7 @@ def test_repr_lock(self): args=(lock,), name=tname) t.start() - time.sleep(0.1) + threading_helper.join_thread(t) self.assertEqual('', repr(lock)) lock.release() @@ -3175,25 +3179,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() @@ -3213,23 +3220,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")) @@ -3435,8 +3448,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. @@ -4030,7 +4046,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) @@ -4331,11 +4347,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.