[3.15] gh-158975: Replace fixed sleeps in multiprocessing tests (gh-158977)

Lib/test/_test_multiprocessing.py

@@ -1221,8 +1221,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)
@@ -1343,8 +1345,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,))
@@ -1567,15 +1571,15 @@ 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()
 
         t = threading.Thread(target=self._acquire,
                              args=(lock,),
                              name=tname)
         t.start()
-        time.sleep(0.1)
+        threading_helper.join_thread(t)
         self.assertEqual('<Lock(owner=SomeOtherThread)>', repr(lock))
         lock.release()
 
@@ -3251,8 +3255,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.
@@ -3744,7 +3751,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)
@@ -4045,11 +4052,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()

Misc/NEWS.d/next/Tests/2026-10-07-12-00-58.gh-issue-158975.yR8q2L.rst

@@ -0,0 +1,2 @@
+Replace fixed sleeps used for synchronization in multiprocessing tests with
+bounded waits to avoid failures on slow or overloaded systems.