Concurrency, Parallelism & Async

A unified API over threads or processes

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

def work(n):
    return n * n

# Thread-backed -- good for I/O-bound work
with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(work, range(10)))

# Process-backed -- good for CPU-bound work; SAME calling code
with ProcessPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(work, range(10)))

Both executors expose the same interface (submit, map, context-manager shutdown), so the choice of thread vs. process pool is largely a config decision based on whether the workload is I/O-bound or CPU-bound, not a rewrite of the calling code.

Future objects: getting results as they complete

futures = [pool.submit(work, n) for n in range(10)]

for future in as_completed(futures):    # yields futures as they finish, not in order
    print(future.result())               # blocks until this specific future is done

submit() returns a Future immediately (non-blocking); .result() blocks until that particular future completes (re-raising any exception the task raised); as_completed() yields futures in completion order, useful when you want to process results as soon as any is ready rather than waiting for all of them in submission order (which is what .map() effectively does).

Key differences beyond thread vs. process

Exception handling

future = pool.submit(lambda: 1 / 0)
future.result()   # raises ZeroDivisionError -- exceptions propagate through .result()

Exceptions raised inside the worker are captured and re-raised when you call .result(), rather than crashing the worker pool silently — a common gotcha is calling .submit() in a loop and never checking .result(), silently swallowing failures.

Interview-ready summary: concurrent.futures gives one API (submit/map/Future) for both thread- and process-backed concurrency — pick ThreadPoolExecutor for I/O-bound work and ProcessPoolExecutor for CPU-bound work, and remember that process pool arguments/results must be picklable since they cross a real process boundary.

Why the GIL doesn't prevent race conditions

import threading

counter = 0

def increment():
    global counter
    for _ in range(100_000):
        counter += 1   # NOT atomic: read counter, add 1, write back -- 3 separate steps

threads = [threading.Thread(target=increment) for _ in range(4)]
[t.start() for t in threads]
[t.join() for t in threads]
print(counter)   # often less than 400,000 -- lost updates!

counter += 1 compiles to multiple bytecode instructions (load, add, store). The GIL guarantees each individual bytecode instruction runs atomically, but a thread can be swapped out between those instructions — so two threads can both read the same value before either writes back the incremented result, and one increment gets lost.

Fixing it with a Lock

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100_000):
        with lock:              # only one thread executes this block at a time
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(4)]
[t.start() for t in threads]
[t.join() for t in threads]
print(counter)   # always exactly 400,000

with lock: acquires the lock on entry and releases it on exit (even if an exception occurs) — while held, no other thread can enter a block guarded by the same lock, making the read-modify-write sequence atomic with respect to other threads using that lock.

Other synchronization primitives

• RLock (reentrant lock): the same thread can acquire it multiple times (e.g., recursive functions or nested methods that both need the lock) without deadlocking itself; a plain Lock would deadlock in that case.
• Semaphore(n): allows up to n threads to hold it simultaneously — useful for capping concurrent access to a limited resource (e.g., at most 5 concurrent connections to a rate-limited service).
• Condition: lets threads wait for some condition to become true, signaled by another thread (.wait()/.notify()) — the basis for producer/consumer patterns.
• Event: a simple flag threads can wait on until another thread sets it (.set()), useful for one-shot "start now" or "shutdown" signals.

Avoiding locks altogether: prefer thread-safe data structures/patterns

from queue import Queue

q = Queue()   # thread-safe by design -- internally uses its own locking

def producer():
    q.put("item")

def consumer():
    item = q.get()

queue.Queue is thread-safe internally, so a producer/consumer pattern built around it needs no manual locking at all — preferring built-in thread-safe structures (Queue, queue.LifoQueue) over hand-rolled locking is usually safer and simpler than reasoning about locks directly.

Interview-ready summary: The GIL makes individual bytecode instructions atomic, but not multi-instruction operations like x += 1 — race conditions are still possible and must be guarded with threading.Lock (or a higher-level primitive like Queue) around any critical section that reads and writes shared mutable state.

Pitfall 1: blocking calls freeze the whole event loop

import asyncio
import time

async def bad():
    time.sleep(5)     # BLOCKS the entire event loop -- every other task stalls too!

async def good():
    await asyncio.sleep(5)   # yields control -- other tasks keep running

time.sleep (or any synchronous, blocking I/O call, or CPU-heavy computation) doesn't await anything, so the event loop has no opportunity to run other coroutines while it executes — a single blocking call anywhere stalls every pending task, not just the one that made it. Offload genuinely blocking/synchronous work with loop.run_in_executor(None, blocking_func) instead.

