### Install aiorwlock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

Install the aiorwlock library using pip.

```bash
pip install aiorwlock
```

--------------------------------

### RWLock Default Mode Example (fast=False)

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Demonstrates the default RWLock behavior where context switching occurs automatically after lock acquisition, ensuring fairness. This example shows multiple readers being scheduled fairly.

```python
import asyncio
import aiorwlock

async def reader(rwlock, reader_id):
    async with rwlock.reader:
        # Context switch happened automatically after acquiring
        # Other tasks might have run here
        print(f'Reader {reader_id}')

async def main():
    rwlock = aiorwlock.RWLock(fast=False) 
    
    # Multiple readers get scheduled fairly
    tasks = [reader(rwlock, i) for i in range(3)]
    await asyncio.gather(*tasks)

asyncio.run(main())
```

--------------------------------

### Complete Usage Example for aiorwlock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/api-reference.md

Demonstrates how to use aiorwlock.RWLock for managing concurrent read access and exclusive write access. This example showcases multiple readers and writers interacting with the lock.

```python
import asyncio
import aiorwlock


async def read_data(rwlock, reader_id):
    """Multiple readers can execute concurrently"""
    async with rwlock.reader:
        print(f'Reader {reader_id}: acquiring read lock')
        await asyncio.sleep(0.1)
        print(f'Reader {reader_id}: releasing read lock')


async def write_data(rwlock, writer_id):
    """Only one writer executes at a time"""
    async with rwlock.writer:
        print(f'Writer {writer_id}: acquiring write lock')
        await asyncio.sleep(0.1)
        print(f'Writer {writer_id}: releasing write lock')


async def main():
    rwlock = aiorwlock.RWLock()
    
    # Create multiple concurrent tasks
    tasks = [
        read_data(rwlock, 1),
        read_data(rwlock, 2),
        write_data(rwlock, 1),
        read_data(rwlock, 3),
    ]
    
    await asyncio.gather(*tasks)


asyncio.run(main())
```

--------------------------------

### Quick Start: Basic RWLock Usage

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/00-START-HERE.md

Demonstrates the fundamental usage of RWLock for concurrent read and write operations. Use this for a rapid introduction to the library's core functionality.

```python
import asyncio
import aiorwlock

async def main():
    # Create a lock
    rwlock = aiorwlock.RWLock()
    
    # Read: multiple tasks can do this together
    async with rwlock.reader:
        data = shared_resource['key']
    
    # Write: only one task at a time
    async with rwlock.writer:
        shared_resource['key'] = new_value

asyncio.run(main())
```

--------------------------------

### Multiple Lock Coordination Example

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

Demonstrates how to manage multiple independent resources using separate RWLock instances for each. This pattern is useful for complex operations involving checks and modifications across different data structures.

```python
import asyncio
import aiorwlock

class MultiResourceManager:
    """Manage multiple independent resources"""
    
    def __init__(self):
        # Separate lock per resource
        self.users_lock = aiorwlock.RWLock()
        self.products_lock = aiorwlock.RWLock()
        self.orders_lock = aiorwlock.RWLock()
        
        self.users = {}
        self.products = {}
        self.orders = {}
    
    async def get_user(self, user_id: str):
        async with self.users_lock.reader:
            return self.users.get(user_id)
    
    async def add_user(self, user_id: str, user_data: dict):
        async with self.users_lock.writer:
            self.users[user_id] = user_data
    
    async def get_product(self, product_id: str):
        async with self.products_lock.reader:
            return self.products.get(product_id)
    
    async def add_product(self, product_id: str, product_data: dict):
        async with self.products_lock.writer:
            self.products[product_id] = product_data
    
    async def create_order(self, user_id: str, product_id: str):
        """Complex operation using multiple locks"""
        # Check user exists (read lock)
        async with self.users_lock.reader:
            user = self.users.get(user_id)
            if not user:
                raise ValueError(f"User {user_id} not found")
        
        # Check product exists (read lock)
        async with self.products_lock.reader:
            product = self.products.get(product_id)
            if not product:
                raise ValueError(f"Product {product_id} not found")
        
        # Create order (write lock)
        async with self.orders_lock.writer:
            order_id = len(self.orders)
            self.orders[order_id] = {
                'user_id': user_id,
                'product_id': product_id,
            }
            return order_id

async def multi_lock_example():
    manager = MultiResourceManager()
    
    # Add initial data
    await manager.add_user('user1', {'name': 'Alice'})
    await manager.add_product('prod1', {'price': 99.99})
    
    # Concurrent operations on different resources
    async def customer_browse(customer_id: int):
        product = await manager.get_product('prod1')
        print(f'Customer {customer_id}: product = {product}')
    
    async def admin_add_product(prod_id: str):
        await manager.add_product(prod_id, {'price': 49.99})
        print(f'Admin: added {prod_id}')
    
    async def place_order(order_id: int):
        try:
            result = await manager.create_order('user1', 'prod1')
            print(f'Order {order_id}: created order {result}')
        except ValueError as e:
            print(f'Order {order_id}: error = {e}')
    
    await asyncio.gather(
        customer_browse(1),
        customer_browse(2),
        admin_add_product('prod2'),
        place_order(1),
    )

asyncio.run(multi_lock_example())
```

--------------------------------

