Fix waiter cancellation in asyncio.Lock (#1031)

Avoid a deadlock when the waiter who is about to take the lock is
cancelled

Issue #27585

Lib/asyncio/locks.py

@@ -176,6 +176,10 @@ class Lock(_ContextManagerMixin):
             yield from fut
             self._locked = True
             return True
+        except futures.CancelledError:
+            if not self._locked:
+                self._wake_up_first()
+            raise
         finally:
             self._waiters.remove(fut)
 
@@ -192,14 +196,17 @@ class Lock(_ContextManagerMixin):
         """
         if self._locked:
             self._locked = False
-            # Wake up the first waiter who isn't cancelled.
-            for fut in self._waiters:
-                if not fut.done():
-                    fut.set_result(True)
-                    break
+            self._wake_up_first()
         else:
             raise RuntimeError('Lock is not acquired.')
 
+    def _wake_up_first(self):
+        """Wake up the first waiter who isn't cancelled."""
+        for fut in self._waiters:
+            if not fut.done():
+                fut.set_result(True)
+                break
+
 
 class Event:
     """Asynchronous equivalent to threading.Event.

Lib/test/test_asyncio/test_locks.py

@@ -176,6 +176,28 @@ class LockTests(test_utils.TestCase):
         self.assertTrue(tb.cancelled())
         self.assertTrue(tc.done())
 
+    def test_finished_waiter_cancelled(self):
+        lock = asyncio.Lock(loop=self.loop)
+
+        ta = asyncio.Task(lock.acquire(), loop=self.loop)
+        test_utils.run_briefly(self.loop)
+        self.assertTrue(lock.locked())
+
+        tb = asyncio.Task(lock.acquire(), loop=self.loop)
+        test_utils.run_briefly(self.loop)
+        self.assertEqual(len(lock._waiters), 1)
+
+        # Create a second waiter, wake up the first, and cancel it.
+        # Without the fix, the second was not woken up.
+        tc = asyncio.Task(lock.acquire(), loop=self.loop)
+        lock.release()
+        tb.cancel()
+        test_utils.run_briefly(self.loop)
+
+        self.assertTrue(lock.locked())
+        self.assertTrue(ta.done())
+        self.assertTrue(tb.cancelled())
+
     def test_release_not_acquired(self):
         lock = asyncio.Lock(loop=self.loop)