Pitfall 2: creating a coroutine without awaiting it

async def fetch():
    ...

async def main():
    fetch()          # BUG: creates a coroutine object but never runs it!
    # RuntimeWarning: coroutine 'fetch' was never awaited

Calling fetch() just constructs a coroutine object — it does nothing until awaited, scheduled as a task, or passed to asyncio.gather(). Forgetting the await is one of the most common asyncio bugs, and Python does emit a RuntimeWarning for it, but it's easy to miss in noisy logs.

Pitfall 3: fire-and-forget tasks losing their reference

async def main():
    asyncio.create_task(background_job())   # BUG: no reference kept!
    await asyncio.sleep(10)
    # background_job's task can be garbage-collected before it finishes,
    # silently cancelling it, and any exception it raised is never surfaced

async def main():
    task = asyncio.create_task(background_job())   # keep a reference
    try:
        await asyncio.sleep(10)
    finally:
        await task   # ensure it completes, and its exceptions propagate

The asyncio docs explicitly warn: "Save a reference to the result of this function, to avoid a task disappearing mid-execution." Without a kept reference, the event loop has no obligation to keep the task alive, and exceptions raised inside an un-awaited, unreferenced task are simply logged (via asyncio's default exception handler) rather than raised anywhere your code can catch them.

Pitfall 4: mixing asyncio.run() calls / event loop confusion

async def main():
    ...

asyncio.run(main())
asyncio.run(main())   # fine -- each call creates and closes its own loop

# But calling asyncio.run() from WITHIN a running coroutine is an error:
async def bad():
    asyncio.run(other_coro())   # RuntimeError: asyncio.run() cannot be called
                                  # from a running event loop

asyncio.run() is meant to be the single top-level entry point per program/script — nesting it inside already-running async code doesn't work; use await other_coro() instead.

Interview-ready summary: The recurring theme across asyncio pitfalls is forgetting that cooperative scheduling depends entirely on await points: blocking calls without an await freeze everything, coroutines created without await/create_task never run, and tasks created without a kept reference can vanish silently along with any exception they raised.

The default: no shared memory, everything is copied

from multiprocessing import Process

data = [1, 2, 3]

def worker(data):
    data.append(4)   # mutates the CHILD process's own copy only

p = Process(target=worker, args=(data,))
p.start()
p.join()
print(data)   # [1, 2, 3] -- parent's list is untouched; child had a separate copy

Unlike threads (which share the same memory space), each process gets its own independent memory — arguments passed to Process/Pool are pickled, sent to the child, and unpickled there, so mutations in the child never affect the parent's original objects.

Queue and Pipe: message passing

from multiprocessing import Process, Queue

def worker(q):
    q.put("result from child")

q = Queue()
p = Process(target=worker, args=(q,))
p.start()
print(q.get())   # 'result from child'
p.join()

Queue is a process-safe FIFO for passing arbitrary picklable objects between processes — the standard way to send results back from workers, or to distribute work items to them. Pipe() provides a lower-level, two-endpoint duplex connection between exactly two processes.

Value/Array: real shared memory for simple types

from multiprocessing import Process, Value

def worker(counter):
    with counter.get_lock():        # Value provides a built-in lock
        counter.value += 1

counter = Value("i", 0)             # 'i' = ctypes int, backed by shared memory
processes = [Process(target=worker, args=(counter,)) for _ in range(4)]
[p.start() for p in processes]
[p.join() for p in processes]
print(counter.value)   # 4

Value/Array allocate memory in a shared segment (via ctypes) that multiple processes can read/write directly — much faster than pickling through a Queue for simple numeric/fixed-type shared state, but limited to ctypes-compatible types.

Manager: shared Python objects (dict, list, etc.)

from multiprocessing import Process, Manager

def worker(shared_dict, key, value):
    shared_dict[key] = value

with Manager() as manager:
    shared_dict = manager.dict()
    processes = [Process(target=worker, args=(shared_dict, i, i * i)) for i in range(4)]
    [p.start() for p in processes]
    [p.join() for p in processes]
    print(dict(shared_dict))   # {0: 0, 1: 1, 2: 4, 3: 9}

A Manager runs a separate server process holding the real object; other processes get a proxy that forwards operations to it over IPC — more flexible than Value/Array (supports dicts, lists, arbitrary picklable values) but slower, since every access is a message round-trip, not a direct memory read.

Interview-ready summary: Processes don't share memory by default, so multiprocessing provides explicit channels: Queue/Pipe for message passing, Value/Array for fast shared memory of simple ctypes types, and Manager for shared, proxied Python objects when you need richer data structures at the cost of IPC overhead.

What changes

# Standard build -- has the GIL
python3.13 my_script.py

# Free-threaded build -- experimental, no GIL
python3.13t my_script.py

In a free-threaded build, the single global lock is replaced by finer-grained mechanisms: biased reference counting (a fast path for the common case where an object is only touched by the thread that owns it, falling back to atomic operations when shared across threads) and per-object or per-data-structure locking where needed, instead of one lock protecting everything.

The practical implication: threads can now use multiple cores

from concurrent.futures import ThreadPoolExecutor

def cpu_heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(cpu_heavy, [10**7] * 4))
    # on a free-threaded build, this can now actually run ~4x faster;
    # on a standard GIL build, it doesn't (see the GIL question)

