Skip to content

Concurrency for LLD in Python

TL;DR

The GIL serialises bytecodes, not your invariants, so += still loses updates and shared mutable state still needs a lock. Know four things cold: which lock protects which state, Condition for producer-consumer, compare-and-set versus a held lock, and lock ordering for deadlocks. Threads for blocking I/O, asyncio for thousands of waits, multiprocessing for CPU work.

Concepts

Every artifact below lives in code/fundamentals/concurrency.py, tested in code/fundamentals/tests/test_concurrency.py without a single sleep — the races are forced with a Barrier, not waited for.

The GIL protects the interpreter, not your invariants

One thread executes Python bytecode at a time, and CPython switches threads every few milliseconds or whenever a thread blocks. That buys two things: the interpreter's own structures stay consistent, and an operation that finishes inside one C call — list.append — is effectively atomic. It buys nothing at the level you care about.

self.value += 1 compiles to load, add, store. A thread can be preempted between the load and the store, and the value it eventually stores overwrites whatever another thread wrote in between:

code/fundamentals/concurrency.py — the same field, with and without a lock
class UnsafeCounter:
    """``value += 1`` is load, add, store. The GIL serialises *bytecodes*, not
    statements, so a thread can run between the two and lose someone's write."""

    def __init__(self) -> None:
        self.value = 0

    def increment(self) -> None:
        self.value += 1


class SafeCounter:
    """The same field with one lock. ``with`` releases it even when the body raises."""

    def __init__(self) -> None:
        self._value = 0
        self._lock = threading.Lock()

    def increment(self) -> None:
        with self._lock:
            self._value += 1

    @property
    def value(self) -> int:
        with self._lock:
            return self._value


def forced_lost_update() -> int:
    """Pin the interleaving instead of hoping for it: both threads read, meet at
    the barrier, then write. Two increments applied, counter ends at 1, every run."""
    counter = UnsafeCounter()
    gate = threading.Barrier(2)

    def read_then_write() -> None:
        seen = counter.value
        gate.wait(timeout=BARRIER_TIMEOUT_SECONDS)
        counter.value = seen + 1

    with ThreadPoolExecutor(max_workers=2) as pool:
        for future in [pool.submit(read_then_write) for _ in range(2)]:
            future.result()
    return counter.value

forced_lost_update is what makes this teachable: rather than running a million increments and hoping for a race, both threads read, meet at a Barrier, then write. Two increments applied, counter ends at 1, every run. Note the cost side: an uncontended lock costs about 17 ns, so the lock is almost never the performance problem; contention is.

The primitives, and when each is the right answer

Primitive Reach for it when
Lock One thread at a time in a critical section — the default
RLock The same thread may re-enter (one public method calls another)
Condition A thread must wait for a predicate, not just for a lock
Semaphore At most N in flight (a connection pool, a rate cap)
BoundedSemaphore The same, but an unmatched release raises instead of leaking
Event One-shot broadcast — started, shutting down
Barrier All N arrive before any proceeds — the test-writing tool
Timer Fire once, later; cancellable
queue.Queue Hand-off between threads with the locking already written
ThreadPoolExecutor You want results and exceptions back, not threads

Two rules make the table usable. Always use the with form, so an exception cannot leave a lock held. And always wait inside a while, never an if: a wakeup says the predicate might hold, not that it does.

Producer-consumer with a Condition

queue.Queue is what you ship; a bounded buffer is what the interviewer wants to watch you write. The instructive version uses two Condition objects over one Lock, so a put wakes a waiting consumer rather than every blocked producer too.

