Concurrency in Python

Concurrency in Python refers to the coordinated progress of multiple units of work that may be implemented using threads, processes, executors, multiple interpreters, asynchronous tasks, and context-local state. This coordination allows overlapping progress of operations without necessarily executing them simultaneously in parallel. Concurrent progress emphasizes structuring and managing multiple active tasks, while parallel execution means those tasks actually run at the same time, often on multiple CPU cores.

Foundations of Concurrency in Python

Concurrency is the concept of overlapping progress among multiple tasks, allowing programs to initiate or continue multiple operations without waiting for each to complete sequentially. Parallelism is a subset of concurrency where multiple tasks execute simultaneously on separate processing units. A program can be concurrent without being parallel (e.g., interleaved execution on a single core), and parallel execution can arise through different Python execution models such as multithreading, multiprocessing, or multiple interpreters.

Tasks represent units of work that can be scheduled or executed independently. Threads are execution units within the same process that share memory and Python objects. Processes are separate operating-system instances of the Python runtime with isolated memory. Interpreter-isolated workers run in separate Python interpreters within the same process, isolating runtime state and interpreter locks. Asynchronous coroutines are cooperative multitasking units managed by an event loop without OS threads. Executors are abstractions that submit callables to worker threads or processes and manage their lifecycle and results.

Major reasons for concurrency include overlapping I/O waits to improve efficiency, distributing independent CPU-bound work across multiple cores or processors, maintaining application responsiveness, coordinating producer-consumer workflows, isolating failure or mutable state to limit error impact, and structuring applications as independently progressing operations.

Two main coordination models exist: shared-state concurrency, where multiple workers access and mutate common objects requiring synchronization; and message-oriented concurrency, where workers communicate explicitly through message passing or queues, isolating state and reducing synchronization complexity. Shared state offers convenient direct access but requires careful synchronization to avoid race conditions. Message passing enforces communication boundaries, improving modularity and safety at the cost of explicit coordination.

A Python program can branch into different concurrency approaches:

Choosing the appropriate concurrency model depends on workload behavior (CPU-bound vs I/O-bound), isolation requirements, communication cost, cancellation needs, library compatibility, platform constraints, and whether actual parallel execution is required.

Concurrency Safety in Python

Concurrency safety means preserving the required invariants and correctness when multiple concurrent execution units (threads, processes, tasks) may overlap in time, interleave operations, or observe shared state asynchronously.

A race condition occurs when program behavior or outcomes depend incorrectly on the relative timing or interleaving of concurrent operations, causing subtle bugs. This differs from nondeterministic completion order, which is expected and safe when tasks complete independently.

Example of unsafe read-modify-write on shared state without synchronization:

import threading

counter = 0

def unsafe_increment():
    global counter
    for _ in range(100000):
        counter += 1  # Not atomic, unsafe in threads

