Repository: agronholm/anyio

Issue:
Potential race condition in `Lock`
Hello!

I'm trying to debug an issue we sometimes hit in production through httpx. We're not using anyio ourselves directly (it's regular asyncio), but httpx is based on it.

I'm not 100% sure the scenario I'm documenting here is the exact issue we're seeing, but there's a good chance. The symptoms we're seeing is that a lock instance gets stuck in the locked state with a finished task still set as the owner.

Here's the rough sketch, involving two tasks:

Task 1:
* acquires the lock

Task 2:
* tries acquiring the lock, fails, creates an event and starts waiting on it

Task 1:
* cancels Task 2, releases the lock setting the lock `_owner_task` to Task 2, gets Task 2's event and sets it

Task 2:
* wakes up with CancelledError on `event.wait()`, propagates it

The lock `_owner_task` is still set to Task 2, rendering the lock stuck.

In our production code it's not Task 1 cancelling Task 2, but aiohttp (but that's not relevant to the example).

Here's a Python snippet demonstrating the problem:
```
from asyncio import CancelledError, create_task, run, sleep
from contextlib import suppress

from anyio import Lock


async def main():
    lock = Lock()

    async def task_b():
        await sleep(0.1)  # Sleep to allow task_a to set up
        async with lock:
            pass

    t = create_task(task_b())

    async def task_a():
        async with lock:
            await sleep(0.1)  # Sleep to allow task_b to try locking
            t.cancel()
        await sleep(0.2)
        print(t.done())  # True
        print(lock._owner_task)  # TaskInfo(id=4337401360, name='Task-2')

    await task_a()
    with suppress(CancelledError):
        await t


run(main())
```
The lock should end up in a free state, but it gets stuck being owned by `task_b`.

Discussion:

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index d198c4b..d129ca3 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 race condition in ``Lock`` and ``Semaphore`` classes when a task waiting on ``acquire()``
+  is cancelled while another task is waiting to acquire the same primitive
+  (`#387 <https://github.com/agronholm/anyio/issues/387>`_)
+
 **3.3.4**
 
 - Fixed ``BrokenResourceError`` instead of ``EndOfStream`` being raised in ``TLSStream`` when the
diff --git a/src/anyio/_core/_synchronization.py b/src/anyio/_core/_synchronization.py
index 6c69177..0898ea3 100644
--- a/src/anyio/_core/_synchronization.py
+++ b/src/anyio/_core/_synchronization.py
@@ -127,6 +127,8 @@ class Lock:
             except BaseException:
                 if not event.is_set():
                     self._waiters.remove(token)
+                elif self._owner_task == task:
+                    self.release()
 
                 raise
 
@@ -302,6 +304,8 @@ class Semaphore:
             except BaseException:
                 if not event.is_set():
                     self._waiters.remove(event)
+                else:
+                    self.release()
 
                 raise
         else:


Test patch:
diff --git a/tests/test_synchronization.py b/tests/test_synchronization.py
index 747ee93..1bbfa95 100644
--- a/tests/test_synchronization.py
+++ b/tests/test_synchronization.py
@@ -5,7 +5,7 @@ import pytest
 from anyio import (
     CancelScope, Condition, Event, Lock, Semaphore, WouldBlock, create_task_group, to_thread,
     wait_all_tasks_blocked)
-from anyio.abc import CapacityLimiter
+from anyio.abc import CapacityLimiter, TaskStatus
 
 pytestmark = pytest.mark.anyio
 
@@ -65,23 +65,34 @@ class TestLock:
             assert lock.locked()
             tg.start_soon(try_lock)
 
-    async def test_cancel(self) -> None:
-        task_started = got_lock = False
+    @pytest.mark.parametrize('release_first', [
+        pytest.param(False, id='releaselast'),
+        pytest.param(True, id='releasefirst')
+    ])
+    async def test_cancel_during_acquire(self, release_first: bool) -> None:
+        acquired = False
 
-        async def task() -> None:
-            nonlocal task_started, got_lock
-            task_started = True
+        async def task(*, task_status: TaskStatus) -> None:
+            nonlocal acquired
+            task_status.started()
             async with lock:
-                got_lock = True
+                acquired = True
 
         lock = Lock()
         async with create_task_group() as tg:
-            async with lock:
-                tg.start_soon(task)
-                tg.cancel_scope.cancel()
+            await lock.acquire()
+            await tg.start(task)
+            tg.cancel_scope.cancel()
+            with CancelScope(shield=True):
+                if release_first:
+                    lock.release()
+                    await wait_all_tasks_blocked()
+                else:
+                    await wait_all_tasks_blocked()
+                    lock.release()
 
-        assert task_started
-        assert not got_lock
+        assert not acquired
+        assert not lock.locked()
 
     async def test_statistics(self) -> None:
         async def waiter() -> None:
@@ -282,24 +293,34 @@ class TestSemaphore:
         assert semaphore.value == 0
         pytest.raises(WouldBlock, semaphore.acquire_nowait)
 
-    async def test_acquire_cancel(self) -> None:
-        local_scope = acquired = None
+    @pytest.mark.parametrize('release_first', [
+        pytest.param(False, id='releaselast'),
+        pytest.param(True, id='releasefirst')
+    ])
+    async def test_cancel_during_acquire(self, release_first: bool) -> None:
+        acquired = False
 
-        async def task() -> None:
-            nonlocal local_scope, acquired
-            with CancelScope() as local_scope:
-                async with semaphore:
-                    acquired = True
+        async def task(*, task_status: TaskStatus) -> None:
+            nonlocal acquired
+            task_status.started()
+            async with semaphore:
+                acquired = True
 
         semaphore = Semaphore(1)
         async with create_task_group() as tg:
-            async with semaphore:
-                tg.start_soon(task)
-                await wait_all_tasks_blocked()
-                assert local_scope is not None
-                local_scope.cancel()
+            await semaphore.acquire()
+            await tg.start(task)
+            tg.cancel_scope.cancel()
+            with CancelScope(shield=True):
+                if release_first:
+                    semaphore.release()
+                    await wait_all_tasks_blocked()
+                else:
+                    await wait_all_tasks_blocked()
+                    semaphore.release()
 
         assert not acquired
+        assert semaphore.value == 1
 
     @pytest.mark.parametrize('max_value', [2, None])
     async def test_max_value(self, max_value: Optional[int]) -> None: