Repository: agronholm/anyio

Issue:
Deadlock in anyio.Lock
Related to this issue in httpcore: https://github.com/encode/httpcore/issues/454

TL;DR: With the right timing, the lock **acquire** step can be cancelled and the lock remain locked for other tasks because when using the lock context manager, if the task is cancelled during the `__aenter__` , `__aexit__` is not executed.

This snippet reproduces the issue:

```python
import asyncio
import anyio

class Holder:
    def __init__(self):
        # self.lock = asyncio.Lock()
        self.lock = anyio.Lock()

    async def work(self, index):
        print(f"Trying to acquire {index}")
        async with self.lock:
            print(f"Doing index {index}")
            await asyncio.sleep(5)


async def main_anyio():
    holder = Holder()
    t1 = asyncio.create_task(holder.work(1), name="holder 1")
    t2 = asyncio.create_task(holder.work(2), name="holder 2")

    print("Tasks created")
    await asyncio.sleep(0)

    print("Cancelling t1")
    t1.cancel()

    print("Waiting for t2")
    await asyncio.gather(t2)


if __name__ == "__main__":
    asyncio.run(main_anyio())
```

The issue goes away when using `asyncio.Lock` because that lock context manager never blocks during the acquire step. Even if the `__aenter__` method in `asyncio.Lock` is asynchronous, no asynchronous code is executed there.

Discussion:

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index d275dab..df01fcf 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -3,6 +3,12 @@ Version history
 
 This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
+**UNRELEASED**
+
+- Fixed deadlock in synchronization primitives on asyncio which can happen if a task acquiring a
+  primitive is hit with a native (not AnyIO) cancellation with just the right timing, leaving the
+  next acquiring task waiting forever (`#398 <https://github.com/agronholm/anyio/issues/398>`_)
+
 **3.4.0**
 
 - Added context propagation to/from worker threads in ``to_thread.run_sync()``,
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 82df30a..43b4c91 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -1743,7 +1743,11 @@ class CapacityLimiter(BaseCapacityLimiter):
 
             self._borrowers.add(borrower)
         else:
-            await cancel_shielded_checkpoint()
+            try:
+                await cancel_shielded_checkpoint()
+            except BaseException:
+                self.release()
+                raise
 
     def release(self) -> None:
         self.release_on_behalf_of(current_task())
diff --git a/src/anyio/_core/_synchronization.py b/src/anyio/_core/_synchronization.py
index 0898ea3..d75afed 100644
--- a/src/anyio/_core/_synchronization.py
+++ b/src/anyio/_core/_synchronization.py
@@ -134,7 +134,11 @@ class Lock:
 
             assert self._owner_task == task
         else:
-            await cancel_shielded_checkpoint()
+            try:
+                await cancel_shielded_checkpoint()
+            except BaseException:
+                self.release()
+                raise
 
     def acquire_nowait(self) -> None:
         """
@@ -309,7 +313,11 @@ class Semaphore:
 
                 raise
         else:
-            await cancel_shielded_checkpoint()
+            try:
+                await cancel_shielded_checkpoint()
+            except BaseException:
+                self.release()
+                raise
 
     def acquire_nowait(self) -> None:
         """


Test patch:
diff --git a/tests/test_synchronization.py b/tests/test_synchronization.py
index 1bbfa95..6697fbf 100644
--- a/tests/test_synchronization.py
+++ b/tests/test_synchronization.py
@@ -1,3 +1,4 @@
+import asyncio
 from typing import Optional
 
 import pytest
@@ -114,6 +115,22 @@ class TestLock:
         assert not lock.statistics().locked
         assert lock.statistics().tasks_waiting == 0
 
+    @pytest.mark.parametrize('anyio_backend', ['asyncio'])
+    async def test_asyncio_deadlock(self) -> None:
+        """Regression test for #398."""
+        lock = Lock()
+
+        async def acquire() -> None:
+            async with lock:
+                await asyncio.sleep(0)
+
+        loop = asyncio.get_event_loop()
+        task1 = loop.create_task(acquire())
+        task2 = loop.create_task(acquire())
+        await asyncio.sleep(0)
+        task1.cancel()
+        await asyncio.wait_for(task2, 1)
+
 
 class TestEvent:
     async def test_event(self) -> None:
@@ -363,6 +380,22 @@ class TestSemaphore:
             semaphore.release()
             pytest.raises(WouldBlock, semaphore.acquire_nowait)
 
+    @pytest.mark.parametrize('anyio_backend', ['asyncio'])
+    async def test_asyncio_deadlock(self) -> None:
+        """Regression test for #398."""
+        semaphore = Semaphore(1)
+
+        async def acquire() -> None:
+            async with semaphore:
+                await asyncio.sleep(0)
+
+        loop = asyncio.get_event_loop()
+        task1 = loop.create_task(acquire())
+        task2 = loop.create_task(acquire())
+        await asyncio.sleep(0)
+        task1.cancel()
+        await asyncio.wait_for(task2, 1)
+
 
 class TestCapacityLimiter:
     async def test_bad_init_type(self) -> None:
@@ -465,3 +498,19 @@ class TestCapacityLimiter:
 
         assert limiter.statistics().tasks_waiting == 0
         assert limiter.statistics().borrowed_tokens == 0
+
+    @pytest.mark.parametrize('anyio_backend', ['asyncio'])
+    async def test_asyncio_deadlock(self) -> None:
+        """Regression test for #398."""
+        limiter = CapacityLimiter(1)
+
+        async def acquire() -> None:
+            async with limiter:
+                await asyncio.sleep(0)
+
+        loop = asyncio.get_event_loop()
+        task1 = loop.create_task(acquire())
+        task2 = loop.create_task(acquire())
+        await asyncio.sleep(0)
+        task1.cancel()
+        await asyncio.wait_for(task2, 1)