Python Threading & Concurrency Cheatsheet Python Threading & Concurrency Cheatsheet

import threading

def worker(host: str) -> None:
    print(f"[{threading.current_thread().name}] pinging {host}")

t = threading.Thread(target=worker, args=("web1",), name="ping-1", daemon=True)
t.start()
t.join(timeout=5)            # wait for it (with a timeout)
t.is_alive()                 # True if still running

lock = threading.Lock()
counter = 0

def increment() -> None:
    global counter
    with lock:               # always use `with` - releases on exception
        counter += 1

# RLock: same thread may re-acquire (needed when locked methods call each other)
rlock = threading.RLock()

class Registry:
    def __init__(self):
        self._lock = threading.RLock()
        self._data = {}

    def set(self, k, v):
        with self._lock:
            self._data[k] = v

    def set_many(self, items):
        with self._lock:          # re-enters fine because it's an RLock
            for k, v in items:
                self.set(k, v)

# Event - one-shot broadcast flag (shutdown signals, "ready" gates)
stop = threading.Event()

def poller():
    while not stop.is_set():
        check_health()
        stop.wait(timeout=5)  # sleep, but wake immediately on set()

stop.set()                    # tell all waiters to stop

# Semaphore - cap concurrency (e.g. max 5 simultaneous API calls)
sem = threading.BoundedSemaphore(5)

def call_api(url):
    with sem:
        return fetch(url)

# Condition - wait for a state change under a lock
cond = threading.Condition()
items = []

def consumer():
    with cond:
        cond.wait_for(lambda: len(items) > 0, timeout=10)
        item = items.pop()

def producer(x):
    with cond:
        items.append(x)
        cond.notify()         # notify_all() to wake every waiter

import queue, threading

q: queue.Queue = queue.Queue(maxsize=100)   # maxsize gives you backpressure
SENTINEL = object()

def producer():
    for job in jobs:
        q.put(job)                # blocks when full
    q.put(SENTINEL)

def consumer():
    while True:
        job = q.get()             # blocks when empty; q.get(timeout=5) raises queue.Empty
        if job is SENTINEL:
            q.put(SENTINEL)       # let sibling consumers see it too
            break
        try:
            process(job)
        finally:
            q.task_done()

threads = [threading.Thread(target=consumer, daemon=True) for _ in range(4)]
[t.start() for t in threads]
producer()
q.join()                          # blocks until every task_done() is called

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

urls = [f"https://host{i}/health" for i in range(50)]

# map - results in submission order
with ThreadPoolExecutor(max_workers=10) as pool:
    for status in pool.map(fetch, urls, timeout=30):
        print(status)

# submit + as_completed - results as they finish, per-task error handling
with ThreadPoolExecutor(max_workers=10) as pool:
    futures = {pool.submit(fetch, url): url for url in urls}
    for fut in as_completed(futures, timeout=60):
        url = futures[fut]
        try:
            print(url, fut.result())
        except Exception as e:
            print(f"{url} failed: {e}")

# CPU-bound? One-line switch:
with ProcessPoolExecutor(max_workers=4) as pool:
    checksums = list(pool.map(sha256_file, big_files))

import multiprocessing as mp

def crunch(chunk):
    return sum(x * x for x in chunk)

if __name__ == "__main__":                # REQUIRED guard on every entry point
    with mp.Pool(processes=mp.cpu_count()) as pool:
        results = pool.map(crunch, chunks)

    # Explicit Process + IPC
    q = mp.Queue()
    p = mp.Process(target=worker, args=(q,))
    p.start(); p.join()

    # Shared state (rarely needed - prefer passing data)
    counter = mp.Value("i", 0)
    with counter.get_lock():
        counter.value += 1

import asyncio
import aiohttp   # pip install aiohttp - requests is blocking, don't use it here

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as resp:
        return resp.status

async def main() -> None:
    async with aiohttp.ClientSession() as session:
        # gather - run concurrently, keep order
        statuses = await asyncio.gather(
            *(fetch(session, u) for u in urls),
            return_exceptions=True,          # exceptions become results, not crashes
        )

        # Tasks - start now, await later
        task = asyncio.create_task(fetch(session, "https://x/health"))
        status = await task

        # Timeouts
        try:
            async with asyncio.timeout(5):           # 3.11+
                await fetch(session, slow_url)
        except TimeoutError:
            print("gave up")
        # pre-3.11 equivalent: await asyncio.wait_for(coro, timeout=5)

        # Structured concurrency (3.11+) - all-or-nothing task group
        async with asyncio.TaskGroup() as tg:
            for u in urls:
                tg.create_task(fetch(session, u))

asyncio.run(main())

# Other essentials
await asyncio.sleep(1)                       # never time.sleep() in async code
sem = asyncio.Semaphore(10)                  # cap concurrent coroutines
async with sem: ...
await asyncio.to_thread(blocking_func, arg)  # run blocking code without freezing the loop