wip

Author: YvesDup

Lib/multiprocessing/synchronize.py

@@ -280,7 +280,10 @@ def wait(self, timeout=None):
                 self._woken_count.release()
             else:
                 # we timed out, so remove ourselves from the sleeping count
-                self._sleeping_count.acquire()
+                #  if possible.
+                if not self._sleeping_count.acquire(False):
+                    print('+')
+                    self._woken_count.release()
 
             # reacquire lock
             for i in range(count):
@@ -294,10 +297,13 @@ def notify(self, n=1):
 
         # to take account of timeouts since last notify*() we subtract
         # woken_count from sleeping_count and rezero woken_count
-        #while self._woken_count.acquire(False):
-        #    res = self._sleeping_count.acquire(False)
-        #    assert res, ('notify: Bug in sleeping_count.acquire'
-        #                 + '- res should not be False')
+        i = 0
+        while self._woken_count.acquire(False):
+            res = self._sleeping_count.acquire(False)
+            assert res, ('notify: Bug in sleeping_count.acquire'
+                         + '- res should not be False')
+            i += 1
+        print('.'*i, end='')
 
         sleepers = 0
         while sleepers < n and self._sleeping_count.acquire(False):
@@ -310,6 +316,7 @@ def notify(self, n=1):
 
             # rezero wait_semaphore in case some timeouts just happened
             while self._wait_semaphore.acquire(False):
+                print(':', end='')
                 pass
 
     def notify_all(self):

Lib/test/_test_multiprocessing.py

@@ -2042,7 +2042,9 @@ def _just_wait(cls, cond, sleeping, delay=None):
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_sleeping_count_invariant_when_elapsed_timeouts(self):
-        # See gh-69655.
+        # See gh-69655: In the `wait` method, exclude processes/threads
+        # that have exceeded the timeout from the list of pendings.
+
         if self.TYPE != 'processes':
             self.skipTest('test not appropriate for {}'.format(self.TYPE))
 
@@ -2052,8 +2054,8 @@ def test_sleeping_count_invariant_when_elapsed_timeouts(self):
         for _ in range(n):
             with cond:
                 cond.wait(0.0)
-        # Sempahore.locked() is equivalent to
-        # Semaphore._semlock._is_zero()
+
+        # Check there is no sleeper.
         self.assertTrue(cond._sleeping_count.locked())
         self.assertTrue(cond._woken_count.locked())