### Get Library Version

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/00-START-HERE.md

Retrieve the installed version of the aiorwlock library.

```python
# Get version
version = aiorwlock.__version__
```

--------------------------------

### Typical Usage Pattern for Cache with RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/quick-reference.md

Demonstrates a common pattern for using `RWLock` to protect a cache. It shows how to use the reader lock for `get` operations and the writer lock for `set` operations, ensuring thread-safe access.

```python
import asyncio
import aiorwlock

class Cache:
    def __init__(self):
        self.lock = aiorwlock.RWLock()
        self.data = {}
    
    async def get(self, key):
        async with self.lock.reader:
            return self.data.get(key)
    
    async def set(self, key, value):
        async with self.lock.writer:
            self.data[key] = value

async def main():
    cache = Cache()
    
    # Many readers
    reads = [cache.get(f'key{i}') for i in range(5)]
    
    # One writer
    writes = [cache.set('key', f'value{i}') for i in range(2)]
    
    await asyncio.gather(*reads, *writes)

asyncio.run(main())
```

--------------------------------

### Reader-Writer Data Store Coordination

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

This example demonstrates a DataStore class using aiorwlock.RWLock to manage concurrent read and write operations. It's suitable for scenarios where reads are frequent and writes are less frequent, optimizing performance by allowing multiple readers simultaneously.

```python
import asyncio
import aiorwlock

class DataStore:
    """Coordinated reader/writer data store"""
    
    def __init__(self):
        self.rwlock = aiorwlock.RWLock()
        self.data = []
        self.stats = {
            'reads': 0,
            'writes': 0,
        }
    
    async def read_item(self, index: int):
        """Read single item with stats"""
        async with self.rwlock.reader:
            self.stats['reads'] += 1
            if index < len(self.data):
                return self.data[index]
            return None
    
    async def write_item(self, index: int, value: any):
        """Write single item with stats"""
        async with self.rwlock.writer:
            self.stats['writes'] += 1
            while len(self.data) <= index:
                self.data.append(None)
            self.data[index] = value
    
    async def batch_read(self, indices: list[int]):
        """Read multiple items efficiently"""
        async with self.rwlock.reader:
            return [
                self.data[i] if i < len(self.data) else None
                for i in indices
            ]
    
    async def batch_write(self, items: dict[int, any]):
        """Write multiple items efficiently"""
        async with self.rwlock.writer:
            for index, value in items.items():
                while len(self.data) <= index:
                    self.data.append(None)
                self.data[index] = value
    
    async def get_stats(self):
        """Get statistics (read operation)"""
        async with self.rwlock.reader:
            return self.stats.copy()

async def coordination_example():
    store = DataStore()
    
    async def reader(reader_id: int):
        for i in range(3):
            value = await store.read_item(i)
            print(f'Reader {reader_id}: data[{i}] = {value}')
            await asyncio.sleep(0.02)
    
    async def writer(writer_id: int):
        items = {writer_id * 10 + i: f'value_{writer_id}_{i}' 
                 for i in range(3)}
        await store.batch_write(items)
        print(f'Writer {writer_id}: wrote {len(items)} items')
    
    # Start concurrent operations
    tasks = [
        reader(1), reader(2),
        writer(1), writer(2),
        store.get_stats()
    ]
    
    results = await asyncio.gather(*tasks)
    
    # Get final stats
    final_stats = await store.get_stats()
    print(f'Final stats: {final_stats}')

asyncio.run(coordination_example())

```

--------------------------------

### Basic RWLock Usage

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/README.md

Shows how to create and use a Read-Write Lock for concurrent read access and exclusive write access to a shared resource. This example uses the default safe mode.

```python
import asyncio
import aiorwlock

async def example():
    # Create a lock
    rwlock = aiorwlock.RWLock()
    
    # Read: multiple tasks can be here
    async with rwlock.reader:
        data = read_shared_resource() # Assume this function exists
    
    # Write: only one task at a time
    async with rwlock.writer:
        write_shared_resource(new_data) # Assume this function exists

asyncio.run(example())
```

--------------------------------

### Double Release

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

This example illustrates the RuntimeError that occurs when `release()` is called twice for a single `acquire()` call. Each `acquire()` must have a corresponding `release()`.

```python
rwlock = aiorwlock.RWLock()

async def main():
    await rwlock.reader.acquire()
    rwlock.reader.release()
    rwlock.reader.release()  # RuntimeError: second release has no acquire
```

--------------------------------

### Demonstrate Task Cancellation While Waiting for Lock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

This example shows a task being cancelled while it is waiting to acquire a writer lock. It demonstrates how to catch the asyncio.CancelledError.

```python
import asyncio
import aiorwlock

rwlock = aiorwlock.RWLock()

async def main():
    task = asyncio.create_task(acquire_and_wait())
    await asyncio.sleep(0.1)  # Let it start waiting
    task.cancel()  # Cancel the waiting task
    
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled while waiting for lock")

async def acquire_and_wait():
    # This will block waiting for the lock
    await rwlock.writer.acquire()
    try:
        await asyncio.sleep(10)  # Long operation
    finally:
        rwlock.writer.release()

asyncio.run(main())
```

--------------------------------

### Get aiorwlock Package Version

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/api-reference.md

