"Named" locks

[davidbrochart]

It might be out of AnyIO's scope, but I often need a lock that is tied to an ID (or more generally a hashable object). For instance, resources on a file system could be represented by their path, and if I want to prevent concurrent access to the same resource, I'd like to write something like:

resource_lock = ResourceLock()
async with resource_lock("path/to/file"):
    ...

Maybe I'm missing something, but nothing in the synchronization primitives seems to offer that functionality easily. Here is what I currently use:

from typing import Any, Dict

from anyio import Event


class ResourceLock:
    _locks: Dict[Any, Event]

    def __init__(self):
        self._locks = {}

    def __call__(self, idx: Any):
        return _ResourceLock(idx, self._locks)


class _ResourceLock:
    _idx: Any
    _locks: Dict[Any, Event]
    _lock: Event

    def __init__(self, idx: Any, locks: Dict[Any, Event]):
        self._idx = idx
        self._locks = locks

    async def __aenter__(self):
        while True:
            if self._idx in self._locks:
                await self._locks[self._idx].wait()
            else:
                break
        self._locks[self._idx] = self._lock = Event()

    async def __aexit__(self, exc_type, exc_value, exc_tb):
        self._lock.set()
        del self._locks[self._idx]

It actually doesn't use any Lock. Do you see something wrong with that approach? Or do you think it could be better implemented?

[agronholm]

I think this is indeed beyond AnyIO's scope, as it doesn't seem generic enough to warrant inclusion in this project. I also take issue with the fact that if the lock is contested by multiple tasks, it wakes up every single one of them when a lock is exited even when only one task just needs to be woken up. That while loop should not be necessary. On top of that, it doesn't guarantee FIFO ordering for lock acquisition.

[davidbrochart]

Right, actually it's better to use a Lock instead of an Event:

from typing import Any

from anyio import Lock


class ResourceLock:
    _locks: dict[Any, Lock]

    def __init__(self):
        self._locks = {}

    def __call__(self, idx: Any):
        return _ResourceLock(idx, self._locks)


class _ResourceLock:
    _idx: Any
    _locks: dict[Any, Lock]
    _lock: Lock

    def __init__(self, idx: Any, locks: dict[Any, Lock]):
        self._idx = idx
        self._locks = locks

    async def __aenter__(self):
        if self._idx not in self._locks:
            self._locks[self._idx] = Lock()
        self._lock = self._locks[self._idx]
        await self._lock.acquire()

    async def __aexit__(self, exc_type, exc_value, exc_tb):
        self._lock.release()
        if self._idx in self._locks:
            del self._locks[self._idx]