code/fundamentals/concurrency.py — the bounded buffer
class BoundedBuffer[T]:
    """Producer-consumer: two condition variables sharing one lock.

    Both waits sit inside a ``while`` because a wakeup is a hint, not a promise,
    and two conditions mean a ``put`` wakes a consumer, not every producer too.
    """

    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            raise ValidationError("capacity must be positive")
        self._capacity = capacity
        self._items: deque[T] = deque()
        self._lock = threading.Lock()
        self._not_full = threading.Condition(self._lock)
        self._not_empty = threading.Condition(self._lock)

    def put(self, item: T) -> None:
        with self._not_full:
            while len(self._items) == self._capacity:
                self._not_full.wait()
            self._items.append(item)
            self._not_empty.notify()

    def get(self) -> T:
        with self._not_empty:
            while not self._items:
                self._not_empty.wait()
            item = self._items.popleft()
            self._not_full.notify()
            return item

    def __len__(self) -> int:
        with self._lock:
            return len(self._items)

A producer blocking on a full buffer, and the consumer that frees it.

sequenceDiagram
    autonumber
    participant P as Producer
    participant B as BoundedBuffer
    participant C as Consumer
    P->>B: put(item), acquires the shared lock
    alt buffer is full
        B->>B: not_full.wait() releases the lock and parks the producer
        C->>B: get() now acquires the same lock
        B->>B: popleft then not_full.notify()
        B-->>C: item
        B->>B: producer wakes, re-tests the while condition
    end
    B->>B: append(item) then not_empty.notify()
    B-->>P: return

Two things to say aloud: wait() releases the lock while it parks (otherwise nothing could make the predicate true), and the woken producer re-checks capacity, because a third producer may have taken the slot first.

A read-write lock you can write from memory

Many readers share; a writer needs the place to itself. The right structure whenever reads dominate writes — a routing table, a configuration registry.

code/fundamentals/concurrency.py — writer-preferring read-write lock
class ReadWriteLock:
    """Many concurrent readers, one exclusive writer, writers preferred.

    The condition's own lock guards ``_readers``, ``_writer`` and
    ``_waiting_writers``. A reader waits while a writer is active *or* queued,
    so readers cannot starve a writer; the mirror risk is the trade you accept.
    Not reentrant: re-taking the read lock deadlocks once a writer has queued.
    """

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

    def acquire_read(self) -> None:
        with self._condition:
            while self._writer or self._waiting_writers:
                self._condition.wait()
            self._readers += 1

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

    def acquire_write(self) -> None:
        with self._condition:
            self._waiting_writers += 1
            try:
                while self._writer or self._readers:
                    self._condition.wait()
            finally:
                self._waiting_writers -= 1
            self._writer = True

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

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

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

The decision worth naming is preference. Here a reader waits while a writer is active or queued, so a stream of readers cannot starve a writer; readers starved by constant writes is the trade you accept. An interviewer who asks "what if reads never stop?" is checking that you chose rather than copied. It is also not reentrant: re-taking the read lock deadlocks once a writer has queued.

Thread-safe singletons and double-checked locking

If a lazily created global must exist, double-checked locking is the idiom: an unlocked check for the common case, then a second check under the lock that is the correct one.

code/fundamentals/concurrency.py — double-checked locking
class MetricsRegistry:
    """One instance per process, published under double-checked locking.

    The unlocked first check answers the common case; the second, under
    ``_instance_lock``, is the correct one. On CPython the store to
    ``_instance`` happens only after ``cls()`` returns, so no thread sees a
    half-built registry. ``_instance_lock`` guards creation, ``_lock`` the counters.
    """

    _instance: ClassVar[MetricsRegistry | None] = None
    _instance_lock: ClassVar[threading.Lock] = threading.Lock()

    def __init__(self) -> None:
        self._counters: dict[str, int] = {}
        self._lock = threading.Lock()

    @classmethod
    def instance(cls) -> MetricsRegistry:
        if cls._instance is None:
            with cls._instance_lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance

    @classmethod
    def reset(cls) -> None:
        with cls._instance_lock:  # test hook; needing one hints that injection is simpler
            cls._instance = None

    def increment(self, name: str, by: int = 1) -> None:
        with self._lock:
            self._counters[name] = self._counters.get(name, 0) + by

    def snapshot(self) -> dict[str, int]:
        with self._lock:
            return dict(self._counters)

