bpo-30457: Add new pending method for asyncio lock primitives

Lib/asyncio/locks.py

@@ -155,6 +155,10 @@ def __repr__(self):
             extra = '{},waiters:{}'.format(extra, len(self._waiters))
         return '<{} [{}]>'.format(res[1:-1], extra)
 
+    def pending(self):
+        """Return the num of waiters pending to be woken up"""
+        return len(self._waiters)
+
     def locked(self):
         """Return True if lock is acquired."""
         return self._locked
@@ -225,6 +229,10 @@ def __repr__(self):
             extra = '{},waiters:{}'.format(extra, len(self._waiters))
         return '<{} [{}]>'.format(res[1:-1], extra)
 
+    def pending(self):
+        """Return the num of waiters pending to be woken up"""
+        return len(self._waiters)
+
     def is_set(self):
         """Return True if and only if the internal flag is true."""
         return self._value
@@ -303,6 +311,10 @@ def __repr__(self):
             extra = '{},waiters:{}'.format(extra, len(self._waiters))
         return '<{} [{}]>'.format(res[1:-1], extra)
 
+    def pending(self):
+        """Return the num of waiters pending to be woken up"""
+        return len(self._waiters)
+
     @coroutine
     def wait(self):
         """Wait until notified.
@@ -424,6 +436,10 @@ def _wake_up_next(self):
                 waiter.set_result(None)
                 return
 
+    def pending(self):
+        """Return the num of waiters pending to be woken up"""
+        return len(self._waiters)
+
     def locked(self):
         """Returns True if semaphore can not be acquired immediately."""
         return self._value == 0

Lib/test/test_asyncio/test_locks.py

@@ -232,6 +232,23 @@ def test_context_manager_no_yield(self):
 
         self.assertFalse(lock.locked())
 
+    def test_pending(self):
+        lock = asyncio.Lock(loop=self.loop)
+        self.loop.run_until_complete(lock.acquire())
+        self.assertEqual(0, lock.pending())
+
+        @asyncio.coroutine
+        def c1():
+            yield from lock.acquire()
+
+        asyncio.Task(c1(), loop=self.loop)
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(1, lock.pending())
+
+        lock.release()
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(0, lock.pending())
+
 
 class EventTests(test_utils.TestCase):
 
@@ -362,6 +379,22 @@ def c1(result):
         self.assertTrue(t.done())
         self.assertTrue(t.result())
 
+    def test_pending(self):
+        ev = asyncio.Event(loop=self.loop)
+        self.assertEqual(0, ev.pending())
+
+        @asyncio.coroutine
+        def c1():
+            yield from ev.wait()
+
+        asyncio.Task(c1(), loop=self.loop)
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(1, ev.pending())
+
+        ev.set()
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(0, ev.pending())
+
 
 class ConditionTests(test_utils.TestCase):
 
@@ -698,6 +731,25 @@ def test_ambiguous_loops(self):
         with self.assertRaises(ValueError):
             asyncio.Condition(lock, loop=loop)
 
+    def test_pending(self):
+        cond = asyncio.Condition(loop=self.loop)
+
+        @asyncio.coroutine
+        def c1():
+            yield from cond.acquire()
+            yield from cond.wait()
+
+        asyncio.Task(c1(), loop=self.loop)
+
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(1, cond.pending())
+
+        self.loop.run_until_complete(cond.acquire())
+        cond.notify()
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(0, cond.pending())
+        cond.release()
+
 
 class SemaphoreTests(test_utils.TestCase):
 
@@ -916,6 +968,22 @@ def test_context_manager_no_yield(self):
 
         self.assertEqual(2, sem._value)
 
+    def test_pending(self):
+        sem = asyncio.Semaphore(0, loop=self.loop)
+
+        @asyncio.coroutine
+        def c1():
+            yield from sem.acquire()
+
+        asyncio.Task(c1(), loop=self.loop)
+
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(1, sem.pending())
+
+        sem.release()
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(0, sem.pending())
+
 
 if __name__ == '__main__':
     unittest.main()