Concurrency: Threads, Locks and Async I/O

A server handles many requests at once, so shared state, ordering and waiting are everyday concerns. Concurrency questions test whether you can spot race conditions, choose between threads, processes and async I/O, use locks and queues correctly, and avoid deadlocks. The examples here are in Python because they are easy to run and check; the concepts apply equally to Java, Go and Node.js, and the notes say where each differs.

1. Concurrency versus parallelism

• Concurrency: structuring a program so several tasks make progress in overlapping time periods (possibly on one core, by interleaving).
• Parallelism: tasks literally executing at the same instant on multiple cores.

A web server mostly waits (network, disk, database). Waiting-heavy (I/O-bound) work benefits from concurrency even on one core. Computation-heavy (CPU-bound) work needs parallelism across cores.

Python note: the Global Interpreter Lock (GIL) lets only one thread run Python bytecode at a time in standard CPython, so threads help I/O-bound code but not CPU-bound code; use multiprocessing for the latter. (Free-threaded builds are emerging, so check the version you use.) Java and Go run threads and goroutines in parallel on all cores. Node.js runs JavaScript on one thread with an event loop and offloads blocking work to a thread pool and worker threads.

2. Race conditions

A race condition is a bug whose outcome depends on the timing of concurrent operations. The classic form is check-then-act or read-modify-write on shared state without protection.

import threading, time

class Counter:
    def __init__(self):
        self.n = 0

def unsafe_increment(counter, times):
    for _ in range(times):
        current = counter.n          # read
        time.sleep(0)                # yield the CPU, as a real scheduler might at any moment
        counter.n = current + 1      # write: overwrites increments made in between

c = Counter()
threads = [threading.Thread(target=unsafe_increment, args=(c, 200)) for _ in range(4)]
[t.start() for t in threads]; [t.join() for t in threads]
assert c.n < 800                     # increments were lost: the result is wrong and nondeterministic

n += 1 looks like one operation but is three steps (load, add, store), and another thread can run between them.

Fix 1: a lock (mutual exclusion)

Only one thread at a time may hold the lock; the critical section it protects runs atomically with respect to other holders.

class SafeCounter:
    def __init__(self):
        self.n = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:                  # released automatically, even if an exception occurs
            current = self.n
            time.sleep(0)
            self.n = current + 1

sc = SafeCounter()
def worker():
    for _ in range(200):
        sc.increment()
threads = [threading.Thread(target=worker) for _ in range(4)]
[t.start() for t in threads]; [t.join() for t in threads]
assert sc.n == 800                       # every increment counted

Keep critical sections small: hold the lock only around the shared-state access, never around slow I/O.

Other fixes

