Python Multiprocessing & Queues

The Story That Explains Multiprocessing

Threads vs Processes — The Fundamental Difference

Your First Process — The Basics

The multiprocessing.Process class mirrors threading.Thread almost exactly. If you know threads, you already know 80% of processes.

from multiprocessing import Process
import os
import time

def worker(name, duration):
    print(f"[{name}] pid={os.getpid()} starting")
    time.sleep(duration)
    print(f"[{name}] pid={os.getpid()} done")

if __name__ == "__main__":                # REQUIRED on Windows and macOS
    print(f"Main pid={os.getpid()}")

    processes = []
    for i in range(3):
        p = Process(target=worker, args=(f"P{i}", 2))
        p.start()
        processes.append(p)

    for p in processes:
        p.join()

    print("All processes complete.")

Daemon Processes — Background Workers That Die With The Parent

Daemon in Action

from multiprocessing import Process
import time

def background_monitor():
    while True:
        print("  [daemon] still watching...")
        time.sleep(1)

if __name__ == "__main__":
    p = Process(target=background_monitor)
    p.daemon = True                     # MUST be set before start()
    p.start()

    print("[main] doing work for 3 seconds...")
    time.sleep(3)
    print("[main] exiting — daemon will be killed automatically")
    # No p.join() needed — daemon dies with parent

The Same Program WITHOUT daemon flag

if __name__ == "__main__":
    p = Process(target=background_monitor)
    # p.daemon = True   <-- omitted!
    p.start()

    time.sleep(3)
    print("[main] want to exit...")
    # PROGRAM HANGS FOREVER — child runs infinite loop, parent waits

When To Use Each

from multiprocessing import Process, Event
import time
import atexit

def graceful_monitor(stop_event):
    try:
        while not stop_event.is_set():
            print("  [worker] doing work...")
            stop_event.wait(timeout=1)     # wakes early if signalled
    finally:
        print("  [worker] shutting down cleanly — flushing state")

if __name__ == "__main__":
    stop = Event()
    p = Process(target=graceful_monitor, args=(stop,))
    p.start()

    atexit.register(lambda: (stop.set(), p.join(timeout=5)))

    time.sleep(3)
    print("[main] done")

Fork vs Spawn — How Processes Are Created

import multiprocessing as mp

# Set start method explicitly for consistent behavior across platforms
if __name__ == "__main__":
    mp.set_start_method("spawn")     # or "fork", "forkserver"

    # Check what's available on this platform
    print("Available:", mp.get_all_start_methods())
    print("Current:  ", mp.get_start_method())

Pool — The Workhorse of Parallel Python

Manually creating and joining processes gets tedious for anything beyond trivial scripts. multiprocessing.Pool gives you a persistent pool of worker processes and a batch of high-level methods to distribute work across them.

from multiprocessing import Pool
import time

def heavy_compute(n):
    # Simulated CPU work
    total = 0
    for i in range(10_000_000):
        total += i * n
    return n, total

if __name__ == "__main__":
    numbers = list(range(1, 9))

    # --- Sequential ---
    start = time.perf_counter()
    results = [heavy_compute(n) for n in numbers]
    print(f"Sequential: {time.perf_counter() - start:.2f}s")

    # --- Pool with 4 workers ---
    start = time.perf_counter()
    with Pool(processes=4) as pool:
        results = pool.map(heavy_compute, numbers)
    print(f"Pool(4):    {time.perf_counter() - start:.2f}s")

    for n, total in results:
        print(f"  {n}: {total}")

Pool Methods — Choosing The Right One

Practical Examples of Each

from multiprocessing import Pool

def square(x): return x * x
def add(a, b): return a + b

if __name__ == "__main__":
    with Pool(4) as pool:

        # 1. map — simplest, blocks until done, results in order
        print(pool.map(square, [1, 2, 3, 4, 5]))
        # [1, 4, 9, 16, 25]

        # 2. starmap — for multi-arg functions
        print(pool.starmap(add, [(1, 2), (3, 4), (5, 6)]))
        # [3, 7, 11]

        # 3. imap_unordered — stream results as they finish
        for result in pool.imap_unordered(square, range(10)):
            print(result, end=" ")   # might print: 0 4 1 9 25 16 36 49 64 81

        # 4. apply_async — fire off a single task without blocking
        r = pool.apply_async(square, (10,))
        # ... do other work ...
        print(r.get(timeout=5))   # 100

        # 5. map_async with callback
        def on_complete(results):
            print(f"Done! {len(results)} results")

        r = pool.map_async(square, range(100), callback=on_complete)
        r.wait()

multiprocessing.Queue — Inter-Process Communication

Producer-Consumer Between Processes

from multiprocessing import Process, Queue
import time
import os

def producer(q, count):
    for i in range(count):
        item = {"id": i, "pid": os.getpid(), "payload": i * 2}
        q.put(item)
        print(f"[P] produced {item}")
        time.sleep(0.1)
    q.put(None)                              # poison pill

