Lesson 35. Threads and Shared State

Mission link: Threads are the right tool for waiting, which is most of what a service does. The failures are not exotic: an exception nobody looked at, a lock held too long, and shared state that did not need to be shared.
Primary source: The Python Standard Library, concurrent.futures
Prerequisites: Lesson 34

Warm-up

1. ▢ Lesson 34: eight threads ran a check-then-act on one cache key. How many objects did they create?

2. ▢ What is a thread doing, in a program where threads help?

Know this

Start with a pool, not a thread

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(fetch, urls))

ThreadPoolExecutor bounds the concurrency, reuses threads, propagates return values, and joins everything on exit. Raw threading.Thread is for the long-lived background worker that a pool does not model: a scheduler loop, a log flusher, a signal handler.

Sizing: for waiting work, more workers than cores is correct, and the right number is set by the thing you are waiting on. Eight database connections in the pool means eight workers, not fifty, and exceeding the remote service's limits turns concurrency into rate-limit errors.

The exception nobody saw

This is the failure that ships:

with ThreadPoolExecutor(3) as pool:
    futures = [pool.submit(process, item) for item in items]

submitted and pool closed, no exception raised yet

The block exits cleanly, the process continues, and one item raised ValueError that nobody will ever see. A future holds its exception until someone asks:

for future in futures:
    future.result()          # re-raises here, in this thread

pool.map is safer, because iterating the result raises, but only when you actually iterate it and only up to the first failure. When every item matters, use as_completed and handle each one:

from concurrent.futures import as_completed

for future in as_completed(futures):
    try:
        handle(future.result())
    except SomeError:
        log.exception("item failed")

Rule: every submit needs a matching result() somewhere, or you have written except Exception: pass by accident.

Do not share; hand over

The best fix for shared mutable state is not a lock, it is a queue.

import queue

q: queue.Queue[Item | None] = queue.Queue()

def worker():
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        results.append(transform(item))     # append is safe; see lesson 34
        q.task_done()

queue.Queue is thread-safe by design and is the channel between producers and consumers. Verified: three workers draining 100 items produced exactly 100 results, with no lock in the code.

The other way to remove sharing is to return instead of accumulate. Each worker computes its own subtotal, the caller sums them. No lock, no queue, no ordering question.

Locks, when the invariant is genuinely shared

lock = threading.Lock()

with lock:
    if key not in cache:
        cache[key] = build(key)

Verified: a plain Lock acquired twice from one thread reported True then False with a timeout, and without a timeout the second acquire waits forever. An RLock nested inside itself is fine. That is the whole difference, and it matters when a locked method calls another locked method of the same object.

The call sequence is identical on both sides; only what the second acquire does differs. That is what makes the plain Lock version so hard to spot in review: neither method is wrong on its own, and the deadlock exists only in the fact that one of them calls the other.

Four rules that prevent nearly all lock bugs:

1. Hold it for as little as possible. Never across I/O if it can be avoided; compute outside, assign inside.
2. Always use with. A finally you wrote by hand will eventually be skipped by an early return.
3. One order, everywhere. Deadlock needs two locks taken in opposite orders by two threads. Pick a global order and document it.
4. Guard the invariant, not the object. The lock exists for the rule that spans statements.

lock.acquire(timeout=...) turns a deadlock into a detectable failure, which is worth it in a long-running service.

Thread-local state

data = threading.local()
data.request_id = generate()        # each thread has its own

Legitimate for per-thread context: a request id, a database connection, a transaction. Verified: five threads each kept their own value across a sleep. It is still global state, so it makes functions depend on invisible context and makes tests order-dependent; prefer passing the value explicitly, and use contextvars instead when asyncio is involved, since a thread-local is shared by every task on one event-loop thread.

Daemon threads and shutdown

t = threading.Thread(target=flush_loop, daemon=True)

A daemon thread does not keep the process alive, so the interpreter exits without waiting for it, and its finally blocks may not run. That makes it wrong for anything that must finish, such as flushing a buffer. The alternative is an Event:

stop = threading.Event()

def flush_loop():
    while not stop.wait(timeout=5.0):
        flush()

stop.set(); t.join(timeout=10)

Event.wait with a timeout is the correct idiom for a periodic loop: it sleeps and is interruptible, where time.sleep in a loop delays shutdown by up to one interval.

Practice

1. ▢ Find the defect.

   with ThreadPoolExecutor(8) as pool:
       for order in orders:
           pool.submit(charge, order)
   log.info("charged %d orders", len(orders))
   

with ThreadPoolExecutor(8) as pool:
    futures = {pool.submit(charge, o): o for o in orders}

failed = []
for future in as_completed(futures):
    try:
        future.result()
    except ChargeError:
        log.exception("charge failed for %s", futures[future].id)
        failed.append(futures[future])
log.info("charged %d orders, %d failed", len(orders) - len(failed), len(failed))

2. ▢ Rewrite without a lock.

   totals = {}
   lock = threading.Lock()
   
   def process(chunk):
       for row in chunk:
           with lock:
               totals[row.country] = totals.get(row.country, 0) + row.amount
   

from collections import Counter

def process(chunk) -> Counter[str]:
    local = Counter()
    for row in chunk:
        local[row.country] += row.amount
    return local

with ThreadPoolExecutor(8) as pool:
    totals = sum(pool.map(process, chunks), Counter())

3. ▢ Match the primitive.

   • a) At most five concurrent calls to a rate-limited API
   • b) Tell four worker threads to shut down
   • c) A method holding a lock that calls another method of the same object which also locks
   • d) A consumer that should sleep until there is work
   • e) A per-request correlation id available to every function in the call stack

4. ▢ Why is this dangerous, and what does the fix cost?

   with lock:
       response = http.get(url)          # network call, inside the lock
       cache[url] = response.json()
   

data = http.get(url).json()          # outside
with lock:
    cache.setdefault(url, data)      # inside, and cheap

5. ▢ A background thread writes metrics every five seconds and is created with daemon=True. On shutdown, up to five seconds of metrics are lost. Fix it.

stop = threading.Event()

def metrics_loop():
    try:
        while not stop.wait(5.0):
            flush()
    finally:
        flush()                      # the last one

t = threading.Thread(target=metrics_loop)    # not a daemon
t.start()
...
stop.set()
t.join(timeout=10)

6. ▢ A service uses 200 threads and gets slower than with 20. Give three explanations.

Real-world reps

• [ ] Grep for pool.submit and executor.submit in code you own and check that every one has a result() reached on some path. Each that does not is a silent failure.
• [ ] Find a lock in your code and measure how long it is held. If a network or database call happens inside it, move the call out and decide what duplication costs.
• [ ] Replace one lock-protected shared accumulator with per-worker results merged by the caller.
• [ ] Find every daemon=True thread and ask what it loses if the process exits mid-iteration.
• [ ] Tomorrow: set your pool size from the downstream limit rather than from the number of items, and see whether throughput changes.

Going further

• concurrent.futures: executors, futures, as_completed, and the exception semantics
• threading: every primitive, and the notes on daemon threads and shutdown
• queue: the thread-safe channel, including task_done and join
• contextvars: the per-task alternative to thread-local state
• Resources

Not landing? Reread the primary source at the top, since this lesson compresses it and compression is where understanding leaks. Check the glossary for any term that felt slippery.

If the lesson itself is unclear rather than the material, that is a defect: open an issue.