Access the installed version of the aiorwlock package. This is useful for dependency checking or logging.

```python
import aiorwlock

print(aiorwlock.__version__)  # e.g., "1.5.1"
```

--------------------------------

### Measure RWLock Performance Modes

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Compares the performance of aiorwlock in 'fast=False' (default) and 'fast=True' modes. This example measures the number of iterations completed within a given duration for both modes.

```python
import asyncio
import aiorwlock
import time

async def measure_mode(fast_mode, duration=1.0):
    rwlock = aiorwlock.RWLock(fast=fast_mode)
    count = 0
    start = time.time()
    
    async def worker():
        nonlocal count
        while time.time() - start < duration:
            async with rwlock.reader:
                await asyncio.sleep(0.0001)  # Small async operation
            count += 1
    
    await asyncio.gather(
        worker(),
        worker(),
        worker(),
    )
    
    elapsed = time.time() - start
    print(f"fast={fast_mode}: {count} iterations in {elapsed:.2f}s")

asyncio.run(measure_mode(False))  # fast=False
asyncio.run(measure_mode(True))   # fast=True
```

--------------------------------

### Invalid Fast Mode RWLock Usage Examples (fast=True)

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Shows unsafe usage patterns in fast mode where the locked code lacks explicit await statements, potentially leading to task starvation. For such cases, use `fast=False`.

```python
import asyncio
import aiorwlock

async def unsafe_fast_mode():
    rwlock = aiorwlock.RWLock(fast=True)
    
    # UNSAFE: no await statement - OTHER TASKS MIGHT STARVE
    async with rwlock.reader:
        # Synchronous computation with no yields
        for i in range(1000):
            data = i * i
        print(data)
    
    # UNSAFE: no await statement
    async with rwlock.reader:
        dict_lookup = my_dict.get('key')  # Synchronous
        print(dict_lookup)

async def main():
    # If you write this code above, use fast=False instead:
    rwlock = aiorwlock.RWLock(fast=False)
    
    async with rwlock.reader:
        for i in range(1000):
            data = i * i
```

--------------------------------

### Basic Read-Write Pattern with aiorwlock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

Protects a shared counter using a reader-writer lock, allowing concurrent reads and exclusive writes. Ensure aiorwlock is installed.

```python
import asyncio
import aiorwlock

async def basic_example():
    """Protect a counter with reader and writer access"""
    
    rwlock = aiorwlock.RWLock()
    counter = {'value': 0}
    
    async def reader(reader_id):
        """Multiple readers can execute concurrently"""
        for _ in range(3):
            async with rwlock.reader:
                current = counter['value']
                await asyncio.sleep(0.01)  # Simulate work
                print(f"Reader {reader_id}: count = {current}")
    
    async def writer(writer_id):
        """Only one writer executes at a time"""
        for i in range(3):
            async with rwlock.writer:
                counter['value'] += 1
                await asyncio.sleep(0.01)  # Simulate work
                print(f"Writer {writer_id}: incremented to {counter['value']}")
    
    # Create concurrent tasks
    await asyncio.gather(
        reader(1),
        reader(2),
        writer(1),
        reader(3),
        writer(2),
    )

asyncio.run(basic_example())
```

--------------------------------

### Handling RuntimeError: Safe Upgrade Logic

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

This example shows how to safely handle the potential RuntimeError when attempting to upgrade a read lock to a write lock. It catches the error and proceeds to acquire the write lock after the read lock's scope is exited.

```python
async def safe_upgrade(rwlock):
    async with rwlock.reader:
        data = read_operation()
    
    if needs_modification(data):
        try:
            # Try to upgrade - this will fail
            await rwlock.writer.acquire()
        except RuntimeError as e:
            if "Cannot upgrade" in str(e):
                print("Upgrading read to write not allowed")
                # Release read lock first (it's already released in context manager)
                # Then acquire write lock
                async with rwlock.writer:
                    perform_write() 
```

--------------------------------

### Valid Fast Mode RWLock Usage Examples (fast=True)

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Illustrates correct usage of RWLock in fast mode, where the locked code includes explicit context switches like `await asyncio.sleep(0)`, awaiting other coroutines, or using `async for`.

```python
import asyncio
import aiorwlock

async def safe_fast_mode():
    rwlock = aiorwlock.RWLock(fast=True)
    
    # Safe: contains await
    async with rwlock.reader:
        await asyncio.sleep(0)
        print('has context switch')
    
    # Safe: contains await
    async with rwlock.reader:
        result = await some_coroutine()
        print(result)
    
    # Safe: contains async iteration
    async with rwlock.reader:
        async for item in async_iterable:
            process(item)

async def safe_fast_nested():
    """Safe: nested async calls provide context switches"""
    rwlock = aiorwlock.RWLock(fast=True)
    
    async def do_work():
        await asyncio.sleep(0.01)
        return "result"
    
    async with rwlock.reader:
        result = await do_work()  # Context switch happens here
        print(result)
```

--------------------------------

### Task B Attempts to Release Lock Acquired by Task A

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

This example demonstrates a common scenario where one task attempts to release a lock that was acquired by a different task, leading to a RuntimeError. Ensure locks are released by the same task that acquired them.

