Concurrency Design Guidelines

Updated

September 8, 2026

Concurrency Design Guidelines

Concurrency is a tool you add after a sequential program is correct. The boring defaults: do not share mutable state, bound the work, and stay sequential until you have a measurement. This chapter shows a leak, then the fix, three times.

Mental model

Shared memory is the bug factory. If two threads own a list, you now own a lock story. If a parent can get a return value instead, do that.

Unbounded Thread() or create_task in a for over an open-ended rail will take the process down. A pool with max_workers (or a TaskGroup over a slice) is a budget.

time.perf_counter around the sequential version is cheaper than a redesign. If 3 milliseconds became 4 with threads, you bought noise.

Worked examples

Case 1: Leak — shared list; fix — return a value

Save as shared_rail.py. Workers append to one list with no lock. The length is often right; the contents and order are not a contract.

# shared_rail.py
from threading import Thread


def fire(ticket, bucket):
    bucket.append(f"fired {ticket}")


def main():
    tickets = ["T-11", "T-12", "T-13"]
    bucket = []
    workers = [Thread(target=fire, args=(t, bucket)) for t in tickets]
    for worker in workers:
        worker.start()
    for worker in workers:
        worker.join()
    print(bucket)


if __name__ == "__main__":
    main()

Run:

uv run python shared_rail.py

Output (order may vary):

['fired T-11', 'fired T-12', 'fired T-13']

Save as return_rail.py. The executor owns the workers. Each call returns a string. The parent prints in input order. No shared list.

# return_rail.py
from concurrent.futures import ThreadPoolExecutor


def fire(ticket):
    return f"fired {ticket}"


def main():
    tickets = ["T-11", "T-12", "T-13"]
    with ThreadPoolExecutor(max_workers=3) as pool:
        for line in pool.map(fire, tickets):
            print(line)


if __name__ == "__main__":
    main()

Run:

uv run python return_rail.py

Output:

fired T-11
fired T-12
fired T-13

If you need a running total, sum the returned numbers in the parent. Do not publish a counter every thread bumps.

Case 2: Leak — unbounded threads; fix — a bound pool

Save as unbounded.py. One thread per ticket, no cap. Fine for three. Fatal for a file with 50_000 lines.

# unbounded.py
from threading import Thread


def fire(ticket, bucket):
    bucket.append(ticket)


def main():
    tickets = [f"T-{n}" for n in range(3)]
    bucket = []
    workers = []
    for ticket in tickets:
        worker = Thread(target=fire, args=(ticket, bucket))
        worker.start()
        workers.append(worker)
    for worker in workers:
        worker.join()
    print(len(bucket))


if __name__ == "__main__":
    main()

Run:

uv run python unbounded.py

Output:

Save as bounded.py. max_workers=2 is the budget. The rest of the rail waits in the queue inside the executor.

# bounded.py
from concurrent.futures import ThreadPoolExecutor


def fire(ticket):
    return ticket


def main():
    tickets = [f"T-{n}" for n in range(3)]
    with ThreadPoolExecutor(max_workers=2) as pool:
        print(len(list(pool.map(fire, tickets))))


if __name__ == "__main__":
    main()

Run:

uv run python bounded.py

Output:

Same three tickets. The bound is the design. Pick max_workers from the I/O you are waiting on (a handful), not from the size of the input.

Case 3: Leak — threads for a cheap loop; fix — sequential until measured

Save as needless_pool.py. The work is a few additions. The pool is theatre.

# needless_pool.py
from concurrent.futures import ThreadPoolExecutor


def total_for_table(orders):
    return sum(orders)


def main():
    tables = [[1200, 800], [400], [500, 500]]
    with ThreadPoolExecutor(max_workers=3) as pool:
        print(sum(pool.map(total_for_table, tables)))


if __name__ == "__main__":
    main()

Run:

uv run python needless_pool.py

Output:

Save as measured.py. Same arithmetic. One stack. If this is slow, the next move is a better loop or a process pool after a timer, not more threads.

# measured.py
import time


def total_for_table(orders):
    return sum(orders)


def main():
    tables = [[1200, 800], [400], [500, 500]]
    started = time.perf_counter()
    grand = sum(total_for_table(orders) for orders in tables)
    elapsed = time.perf_counter() - started
    print(grand)
    print(f"elapsed < 0.01: {elapsed < 0.01}")


if __name__ == "__main__":
    main()

Run:

uv run python measured.py

Output:

3400
elapsed < 0.01: True

When elapsed < 0.01 is true, you do not have a concurrency problem.

The trap

Mixing all three leaks: a shared dict, a new Thread per key, no join on error paths, no bound. The fix is not a bigger lock. The fix is Case 1’s return values, Case 2’s pool, and Case 3’s timer, in that order.

Save as three_leaks.py only as a warning. Prefer not to run it as a template.

# three_leaks.py
from threading import Thread


def bump(counts, table):
    counts[table] = counts.get(table, 0) + 1


def main():
    counts = {}
    workers = []
    for table in [4, 2, 4, 2, 4]:
        worker = Thread(target=bump, args=(counts, table))
        worker.start()
        workers.append(worker)
    for worker in workers:
        worker.join()
    print(counts)


if __name__ == "__main__":
    main()

Run:

uv run python three_leaks.py

Output (you might get this; you might not):

{4: 3, 2: 2}

dict.get plus assign is not atomic. The print can lie. The sequential fix:

# three_fixes.py
from collections import Counter


def main():
    tables = [4, 2, 4, 2, 4]
    print(dict(Counter(tables)))


if __name__ == "__main__":
    main()

Run:

uv run python three_fixes.py

Output:

{4: 3, 2: 2}

No threads. Same answer every run. That is the design.

The boring rule

Do not share mutable state. Return values; sum in the parent.
Bound workers (max_workers, a TaskGroup over a known list). Join or shutdown on every path.
Sequential until perf_counter says you are waiting or burning CPU.
Threads for I/O, processes for CPU, asyncio for many I/O waits in one process. Pick one style per program.
A green print from a racy program is not a test. If order or totals matter, make them deterministic.

Try this

Add a Lock to shared_rail.py. Confirm it still does not give you input order. Then switch to return_rail.py.
Change bounded.py to 20 tickets and max_workers=4. Print each line from .map.
Time needless_pool.py and measured.py with perf_counter. Keep the faster one.
Rewrite three_leaks.py with ThreadPoolExecutor and a worker that returns table. Count in the parent with Counter.