Function chain_reader

tests/test_read_write.py:175–204

(
    reader_index: int,
    lock_file: str,
    release_count: Synchronized[int],
    start_signal: EventType,
    release_signal: EventType,
    next_reader_signal: EventType | None,
    writer_or_prev_signal: EventType,
)

def chain_reader(
    reader_index: int,
    lock_file: str,
    release_count: Synchronized[int],
    start_signal: EventType,
    release_signal: EventType,
    next_reader_signal: EventType | None,
    writer_or_prev_signal: EventType,
) -> None:
    start_signal.wait(timeout=10)

    lock = ReadWriteLock(lock_file)
    with lock.read_lock():
        if reader_index > 0:
            # delay so writer can attempt acquisition while readers overlap
            time.sleep(2)

        if next_reader_signal is not None:
            next_reader_signal.set()

        if reader_index == 0:
            # first reader holds lock briefly then signals writer
            time.sleep(1)

        writer_or_prev_signal.set()

        release_signal.wait(timeout=10)

        with release_count.get_lock():
            release_count.value += 1


@pytest.mark.timeout(40)