```python
import asyncio
import aiorwlock

rwlock = aiorwlock.RWLock()

async def task_a():
    await rwlock.reader.acquire()
    # ... do work ...
    # Forgot to release!

async def task_b():
    # Trying to release a lock acquired by task_a
    rwlock.reader.release()  # RuntimeError!

async def main():
    await asyncio.gather(task_a(), task_b())
```

--------------------------------

### Import and Use RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/README.md

Demonstrates how to import the RWLock class and its version, create a lock instance, and use it for reader and writer access. Accessing the version is also shown.

```python
from aiorwlock import RWLock, __version__

# Create a lock
lock = RWLock(fast=False)

# Access reader
async with lock.reader:
    data = await read_operation()

# Access writer  
async with lock.writer:
    data = await write_operation()

# Get version
print(RWLock.__version__)
```

--------------------------------

### Access RWLock and Version

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

Demonstrates how to instantiate the RWLock class, acquire and release reader and writer locks, and access the package version. Ensure the aiorwlock library is imported.

```python
import aiorwlock

# Main class
lock = aiorwlock.RWLock(fast=False)

# Access reader lock
reader = lock.reader  # or lock.reader_lock
await reader.acquire()
reader.release()

# Access writer lock
writer = lock.writer  # or lock.writer_lock
await writer.acquire()
writer.release()

# Get version
version = aiorwlock.__version__
```

--------------------------------

### Initialize RWLock with Default Configuration (fast=False)

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Instantiate RWLock using default settings for safe, fair locking. This is recommended for general use and when locked code may not contain explicit await statements.

```python
rwlock = aiorwlock.RWLock()  # Default: fast=False
```

```python
rwlock = aiorwlock.RWLock(fast=False)  # Explicit
```

--------------------------------

### Create a RWLock Instance

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/quick-reference.md

Instantiate a RWLock object. Use the default safe mode or enable fast mode for explicit awaits.

```python
rwlock = aiorwlock.RWLock()              # Default: safe mode
```

```python
rwlock = aiorwlock.RWLock(fast=True)     # Fast mode: requires explicit awaits
```

--------------------------------

### Initialize RWLock with Fast Configuration (fast=True)

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Instantiate RWLock with `fast=True` for performance-critical code. This mode skips automatic context switching, requiring explicit await statements within the locked code to prevent task starvation.

```python
rwlock = aiorwlock.RWLock(fast=True)
```

--------------------------------

### Import and Create RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/00-START-HERE.md

Import the aiorwlock library and create a new RWLock instance. You can choose between the default safe mode or the performance-oriented fast mode.

```python
# Import
import aiorwlock

# Create lock
lock = aiorwlock.RWLock()           # Default: safe
lock = aiorwlock.RWLock(fast=True)  # Performance mode
```

--------------------------------

### Initialize RWLock with Fast Mode and Safe Fallback

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Initialize an RWLock with a hint for fast mode, allowing it to be switched to safe mode if necessary. This provides a performance optimization with a fallback mechanism.

```python
import asyncio
import aiorwlock

class SmartLock:
    def __init__(self, fast_mode_hint=True):
        self.rwlock = aiorwlock.RWLock(fast=fast_mode_hint)
        self.fast_mode = fast_mode_hint
    
    def set_safe_mode(self):
        """Switch to safe mode if needed"""
        if self.fast_mode:
            self.rwlock = aiorwlock.RWLock(fast=False)
            self.fast_mode = False
```

--------------------------------

### Basic aiorwlock Usage

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

Demonstrates basic usage of aiorwlock with reader and writer locks using async context managers. Multiple tasks can acquire the reader lock concurrently, while the writer lock grants exclusive access.

```python
import asyncio
import aiorwlock

async def example():
    # Create a lock
    rwlock = aiorwlock.RWLock()
    
    # Use reader lock for read access
    async with rwlock.reader:
        # Multiple tasks can be here simultaneously
        data = read_shared_resource()
    
    # Use writer lock for write access
    async with rwlock.writer:
        # Only one task can be here
        write_shared_resource(new_data)

asyncio.run(example())
```

--------------------------------

### Configuration Manager with Concurrent Reads and Writes

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

Demonstrates a thread-safe configuration manager using aiorwlock. It allows multiple concurrent readers to access configuration data while ensuring exclusive access for writers during updates.

```python
import asyncio
import aiorwlock
from dataclasses import dataclass, asdict
from typing import Any

@dataclass
class Config:
    """Application configuration"""
    debug: bool = False
    database_url: str = "postgres://localhost"
    cache_ttl: int = 3600
    max_connections: int = 100

class ConfigManager:
    """Thread-safe configuration manager"""
    
    def __init__(self, initial_config: Config):
        self.rwlock = aiorwlock.RWLock()
        self._config = initial_config
    
    async def get_config(self) -> Config:
        """Read current configuration (many concurrent readers)"""
        async with self.rwlock.reader:
            # Simulate reading multiple fields
            await asyncio.sleep(0.001)
            return self._config
    
    async def get_value(self, key: str) -> Any:
        """Get single configuration value"""
        async with self.rwlock.reader:
            config_dict = asdict(self._config)
            return config_dict.get(key)
    
    async def update_config(self, **kwargs) -> None:
        """Update configuration (exclusive write)"""
        async with self.rwlock.writer:
            for key, value in kwargs.items():
                if hasattr(self._config, key):
                    setattr(self._config, key, value)
            print(f'Config updated: {kwargs}')
    
    async def reload_config(self, new_config: Config) -> None:
        """Replace entire configuration"""
        async with self.rwlock.writer:
            self._config = new_config
            print('Config reloaded')

async def config_manager_example():
    config = ConfigManager(Config())
    
    async def service_reader(service_id: int):
        """Simulate service reading configuration"""
        for i in range(3):
            config_data = await config.get_config()
            print(f'Service {service_id}: debug={config_data.debug}, ' 
                  f'max_conn={config_data.max_connections}')
            await asyncio.sleep(0.05)
    
    async def config_updater():
        """Simulate configuration updates"""
        await asyncio.sleep(0.1)
        await config.update_config(debug=True, max_connections=200)
        await asyncio.sleep(0.1)
        await config.update_config(cache_ttl=7200)
    
    await asyncio.gather(
        service_reader(1),
        service_reader(2),
        service_reader(3),
        config_updater(),
    )

asyncio.run(config_manager_example())
```

