Resilience — Race Conditions¶

import sys
import threading

N, THREADS = 1_000_000, 4
counter = 0
lock = threading.Lock()


def plus_one(value: int) -> int:
    return value + 1


def increment() -> None:
    global counter
    for _ in range(N):
        counter += 1                            # load, add, store


def increment_via_call() -> None:
    global counter
    for _ in range(N):
        counter = plus_one(counter)             # a function call between read and write


def increment_locked() -> None:
    global counter
    for _ in range(N):
        with lock:
            counter += 1


for target in (increment, increment_via_call, increment_locked):
    counter = 0
    threads = [threading.Thread(target=target) for _ in range(THREADS)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"{target.__name__:20} {counter:>9,} of {N * THREADS:,}  (GIL enabled: {sys._is_gil_enabled()})")

# app/inventory.py
import asyncio
import threading
from collections.abc import Callable


def _no_pause() -> None:
    pass


class Inventory:
    """Racy on purpose: check-then-act with a pause between the check and the write."""

    def __init__(self, stock: int, pause: Callable[[], None] = _no_pause) -> None:
        self.stock = stock
        self.sold = 0
        self._pause = pause                      # stands in for a log call, DB read, HTTP call

    def buy(self) -> bool:
        if self.stock > 0:                       # check
            self._pause()
            self.stock -= 1                      # act
            self.sold += 1
            return True
        return False


class SafeInventory(Inventory):
    def __init__(self, stock: int, pause: Callable[[], None] = _no_pause) -> None:
        super().__init__(stock, pause)
        self._lock = threading.Lock()

    def buy(self) -> bool:
        with self._lock:                         # check and act are one step for other threads
            return super().buy()


class Wallet:
    def __init__(self, balance: int) -> None:
        self.balance = balance
        self._lock = asyncio.Lock()

    async def withdraw_racy(self, amount: int) -> bool:
        if self.balance >= amount:
            await asyncio.sleep(0)               # any await: DB call, HTTP call, log shipping
            self.balance -= amount
            return True
        return False

    async def withdraw(self, amount: int) -> bool:
        async with self._lock:
            return await self.withdraw_racy(amount)

import threading
import time

from app.inventory import Inventory, SafeInventory


def run_threads(target, threads: int = 8, calls: int = 200) -> None:
    workers = [threading.Thread(target=lambda: [target() for _ in range(calls)]) for _ in range(threads)]
    for w in workers:
        w.start()
    for w in workers:
        w.join()


for cls in (Inventory, SafeInventory):
    inventory = cls(stock=1000, pause=lambda: time.sleep(0))   # sleep(0) releases the GIL
    run_threads(inventory.buy)
    print(cls.__name__, "stock", inventory.stock, "sold", inventory.sold)
# Inventory stock -7 sold 1007
# SafeInventory stock 0 sold 1000

uv python install 3.14t
uv run --python 3.14t python -c "import sys; print(sys._is_gil_enabled())"    # False

import asyncio

from app.inventory import Wallet


async def main() -> None:
    wallet = Wallet(balance=100)
    print(await asyncio.gather(wallet.withdraw_racy(80), wallet.withdraw_racy(80)), wallet.balance)
    # [True, True] -60

    wallet = Wallet(balance=100)
    print(await asyncio.gather(wallet.withdraw(80), wallet.withdraw(80)), wallet.balance)
    # [True, False] 20


asyncio.run(main())

import threading


class Cache:
    def __init__(self) -> None:
        self._lock = threading.RLock()               # re-entrant: the same thread may enter again
        self._data: dict[str, str] = {}

    def get_or_load(self, key: str) -> str:
        with self._lock:
            if key not in self._data:
                self.put(key, key.upper())           # put() takes the same lock: fine with RLock
            return self._data[key]

    def put(self, key: str, value: str) -> None:
        with self._lock:
            self._data[key] = value


class Gate:
    """Wait until a condition on shared state is true (e.g. N workers registered)."""

    def __init__(self) -> None:
        self._cond = threading.Condition()
        self.ready = 0

    def register(self) -> None:
        with self._cond:
            self.ready += 1
            self._cond.notify_all()

    def wait_for(self, n: int, timeout: float) -> bool:
        with self._cond:
            return self._cond.wait_for(lambda: self.ready >= n, timeout=timeout)


print(Cache().get_or_load("a"))                      # A
gate = Gate()
for _ in range(3):
    threading.Thread(target=gate.register).start()
print(gate.wait_for(3, timeout=2))                   # True

stop = threading.Event()                             # one-way flag: "shut down now"
worker = threading.Thread(target=lambda: stop.wait(10))
worker.start()
stop.set()
worker.join(1)
print("worker alive:", worker.is_alive())            # worker alive: False

