### Install and Test ReaderWriterLock from Wheel File (Bash)

Source: https://github.com/elarivie/pyreaderwriterlock/blob/master/tests/test_manual.md

This snippet outlines the steps to install the readerwriterlock package from a downloaded wheel file and then test its basic functionality. It ensures the package is not already installed, installs the specified wheel file, and verifies the import and lock acquisition.

```bash
python3 -m pip uninstall readerwriterlock typing_extensions
```

```bash
python3 -m pip install readerwriterlock-#.#.#.whl
```

```bash
python3 -c "import readerwriterlock; readerwriterlock.RWLockFair().gen_wlock().acquire();"
```

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

### Install readerwriterlock library

Source: https://github.com/elarivie/pyreaderwriterlock/blob/master/README.md

Command to install the readerwriterlock package using pip.

```bash
python3 -m pip install -U readerwriterlock
```

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

### Implement Multi-threaded Reader-Writer Synchronization

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

A complete example showing how to coordinate multiple reader and writer threads accessing a shared resource using an RWLockFair instance.

```python
from readerwriterlock import rwlock
import threading
import time

def multi_threaded_example():
    my_rwlock = rwlock.RWLockFair()
    shared_counter = {"value": 0}

    def writer_task(writer_id):
        lock = my_rwlock.gen_wlock()
        for _ in range(5):
            with lock:
                shared_counter["value"] += 1
                print(f"Writer {writer_id}: counter = {shared_counter['value']}")
            time.sleep(0.01)

    def reader_task(reader_id):
        lock = my_rwlock.gen_rlock()
        for _ in range(5):
            with lock:
                print(f"Reader {reader_id}: sees counter = {shared_counter['value']}")
            time.sleep(0.01)

    threads = []
    for i in range(2):
        t = threading.Thread(target=writer_task, args=(i,))
        threads.append(t)
    for i in range(3):
        t = threading.Thread(target=reader_task, args=(i,))
        threads.append(t)
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    print(f"Final counter value: {shared_counter['value']}")

multi_threaded_example()
```

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

### Install and Test ReaderWriterLock from Source Tarball (Bash)

Source: https://github.com/elarivie/pyreaderwriterlock/blob/master/tests/test_manual.md

This snippet demonstrates the process of installing the readerwriterlock package from a downloaded source tarball and performing a basic import and lock acquisition test. It includes uninstalling previous versions, installing the local package, and verifying the import and lock functionality.

```bash
python3 -m pip uninstall readerwriterlock typing_extensions;
python3 -m pip install .;
python3 -c "from readerwriterlock import rwlock";
python3 -m pip uninstall readerwriterlock;
```

```bash
python3 -c "from readerwriterlock import rwlock; rwlock.RWLockFair().gen_wlock().acquire();"
```

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

### Initialize and use Reader-Writer locks

Source: https://github.com/elarivie/pyreaderwriterlock/blob/master/README.md

Demonstrates how to instantiate a lock class, generate read/write locks, and use them with context managers or manual acquisition.

```python
from readerwriterlock import rwlock
a = rwlock.RWLockFairD()

# Generate locks
a_reader_lock = a.gen_rlock()
a_writer_lock = a.gen_wlock()

# Pythonic usage
with a.gen_rlock():
  # Read stuff
  pass

with a.gen_wlock():
  # Write stuff
  pass

# Timeout usage
b = a.gen_wlock()
if b.acquire(blocking=True, timeout=5):
  try:
    # Do stuff
    pass
  finally:
    b.release()

# Downgrade usage
b = a.gen_wlock()
if b.acquire():
  try:
    # Read/Write stuff
    b = b.downgrade()
    # Read stuff
  finally:
    b.release()
```

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

### Configure Custom Lock Factory and Time Source

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Demonstrates how to initialize an RWLockFair instance with custom threading locks and time sources for specialized testing or synchronization requirements.

```python
from readerwriterlock import rwlock
import threading
import time

# Use custom lock factory and time source
custom_rwlock = rwlock.RWLockFair(
    lock_factory=threading.Lock,
    time_source=time.perf_counter
)

reader = custom_rwlock.gen_rlock()
writer = custom_rwlock.gen_wlock()

with writer:
    print("Using custom lock factory")
```

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

### Run library tests

Source: https://github.com/elarivie/pyreaderwriterlock/blob/master/README.md