threads = [threading.Thread(target=unsafe_increment) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Final counter value (unsafe): {counter}")

Corrected version using a lock to synchronize the critical section:

import threading

counter = 0
lock = threading.Lock()

def safe_increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1  # Protected critical section

threads = [threading.Thread(target=safe_increment) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Final counter value (safe): {counter}")

Critical sections are code regions that must execute atomically relative to other concurrent workers to preserve shared state invariants. Mutual exclusion mechanisms (locks) enforce that only one worker can enter a critical section at a time.

Relying on incidental atomicity of single operations (e.g., increments appearing atomic) is unsafe when multiple steps are required to maintain invariants. Explicit synchronization is necessary for multi-step state transitions.

Deadlock arises when two or more concurrent operations wait indefinitely for resources held by each other, often caused by inconsistent lock acquisition order, nested lock requests, or blocking dependencies that form cycles.

Other concurrency failures include starvation (some tasks never progress), livelock (tasks continuously retry without progress), unfair progress (some tasks favored over others), and excessive contention (too many threads or locks causing overhead).

Reducing shared mutable state, transferring ownership of data rather than sharing, using immutable values, and communicating through queues or explicit message passing reduce synchronization complexity and race conditions.

Thread safety, process safety, asynchronous-task safety, and multi-interpreter safety are distinct correctness properties. Correctness in one concurrency model does not imply correctness in another due to differences in memory sharing, scheduling, and execution semantics.

Concurrency safety also involves cleanup and failure propagation to ensure synchronization resources are released, workers terminate predictably, and partially completed shared-state transitions are not abandoned, avoiding inconsistent states or resource leaks.

Thread-Based Concurrency in Python

Threads are concurrently scheduled execution units within a single process that share access to the same Python objects and process resources.

Python Thread Lifecycle

Threads are constructed by creating threading.Thread objects with a target callable and optional arguments. Calling start() schedules the thread for execution, invoking the target function in a new thread. The thread runs concurrently until the target returns or raises an uncaught exception, at which point the thread completes. Other threads may join() to wait for completion. Threads can be marked daemon to indicate they should not block process exit. Threads have identity accessible via threading.current_thread().

Example creating and running multiple named threads:

import threading
import time

def worker(name, delay):
    for i in range(3):
        print(f"{name} working iteration {i}")
        time.sleep(delay)
    print(f"{name} done")

threads = [
    threading.Thread(target=worker, args=(f"Worker-{i}", 0.5), name=f"Worker-{i}")
    for i in range(3)
]

for t in threads:
    t.start()

for t in threads:
    t.join()

print("All workers completed.")

Calling start() schedules the thread for concurrent execution. Calling the target function directly runs it synchronously in the current thread. join() waits for the thread to finish but does not start or terminate it.

Daemon threads do not prevent the Python process from exiting. However, daemon status is not a substitute for explicit management of thread lifecycle and cleanup, as daemon threads may be abruptly stopped without cleanup.

Thread Synchronization in Python

Locks provide mutual exclusion to protect shared-state invariants by ensuring only one thread executes a critical section at a time.

Example using threading.Lock as a context manager to protect a shared counter:

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Counter after synchronized increments: {counter}")

Other synchronization primitives include:

• RLock: Reentrant lock allowing the same thread to acquire multiple times.
• Semaphore: Limits concurrent access to a resource by a fixed number.
• BoundedSemaphore: Like a semaphore but raises error if released too many times.
• Event: One thread signals one or more waiting threads.
• Condition: Allows threads to wait for certain state changes, supporting notify/wait.
• Barrier: Synchronizes a fixed number of threads to wait for each other.

Example using an Event to avoid busy polling:

import threading
import time

event = threading.Event()

def waiter():
    print("Waiting for event...")
    event.wait()
    print("Event received!")

threading.Thread(target=waiter).start()
time.sleep(1)
event.set()

Thread Communication with Queues in Python

queue.Queue provides synchronized FIFO queues for transferring data or work between threads, coordinating producer and consumer access.

Example producer-consumer with sentinel shutdown:

import threading
import queue
import time

q = queue.Queue()

def producer():
    for i in range(5):
        q.put(i)
        print(f"Produced {i}")
        time.sleep(0.5)
    q.put(None)  # Sentinel to signal shutdown

def consumer():
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        print(f"Consumed {item}")
        q.task_done()

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

q.join()  # Wait until all items processed
producer_thread.join()
consumer_thread.join()
print("Producer-consumer complete.")

Bounded queues limit capacity, causing put() to block when full and get() to block when empty, providing backpressure. Queue capacity is independent of the number of active workers.

Thread-Local State in Python

threading.local provides per-thread attribute storage accessible through a shared object, allowing threads to maintain independent state.

Example:

import threading

local_data = threading.local()

def worker(num):
    local_data.value = num
    print(f"Thread {threading.current_thread().name} has value {local_data.value}")

threads = [threading.Thread(target=worker, args=(i,), name=f"Thread-{i}") for i in range(3)]

for t in threads:
    t.start()
for t in threads:
    t.join()

Thread-local state is limited when logical concurrency crosses thread boundaries or involves asynchronous tasks. For such cases, context variables are a better abstraction.

Thread Parallelism in Python

CPython's Global Interpreter Lock (GIL) restricts execution of Python bytecode to one thread at a time, limiting CPU-bound thread parallelism. However, threads can overlap I/O-bound work where the GIL is released during blocking I/O or in native code extensions.

Free-threaded CPython builds disable the GIL to allow true parallel bytecode execution but require more explicit thread safety.

Example comparing I/O-bound thread concurrency with sequential execution:

import threading
import time

def io_bound_task():
    time.sleep(1)
    print(f"Task completed in thread {threading.current_thread().name}")

start = time.time()

threads = [threading.Thread(target=io_bound_task, name=f"T{i}") for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Elapsed time with threads: {time.time() - start:.2f}s")

start = time.time()

for _ in range(3):
    io_bound_task()

print(f"Elapsed time sequentially: {time.time() - start:.2f}s")

Measured performance depends heavily on environment and workload characteristics.

Process-Based Concurrency in Python

Process-based concurrency uses separate OS processes, each with independent Python runtime and memory space. Communication and coordination require explicit IPC mechanisms.

Python Process Lifecycle

Processes are created by instantiating multiprocessing.Process objects with target callables. Starting the process launches a child executing the target independently. Processes have unique IDs, exit with status codes, and can be joined to wait for completion. Processes can also be terminated prematurely.

Example:

from multiprocessing import Process

def worker(name):
    print(f"Process {name} starting")
    import time
    time.sleep(1)
    print(f"Process {name} done")

if __name__ == "__main__":
    procs = [Process(target=worker, args=(f"P{i}",)) for i in range(3)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()
    print("All processes complete")

Top-level functions must be importable, and the if __name__ == "__main__": guard prevents unintended recursive process spawning on some platforms.

Python Multiprocessing Start Methods

Start methods determine how new processes are initialized:

Explicit multiprocessing contexts can be created with multiprocessing.get_context() to select start method per usage.

Example creating process and queue with explicit context:

import multiprocessing

def worker(q):
    q.put("Hello from process")

if __name__ == "__main__":
    ctx = multiprocessing.get_context('spawn')
    q = ctx.Queue()
    p = ctx.Process(target=worker, args=(q,))
    p.start()
    print(q.get())
    p.join()

Interprocess Communication in Python

Processes communicate via queues, pipes, and connection objects. Data is serialized (pickled) when sent, transferring a copy rather than sharing memory.

Example sending results from workers:

import multiprocessing

def worker(i, q):
    q.put(i * 2)

if __name__ == "__main__":
    q = multiprocessing.Queue()
    procs = [multiprocessing.Process(target=worker, args=(i, q)) for i in range(4)]
    for p in procs:
        p.start()
    results = [q.get() for _ in procs]
    for p in procs:
        p.join()
    print("Results:", results)

Shared State Between Python Processes

Shared-memory objects such as multiprocessing.Value and Array allow explicit shared state. Manager-backed proxies provide access to objects via a server process.

Example using shared value with synchronization:

import multiprocessing

def increment(shared_val, lock):
    for _ in range(10000):
        with lock:
            shared_val.value += 1

if __name__ == "__main__":
    lock = multiprocessing.Lock()
    shared_val = multiprocessing.Value('i', 0)
    procs = [multiprocessing.Process(target=increment, args=(shared_val, lock)) for _ in range(2)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()
    print(f"Shared counter: {shared_val.value}")

True shared memory offers lower latency but requires explicit synchronization. Manager proxies incur communication overhead but simplify sharing complex objects.

Process Synchronization in Python

Process synchronization primitives mirror those for threads, supporting locks, semaphores, events, conditions, and barriers adapted for interprocess use.

Example:

import multiprocessing
import time

def worker(event):
    print("Worker waiting for event")
    event.wait()
    print("Worker proceeding")

if __name__ == "__main__":
    event = multiprocessing.Event()
    p = multiprocessing.Process(target=worker, args=(event,))
    p.start()
    time.sleep(1)
    event.set()
    p.join()

Synchronization objects are tied to multiprocessing contexts; mixing objects from incompatible start methods is unsafe.

Process termination hazards include abandoning locks, queues, or partially completed state. Cooperative lifecycle management is preferred.

Executor-Based Concurrency in Python

Executors provide high-level abstractions accepting callable tasks, managing worker pools, and representing results through future objects.

The submit method schedules a callable, returning a Future representing its eventual result or exception. map schedules multiple calls and returns results in submission order. Executors support shutdown and context management to control worker lifecycle. Some executors support worker initialization hooks.

Python Concurrent Future Objects

concurrent.futures.Future objects track states: pending, running, finished. They provide methods for retrieving results or exceptions, cancellation requests, registering callbacks, and timeout-aware waiting.

Example submitting tasks and handling futures:

from concurrent.futures import ThreadPoolExecutor, as_completed

def task(n):
    if n == 3:
        raise ValueError("Error in task 3")
    return n * 2

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]
    for future in as_completed(futures):
        try:
            result = future.result()
            print(f"Result: {result}")
        except Exception as e:
            print(f"Caught exception: {e}")

wait blocks until all or some futures complete, whereas as_completed yields futures as they finish, regardless of submission order.

concurrent.futures.Future differs from asyncio.Future; they belong to separate concurrency environments and are incompatible.

Thread Pool Execution in Python

ThreadPoolExecutor manages a pool of worker threads for asynchronous callable execution.

Example:

from concurrent.futures import ThreadPoolExecutor
import time

def io_task(n):
    time.sleep(0.5)
    return f"Task {n} done"

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(io_task, range(5)))
    print(results)

submit returns futures immediately for individual tasks; map returns results in order, blocking until all complete.

Thread-pool deadlocks can occur if tasks wait synchronously on futures that require idle workers in the same pool, exhausting available threads.

Process Pool Execution in Python

ProcessPoolExecutor manages worker processes to bypass the GIL for CPU-bound parallelism.

Example:

from concurrent.futures import ProcessPoolExecutor

def cpu_task(n):
    return n * n

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = [executor.submit(cpu_task, i) for i in range(5)]
        results = [f.result() for f in futures]
        print(results)

Workers require importable callables, appropriate main-module guards, picklable arguments, and clean process startup context. Worker failure and communication overhead are important considerations.

Interpreter Pool Execution in Python

InterpreterPoolExecutor (available in newer Python versions) runs workers in multiple isolated interpreters within one process, combining interpreter isolation with thread-based scheduling.

Interpreter pool workers have separate interpreter state and locks, allowing true multi-core parallelism. They require stronger isolation and explicit data transfer compared to threads.

Example (if supported):

from concurrent.interpreter import InterpreterPoolExecutor

def task(n):
    return n + 1

with InterpreterPoolExecutor() as executor:
    futures = [executor.submit(task, i) for i in range(3)]
    results = [f.result() for f in futures]
    print(results)

Selecting an executor depends on workload characteristics, isolation needs, communication overhead, and parallel execution requirements.

Multiple-Interpreter Concurrency in Python

Multiple Python interpreters within one process provide isolated execution contexts with separate runtime states, including imports, builtins, and module namespaces.

Interpreter isolation alone does not schedule concurrent work; actual concurrency requires threads or another mechanism to run code in multiple interpreters simultaneously.

Modern multiple-interpreter support is available through concurrent.interpreters, allowing creation, lifecycle management, and execution within interpreter contexts.

Example creating an interpreter, running a callable, and closing it (version dependent):

import concurrent.interpreters

interp = concurrent.interpreters.create()
future = concurrent.interpreters.run(interp, lambda: 42)
result = future.result()
print(f"Result from interpreter: {result}")
concurrent.interpreters.close(interp)

Communication between interpreters requires explicit transfer or message passing. Transferred objects are copies or proxies; ordinary Python object identity is not shared.

Example conceptual queue usage for inter-interpreter communication:

# Pseudocode conceptual example
queue = concurrent.interpreters.Queue()
concurrent.interpreters.run(interp1, lambda: queue.put("Hello"))
result = concurrent.interpreters.run(interp2, lambda: queue.get())
print(result)

Many mutable Python objects cannot be shared across interpreters without serialization or compatible immutable types.

Interpreter parallelism uses separate interpreter locks, differing from free-threaded single interpreter or separate-process parallelism.

Extension modules and third-party libraries may not be safe across multiple interpreters, limiting compatibility.

Asynchronous Concurrency with asyncio

asyncio enables cooperative concurrency using an event loop that schedules coroutines, tasks, callbacks, and I/O readiness without requiring one OS thread per asynchronous task.

Python asyncio Execution Model

Coroutines are special generator-like objects defined with async def. Using await suspends coroutine execution at suspension points. The event loop schedules runnable tasks and resumes them when I/O or timers are ready, allowing multiple coroutines to progress cooperatively without preemption.

Concurrency here refers to interleaved progress without automatic parallel execution.

Example of overlapping delays with asyncio.run:

import asyncio

async def coro(name, delay):
    print(f"{name} started")
    await asyncio.sleep(delay)
    print(f"{name} finished")

async def main():
    await asyncio.gather(
        coro("Task1", 2),
        coro("Task2", 1),
        coro("Task3", 3),
    )

asyncio.run(main())

CPU-bound or blocking synchronous code executed in the event-loop thread blocks all other asyncio tasks.

Python asyncio Tasks and Futures

asyncio.Task wraps coroutine execution scheduled by the event loop. asyncio.Future represents an eventual result or exception at a low level.

Example using asyncio.create_task:

import asyncio

async def work(n):
    await asyncio.sleep(n)
    return n * 2

async def main():
    tasks = [asyncio.create_task(work(i), name=f"task-{i}") for i in range(3)]
    for task in tasks:
        result = await task
        print(f"{task.get_name()} result: {result}")

asyncio.run(main())

Task lifecycle states include created, scheduled, running, done, and cancelled. Creating a coroutine object differs from scheduling it as a task.

Concurrent Task Coordination in asyncio

Tools like gather, wait, and as_completed coordinate multiple tasks with different ordering and failure semantics.

Example comparing gather and completion-based processing:

import asyncio
import random

async def work(i):
    await asyncio.sleep(random.uniform(0.1, 1))
    return i

async def gather_example():
    results = await asyncio.gather(*(work(i) for i in range(5)))
    print("Gather order:", results)

async def as_completed_example():
    tasks = [asyncio.create_task(work(i)) for i in range(5)]
    results = []
    for task in asyncio.as_completed(tasks):
        result = await task
        results.append(result)
    print("Completion order:", results)

async def main():
    await gather_example()
    await as_completed_example()

asyncio.run(main())

Creating many tasks without limits can overwhelm external services or system resources despite efficient event-loop scheduling.

Structured Concurrency with asyncio Task Groups

asyncio.TaskGroup manages a dynamic collection of child tasks, waiting for all to finish before exiting.

Example:

import asyncio

async def child(n):
    await asyncio.sleep(n)
    return n * 10

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(child(i)) for i in range(3)]
    results = [task.result() for task in tasks]
    print("TaskGroup results:", results)

asyncio.run(main())

TaskGroup failure cancels siblings on non-cancellation exceptions and collects multiple failures into exception groups.

Example with failure handling:

import asyncio

async def child(i):
    if i == 1:
        raise ValueError("Failure in child 1")
    await asyncio.sleep(1)
    return i

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(child(0))
            tg.create_task(child(1))
            tg.create_task(child(2))
    except* ValueError as e:
        print(f"Caught grouped exception: {e}")

asyncio.run(main())

Task Cancellation and Timeouts in asyncio

Calling Task.cancel() requests cancellation by raising CancelledError at an await suspension point within the task. Proper cleanup uses try/finally, and caught CancelledError should normally be re-raised.

Example:

import asyncio

async def cancellable():
    try:
        print("Starting work")
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        print("Cleanup on cancellation")
        raise
    finally:
        print("Finally block executed")

async def main():
    task = asyncio.create_task(cancellable())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")

asyncio.run(main())

Timeouts can be imposed using asyncio.timeout() or wait_for(), differentiating deadline imposition from ordinary completion.

Shielding protects awaitables from propagation of caller cancellation but does not make them universally uncancellable.

Synchronization in asyncio

Asyncio provides coordination primitives whose waiting suspends only the current task, not the event loop thread:

Example using asyncio.Lock to protect shared state:

import asyncio

counter = 0
lock = asyncio.Lock()

async def increment():
    global counter
    for _ in range(1000):
        async with lock:
            counter += 1

async def main():
    await asyncio.gather(increment(), increment())
    print(f"Counter value: {counter}")

asyncio.run(main())

Threading and asyncio synchronization primitives operate in different scheduling domains and are not interchangeable.

Asyncio Queues

asyncio.Queue enables asynchronous producer-consumer coordination with awaitable put and get, optional bounded capacity, and task-completion tracking.

Example async producer-consumer with shutdown:

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        print(f"Produced {i}")
        await asyncio.sleep(0.5)
    await q.put(None)  # Sentinel

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            q.task_done()
            break
        print(f"Consumed {item}")
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=3)
    prod = asyncio.create_task(producer(q))
    cons = asyncio.create_task(consumer(q))
    await asyncio.gather(prod, cons)
    await q.join()
    print("Async producer-consumer complete")

asyncio.run(main())

Queue backpressure suspends tasks awaiting put() when full, without blocking event loop threads.

Integrating Blocking Work with asyncio

Blocking calls must be moved off the event-loop thread to avoid stalling all tasks.

asyncio.to_thread runs a blocking function in a separate thread, returning an awaitable.

Example:

import asyncio
import time

def blocking_io():
    time.sleep(2)
    return "blocking result"

async def main():
    task = asyncio.create_task(asyncio.to_thread(blocking_io))
    print("Running other async work")
    result = await task
    print(f"Got result: {result}")

asyncio.run(main())

Executor offloading integrates blocking work but does not convert it into native async I/O. Excessive offloading may hit thread or resource limits.

Context Variables in Concurrent Python

Context variables provide context-local state logically associated with concurrent execution flows, independent of global or thread-local state.

ContextVar objects are created with optional defaults. get() retrieves the current value, set() assigns a new value returning a token, and tokens can be used to reset to previous values, enabling scoped restoration.

Example:

from contextvars import ContextVar

var = ContextVar('var', default='default')

print(var.get())  # default
token = var.set('new value')
print(var.get())  # new value
var.reset(token)
print(var.get())  # default

Context and copy_context() capture and run code under a saved context snapshot, differentiating copying from sharing mutable namespaces.

Context variables integrate with asyncio tasks, propagating context on task creation and isolating later changes between tasks.

Example with asyncio:

import asyncio
from contextvars import ContextVar

var = ContextVar('var', default='initial')

async def worker(n):
    print(f"Worker {n} initial: {var.get()}")
    var.set(f"value-{n}")
    await asyncio.sleep(0.1)
    print(f"Worker {n} after set: {var.get()}")

async def main():
    await asyncio.gather(*(worker(i) for i in range(3)))

asyncio.run(main())

Context variables differ from threading.local in that they track logical execution context rather than OS threads.

Solved Concurrency Exercises in Python

Exercise 1: Bounded Thread-Based Producer-Consumer System

This exercise implements a producer-consumer system using threads and a bounded queue, with multiple consumers, explicit shutdown signaling, synchronized shared aggregate state, and deterministic joining.

import threading
import queue

NUM_WORKERS = 3
ITEMS_TO_PRODUCE = 10

# Shared aggregate state with lock protection
total_sum = 0
sum_lock = threading.Lock()

def producer(q):
    for i in range(ITEMS_TO_PRODUCE):
        q.put(i)
    # Send sentinel None to signal consumers to exit
    for _ in range(NUM_WORKERS):
        q.put(None)

def consumer(q):
    global total_sum
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        # Process item (here summing)
        with sum_lock:
            total_sum += item
        q.task_done()

def main():
    q = queue.Queue(maxsize=5)
    consumers = [threading.Thread(target=consumer, args=(q,)) for _ in range(NUM_WORKERS)]
    for c in consumers:
        c.start()

    producer(q)
    q.join()  # Wait until all items processed

    for c in consumers:
        c.join()

    print(f"Total sum: {total_sum}")
    expected = sum(range(ITEMS_TO_PRODUCE))
    print(f"Expected sum: {expected}")
    assert total_sum == expected, "Sum mismatch!"

if __name__ == "__main__":
    main()

Step-by-step explanation:

• Threads are started for consumers before producing items.
• The queue is bounded to limit in-flight items, applying backpressure.
• Producer places data and sentinel values into the queue.
• Consumers process items, updating shared state under lock.
• queue.task_done() and queue.join() coordinate completion.
• Consumers exit cleanly on sentinel.
• Threads join ensures deterministic shutdown.
• Final sum verifies correctness.

Exercise 2: Process-Based CPU-Oriented Work with Executor

This example uses ProcessPoolExecutor to distribute CPU-bound tasks, with importable worker function, main-module protection, and result collection.

from concurrent.futures import ProcessPoolExecutor

def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = [executor.submit(fib, n) for n in range(10, 15)]
        for future in futures:
            try:
                print(f"Fib result: {future.result()}")
            except Exception as e:
                print(f"Task failed: {e}")

Explanation:

• The worker fib function is top-level and importable.
• The main guard protects process spawning.
• ProcessPoolExecutor manages process workers.
• Futures provide result retrieval and exception propagation.
• Tasks run in parallel on multiple cores.
• Clean shutdown occurs via context manager.

Exercise 3: Asyncio TaskGroup with Queue, Context Variables, and Cancellation

This exercise uses asyncio.TaskGroup to coordinate bounded asynchronous concurrency with a queue, context variables for per-task context, cancellation-safe cleanup, and offloading blocking work with asyncio.to_thread.

import asyncio
from contextvars import ContextVar

request_id = ContextVar('request_id')

async def blocking_io(n):
    await asyncio.sleep(1)
    return f"Processed {n}"

async def worker(name, q):
    while True:
        try:
            item = await q.get()
            if item is None:
                q.task_done()
                break
            rid = request_id.get()
            print(f"{name} processing item {item} with request_id {rid}")
            # Offload blocking I/O
            result = await asyncio.to_thread(blocking_io, item)
            print(f"{name} got result: {result}")
            q.task_done()
        except asyncio.CancelledError:
            print(f"{name} received cancellation")
            raise
        finally:
            # Cleanup if needed
            pass

async def main():
    q = asyncio.Queue(maxsize=3)
    async with asyncio.TaskGroup() as tg:
        # Start multiple workers
        for i in range(2):
            tg.create_task(worker(f"worker-{i}", q))

        # Produce items with context variable set
        for i in range(5):
            token = request_id.set(f"req-{i}")
            await q.put(i)
            request_id.reset(token)

        # Send shutdown signals
        for _ in range(2):
            await q.put(None)

    await q.join()
    print("All tasks complete")

asyncio.run(main())

Explanation:

• The queue limits concurrent tasks, providing backpressure.
• asyncio.TaskGroup ensures structured concurrency; all workers finish before exit.
• Context variable request_id tracks per-task logical context.
• asyncio.to_thread offloads blocking operation, allowing other tasks to progress.
• Cancellation is handled safely with proper cleanup.
• Shutdown uses sentinel values.
• Final await q.join() confirms all work processed.