This post is about cocotb Lock trigger which is used as a mutex between cocotb coroutines.

example Link to heading

The typical use case is locking while writing by defining the lock and using async with.

        self.write_address_busy = Lock()

        async with self.write_address_busy:
            ...

Deep dive Link to heading

Looking at the code for Lock, the docstring has the description and equivalent code for Lock.

class Lock(AsyncContextManager[None]):
    """A mutual exclusion lock.

    Guarantees fair scheduling.
    Lock acquisition is given in order of attempted lock acquisition.

    Usage:
        By directly calling :meth:`acquire` and :meth:`release`.

        .. code-block:: python

            lock = Lock()
            ...
            await lock.acquire()
            try:
                # do some stuff
                ...
            finally:
                lock.release()

        Or...

        .. code-block:: python

            async with Lock():
                # do some stuff
                ...
        The lock can be used as an asynchronous context manager in an
        :keyword:`async with` statement

As expected, there are __aenter__ and __aexit__ which call internal acquire and release.

    async def __aenter__(self) -> None:
        await self.acquire()

    async def __aexit__(self, *args: object) -> None:
        self.release()

when released, it pops from _pending_primed and things are delegated to _Lock

    def _acquire_and_fire(self, lock: _Lock) -> None:
        self._locked = True
        lock._callback(lock)

...
...
    def acquire(self) -> Trigger:
        """Produce a trigger which fires when the lock is acquired."""
        trig = _Lock(self)
        return trig

    def release(self) -> None:
        """Release the lock."""
        if not self._locked:
            raise RuntimeError(f"Attempt to release an unacquired Lock {self!s}")

        self._locked = False

        # nobody waiting for this lock
        if not self._pending_primed:
            return

        lock = self._pending_primed.pop(0)
        self._acquire_and_fire(lock)