import queue
import threading

results: list[int] = []
jobs: queue.Queue[int] = queue.Queue(maxsize=100)     # bounded: producers block when it is full


def worker() -> None:
    while True:
        try:
            item = jobs.get()
        except queue.ShutDown:                         # Python 3.13+
            return
        results.append(item * item)                    # one append: safe on both builds
        jobs.task_done()


threads = [threading.Thread(target=worker) for _ in range(4)]
for t in threads:
    t.start()
for i in range(1000):
    jobs.put(i)
jobs.join()                                            # wait until every item was processed
jobs.shutdown()                                        # wake workers blocked in get()
for t in threads:
    t.join()
print(len(results), sum(results) == sum(i * i for i in range(1000)))   # 1000 True

import asyncio


async def worker(name: str, q: asyncio.Queue[int], out: list[str]) -> None:
    while True:
        try:
            item = await q.get()
        except asyncio.QueueShutDown:                  # Python 3.13+
            return
        await asyncio.sleep(0.01)
        out.append(f"{name}:{item}")
        q.task_done()


async def main() -> None:
    q: asyncio.Queue[int] = asyncio.Queue(maxsize=10)
    out: list[str] = []
    async with asyncio.TaskGroup() as tg:
        for i in range(3):
            tg.create_task(worker(f"w{i}", q, out))
        for item in range(30):
            await q.put(item)                          # waits while the queue is full: backpressure
        await q.join()
        q.shutdown()
    print(len(out), "items processed")                 # 30 items processed


asyncio.run(main())

import asyncio
import contextvars

current_request: str | None = None                                        # global: shared by all tasks
request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id")


async def handle_global(rid: str) -> str:
    global current_request
    current_request = rid
    await asyncio.sleep(0.01)                     # another request runs here and overwrites it
    return f"{rid} logged as {current_request}"


async def handle_ctx(rid: str) -> str:
    request_id.set(rid)                           # each task runs in a copy of the context
    await asyncio.sleep(0.01)
    return f"{rid} logged as {request_id.get()}"


async def main() -> None:
    print(await asyncio.gather(handle_global("req-1"), handle_global("req-2")))
    # ['req-1 logged as req-2', 'req-2 logged as req-2']
    print(await asyncio.gather(handle_ctx("req-1"), handle_ctx("req-2")))
    # ['req-1 logged as req-1', 'req-2 logged as req-2']


asyncio.run(main())

import multiprocessing as mp


def add_racy(counter, n: int) -> None:
    for _ in range(n):
        counter.value += 1                        # read, add, write: three steps


def add_locked(counter, n: int) -> None:
    for _ in range(n):
        with counter.get_lock():                  # Value() comes with its own lock
            counter.value += 1


if __name__ == "__main__":
    for target in (add_racy, add_locked):
        counter = mp.Value("i", 0)
        procs = [mp.Process(target=target, args=(counter, 50_000)) for _ in range(4)]
        for p in procs:
            p.start()
        for p in procs:
            p.join()
        print(f"{target.__name__}: {counter.value} (expected 200000)")
# add_racy: 70237 (expected 200000)
# add_locked: 200000 (expected 200000)

import os
import tempfile
import threading
from pathlib import Path


def claim_racy(path: Path, owner: str, barrier: threading.Barrier) -> bool:
    if not path.exists():                          # time of check
        barrier.wait()                             # both threads are now past the check
        path.write_text(owner)                     # time of use: the second write wins
        return True
    return False


def claim(path: Path, owner: str, barrier: threading.Barrier) -> bool:
    barrier.wait()
    try:
        with path.open("x") as f:                  # O_CREAT | O_EXCL: create or fail, atomically
            f.write(owner)
        return True
    except FileExistsError:
        return False


def write_atomic(path: Path, text: str) -> None:
    """Readers see the old file or the new file, never a half-written one."""
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
    with os.fdopen(fd, "w") as f:
        f.write(text)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)                          # atomic rename on the same file system