--------------------------------

### RWLock Constructor

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/api-reference.md

Initializes a new RWLock instance. The `fast` parameter controls context switching behavior for performance optimization.

```APIDOC
## RWLock Constructor

### Description
Creates a new RWLock instance with separate reader and writer lock objects.

### Parameters
#### Keyword Arguments
- **fast** (bool) - Optional - When `False` (default), the lock yields context to other tasks after acquiring. When `True`, context switching is disabled for minor performance gain. Use `True` only if your locked code contains explicit context switches (await, async with, async for, yield from).

### Example
```python
import asyncio
import aiorwlock

async def main():
    # Default mode: safe context switching
    rwlock = aiorwlock.RWLock()
    
    # Fast mode: skip context switching for performance
    rwlock_fast = aiorwlock.RWLock(fast=True)

asyncio.run(main())
```
```

--------------------------------

### Basic RWLock Usage in asyncio

Source: https://github.com/aio-libs/aiorwlock/blob/master/README.rst

Demonstrates acquiring and releasing both reader and writer locks using `async with`. Ensure the task that acquires the lock is also responsible for releasing it.

```python
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())
```

--------------------------------

### Protected Shared State with RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

This pattern shows how to protect shared data (like a cache) using a reader-writer lock. Multiple readers can access the data concurrently, while writers get exclusive access. This is useful for managing shared state in asynchronous applications.

```python
import asyncio
import aiorwlock

class SharedCache:
    def __init__(self):
        self.rwlock = aiorwlock.RWLock()
        self.data = {}
    
    async def get(self, key):
        """Read with multiple concurrent accessors"""
        async with self.rwlock.reader:
            return self.data.get(key)
    
    async def set(self, key, value):
        """Write with exclusive access"""
        async with self.rwlock.writer:
            self.data[key] = value

async def main():
    cache = SharedCache()
    
    # Many readers can run concurrently
    read_tasks = [cache.get('key') for _ in range(10)]
    
    # Writer gets exclusive access
    write_task = cache.set('key', 'value')
    
    await asyncio.gather(*read_tasks, write_task)

asyncio.run(main())
```

--------------------------------

### Safe Fast Mode with RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

This pattern illustrates how to use the `fast=True` option for `RWLock` to potentially improve performance. It emphasizes that `fast` mode requires explicit context switches, especially when awaiting within a read lock, to maintain safety.

```python
async def fast_mode_safe(rwlock):
    """Fast mode requires explicit context switches"""
    rwlock_fast = aiorwlock.RWLock(fast=True)
    
    async with rwlock_fast.reader:
        # Safe: contains await
        result = await fetch_data()
        await process(result)
    
    async with rwlock_fast.reader:
        # Safe: context switch in loop
        async for item in get_items():
            process(item)
```

--------------------------------

### RWLock Constructor

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/api-reference.md

Creates a new RWLock instance. Use the 'fast' parameter to control context switching behavior.

```python
def __init__(self, *, fast: bool = False) -> None
```

```python
import asyncio
import aiorwlock

async def main():
    # Default mode: safe context switching
    rwlock = aiorwlock.RWLock()
    
    # Fast mode: skip context switching for performance
    rwlock_fast = aiorwlock.RWLock(fast=True)

asyncio.run(main())
```

--------------------------------

### Fast Mode vs. Default Mode Comparison

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

Compares the behavior of RWLock in safe mode (fast=False) versus performance mode (fast=True). Safe mode ensures context switches after acquire, while fast mode does not, which can lead to issues with synchronous work.

```python
import asyncio
import aiorwlock
import time

async def compare_modes():
    """Compare fast=False (safe) vs fast=True (performance)"""
    
    # Safe mode - context switches after acquire
    rwlock_safe = aiorwlock.RWLock(fast=False)
    
    # Fast mode - no context switching
    rwlock_fast = aiorwlock.RWLock(fast=True)
    
    async def work_with_await(rwlock, mode_name):
        """Safe in both modes: contains await"""
        count = 0
        for _ in range(100):
            async with rwlock.reader:
                count += 1
                await asyncio.sleep(0)  # Context switch
        print(f'{mode_name} with await: {count} iterations')
    
    async def work_without_await(rwlock, mode_name):
        """Unsafe in fast mode: no context switch"""
        count = 0
        for _ in range(100):
            async with rwlock.reader:
                count += 1
                # Synchronous work only
        print(f'{mode_name} without await: {count} iterations')
    
    # Safe examples
    await asyncio.gather(
        work_with_await(rwlock_safe, 'Safe mode'),
        work_with_await(rwlock_fast, 'Fast mode'),
    )
    
    # Unsafe example - only in safe mode
    await work_without_await(rwlock_safe, 'Safe mode')
    # Don't do: work_without_await(rwlock_fast, 'Fast mode')

asyncio.run(compare_modes())
```