Command to execute the project test suite with coverage reporting.

```bash
make check.test.coverage
```

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

### Configure Lock Acquisition with Timeouts

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Shows how to use non-blocking acquisition and timeouts for controlled lock access.

```python
from readerwriterlock import rwlock

my_rwlock = rwlock.RWLockFair()
writer_lock = my_rwlock.gen_wlock()

if writer_lock.acquire(blocking=False):
    try:
        print("Got writer lock immediately")
    finally:
        writer_lock.release()

with writer_lock:
    other_reader = my_rwlock.gen_rlock()
    acquired = other_reader.acquire(blocking=True, timeout=0.5)
    if acquired:
        try:
            print("Got reader lock within timeout")
        finally:
            other_reader.release()
```

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

### Implement Fair Lock

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Demonstrates using RWLockFair to process acquisition requests in FIFO order to prevent starvation.

```python
from readerwriterlock import rwlock

my_rwlock = rwlock.RWLockFair()
reader_lock = my_rwlock.gen_rlock()
writer_lock = my_rwlock.gen_wlock()

cache = {}

with writer_lock:
    cache["key"] = "value"

with reader_lock:
    print(f"Cache contents: {cache}")
```

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

### Implement Writer-Priority Lock

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Demonstrates using RWLockWrite to ensure writers take precedence over waiting readers.

```python
from readerwriterlock import rwlock

my_rwlock = rwlock.RWLockWrite()
reader_lock = my_rwlock.gen_rlock()
writer_lock = my_rwlock.gen_wlock()

shared_resource = []

with writer_lock:
    shared_resource.append("data")
    print(f"Writer appended data: {shared_resource}")

with reader_lock:
    print(f"Reader sees: {shared_resource}")
```

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

### Implement Reader-Priority Lock

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Demonstrates using RWLockRead to allow multiple concurrent readers while preventing writer starvation.

```python
from readerwriterlock import rwlock

my_rwlock = rwlock.RWLockRead()
reader_lock = my_rwlock.gen_rlock()
writer_lock = my_rwlock.gen_wlock()

shared_data = {"counter": 0}

with writer_lock:
    shared_data["counter"] += 1
    print(f"Writer updated counter to: {shared_data['counter']}")

with reader_lock:
    print(f"Reader sees counter: {shared_data['counter']}")
```

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

### Utilize Protocol Types for Static and Runtime Checking

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Shows how to use library-defined Protocol types to enforce interface compliance for locks, enabling both static type checking and runtime verification.

```python
from readerwriterlock import rwlock

def use_any_rwlock(rw: rwlock.RWLockable):
    reader = rw.gen_rlock()
    writer = rw.gen_wlock()
    with writer:
        print("Writing...")
    with reader:
        print("Reading...")

use_any_rwlock(rwlock.RWLockRead())
use_any_rwlock(rwlock.RWLockWrite())
use_any_rwlock(rwlock.RWLockFair())

lock = rwlock.RWLockFair()
print(f"Is RWLockable: {isinstance(lock, rwlock.RWLockable)}")
```

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

### Atomic Writer-to-Reader Lock Downgrade

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Demonstrates how to acquire an exclusive writer lock, perform modifications, and atomically downgrade to a reader lock to allow shared access without releasing the lock entirely.

```python
writer_lock.acquire()
try:
    shared_state["data"] = "important value"
    shared_state["initialized"] = True
    reader_lock = writer_lock.downgrade()
    print(f"After downgrade, reading: {shared_state}")
finally:
    reader_lock.release()
```

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

### Asyncio Writer-Priority and Fair Locks

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Basic implementations of asynchronous writer-priority and fair locks for managing shared state in concurrent asyncio tasks.

```python
import asyncio
from readerwriterlock import rwlock_async
async def main():
    my_rwlock = rwlock_async.RWLockWrite()
    writer_lock = await my_rwlock.gen_wlock()
    reader_lock = await my_rwlock.gen_rlock()
    results = []
    async with writer_lock:
        results.append("written")
    async with reader_lock:
        print(f"Results: {results}")
asyncio.run(main())
```

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

### Lock Acquisition and Management

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Methods for initializing different lock strategies and managing acquisition modes including blocking, non-blocking, and timeouts.

