Multithreading, Multiprocessing, and Process Pools

Multithreading, Multiprocessing, and Process Pools

After reading this chapter, you will master parallel and concurrent execution in Python: implement thread-safe producer-consumer pipelines with queue.Queue, scale CPU-bound algorithms across cores using concurrent.futures.ProcessPoolExecutor, and achieve zero-copy inter-process communication using multiprocessing.shared_memory.

Mental model

Choosing between threads and processes depends entirely on whether your workload is I/O-bound or CPU-bound:

Workload Type                Recommended Primitive             Architecture
─────────────────────────────────────────────────────────────────────────────────
I/O-Bound (Sockets, DB, HTTP) ──▶ ThreadPoolExecutor / threading  ──▶ Single process, shared RAM
CPU-Bound (Hashing, Math, ML) ──▶ ProcessPoolExecutor             ──▶ Multi-process, isolated RAM
Zero-Copy IPC Across Cores    ──▶ multiprocessing.shared_memory   ──▶ Shared OS memory block

Threads vs Processes:
  Process 1 (Threads share heap):
    [ Heap Memory: Variables, Objects ]
         ├── Thread 1 (Stack 1)
         └── Thread 2 (Stack 2)

  Separate Processes (Isolated heap, parallel cores):
    [ Process 1 Heap ] ──(IPC Pipe / Shared Memory)──▶ [ Process 2 Heap ]
       Core 0                                             Core 1

---

Minimal example

Save as threads_and_processes.py:

# threads_and_processes.py
import concurrent.futures
import math
import time

def simulate_io_fetch(task_id: int) -> str:
    """Simulate I/O-bound network latency."""
    time.sleep(0.05)
    return f"Result-IO-{task_id}"

def compute_heavy_math(n: int) -> int:
    """Simulate CPU-bound mathematical computation."""
    return sum(int(math.sqrt(x)) for x in range(n))

def main() -> None:
    # 1. ThreadPoolExecutor for I/O-bound tasks
    print("--- Running I/O tasks with ThreadPoolExecutor ---")
    t0 = time.perf_counter()
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        io_results = list(executor.map(simulate_io_fetch, range(8)))
    print(f"Completed 8 I/O tasks in {time.perf_counter() - t0:.2f}s: {io_results[:3]}...")

    # 2. ProcessPoolExecutor for CPU-bound tasks (bypasses GIL across cores)
    print("\n--- Running CPU tasks with ProcessPoolExecutor ---")
    t0 = time.perf_counter()
    work_sizes = [500_000] * 4
    with concurrent.futures.ProcessPoolExecutor() as executor:
        cpu_results = list(executor.map(compute_heavy_math, work_sizes))
    print(f"Completed 4 CPU tasks in {time.perf_counter() - t0:.2f}s: {cpu_results}")

if __name__ == "__main__":
    main()

Run via uv run python threads_and_processes.py:

--- Running I/O tasks with ThreadPoolExecutor ---
Completed 8 I/O tasks in 0.10s: ['Result-IO-0', 'Result-IO-1', 'Result-IO-2']...

--- Running CPU tasks with ProcessPoolExecutor ---
Completed 4 CPU tasks in 0.08s: [...]

---

Worked examples

Case 1: Thread-Safe Producer-Consumer Pipeline with queue.Queue

Decoupling data acquisition (producers) from data processing (consumers) prevents slow operations from stalling ingestion:

# producer_consumer_queue.py
import queue
import threading
import time

def producer(q: queue.Queue[str | None], count: int) -> None:
    for i in range(count):
        item = f"packet_{i:03d}"
        q.put(item)
        print(f"  [Producer] Ingested {item}")
        time.sleep(0.01)
    # Poison pill to signal consumer termination
    q.put(None)

def consumer(q: queue.Queue[str | None]) -> None:
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        print(f"    [Consumer] Processed {item}")
        q.task_done()

if __name__ == "__main__":
    work_queue: queue.Queue[str | None] = queue.Queue(maxsize=10)
    
    prod_thread = threading.Thread(target=producer, args=(work_queue, 5))
    cons_thread = threading.Thread(target=consumer, args=(work_queue,))

    prod_thread.start()
    cons_thread.start()

    prod_thread.join()
    cons_thread.join()
    print("Pipeline completed gracefully.")

Run:

uv run python producer_consumer_queue.py

Output:

  [Producer] Ingested packet_000
    [Consumer] Processed packet_000
  [Producer] Ingested packet_001
    [Consumer] Processed packet_001
  [Producer] Ingested packet_002
    [Consumer] Processed packet_002
  [Producer] Ingested packet_003
    [Consumer] Processed packet_003
  [Producer] Ingested packet_004
    [Consumer] Processed packet_004
Pipeline completed gracefully.

Case 2: Zero-Copy Shared Memory Across Processes

Passing large datasets through standard multiprocessing pipes (pickle) introduces severe serialization overhead. multiprocessing.shared_memory maps raw OS memory across process boundaries:

# shared_memory_demo.py
from multiprocessing import shared_memory

def main() -> None:
    # 1. Allocate 1000 bytes in system shared memory
    shm_parent = shared_memory.SharedMemory(create=True, size=1000)
    print(f"Allocated shared memory block: {shm_parent.name}")

    try:
        # Write bytes directly to shared buffer
        shm_parent.buf[:18] = b"CLUSTER_STATE_LIVE"

        # 2. Attach from another context using the memory block name
        shm_child = shared_memory.SharedMemory(name=shm_parent.name)
        data = bytes(shm_child.buf[:18])
        print(f"Read from child memory block : {data.decode()}")

        shm_child.close()
    finally:
        shm_parent.close()
        shm_parent.unlink()  # Release system shared memory segment

if __name__ == "__main__":
    main()

Run:

uv run python shared_memory_demo.py

Output:

Allocated shared memory block: ...
Read from child memory block : CLUSTER_STATE_LIVE

---

Pitfalls

Pitfall 1: Forgetting if __name__ == "__main__": in Multiprocessing

On Windows and modern macOS/Linux (spawn start method), child processes re-import the main module. Omitting if __name__ == "__main__": causes an infinite process spawning fork bomb (RuntimeError).

Pitfall 2: High IPC Pickling Overhead

Passing large objects into executor.map() pickles the data, sends it over an OS pipe, and unpickles it in the worker process. If data transfer takes longer than the computation, multiprocessing will be slower than single-threaded execution. Use shared_memory for large arrays.

---

Exercises

1. Build a concurrent website health checker using ThreadPoolExecutor that polls 10 URLs and reports their HTTP response codes.
2. Implement a parallel image/matrix processor using ProcessPoolExecutor that chunks a large list across all available CPU cores.
3. Use threading.Event to coordinate a graceful shutdown signal across multiple running worker threads.
4. Share a 1MB byte array across two processes using multiprocessing.shared_memory.SharedMemory without pickling.

---

Further reading

• Python Standard Library: concurrent.futures, threading, multiprocessing.
• Python Documentation: multiprocessing.shared_memory — Shared memory for direct process access.