--------------------------------

### Correct Lock Usage Patterns

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

Demonstrates the recommended ways to correctly acquire and release locks in aiorwlock. Pattern 1 uses explicit acquire/release with a try-finally block, while Pattern 2 utilizes the recommended async context manager.

```python
async def correct_usage():
    rwlock = aiorwlock.RWLock()
    
    # Pattern 1: explicit acquire/release in same task
    await rwlock.reader.acquire()
    try:
        # do work
        pass
    finally:
        rwlock.reader.release()
    
    # Pattern 2: async context manager (recommended)
    async with rwlock.reader:
        # do work
        pass
```

--------------------------------

### Initialize Multiple RWLocks for Independent Resources

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Create separate RWLock instances for different sets of resources to allow independent access. This improves concurrency by not blocking unrelated operations.

```python
import aiorwlock

class MultiResourceData:
    def __init__(self):
        self.users_lock = aiorwlock.RWLock()
        self.config_lock = aiorwlock.RWLock()
        self.users = {}
        self.config = {}
    
    async def read_user(self, user_id):
        async with self.users_lock.reader:
            return self.users.get(user_id)
    
    async def read_config(self, key):
        async with self.config_lock.reader:
            return self.config.get(key)
```

--------------------------------

### Explicit Acquire and Release of RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

Demonstrates manual acquisition and release of reader and writer locks. Use this pattern for complex control flow or when context managers are not suitable.

```python
import asyncio
import aiorwlock

async def explicit_acquire_release():
    """Using explicit acquire and release calls"""
    
    rwlock = aiorwlock.RWLock()
    data = {'count': 0}
    
    # Reader with explicit acquire/release
    try:
        await rwlock.reader.acquire()
        print(f'Current count: {data["count"]}')
    finally:
        rwlock.reader.release()
    
    # Writer with explicit acquire/release
    try:
        await rwlock.writer.acquire()
        data['count'] += 1
        print(f'Incremented to: {data["count"]}')
    finally:
        rwlock.writer.release()

asyncio.run(explicit_acquire_release())
```

--------------------------------

### Acquire Read Access

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/00-START-HERE.md

Use the 'async with lock.reader:' context manager to acquire read access to a shared resource. This allows multiple concurrent tasks to read simultaneously.

```python
# Read access (many concurrent tasks)
async with lock.reader:
    data = shared_resource
```

--------------------------------

### Import aiorwlock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/quick-reference.md

Import the aiorwlock library into your Python script.

```python
import aiorwlock
```

--------------------------------

### Use WriterLock as Async Context Manager

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/api-reference.md

Demonstrates using the WriterLock with an `async with` statement for automatic acquisition and release. The lock is acquired on entering the block and released on exiting.

```python
async with rwlock.writer:
    # Lock is acquired here
    await perform_write_operation()
    # Lock is automatically released here
```

--------------------------------

### Acquire Write Access

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/00-START-HERE.md

Use the 'async with lock.writer:' context manager to acquire exclusive write access to a shared resource. Only one task can write at a time.

```python
# Write access (exclusive)
async with lock.writer:
    shared_resource = new_data
```

--------------------------------

### Type Relationships Diagram

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/types.md

Illustrates the hierarchical and shared instance relationships between RWLock, _ReaderLock, _WriterLock, and _RWLockCore.

```text
RWLock
├── _reader_lock: _ReaderLock
│   └── _lock: _RWLockCore
└── _writer_lock: _WriterLock
    └── _lock: _RWLockCore (same instance)

_ReaderLock and _WriterLock
└── inherit from _ContextManagerMixin

Both `_ReaderLock` and `_WriterLock` reference the same `_RWLockCore` instance, allowing coordination between readers and writers.
```

--------------------------------

### Acquire and Release Write Lock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/api-reference.md

Demonstrates the basic pattern for acquiring and releasing the write lock using acquire() and release() methods. Ensure to release the lock in a finally block.

```python
rwlock = aiorwlock.RWLock()

# Simple acquire pattern
await rwlock.writer.acquire()
try:
    # Write operation
    shared_resource = new_value
finally:
    rwlock.writer.release()
```

--------------------------------

### Explicit Acquire/Release with RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

This pattern shows how to use explicit `acquire()` and `release()` methods for the reader lock when the `async with` statement is not suitable. It's crucial to use a `try...finally` block to ensure the lock is always released.

```python
async def explicit_locking(rwlock):
    """Use acquire/release when async with doesn't fit"""
    
    try:
        await rwlock.reader.acquire()
        result = read_resource()
        await process_async(result)
    finally:
        rwlock.reader.release()
```

--------------------------------

### Acquire Reader Lock (Async Context Manager)

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/quick-reference.md

