aiorwlock

An RWLock (Read-Write Lock) manages two types of locks: one for read-only operations and one for writing.

• Reader Lock: Multiple reader tasks can hold the lock simultaneously, provided no writer holds the lock.
• Writer Lock: This lock is exclusive; only one coroutine can hold it at a time.

When to use it: Use an RWLock when data is read frequently but modified infrequently. If updates are frequent, the performance benefits of concurrency will be lost as the lock will spend most of its time in exclusive write mode.

By default, RWLock switches context when acquiring a lock. This ensures that other waiting tasks can acquire the lock even if the current holder does not contain explicit context switches (like await).

If you are certain that your code inside the locked section contains context switches (await, async with, async for, or yield from), you can use the fast=True argument to disable this behavior for a minor performance improvement.

Usage: RWLock(fast=True)

The aiorwlock implementation does not support upgrading a lock from a reader to a writer. If a task currently holding a read lock attempts to acquire a write lock, a RuntimeError will be raised.

Error: RuntimeError: Cannot upgrade RWLock from read to write

You can use aiorwlock.RWLock to manage concurrent access to resources. Use rwlock.reader_lock for shared read access and rwlock.writer_lock for exclusive write access.

Important: You must release the lock using the same task that acquired it. Attempting to lock in one task and release in another will raise a RuntimeError.

import asyncio
import aiorwlock


async def go():
    rwlock = aiorwlock.RWLock()

    # acquire reader lock, multiple coroutines allowed to hold the lock
    async with rwlock.reader_lock:
        print('inside reader lock')
        await asyncio.sleep(0.1)

    # acquire writer lock, only one coroutine can hold the lock
    async with rwlock.writer_lock:
        print('inside writer lock')
        await asyncio.sleep(0.1)


asyncio.run(go())

When initializing RWLock, you can pass a fast boolean argument:

• fast=False (default): The lock will perform an await asyncio.sleep(0.0) after acquiring the lock to yield control to the event loop.
• fast=True: The lock skips the explicit yield, which can improve performance in high-throughput scenarios where tasks are frequent.

from aiorwlock import RWLock

# Default behavior (yields after acquire)
lock = RWLock(fast=False)

# High-performance mode (no explicit yield)
fast_lock = RWLock(fast=True)

You can check the status of the locks using the .locked property on the reader or writer objects. This is useful for debugging or conditional logic.

• lock.reader.locked: Returns True if at least one reader holds the lock.
• lock.writer.locked: Returns True if a writer holds the lock.

from aiorwlock import RWLock

lock = RWLock()
print(lock.reader.locked)  # False
print(lock.writer.locked) # False

The RWLock class provides a reader-writer lock for asyncio applications. It allows multiple coroutines to hold a read lock simultaneously, provided no writer holds the lock. A write lock is exclusive and can only be held by one coroutine at a time.

To use it, instantiate RWLock and access either the .reader property for shared access or the .writer property for exclusive access. Both properties return context managers that can be used with async with.

import asyncio
from aiorwlock import RWLock

async def main():
    lock = RWLock()

    # Shared read access
    async with lock.reader:
        # Multiple coroutines can be here at once
        pass

    # Exclusive write access
    async with lock.writer:
        # Only one coroutine can be here
        pass

asyncio.run(main())