def consumer(q):
    while True:
        item = q.get()
        if item is None: break
        print(f"    [C pid={os.getpid()}] got {item}")
        time.sleep(0.15)

if __name__ == "__main__":
    q = Queue(maxsize=10)

    p1 = Process(target=producer, args=(q, 5))
    p2 = Process(target=consumer, args=(q,))

    p1.start(); p2.start()
    p1.join();  p2.join()
    print("Done.")

Pipes — The Faster, Two-Endpoint Alternative

multiprocessing.Pipe() returns a pair of connected endpoints for bidirectional communication between exactly two processes. It's faster than Queue for point-to-point traffic but doesn't have safe multi-producer or multi-consumer semantics.

from multiprocessing import Process, Pipe

def worker(conn):
    conn.send({"status": "ready", "pid": 42})
    reply = conn.recv()                       # wait for parent's message
    print(f"Worker got reply: {reply}")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()

    p = Process(target=worker, args=(child_conn,))
    p.start()

    print("Parent got:", parent_conn.recv())
    parent_conn.send("go ahead")
    p.join()

Shared Memory — When Pickling Is Too Slow

Pickling a 1 GB NumPy array to send across a Queue is absurdly slow. For high-volume data, Python offers real shared memory — a chunk of RAM that multiple processes can read and write directly, no serialisation needed.

Shared Memory — Value & Array (Simple C Types)

For a shared counter or a small fixed-size array of numbers, use Value and Array. They wrap C-level primitives with an optional lock for atomicity.

from multiprocessing import Process, Value, Array, Lock

def increment(counter, lock, n):
    for _ in range(n):
        with lock:                              # atomic increment
            counter.value += 1

def square_in_place(arr, start, end):
    for i in range(start, end):
        arr[i] = arr[i] * arr[i]

if __name__ == "__main__":
    # --- Shared counter ---
    counter = Value('i', 0)                    # 'i' = signed int
    lock = Lock()

    procs = [Process(target=increment, args=(counter, lock, 10_000))
             for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"Final counter: {counter.value}")   # 40000 (deterministic!)

    # --- Shared array ---
    arr = Array('d', range(20))               # 'd' = double

    p1 = Process(target=square_in_place, args=(arr, 0, 10))
    p2 = Process(target=square_in_place, args=(arr, 10, 20))
    p1.start(); p2.start()
    p1.join();  p2.join()
    print(list(arr[:]))

Shared Memory — Manager (Any Python Object)

When you need to share dicts, lists, or arbitrary Python objects, use Manager. It runs a separate server process that owns the real object; every access from a worker is a small RPC call.

from multiprocessing import Process, Manager

def worker(shared_dict, shared_list, worker_id):
    shared_dict[worker_id] = f"done by worker {worker_id}"
    shared_list.append(worker_id ** 2)

if __name__ == "__main__":
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()

        procs = [Process(target=worker, args=(shared_dict, shared_list, i))
                 for i in range(5)]

        for p in procs: p.start()
        for p in procs: p.join()

        print("Dict:", dict(shared_dict))
        print("List:", list(shared_list))

Shared Memory — The Modern shared_memory Module

Introduced in Python 3.8, multiprocessing.shared_memory lets processes share a raw block of memory with zero pickling overhead. Combined with NumPy, this is the fastest way to share large numeric arrays.

from multiprocessing import Process
from multiprocessing.shared_memory import SharedMemory
import numpy as np

def worker(shm_name, shape, dtype, start, end):
    # Attach to existing shared memory block by name
    existing = SharedMemory(name=shm_name)
    arr = np.ndarray(shape, dtype=dtype, buffer=existing.buf)

    # Mutate in place — no copy, no pickle
    arr[start:end] *= 2
    existing.close()

if __name__ == "__main__":
    # Allocate a 1 million-element float64 array in shared memory
    shape = (1_000_000,)
    dtype = np.float64
    nbytes = np.zeros(shape, dtype=dtype).nbytes

    shm = SharedMemory(create=True, size=nbytes)
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    arr[:] = np.arange(1_000_000)

    # Spawn 4 workers, each mutating a slice
    chunk = 250_000
    procs = [Process(target=worker,
                     args=(shm.name, shape, dtype, i * chunk, (i + 1) * chunk))
             for i in range(4)]

    for p in procs: p.start()
    for p in procs: p.join()

    print("First 5 :", arr[:5])           # [0, 2, 4, 6, 8]
    print("Last 5  :", arr[-5:])          # [1999990, ..., 1999998]

    shm.close()
    shm.unlink()                             # free the shared memory block

ShareableList — Simpler Alternative

from multiprocessing import Process
from multiprocessing.shared_memory import ShareableList

def worker(name):
    sl = ShareableList(name=name)
    sl[0] = "modified by child"
    sl[1] += 100
    sl.shm.close()