Use the async context manager for recommended read access, ensuring proper lock acquisition and release.

```python
async with rwlock.reader:
    data = read_shared_resource()
```

--------------------------------

### Check-then-Act Pattern with RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

This pattern demonstrates how to check a condition under a read lock and then perform an update under a write lock if necessary. This ensures that the condition remains valid between the check and the update, preventing race conditions.

```python
async def check_then_update(rwlock, resource):
    """Check condition under read lock, update under write lock"""
    
    # Check condition without blocking other readers
    async with rwlock.reader:
        needs_update = check_condition(resource)
    
    # Update if needed - exclusive write access
    if needs_update:
        async with rwlock.writer:
            perform_update(resource)
```

--------------------------------

### Check RWLock State

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Demonstrates how to check if the reader or writer lock is currently held. Initialize an RWLock and observe its state before and during lock acquisition.

```python
import asyncio
import aiorwlock

async def check_lock_state():
    rwlock = aiorwlock.RWLock()
    
    print(rwlock.reader.locked)  # False
    print(rwlock.writer.locked)  # False
    
    async with rwlock.reader:
        print(rwlock.reader.locked)  # True
        print(rwlock.writer.locked)  # False
    
    async with rwlock.writer:
        print(rwlock.reader.locked)  # False
        print(rwlock.writer.locked)  # True

asyncio.run(check_lock_state())
```

--------------------------------

### Correct Pattern: Check Condition Under Read Lock, Then Write

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

This pattern illustrates how to check a condition using a read lock and then, if necessary, acquire a write lock separately to perform modifications. This avoids holding the read lock during the write attempt.

```python
async def check_then_write():
    rwlock = aiorwlock.RWLock()
    
    # Check under read lock
    async with rwlock.reader:
        needs_update = check_condition()
    
    # If update needed, acquire write lock separately
    if needs_update:
        async with rwlock.writer:
            perform_write_operation()
```

--------------------------------

### Timeout-Protected Acquire with RWLock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

This pattern demonstrates how to acquire a reader lock with a specified timeout using `asyncio.wait_for`. This prevents the application from hanging indefinitely if the lock cannot be acquired within the given time.

```python
async def acquire_with_timeout(rwlock, timeout=5.0):
    """Acquire lock with timeout to prevent hanging"""
    
    try:
        # Lock acquisition with timeout
        await asyncio.wait_for(
            rwlock.reader.acquire(),
            timeout=timeout
        )
        try:
            result = read_resource()
            return result
        finally:
            rwlock.reader.release()
    except asyncio.TimeoutError:
        print("Lock acquisition timed out")
        raise
```

--------------------------------

### Timeout-Protected RWLock Acquisition

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

Shows how to acquire locks with a timeout to prevent deadlocks. This is useful when a lock might be held indefinitely by another process.

```python
import asyncio
import aiorwlock

async def timeout_example():
    """Acquire lock with timeout protection"""
    
    rwlock = aiorwlock.RWLock()
    
    async def reader_with_timeout(reader_id: int, timeout: float):
        """Try to acquire reader lock with timeout"""
        try:
            await asyncio.wait_for(
                rwlock.reader.acquire(),
                timeout=timeout
            )
            try:
                print(f'Reader {reader_id}: acquired lock')
                await asyncio.sleep(0.05)
            finally:
                rwlock.reader.release()
        except asyncio.TimeoutError:
            print(f'Reader {reader_id}: timeout waiting for lock')
    
    async def writer_holds_lock():
        """Writer holds lock for a while"""
        async with rwlock.writer:
            print('Writer: holding lock...')
            await asyncio.sleep(1.0)
    
    # Reader times out because writer holds lock
    await asyncio.gather(
        writer_holds_lock(),
        reader_with_timeout(1, timeout=0.5),  # Will timeout
        reader_with_timeout(2, timeout=2.0),  # Will succeed
    )

asyncio.run(timeout_example())
```

--------------------------------

### Initialize Single RWLock for Multiple Resources

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Use a single RWLock instance to protect multiple data access operations within a class. This is useful when all operations need to be coordinated by the same lock.

```python
import asyncio
import aiorwlock

class ReadWriteData:
    def __init__(self, fast=False):
        self.rwlock = aiorwlock.RWLock(fast=fast)
        self.data = {}
    
    async def read_value(self, key):
        async with self.rwlock.reader:
            return self.data.get(key)
    
    async def write_value(self, key, value):
        async with self.rwlock.writer:
            self.data[key] = value
```

--------------------------------

### RWLock Creation

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/quick-reference.md

Creates a new RWLock instance. The `fast` parameter controls the locking mechanism, with `False` (default) prioritizing safety and `True` optimizing for high-contention, all-async scenarios.

```APIDOC
## RWLock(fast=False)

### Description
Creates a new RWLock instance. The `fast` parameter controls the locking mechanism, with `False` (default) prioritizing safety and `True` optimizing for high-contention, all-async scenarios.

### Parameters
- **fast** (bool) - Optional - If `True`, uses a faster, but potentially less safe, locking mechanism suitable for all-async code and high contention. Defaults to `False`.
```

--------------------------------

### Correct Pattern: Release Read Lock, Then Acquire Write Lock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

This pattern shows the correct way to handle operations that require both reading and writing by releasing the read lock before acquiring the write lock in a separate scope.

