Problem 85 · Concurrency & Python

Writer-Preferring Read-Write Lock

HardClassic DSA & systems

01Problem

Implement a read-write lock allowing many concurrent readers or one writer, where waiting writers block new readers so writers cannot starve.

02Key insight

One Condition guards three counters: active readers, writer held, waiting writers. Readers wait while a writer holds OR is waiting (writer preference); writers wait until no readers and no writer. Expose context managers so release always happens.

03Solution

solution.pypython50 lines

import threading
from contextlib import contextmanager
from typing import Iterator


class RWLock:
    def __init__(self) -> None:
        self._cond = threading.Condition()
        self._readers = 0
        self._writer = False
        self._waiting_writers = 0

    def acquire_read(self) -> None:
        with self._cond:
            self._cond.wait_for(lambda: not self._writer and self._waiting_writers == 0)
            self._readers += 1

    def release_read(self) -> None:
        with self._cond:
            self._readers -= 1
            if self._readers == 0:
                self._cond.notify_all()

    def acquire_write(self) -> None:
        with self._cond:
            self._waiting_writers += 1
            self._cond.wait_for(lambda: not self._writer and self._readers == 0)
            self._waiting_writers -= 1
            self._writer = True

    def release_write(self) -> None:
        with self._cond:
            self._writer = False
            self._cond.notify_all()

    @contextmanager
    def read(self) -> Iterator[None]:
        self.acquire_read()
        try:
            yield
        finally:
            self.release_read()

    @contextmanager
    def write(self) -> Iterator[None]:
        self.acquire_write()
        try:
            yield
        finally:
            self.release_write()

Try it first.

TimeO(1) per acquire/release (excluding waiting)

SpaceO(1)

tests.py · 3 passing

tests.pypython28 lines

import threading, time
lk = RWLock()
events = []
lk.acquire_read()
def writer():
    with lk.write():
        events.append("w")
def reader():
    with lk.read():
        events.append("r")
tw = threading.Thread(target=writer); tw.start()
while lk._waiting_writers == 0:
    time.sleep(0.0005)
tr = threading.Thread(target=reader); tr.start()
time.sleep(0.005)
assert events == []          # new reader blocked behind waiting writer
lk.release_read()
tw.join(); tr.join()
assert events == ["w", "r"]  # writer preferred
counter = [0]
def inc():
    for _ in range(200):
        with lk.write():
            counter[0] += 1
ts = [threading.Thread(target=inc) for _ in range(4)]
for t in ts: t.start()
for t in ts: t.join()
assert counter[0] == 800

04Staff-level follow-up

Writer preference can starve readers under constant writes; how would you fix it? Use phase-fair/ticket ordering (alternate batches of readers and writers), and note that in CPython a plain Lock is often faster unless read sections are long or release the GIL.