```APIDOC
## Lock Initialization and Usage

### Description
This section covers the initialization of various lock strategies (Reader-Priority, Writer-Priority, Fair) and standard acquisition patterns.

### Method
N/A (Library Class Methods)

### Endpoint
`readerwriterlock.rwlock`

### Parameters
#### Request Body
- **blocking** (bool) - Optional - If True, wait for the lock; if False, return immediately.
- **timeout** (float) - Optional - Maximum time in seconds to wait for the lock.

### Request Example
```python
from readerwriterlock import rwlock
my_rwlock = rwlock.RWLockFair()
writer_lock = my_rwlock.gen_wlock()
writer_lock.acquire(blocking=True, timeout=0.5)
```

### Response
#### Success Response (200)
- **locked** (bool) - Returns True if the lock is currently held by any thread.
```

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

### Asyncio Reader-Priority Lock

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Asynchronous reader-priority lock implementation for asyncio applications, supporting context manager usage and manual acquisition with timeouts.

```python
import asyncio
from readerwriterlock import rwlock_async
async def main():
    my_rwlock = rwlock_async.RWLockRead()
    shared_data = {"value": 0}
    writer_lock = await my_rwlock.gen_wlock()
    reader_lock = await my_rwlock.gen_rlock()
    async with writer_lock:
        shared_data["value"] = 42
    async with reader_lock:
        print(f"Async reader sees value: {shared_data['value']}")
    await writer_lock.acquire(blocking=True, timeout=1.0)
    try:
        shared_data["value"] += 1
    finally:
        await writer_lock.release()
asyncio.run(main())
```

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

### Asyncio Downgradable Locks

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Asynchronous versions of downgradable locks, allowing tasks to transition from exclusive write access to shared read access atomically.

```python
import asyncio
from readerwriterlock import rwlock_async
async def main():
    rwlock_write = rwlock_async.RWLockWriteD()
    writer_lock = await rwlock_write.gen_wlock()
    state = {"computed": False, "result": None}
    await writer_lock.acquire()
    try:
        state["result"] = sum(range(100))
        state["computed"] = True
        reader_lock = await writer_lock.downgrade()
        print(f"Computed result: {state['result']}")
    finally:
        await reader_lock.release()
asyncio.run(main())
```

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

### RWLockFairD Downgradable Fair Lock

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Provides a fair, FIFO-ordered lock that supports atomic downgrading, ensuring that requests are processed in the order they were received.

```python
from readerwriterlock import rwlock
my_rwlock = rwlock.RWLockFairD()
writer_lock = my_rwlock.gen_wlock()
database = {"users": []}
writer_lock.acquire()
try:
    database["users"].append({"id": 1, "name": "Alice"})
    reader_lock = writer_lock.downgrade()
    user_count = len(database["users"])
    print(f"User count after insert: {user_count}")
finally:
    reader_lock.release()
```

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

### Downgradable Locks

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Using Downgradable locks to atomically convert a writer lock to a reader lock.

```APIDOC
## Downgradable Lock Acquisition

### Description
Allows a thread to convert an exclusive writer lock to a shared reader lock without releasing the lock, ensuring no other writers can intervene.

### Method
N/A

### Endpoint
`readerwriterlock.rwlock.RWLockReadD`

### Parameters
#### Request Body
- **downgrade()** (method) - Required - Atomically transitions the lock state from exclusive to shared.

### Request Example
```python
my_rwlock = rwlock.RWLockReadD()
writer_lock = my_rwlock.gen_wlock()
with writer_lock:
    # Perform write operations
    writer_lock.downgrade()
    # Now holding reader lock
```

### Response
#### Success Response (200)
- **status** (string) - Lock successfully transitioned to shared reader mode.
```

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

### RWLockWriteD Downgradable Writer Priority Lock

Source: https://context7.com/elarivie/pyreaderwriterlock/llms.txt

Implements a writer-priority lock that supports atomic downgrading, ensuring writers have precedence while allowing for efficient transition to shared read access.

```python
from readerwriterlock import rwlock
my_rwlock = rwlock.RWLockWriteD()
writer_lock = my_rwlock.gen_wlock()
configuration = {}
writer_lock.acquire()
try:
    configuration["setting"] = "new_value"
    reader_lock = writer_lock.downgrade()
    assert "setting" in configuration
    print(f"Configuration validated: {configuration}")
finally:
    reader_lock.release()
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.