• Atomic operations (AtomicInteger in Java, sync/atomic in Go) for single counters and flags.
• Immutability: shared data that never changes needs no locking.
• Confinement: give each piece of state to one thread or actor and communicate by messages (Go's "do not communicate by sharing memory; share memory by communicating").
• Database-level atomicity: UPDATE ... SET n = n + 1, transactions and unique constraints for state shared between processes.

3. Deadlock, livelock and starvation

Deadlock needs four conditions at once (Coffman): mutual exclusion, hold and wait, no preemption and circular wait. Break any one to prevent it. The most practical fix is a global lock ordering: everybody acquires locks in the same order.

a, b = threading.Lock(), threading.Lock()
log = []

def transfer_in_order(first, second, name):
    # always acquire the lower-ranked lock first, whatever the direction of the transfer
    ordered = sorted([first, second], key=id)
    with ordered[0]:
        time.sleep(0.01)                             # widen the window where a deadlock would occur
        with ordered[1]:
            log.append(name)

t1 = threading.Thread(target=transfer_in_order, args=(a, b, "a->b"))
t2 = threading.Thread(target=transfer_in_order, args=(b, a, "b->a"))
t1.start(); t2.start()
t1.join(timeout=2); t2.join(timeout=2)
assert not t1.is_alive() and not t2.is_alive()       # both finished: no deadlock
assert sorted(log) == ["a->b", "b->a"]

Without the ordering, thread 1 would hold a and wait for b while thread 2 holds b and waits for a, forever. Other defences: try-lock with timeout and back off, acquire all locks at once, and avoid calling unknown code (callbacks) while holding a lock.

Livelock: threads keep reacting to each other and make no progress (two people stepping aside in a corridor). Starvation: a thread never gets the resource (unfair locks, priority issues).

4. Producer-consumer and thread pools

A bounded queue decouples producers from consumers and provides backpressure: when the queue is full, producers block (or are rejected), so memory stays bounded instead of growing without limit.

import queue

q = queue.Queue(maxsize=5)
results, lock = [], threading.Lock()
SENTINEL = object()

def producer():
    for i in range(20):
        q.put(i)                                    # blocks when the queue is full: backpressure
    for _ in range(3):
        q.put(SENTINEL)                             # one stop signal per consumer

def consumer():
    while True:
        item = q.get()
        if item is SENTINEL:
            break
        with lock:
            results.append(item * 2)

consumers = [threading.Thread(target=consumer) for _ in range(3)]
p = threading.Thread(target=producer)
[t.start() for t in consumers]; p.start()
p.join(); [t.join() for t in consumers]
assert sorted(results) == [i * 2 for i in range(20)]  # every item processed exactly once

A thread pool reuses a fixed set of worker threads for many tasks (ThreadPoolExecutor, Java's ExecutorService, Go's worker goroutines). Size it deliberately: for CPU-bound work about the number of cores; for I/O-bound work larger, from latency and target throughput (Little's law: concurrency = throughput × latency).

from concurrent.futures import ThreadPoolExecutor

def fetch(i):
    time.sleep(0.05)                                 # pretend network call
    return i * i

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=10) as pool:
    out = list(pool.map(fetch, range(20)))
elapsed = time.perf_counter() - start
assert out == [i * i for i in range(20)]            # results keep input order
assert elapsed < 0.5                                 # 20 calls of 50 ms finish in about 100 ms, not 1 second

Little's law states that the average number of items in a system $L$ equals the arrival rate $\lambda$ times the average time spent in the system $W$: $L=\lambda W$. A service that handles 200 requests per second with 50 ms average latency has about $200\times 0.05=10$ requests in flight on average.

def in_flight(requests_per_second, latency_seconds):
    return requests_per_second * latency_seconds
assert abs(in_flight(200, 0.05) - 10) < 1e-9
assert abs(in_flight(1000, 0.2) - 200) < 1e-9        # slower responses mean many more concurrent requests: size pools and limits for it

5. Async I/O and the event loop

Instead of one thread per connection, an event loop runs many tasks on one thread, switching whenever a task awaits I/O. Cheap per-connection cost makes tens of thousands of connections feasible.

import asyncio

async def call_service(name, delay):
    await asyncio.sleep(delay)                       # yields to the loop while "waiting on the network"
    return name

async def sequential():
    return [await call_service("a", 0.05), await call_service("b", 0.05), await call_service("c", 0.05)]

async def concurrent():
    return await asyncio.gather(call_service("a", 0.05), call_service("b", 0.05), call_service("c", 0.05))

t0 = time.perf_counter(); assert asyncio.run(sequential()) == ["a", "b", "c"]; seq = time.perf_counter() - t0
t0 = time.perf_counter(); assert asyncio.run(concurrent()) == ["a", "b", "c"]; con = time.perf_counter() - t0
assert seq > 0.14 and con < 0.1                      # three 50 ms calls: about 150 ms one after another, about 50 ms together

The cardinal rule of async code: never block the event loop. A CPU-heavy loop or a blocking call (a synchronous database driver, time.sleep, a large JSON parse) stops every other task. Offload to a thread or process pool (run_in_executor, worker_threads) or use non-blocking libraries.

async def heartbeat(ticks):
    for _ in range(5):
        await asyncio.sleep(0.01)
        ticks.append(time.perf_counter())

async def blocking_task():
    time.sleep(0.2)                                  # blocks the whole loop

async def demo():
    ticks = []
    start = time.perf_counter()
    await asyncio.gather(heartbeat(ticks), blocking_task())
    return start, ticks

start, ticks = asyncio.run(demo())
assert ticks[0] - start >= 0.2                       # a 10 ms heartbeat could not run until the blocking call released the loop

Async cancellation and timeouts: always bound waits (asyncio.wait_for, AbortController, Go contexts, Java Future.get(timeout)), and make tasks cancellable and cleanup-safe.

async def slow():
    await asyncio.sleep(1)

async def with_timeout():
    try:
        await asyncio.wait_for(slow(), timeout=0.05)
    except asyncio.TimeoutError:
        return "timed out"

assert asyncio.run(with_timeout()) == "timed out"

6. Memory visibility and the memory model

On modern hardware, threads may see stale values because of CPU caches and compiler reordering. In Java and Go, a write by one thread is guaranteed visible to another only if there is a happens-before relationship (a lock, a volatile or atomic access, a channel send and receive). Without it, a flag set by one thread may never be seen by another, or may be seen out of order. The practical rule: do not share mutable state without synchronisation, and prefer higher-level tools (queues, channels, concurrent collections).

Double-checked locking (a lazy singleton) is a classic interview trap: without proper volatile (Java) a thread may see a partially constructed object.

7. Concurrency in different languages

Detect races with tools: ThreadSanitizer, Go's race detector, Java stress tests (jcstress) and careful review.

8. Concurrency across processes and machines

Locks inside one process do not protect against two servers running the same code. For shared resources use:

• Database transactions, unique constraints and row locks (see the transactions chapter).
• Optimistic concurrency with version columns.
• Distributed locks (Redis, ZooKeeper, etcd, database advisory locks) with expiry and fencing tokens.
• Idempotent operations and queues with single consumers per partition to serialise work per key.

# a fencing token: the resource rejects writes from a client holding an out-of-date lock
class Resource:
    def __init__(self):
        self.highest_token, self.value = 0, None

    def write(self, token, value):
        if token < self.highest_token:
            return False                              # a stale lock holder (paused, then resumed) is refused
        self.highest_token, self.value = token, value
        return True

r = Resource()
assert r.write(33, "from client B")                   # B got token 33 after A's lock expired
assert not r.write(32, "from client A, resumed late") # A still thinks it holds the lock; the token proves it does not
assert r.value == "from client B"

9. Common mistakes

• Unsynchronised shared state and check-then-act races.
• Holding a lock while doing I/O or calling external code.
• Inconsistent lock ordering, leading to deadlock.
• Blocking the event loop in async code.
• Unbounded queues and thread pools that hide overload until memory runs out.
• Assuming a lock in one process protects a multi-instance deployment.
• Swallowing exceptions in threads so failures vanish.
• Sleeping to "fix" a race instead of synchronising.
• Thread-per-request designs with no limit on concurrency.

10. Practice questions

1. What is a race condition? Show a read-modify-write example and fix it three ways.
2. List the four conditions for deadlock and how to prevent it in practice.
3. Threads, processes or async: which for a service that mostly calls other services? For image resizing?
4. What happens if you call a blocking function inside an async handler?
5. Implement a bounded blocking queue, or a thread-safe LRU cache.
6. What is Little's law and how does it help size a pool?
7. How do you protect a resource shared by many service instances?
8. Explain what a fencing token solves.