if __name__ == "__main__":
    sl = ShareableList(["hello", 42, 3.14, True])
    print("Before:", list(sl))

    p = Process(target=worker, args=(sl.shm.name,))
    p.start(); p.join()

    print("After :", list(sl))
    sl.shm.close(); sl.shm.unlink()

Synchronisation Primitives

Just like threads, processes sometimes need to coordinate. multiprocessing provides process-safe versions of all the usual primitives.

from multiprocessing import Process, Event, Barrier
import time
import random

def worker(worker_id, start_event, barrier):
    start_event.wait()                          # wait for green light
    print(f"[W{worker_id}] phase 1 starting")
    time.sleep(random.uniform(0.5, 2))
    print(f"[W{worker_id}] phase 1 done, waiting at barrier")

    barrier.wait()                              # wait for all workers
    print(f"[W{worker_id}] phase 2 starting")

if __name__ == "__main__":
    N = 3
    start_event = Event()
    barrier     = Barrier(N)

    procs = [Process(target=worker, args=(i, start_event, barrier))
             for i in range(N)]
    for p in procs: p.start()

    time.sleep(1)                              # let workers set up
    print("All workers ready. GO!")
    start_event.set()                            # release all at once

    for p in procs: p.join()

Real-World Example — Parallel Image Processing

A classic use case for multiprocessing: apply a heavy filter to hundreds of images. Threads would be crippled by the GIL; processes scale linearly with cores.

from multiprocessing import Pool, cpu_count
from PIL import Image, ImageFilter
from pathlib import Path
import time

INPUT_DIR  = Path("input_images")
OUTPUT_DIR = Path("output_images")
OUTPUT_DIR.mkdir(exist_ok=True)

def process_image(input_path):
    """Load, apply expensive filter, save."""
    img = Image.open(input_path)

    # CPU-heavy filter chain
    img = img.filter(ImageFilter.GaussianBlur(radius=5))
    img = img.filter(ImageFilter.EDGE_ENHANCE_MORE)
    img = img.convert("RGB")

    output_path = OUTPUT_DIR / input_path.name
    img.save(output_path, quality=85)
    return input_path.name

if __name__ == "__main__":
    images = list(INPUT_DIR.glob("*.jpg"))
    print(f"Processing {len(images)} images on {cpu_count()} cores")

    start = time.perf_counter()
    with Pool(processes=cpu_count()) as pool:
        # imap_unordered streams results, chunksize batches for efficiency
        for i, name in enumerate(pool.imap_unordered(process_image, images, chunksize=4), 1):
            print(f"  [{i}/{len(images)}] {name}")

    print(f"\nCompleted in {time.perf_counter() - start:.2f}s")

Real-World Example — Parallel Data Aggregation With Shared Memory

Compute per-row statistics on a huge NumPy array using all cores, without copying the data to each worker. Uses shared_memory for the input and a small Array for the output.

from multiprocessing import Process, Array, cpu_count
from multiprocessing.shared_memory import SharedMemory
import numpy as np
import time

def worker(shm_name, shape, dtype, out_arr, start_row, end_row):
    shm = SharedMemory(name=shm_name)
    data = np.ndarray(shape, dtype=dtype, buffer=shm.buf)

    # Compute row means for our slice
    for i in range(start_row, end_row):
        out_arr[i] = data[i].mean()

    shm.close()

if __name__ == "__main__":
    # Simulate a 100k x 1000 dataset (~800 MB float64)
    N_ROWS, N_COLS = 100_000, 1_000
    data = np.random.rand(N_ROWS, N_COLS)

    # Move into shared memory
    shm = SharedMemory(create=True, size=data.nbytes)
    shared = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
    shared[:] = data

    # Shared output — one mean per row
    results = Array('d', N_ROWS)

    # Split work across cores
    n_workers = cpu_count()
    chunk = N_ROWS // n_workers

    start = time.perf_counter()
    procs = []
    for i in range(n_workers):
        s = i * chunk
        e = N_ROWS if i == n_workers - 1 else (i + 1) * chunk
        p = Process(target=worker,
                    args=(shm.name, data.shape, data.dtype, results, s, e))
        procs.append(p); p.start()

    for p in procs: p.join()

    elapsed = time.perf_counter() - start
    print(f"Parallel: {elapsed:.2f}s using {n_workers} workers")
    print(f"First 5 row means: {list(results[:5])}")

    shm.close(); shm.unlink()

Common Pitfalls And Fixes

from multiprocessing import Pool

def safe_task(item):
    try:
        return {"ok": True, "result": process(item)}
    except Exception as e:
        return {"ok": False, "error": repr(e), "item": item}

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        results = pool.map(safe_task, items)

    ok    = [r for r in results if r["ok"]]
    fail  = [r for r in results if not r["ok"]]
    print(f"{len(ok)} succeeded, {len(fail)} failed")

Choosing The Right IPC — Decision Table

Golden Rules