def race(fn) -> tuple[dict[str, bool], str]:
    with tempfile.TemporaryDirectory() as d:
        path = Path(d) / "job.lock"
        barrier = threading.Barrier(2)
        results: dict[str, bool] = {}
        threads = [
            threading.Thread(target=lambda o=o: results.__setitem__(o, fn(path, o, barrier)))
            for o in ("worker-a", "worker-b")
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        return results, path.read_text()


print("racy: ", race(claim_racy))    # both claim it: ({'worker-b': True, 'worker-a': True}, 'worker-a')
print("fixed:", race(claim))         # exactly one:   ({'worker-b': True, 'worker-a': False}, 'worker-b')

# app/stock_db.py
from collections.abc import Callable

import psycopg

SCHEMA = """
CREATE TABLE IF NOT EXISTS products (
    id      int PRIMARY KEY,
    stock   int NOT NULL CHECK (stock >= 0),
    version int NOT NULL DEFAULT 1
);
CREATE TABLE IF NOT EXISTS payments (
    id              bigserial PRIMARY KEY,
    idempotency_key text NOT NULL UNIQUE,
    order_id        int NOT NULL,
    amount_cents    int NOT NULL
);
CREATE TABLE IF NOT EXISTS daily_counts (
    day date PRIMARY KEY,
    n   int NOT NULL
);
"""


def _noop() -> None:
    pass


def buy_racy(conn: psycopg.Connection, product_id: int, after_read: Callable[[], None] = _noop) -> bool:
    """Lost update: read in Python, decide, write back an absolute value."""
    with conn.transaction():
        (stock,) = conn.execute("SELECT stock FROM products WHERE id = %s", (product_id,)).fetchone()
        after_read()
        if stock <= 0:
            return False
        conn.execute("UPDATE products SET stock = %s WHERE id = %s", (stock - 1, product_id))
        return True


def buy_atomic(conn: psycopg.Connection, product_id: int) -> bool:
    """One statement: the database does check and write under its row lock."""
    with conn.transaction():
        cur = conn.execute(
            "UPDATE products SET stock = stock - 1 WHERE id = %s AND stock > 0", (product_id,)
        )
        return cur.rowcount == 1


def buy_for_update(conn: psycopg.Connection, product_id: int, after_read: Callable[[], None] = _noop) -> bool:
    """Pessimistic lock: other transactions wait at SELECT ... FOR UPDATE until we commit."""
    with conn.transaction():
        (stock,) = conn.execute(
            "SELECT stock FROM products WHERE id = %s FOR UPDATE", (product_id,)
        ).fetchone()
        after_read()
        if stock <= 0:
            return False
        conn.execute("UPDATE products SET stock = %s WHERE id = %s", (stock - 1, product_id))
        return True


class ConflictError(Exception):
    """Someone else changed the row since we read it: re-read and try again."""


def buy_optimistic(conn: psycopg.Connection, product_id: int, after_read: Callable[[], None] = _noop) -> bool:
    """Optimistic lock: no lock while thinking, the UPDATE checks that nothing changed."""
    with conn.transaction():
        stock, version = conn.execute(
            "SELECT stock, version FROM products WHERE id = %s", (product_id,)
        ).fetchone()
        after_read()
        if stock <= 0:
            return False
        cur = conn.execute(
            "UPDATE products SET stock = %s, version = version + 1 WHERE id = %s AND version = %s",
            (stock - 1, product_id, version),
        )
        if cur.rowcount == 0:
            raise ConflictError(product_id)
        return True


def record_payment(conn: psycopg.Connection, key: str, order_id: int, amount_cents: int) -> bool:
    """Idempotent insert: the UNIQUE constraint decides, not a SELECT before the INSERT."""
    with conn.transaction():
        cur = conn.execute(
            """
            INSERT INTO payments (idempotency_key, order_id, amount_cents)
            VALUES (%s, %s, %s)
            ON CONFLICT (idempotency_key) DO NOTHING
            """,
            (key, order_id, amount_cents),
        )
        return cur.rowcount == 1                   # False: a duplicate, already recorded


def count_event(conn: psycopg.Connection, day: str) -> int:
    """Upsert: insert the first row or increment the existing one, in one statement."""
    with conn.transaction():
        (n,) = conn.execute(
            """
            INSERT INTO daily_counts (day, n) VALUES (%s, 1)
            ON CONFLICT (day) DO UPDATE SET n = daily_counts.n + 1
            RETURNING n
            """,
            (day,),
        ).fetchone()
        return n

# app/consumer.py
import psycopg

SCHEMA = """
CREATE TABLE IF NOT EXISTS accounts (
    id            int PRIMARY KEY,
    balance_cents bigint NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_messages (
    message_id   text PRIMARY KEY,
    processed_at timestamptz NOT NULL DEFAULT now()
);
"""


def handle_deposit(conn: psycopg.Connection, message_id: str, account_id: int, amount_cents: int) -> bool:
    """Apply a deposit message once, however many times it is delivered."""
    with conn.transaction():                        # the marker and the effect commit together
        cur = conn.execute(
            "INSERT INTO processed_messages (message_id) VALUES (%s) ON CONFLICT DO NOTHING",
            (message_id,),
        )
        if cur.rowcount == 0:
            return False                            # duplicate delivery: already applied
        conn.execute(
            "UPDATE accounts SET balance_cents = balance_cents + %s WHERE id = %s",
            (amount_cents, account_id),
        )
        return True