【Python】asyncio.Lockの内部実装メモ

asyncio.Lock は Python の非同期プログラミングにおける 排他制御 を行うためのロックで、スレッドセーフな排他制御ではなく、イベントループ上のコルーチン同士の競合を防ぐために設計されています。

class Lock:
    def __init__(self, loop=None):
        self._loop = loop if loop is not None else get_running_loop()
        self._locked = False
        self._waiters = deque()

    def locked(self):
        return self._locked

    async def acquire(self):
        if not self._locked:
            self._locked = True
            return True

        fut = self._loop.create_future()
        self._waiters.append(fut)
        try:
            await fut
            return True
        except:
            fut.cancel()
            if not fut.done():
                fut.set_exception(CancelledError())
            raise

    def release(self):
        if not self._locked:
            raise RuntimeError("Lock is not acquired.")

        if self._waiters:
            fut = self._waiters.popleft()
            if not fut.done():
                fut.set_result(True)
        else:
            self._locked = False