Python - Concurrency Quick Reference

1. Function/construct table

threading

Construct	Signature	What it does
Thread	Thread(target=fn, args=()).start() / .join()	Run fn on a new OS thread; no return value
Lock / RLock	.acquire() / .release() or with lock:	Mutual exclusion; RLock = re-entrant (same thread can re-acquire)
Condition	.wait() / .notify() / .notify_all()	Wait/notify coordination over shared state
Event	.set() / .clear() / .wait() / .is_set()	One-shot boolean signal
Semaphore	.acquire() / .release()	Cap concurrent access to N holders
Barrier	Barrier(n), .wait()	Block until N threads all arrive
threading.local()	attribute access	Per-thread storage, no locking needed

multiprocessing

Construct	Signature	What it does
Process	Process(target=fn, args=()).start() / .join()	Run fn on a new OS process; no return value
Pool	Pool(n)	Worker pool for parallel calls
Pool.map	.map(fn, iterable)	1 iterable, 1 arg per call, blocks, returns list
Pool.starmap	.starmap(fn, iterable_of_tuples)	Unpacks each tuple as multi-args
Pool.apply_async	.apply_async(fn, args)	1 async call, returns AsyncResult (has .get())
Value / Array	Value('i', 0)	Shared memory primitives across processes
Lock (mp's own)	same API as threading's	Cross-process mutual exclusion
initializer/initargs	kwarg on Pool/Process	Run setup once per worker at startup

asyncio

Construct	Signature	What it does
async def	—	Defines a coroutine function
await	—	Yield control to loop at a cooperative point
asyncio.run	asyncio.run(coro())	Start loop, run one coroutine to completion
create_task	asyncio.create_task(coro())	Schedule coroutine concurrently, returns Task immediately
gather	await asyncio.gather(*coros)	Run several concurrently, collect all results in order
TaskGroup	async with asyncio.TaskGroup() as tg: tg.create_task(...)	Structured concurrency; auto-cancels siblings on first exception
asyncio.sleep	await asyncio.sleep(n)	Non-blocking wait, yields to loop
Lock/Event/Semaphore	await lock.acquire() etc.	Same concepts as threading's, but await-based
to_thread	await asyncio.to_thread(fn, *args)	Run blocking sync fn on a worker thread
run_in_executor	await loop.run_in_executor(executor, fn, *args)	Same, lets you pass your own Executor (thread/process)

concurrent.futures

Construct	Signature	What it does
ThreadPoolExecutor	ThreadPoolExecutor(max_workers=n)	Pool of reused threads
ProcessPoolExecutor	ProcessPoolExecutor(max_workers=n)	Pool of reused processes
.submit	.submit(fn, *args) → Future	1 call, 1 Future, heterogeneous tasks
.map	.map(fn, *iterables)	Same fn across N inputs (zipped), ordered, lazy iterator
Future.result	.result(timeout=None)	Block for value; re-raises worker's exception
Future.exception	.exception(timeout=None)	Block for exception object; no raise
Future.done/running/cancelled	.done() etc.	State queries
Future.cancel	.cancel() → bool	Works only if still PENDING
Future.add_done_callback	.add_done_callback(fn)	Fire fn(future) on completion
as_completed	as_completed(futures, timeout=None)	Yield futures one at a time, completion order
wait	wait(futures, timeout=None, return_when=...)	One blocking call → (done, not_done); never raises on timeout
initializer/initargs	kwarg on ProcessPoolExecutor	Same pattern as Pool's
chunksize	kwarg on ProcessPoolExecutor.map	Batch items per IPC round-trip

---

2. Heavily commented snippets

threading

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100_000):
        with lock:            # acquire before touching shared state, auto-release after block
            counter += 1      # NOT atomic without the lock — GIL switches mid-operation

threads = [threading.Thread(target=increment) for _ in range(4)]  # 4 OS threads, same process
for t in threads:
    t.start()                 # start() schedules it; does NOT block
for t in threads:
    t.join()                  # block main thread until this one finishes; no return value available
print(counter)                # 400000 — correct only because of the lock

multiprocessing

from multiprocessing import Pool, Value, Lock

def cpu_heavy(n):
    return sum(i * i for i in range(n))   # pure Python CPU work — needs real processes to parallelize

if __name__ == "__main__":                # REQUIRED guard — spawn re-imports this module
    with Pool(processes=4) as pool:        # 4 worker OS processes, separate interpreters/GILs
        results = pool.map(cpu_heavy, [10**6] * 4)   # 1 iterable, 1 arg per call, blocks, returns list
        # pool.starmap(fn, [(a1,b1),(a2,b2)]) -> unpacks each tuple as multiple positional args
    print(results)

    shared_val = Value('i', 0)             # 'i' = C int, lives in shared memory across processes
    shared_lock = Lock()                   # multiprocessing's OWN Lock — NOT threading.Lock, not interchangeable
    # pass both via initializer/initargs to workers that need to mutate shared_val safely

asyncio

import asyncio

async def fetch(n):
    print(f"start {n}")
    await asyncio.sleep(1)     # NON-blocking wait — yields control back to loop, does NOT occupy the thread
    print(f"done {n}")
    return n * 2

async def main():
    # create_task schedules immediately, runs concurrently — NOT the same as multiprocessing's Process/Pool
    task = asyncio.create_task(fetch(99))

    results = await asyncio.gather(*[fetch(i) for i in range(3)])  # run many concurrently, ordered results
    print(results)

    await task                 # must await it eventually or it may be GC'd before completing

    # blocking sync call inside async code — WOULD freeze the whole loop if awaited directly:
    # time.sleep(2)  # <-- never do this inside a coroutine
    await asyncio.to_thread(lambda: None)  # correct way: offload blocking work to a thread, don't block loop

asyncio.run(main())             # starts the loop, runs main() to completion, only entry point needed

concurrent.futures

from concurrent.futures import ThreadPoolExecutor, as_completed, wait, FIRST_EXCEPTION

def fetch(url):
    return f"fetched {url}"    # any regular blocking callable — no async needed

with ThreadPoolExecutor(max_workers=4) as ex:   # __exit__ calls shutdown(wait=True) — blocks til ALL done
    future = ex.submit(fetch, "a.com")          # 1 call -> 1 Future, heterogeneous tasks OK
    print(future.result())                      # blocks for THIS future's value; re-raises its exception

    futures = {ex.submit(fetch, u): u for u in ["a.com", "b.com"]}  # dict: Future -> original input
    for f in as_completed(futures):              # yields in COMPLETION order, not submission order
        print(futures[f], "->", f.result())

    done, not_done = wait(futures, timeout=5, return_when=FIRST_EXCEPTION)
    # wait() NEVER raises on timeout — just returns whatever split exists; not_done may be non-empty
    # FIRST_EXCEPTION stops WAITING early, does NOT cancel/stop the still-running not_done futures