This is the headline benefit: CPU-bound pure-Python code using threading/ThreadPoolExecutor can see genuine multi-core speedups without needing to switch to multiprocessing and its IPC/pickling overhead.

The tradeoffs, as of Python 3.13/3.14

• Single-threaded performance cost: removing the GIL's cheap global lock in favor of finer-grained (sometimes atomic) operations adds overhead to single-threaded code — early free-threaded builds showed a measurable (though steadily improving) single-thread slowdown compared to the standard GIL build.
• C extension compatibility: many existing C extensions assumed GIL protection for their own internal state and need updates (Py_mod_gil slot, thread-safety audits) to be safe under free-threading; the ecosystem is actively migrating but not fully there yet.
• Still opt-in: the GIL-enabled build remains the default and officially supported configuration; free-threading is offered as an alternative build for testing and gradual adoption, not (yet) the default for everyone.

Why it matters for interviews

It's a live, actively-evolving area of CPython — a good signal that a candidate follows the ecosystem, understands why the GIL existed in the first place (reference-counting safety), and can articulate what replacing it actually requires (finer-grained locking, not just deleting a mutex).

Interview-ready summary: PEP 703 introduces an experimental, opt-in build of CPython (3.13+) that removes the GIL via finer-grained synchronization (biased reference counting plus targeted locking), enabling real multi-core parallelism for threaded CPU-bound code — at the cost of some single-threaded overhead and an still-adapting C-extension ecosystem, which is why it isn't the default build yet.

Timeouts with asyncio.wait_for

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2)
    except TimeoutError:
        print("timed out after 2 seconds")

asyncio.run(main())

wait_for runs the coroutine as a task and races it against the timeout; if the timeout elapses first, it cancels the task internally and raises TimeoutError (in Python 3.11+; asyncio.TimeoutError, an alias of the same, in earlier versions) to the caller.

Python 3.11+'s asyncio.timeout(): a context-manager alternative

async def main():
    try:
        async with asyncio.timeout(2):
            await slow_operation()
    except TimeoutError:
        print("timed out")

asyncio.timeout() applies a deadline to an entire async with block (potentially covering multiple awaits), which is more flexible than wait_for when you want one timeout to cover several sequential operations rather than wrapping each individually.

Explicit cancellation with task.cancel()

async def worker():
    try:
        await asyncio.sleep(100)
    except asyncio.CancelledError:
        print("cleaning up before exiting")
        raise             # IMPORTANT: re-raise so the cancellation actually completes

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(1)
    task.cancel()          # requests cancellation
    await task              # propagates CancelledError here, after cleanup runs

task.cancel() doesn't stop the task immediately — it arranges for CancelledError to be raised inside the task at its next await point. The task can catch that to run cleanup (closing a connection, releasing a lock), but must re-raise it afterward; swallowing CancelledError silently prevents the cancellation from actually taking effect and can leave .cancel() callers waiting forever.

Why you must never blanket-catch CancelledError

async def bad_worker():
    try:
        await asyncio.sleep(100)
    except Exception:        # BUG: in Python 3.8+, CancelledError is a
        pass                  # BaseException subclass, NOT caught here --
                                # but a bare `except:` WOULD catch it and break cancellation

Since Python 3.8, asyncio.CancelledError inherits from BaseException (not Exception), specifically so that ordinary except Exception: blocks don't accidentally swallow it — but a bare except: still would, so it's a rule worth stating explicitly: always let cancellation propagate unless you're deliberately intercepting it to clean up, and always re-raise afterward.

Interview-ready summary: Use asyncio.wait_for/asyncio.timeout() for deadline-based cancellation, and task.cancel() for explicit cancellation — both work by raising CancelledError inside the task at its next await. Treat CancelledError as something to clean up around, never to swallow, since eating it silently breaks the cancellation contract for whoever is waiting on the task.