Two details separate a good answer from a recited one. The two locks do different jobs: _instance_lock guards creation, _lock guards the counters afterwards. And this idiom was historically broken in Java, where a partially constructed object could be published before its constructor finished; on CPython the store to _instance happens only after cls() returns, so no thread sees a half-built registry. Then add the judgement: a singleton is global mutable state, reset() exists only for tests, and one instance built in main and injected is the better default.

Optimistic versus pessimistic locking

Pessimistic locking prevents contention by holding a lock across the whole read-modify-write. Optimistic locking detects it: read the value with its version, compute, write only if the version has not moved.

code/fundamentals/concurrency.py — compare-and-set with a bounded retry loop
@dataclass(frozen=True, slots=True)
class Versioned[T]:
    """A value and the version it was read at; immutable, so a stale copy stays stale."""

    value: T
    version: int


class OptimisticStore[T]:
    """Compare-and-set on a version: detect contention instead of preventing it.
    ``_lock`` is an ``RLock`` so the pessimistic path can reuse ``read``."""

    def __init__(self) -> None:
        self._entries: dict[str, Versioned[T]] = {}
        self._lock = threading.RLock()

    def put(self, key: str, value: T) -> Versioned[T]:
        with self._lock:
            entry = Versioned(value, 1)
            self._entries[key] = entry
            return entry

    def read(self, key: str) -> Versioned[T]:
        with self._lock:
            try:
                return self._entries[key]
            except KeyError:
                raise NotFoundError(f"no entry for {key!r}") from None

    def compare_and_set(self, key: str, expected_version: int, value: T) -> Versioned[T]:
        with self._lock:
            current = self.read(key)
            if current.version != expected_version:
                raise ConflictError(
                    f"{key!r} moved on: expected version {expected_version}, found {current.version}"
                )
            updated = Versioned(value, current.version + 1)
            self._entries[key] = updated
            return updated

    def update(self, key: str, change: Callable[[T], T], attempts: int = MAX_CAS_ATTEMPTS) -> Versioned[T]:
        """Read, compute outside the lock, write only if nothing moved underneath.
        Bounded, so a permanently hot key raises instead of spinning forever."""
        for _ in range(attempts):
            current = self.read(key)
            try:
                return self.compare_and_set(key, current.version, change(current.value))
            except ConflictError:
                continue
        raise ConflictError(f"gave up on {key!r} after {attempts} attempts")

    def update_pessimistically(self, key: str, change: Callable[[T], T]) -> Versioned[T]:
        """The pessimistic twin: hold the lock across the whole read-modify-write.
        No retries, no lost updates, but every writer waits for ``change``."""
        with self._lock:
            current = self.read(key)
            return self.compare_and_set(key, current.version, change(current.value))
Pessimistic Optimistic (version / CAS)
Cost when uncontended A lock acquire A lock acquire, plus a version compare
Cost when contended Everyone waits, including for slow work Losers retry, so work is repeated
Best when Writes are frequent or change is slow Conflicts are rare and change is cheap
Failure mode Convoying, and deadlock with two locks Livelock on a hot key, so bound the retries

The retry loop is bounded on purpose: a permanently hot key raises ConflictError instead of spinning. It is the same trade the parking-lot design makes on a lost spot claim, and the one a database makes between row locks and snapshot isolation.

Deadlock: ordering, then timeouts

A deadlock needs a cycle in the lock graph. transfer(a, b) against transfer(b, a) is the textbook one; the cure is to give every thread the same path through the graph by sorting on a stable key:

code/fundamentals/concurrency.py — lock ordering, and bounded acquisition
@dataclass(slots=True)
class Account:
    id: str
    balance: int
    lock: threading.Lock = field(default_factory=threading.Lock)


def transfer(source: Account, target: Account, amount: int) -> None:
    """Deadlock-free because both locks are taken in ``id`` order: ``transfer(a, b)``
    racing ``transfer(b, a)`` is the textbook cycle, and a stable key removes it."""
    if amount <= 0:
        raise ValidationError("transfer amount must be positive")
    if source.id == target.id:
        raise ValidationError("cannot transfer to the same account")
    first, second = sorted((source, target), key=lambda account: account.id)
    with first.lock, second.lock:
        if source.balance < amount:
            raise ConflictError(f"{source.id} holds {source.balance}, not {amount}")
        source.balance -= amount
        target.balance += amount


def try_transfer(source: Account, target: Account, amount: int, timeout: float = LOCK_TIMEOUT_SECONDS) -> bool:
    """The other cure when no global order exists: bounded acquisition, so the
    caller gets a back-off decision instead of a hang to debug."""
    if not source.lock.acquire(timeout=timeout):
        return False
    try:
        if not target.lock.acquire(timeout=timeout):
            return False
        try:
            if source.balance < amount:
                return False
            source.balance -= amount
            target.balance += amount
            return True
        finally:
            target.lock.release()
    finally:
        source.lock.release()

Ordering is the first answer because it removes the cycle rather than surviving it. try_transfer is the fallback when no global order exists: acquire with a timeout and hand back a clean False to retry on, rather than a hang for someone to debug. Name the third option too — hold one lock instead of two, by moving the invariant into a single owner.

Threads, asyncio, or processes

The choice you should be able to draw in ten seconds.

flowchart TD
    d_work{"What is the work?"}
    p_mp["multiprocessing or ProcessPoolExecutor"]
    d_scale{"Thousands of concurrent waits?"}
    p_async["asyncio, one event loop, no locks"]
    p_threads["threading or ThreadPoolExecutor"]
    d_shared{"Shared mutable state?"}
    p_none["No lock: pass values in, return results out"]
    p_queue["queue.Queue or a bounded buffer"]
    p_rw["ReadWriteLock"]
    p_lock["Lock, or CAS when conflicts are rare"]
    d_work -->|"CPU-bound"| p_mp
    d_work -->|"blocking I/O"| d_scale
    d_scale -->|"yes"| p_async
    d_scale -->|"no"| p_threads
    p_threads --> d_shared
    d_shared -->|"none"| p_none
    d_shared -->|"hand-off"| p_queue
    d_shared -->|"read-mostly"| p_rw
    d_shared -->|"read-modify-write"| p_lock

asyncio is cooperative concurrency on one thread: awaits are explicit yield points, so many data races disappear, but one blocking call stalls everything and CPU-bound work gains nothing. multiprocessing sidesteps the GIL at the cost of pickling and separate memory, which is why you share by message, not by object.

Running python -m fundamentals.concurrency prints the same numbers every time:

--- += is not atomic: two threads read 0, both write 1 ---
2 increments applied, counter shows 1
--- one Lock, 8 threads x 5000 increments: expected 40000, got 40000 ---
--- read-write lock ---
5 readers inside at once; value after 4 exclusive writers: 4
--- bounded buffer (capacity 2): 12 items through, 0 left ---
--- singleton: 16 threads race ---
distinct instances: 1; counted: {'requests': 16}
--- optimistic CAS: 4 threads x 25 updates -> value 100 at version 101 ---
--- lock ordering: 8 workers transferring both ways ---
balances 300 and 300, total conserved at 600

Applying it in the interview

Concurrency arrives at minutes 35–42 of the LLD framework, and there are three questions to answer.

Which lock protects what. Name the state, the lock and the granularity in one sentence: "the floor owns a lock over its spot statuses, so two gates racing for the last spot on floor 1 serialise while floor 2 is unaffected." One lock over the whole aggregate loses points — not because it is wrong, but because it shows you did not consider granularity.

Which invariant it defends. A lock only means something next to the rule it enforces — one vehicle per spot, one charge per ticket, no negative balance. If you cannot state the invariant, you cannot argue the lock is in the right place.

How you would test it. The differentiator, and it is cheap: "a ThreadPoolExecutor with 40 arrivals against 10 spots, asserting every spot is used exactly once", or "a Barrier so all 16 threads reach instance() together, then assert one distinct object".

Two smaller moves. Volunteer where you would not lock: immutable value objects need none, nor does thread-confined state. And name the primitive with its reason — RLock because a public method calls another, Condition because the wait is on a predicate, Semaphore because the cap is N rather than one.

Interview tip

When asked "is this thread-safe?", answer in the shape state, lock, invariant, test: "the ticket registry is shared, _tickets_lock guards it, the invariant is that a ticket leaves ACTIVE exactly once — which is why begin_checkout moves it to PAYING under the lock — and the test hands one ticket to two exit gates and asserts one raises." Four clauses, fifteen seconds, everything the rubric asks.

Pitfalls

  • Believing the GIL makes code thread-safe. It makes single bytecodes atomic. +=, check-then-act and every read-modify-write are still races.
  • Waiting on an if. if not self._items: self._condition.wait() is a bug — the item can be taken before the woken thread runs. Always a while.
  • Holding a lock across slow work. I/O or a callback inside the critical section turns one slow listener into a stalled system. Compute outside, mutate inside, notify after.
  • Wrong lock granularity. One lock per aggregate serialises everything; one per element makes any multi-element operation a lock-ordering problem. Pick the middle unit and justify it.
  • Unbounded retry loops. An optimistic update with no attempt cap turns a hot key into a spin. Bound it and raise.
  • Threads for CPU-bound work. Eight threads hashing passwords run slower than one, trading the GIL back and forth. That is multiprocessing.

Common mistake

Testing concurrency with sleep. A time.sleep(0.1) between two threads and an assertion afterwards proves nothing: it passes on your laptop, fails on a loaded CI box, and never forces the interleaving you claim to test. Use a Barrier so every thread reaches the critical section together, ThreadPoolExecutor so worker exceptions re-raise at future.result(), and a short bounded result(timeout=...) when the point is that a thread is blocked.

Exercises

  1. Make a cache thread-safe. An LRU cache uses an OrderedDict: get looks up then calls move_to_end, put inserts then evicts. Which need locking, and which lock?

    Solution

    Both, with one Lock over the whole OrderedDict: each is check-then-act across two operations, so two threads can move_to_end the same key and two puts at capacity can evict twice. A ReadWriteLock looks tempting for a read-mostly cache but does not help — get mutates recency, so it is a writer. An LRU cache has no read-only path, which is why a real one shards its keys into independently locked segments.

  2. Spot the deadlock. Order.cancel() takes the order lock then calls inventory.restock(sku), which takes the inventory lock. Inventory.reserve(sku) takes the inventory lock then calls order.hold(), which takes the order lock. Give two fixes.

    Solution

    The cycle is order-then-inventory against inventory-then-order. Fix one: impose a global order — always take the inventory lock first, on both paths. Fix two, better: stop holding one lock while calling a component that takes another, so cancel() releases the order lock and then restocks, or publishes an event inventory consumes on its own. Fix three, when callers are outside your control: acquire(timeout=...) and retry, which survives the cycle rather than removing it.

  3. Choose the locking style. A booking reads an in-memory seat, checks it is free, and marks it held. Bookings for a hot show arrive in bursts. Optimistic or pessimistic?

    Solution

    Pessimistic per seat. Conflicts on a hot show are common and change is trivially fast, so an optimistic retry storm just burns CPU re-reading the same contended seat. Do not lock the whole show — that serialises every buyer. Moved to a database the same shape becomes SELECT ... FOR UPDATE on the row, or a conditional update on a version column when the row is read far more often than written.

  4. Make a flaky test deterministic. A test starts two threads, sleeps 200 ms, then asserts a counter equals 2. Rewrite it so it cannot flake, and say what each piece buys.

    Solution

    Submit both callables to a ThreadPoolExecutor and call future.result() on each: the pool joins the threads and re-raises what a bare Thread would swallow. If the point is that they overlap, put a threading.Barrier(2) inside the callable so neither finishes before both arrive — that makes overlap a fact rather than a hope, and the barrier timeout turns a hang into a failed assertion. The sleep then has nothing left to do.