Check and Update Under One Lock

Picture a ticket counter for a concert with 100 seats. Many threads serve buyers at once. Each thread asks "is there a seat left?" and, if there is, takes one. You must never sell more than 100 seats.

Two threads can both sell the last seat

Here is the obvious code:

def allow(self) -> bool:
    if self._sold < self._limit:
        self._sold += 1
        return True
    return False

With one seat left, two threads can both run the if before either one runs self._sold += 1. Both see a free seat, both sell it, and you have sold 101. This bug is called check-then-act. By the time you act on the check, its answer is out of date.

A lock around only self._sold += 1 does not fix it. Each increment is now safe, but both threads have already passed the check.

Lock the check and the update together

Hold one lock across the check and the update, so no other thread can get between them:

from threading import Lock
from typing import Final


class CapacityCounter:
    def __init__(self, limit: int) -> None:
        self._limit: Final[int] = limit
        self._lock: Final = Lock()
        self._sold = 0

    def allow(self) -> bool:
        with self._lock:
            if self._sold >= self._limit:
                return False
            self._sold += 1
            return True

The code inside with self._lock: is a critical section, code that only one thread can run at a time. Every method that reads or changes _sold must use the same lock. Here a read without the lock would only get an out-of-date number. Once the state has two fields, as in the exercise, a read without the lock can see one field updated and the other not.

Final tells the type checker that _lock and _limit are never reassigned, so a stray self._lock = Lock() in another method becomes a type error. It does nothing at runtime. The lock does the protecting.

Check your understanding Why use with lock: instead of calling lock.acquire() and lock.release()? Show answer

The with block releases the lock however the block ends, whether by a normal finish, a return, or an exception. With separate calls you need a try/finally to get the same behavior. If you forget the finally, an exception leaves the lock held, so every other thread that needs it waits forever.

Call acquire() yourself only when you need one of its options, such as a timeout. Then release in a finally, and only if acquire() returned True.

Hand out copies, not your internals

Callers often want a snapshot, such as how many seats are sold and what the limit is. If you hand out a dictionary that the counter keeps updating, a caller can change it without the lock, or watch it change later.

Build a new dictionary while you hold the lock, and wrap it in types.MappingProxyType, which is a read-only view of a dictionary:

from collections.abc import Mapping
from types import MappingProxyType


def snapshot(self) -> Mapping[str, int]:
    with self._lock:
        copy = {"sold": self._sold, "limit": self._limit}
    return MappingProxyType(copy)

The Mapping return type tells the type checker the result is read-only, but a plain dict would still accept writes at runtime. The proxy raises TypeError on a write. Wrap a fresh copy, not your live dictionary, because a proxy over the live one shows every later change.

Check your understanding Does a thread-safe dictionary make "check if there is room, then add" safe? Show answer

No. Each dictionary operation can be safe on its own, but your rule spans two operations, and another thread can run between them. Put both inside one lock, or give one thread sole ownership of the data.

Test it, and read it

A stress test with many threads can catch a race, but it can also pass by luck. Write the test, and also read the code. Every change to the shared state must happen inside the same lock.

See Python's threading documentation for the full Lock API.