```python
async def correct_upgrade():
    rwlock = aiorwlock.RWLock()
    
    # First: read operation
    async with rwlock.reader:
        data = shared_resource
    
    # Then: write operation (separate lock scope)
    async with rwlock.writer:
        shared_resource = modified_data
```

--------------------------------

### Demonstrate Reentrant Read Lock Acquisition

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Illustrates that a task holding a read lock can acquire additional read locks without blocking. This behavior is supported by default.

```python
async def reentrant_read():
    rwlock = aiorwlock.RWLock()
    
    async with rwlock.reader:
        # This same task can acquire reader lock again
        async with rwlock.reader:
            print("nested read lock allowed")
```

--------------------------------

### AIORWLock Error Handling Patterns

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/examples.md

This snippet demonstrates several common error conditions when using AIORWLock, such as attempting to acquire/release locks across different event loops or tasks, and releasing a lock without a prior acquire. It also shows the recommended pattern for safe lock release using a `finally` block.

```python
import asyncio
import aiorwlock

async def error_handling_example():
    """Demonstrate error handling patterns"""
    
    rwlock = aiorwlock.RWLock()
    
    # Error 1: Different event loop
    try:
        async def first_loop():
            await rwlock.reader.acquire()
            rwlock.reader.release()
        
        async def second_loop():
            await rwlock.reader.acquire()
        
        asyncio.run(first_loop())
        asyncio.run(second_loop())  # Different loop!
    except RuntimeError as e:
        print(f'Event loop error: {e}')
    
    # Error 2: Release without acquire
    rwlock2 = aiorwlock.RWLock()
    try:
        rwlock2.reader.release()  # No matching acquire
    except RuntimeError as e:
        print(f'Release error: {e}')
    
    # Error 3: Cross-task release
    rwlock3 = aiorwlock.RWLock()
    
    async def task_a():
        await rwlock3.reader.acquire()
        # Task exits without releasing
    
    async def task_b():
        await asyncio.sleep(0.01)
        try:
            rwlock3.reader.release()  # Different task!
        except RuntimeError as e:
            print(f'Cross-task error: {e}')
    
    await asyncio.gather(task_a(), task_b())
    
    # Correct pattern: use finally
    rwlock4 = aiorwlock.RWLock()
    
    async def correct_release():
        await rwlock4.reader.acquire()
        try:
            print('Inside lock')
        finally:
            rwlock4.reader.release()
    
    await correct_release()

asyncio.run(error_handling_example())
```

--------------------------------

### Benchmark aiorwlock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/index.md

Benchmark the aiorwlock library by running tests with both fast=False and fast=True modes. Measure performance with a typical workload to determine the optimal setting for your application.

```python
# Run with fast=False and fast=True
# Measure with typical workload
# Choose based on measured results
```

--------------------------------

### Demonstrate Reentrant Write Lock Acquisition

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/configuration.md

Shows that a task holding a write lock can acquire additional write locks without blocking. This reentrant behavior is supported by default.

```python
async def reentrant_write():
    rwlock = aiorwlock.RWLock()
    
    async with rwlock.writer:
        # This same task can acquire writer lock again
        async with rwlock.writer:
            print("nested write lock allowed")
```

--------------------------------

### Define _ContextManagerMixin Class

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/types.md

Provides the async context manager protocol support for lock objects. It defines __aenter__ and __aexit__ methods for acquiring and releasing locks asynchronously.

```python
class _ContextManagerMixin:
    """Provides __aenter__ and __aexit__ for async context manager protocol."""
    
    __slots__ = ()

```

--------------------------------

### Correct Pattern: Single Event Loop Lock Usage

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/errors.md

This code illustrates the correct way to use aiorwlock by ensuring the lock is created and accessed within a single event loop. This avoids the RuntimeError related to cross-event loop usage.

```python
import asyncio
import aiorwlock

async def main():
    # Create lock in the single event loop
    rwlock = aiorwlock.RWLock()
    
    # Use it within the same loop
    await rwlock.reader.acquire()
    rwlock.reader.release()

asyncio.run(main())
```

--------------------------------

### Check Write Lock Status

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/api-reference.md

Illustrates how to check if the write lock is currently held using the locked property. The status changes based on whether the lock is acquired.

```python
rwlock = aiorwlock.RWLock()

print(rwlock.writer.locked)  # False initially

async with rwlock.writer:
    print(rwlock.writer.locked)  # True inside lock
    
print(rwlock.writer.locked)  # False after release
```

--------------------------------

### Correctly Using RWLock Across Event Loops

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/quick-reference.md

Ensure that a new RWLock instance is created within each asynchronous task to avoid RuntimeError when using locks across different event loops.

```python
# ✓ CORRECT: Create lock inside task
async def task():
    lock = aiorwlock.RWLock()
    async with lock.reader:
        pass
```

--------------------------------

### Correctly Upgrading Read Lock to Write Lock

Source: https://github.com/aio-libs/aiorwlock/blob/master/_autodocs/quick-reference.md

To upgrade a read lock to a write lock, release the read lock first and then acquire the write lock. This prevents a RuntimeError.

```python
# ✓ CORRECT: Release and re-acquire
async with rwlock.reader:
    data = read()

async with rwlock.writer:
    write(data)
```