Function chain_reader

tests/test_read_write.py:563–590

(
    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=_PROCESS_DEADLINE)

    lock = ReadWriteLock(lock_file)
    with lock.read_lock():
        if reader_index > 0:
            time.sleep(2)  # overlap with earlier readers so the writer contends mid-chain

        if next_reader_signal is not None:
            next_reader_signal.set()

        if reader_index == 0:
            time.sleep(1)  # hold briefly before releasing the writer

        writer_or_prev_signal.set()

        release_signal.wait(timeout=_PROCESS_DEADLINE)

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


def try_illegal_transition(lock_file: str, hold_mode: Literal["read", "write"], result: Synchronized[int]) -> None: