Introduction

When datasets grow and computations become heavy, a single Python interpreter process often becomes a bottleneck. Why? Because of the Global Interpreter Lock (GIL) — Python threads don't run Python bytecode in parallel on multiple CPU cores. Enter multiprocessing: a standard library module that runs separate Python processes, enabling true parallelism on multi-core machines.

In this post you'll learn:

• Core concepts and when to use multiprocessing vs threading or async.
• Practical, ready-to-run code examples for CPU-bound and data-intensive tasks.
• How to use shared memory (including NumPy-friendly approaches).
• How to integrate multiprocessing safely in web applications (Django) and event-driven designs (Observer pattern).
• Performance best practices, monitoring, and common pitfalls.

This is targeted at intermediate Python developers who want to convert slow single-process workflows into efficient parallel pipelines.

Prerequisites

• Python 3.8+ recommended (we'll use multiprocessing.shared_memory examples introduced in 3.8).
• Basic knowledge of Python functions and modules.
• Familiarity with NumPy is helpful for the shared-memory example.
• Optional: basic Django knowledge for the integration notes.

If you haven't used multiprocessing before, you should first read Python's documentation:

• Official multiprocessing docs (search "multiprocessing — process-based parallelism" in Python docs)

Core Concepts

• CPU-bound vs I/O-bound: Use multiprocessing for CPU-bound tasks (math, data transforms). For I/O-bound tasks (network, disk), threads or asyncio may be better.
• Process vs Thread: Processes have separate memory spaces; threads share memory but are constrained by the GIL.
• Pickleable data: When sending data to worker processes, arguments and return values are pickled. Complex objects (open file handles, generators, local functions, lambdas) can cause issues.
• spawn vs fork: On Windows, the default is spawn (fresh interpreter). On Unix, fork is available (copy-on-write memory benefits). Use if __name__ == '__main__' to avoid recursion on spawn platforms.
• Shared memory: Use multiprocessing.Value/Array or multiprocessing.shared_memory for large arrays to avoid expensive copies.
• Pool vs Process: Pools provide convenient worker management and mapping functions; Process gives fine control.

High-Level Strategy (Analogy)

Think of your machine as a kitchen and each CPU core as a cook. If you have one cook, preparing a complex dish (heavy computation) takes long. Multiprocessing hires additional cooks, each preparing pieces of multiple dishes simultaneously. But hiring cooks costs overhead (start time, duplicated pantry), so only hire when workload justifies it.

Step-by-Step Examples

Example 1 — CPU-bound: Parallel Prime Counting

A classical demonstration: count primes over ranges using a Pool.

# cpu_primes.py
import math
from multiprocessing import Pool, cpu_count
import time
def is_prime(n: int) -> bool:
    if n < 2:
        return False
    if n % 2 == 0:
        return n == 2
    r = int(math.sqrt(n))
    for i in range(3, r + 1, 2):
        if n % i == 0:
            return False
    return True
def count_primes_in_range(rng):
    start, end = rng
    count = 0
    for n in range(start, end):
        if is_prime(n):
            count += 1
    return count
def split_ranges(start, end, chunks):
    size = (end - start) // chunks
    ranges = []
    s = start
    for i in range(chunks):
        e = s + size if i < chunks - 1 else end
        ranges.append((s, e))
        s = e
    return ranges
if __name__ == "__main__":
    start_val, end_val = 10_000, 100_000
    n_cpus = cpu_count()
    ranges = split_ranges(start_val, end_val, n_cpus)
    t0 = time.time()
    with Pool(processes=n_cpus) as pool:
        counts = pool.map(count_primes_in_range, ranges)
    total_primes = sum(counts)
    t1 = time.time()
    print(f"Primes between {start_val} and {end_val}: {total_primes}")
    print(f"Elapsed: {t1 - t0:.2f} s using {n_cpus} processes")

Explanation (line-by-line):

• import modules: math for sqrt, Pool and cpu_count from multiprocessing, time for timing.
• is_prime: simple deterministic primality check (efficient for moderate n).

- Handles n < 2, even numbers, then tests odd divisors up to sqrt(n).

• count_primes_in_range: counts primes in half-open interval [start, end).
• split_ranges: splits the global interval into roughly equal subranges for each worker.
• In main: compute ranges, create a Pool with one process per CPU, call pool.map to run count_primes_in_range in parallel, sum results, and print elapsed time.

Edge cases:

• Very small ranges • map overhead may dominate and be slower than single-threaded.
• is_prime not optimized for huge numbers (for very large numbers consider optimized algorithms or libraries).

Call to action: Run the script and compare elapsed times with different process counts (try cpu_count() - 1).

Example 2 — Data-Intensive: Chunked CSV Processing with Pool.imap and Partial

When processing a large CSV that doesn't fit in memory, process it in chunks and use Pool.imap or apply_async to stream results.

# chunked_processing.py
import csv
from multiprocessing import Pool
from functools import partial
from typing import List, Dict
import gzip
import json
def process_rows(rows: List[Dict[str, str]]) -> List[Dict[str, str]]:
    # Example transform: compute derived field and filter
    out = []
    for row in rows:
        try:
            value = float(row.get("value", "0"))
            if value > 0:
                row["log_value"] = value  0.5
                out.append(row)
        except ValueError:
            # skip malformed rows
            continue
    return out

def chunk_reader(file_path, chunk_size=10_000):
    open_fn = gzip.open if file_path.endswith(".gz") else open
    with open_fn(file_path, newline='') as fh:
        reader = csv.DictReader(fh)
        chunk = []
        for row in reader:
            chunk.append(row)
            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []
        if chunk:
            yield chunk
if __name__ == "__main__":
    file_path = "big_data.csv"  # or big_data.csv.gz
    pool_size = 4
    pool = Pool(pool_size)
    try:
        for processed in pool.imap(process_rows, chunk_reader(file_path), chunksize=1):
            # write or stream processed rows to disk / DB
            for row in processed:
                print(json.dumps(row))  # placeholder sink
    finally:
        pool.close()
        pool.join()

Explanation:

• process_rows: transforms and filters a list of CSV rows; handles parse errors defensively.
• chunk_reader: lazily yields lists of rows (chunks), supports gzipped files.
• Using Pool.imap: iterates over results as they become available, keeping memory usage bounded.
• Use chunksize=1 for imap to get fine-grained streaming. For performance, adjust chunksize upward.

Edge cases:

• If transform is CPU-light and I/O dominates, multiprocessing might not help.
• Use Manager or database for final sink if results need global aggregation.

SEO tip: chunking large files is crucial in data-intensive multiprocessing to avoid memory blow-ups.

Example 3 — Shared Memory with NumPy (multiprocessing.shared_memory)

Large NumPy arrays can be expensive to copy; use shared_memory to let worker processes access the same buffer.

# shared_numpy.py
import numpy as np
from multiprocessing import Process
from multiprocessing import shared_memory
import time
def worker(name, shape, dtype, start_idx, end_idx):
    shm = shared_memory.SharedMemory(name=name)
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    # operate in-place on slice
    arr[start_idx:end_idx] = arr[start_idx:end_idx]  2
    shm.close()
if __name__ == "__main__":
    size = 10_000_000
    a = np.arange(size, dtype=np.float64)
    shm = shared_memory.SharedMemory(create=True, size=a.nbytes)
    arr = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
    arr[:] = a[:]
    n_workers = 4
    chunk = size // n_workers
    procs = []
    t0 = time.time()
    for i in range(n_workers):
        s = i  chunk
        e = (i + 1)  chunk if i < n_workers - 1 else size
        p = Process(target=worker, args=(shm.name, a.shape, a.dtype, s, e))
        p.start()
        procs.append(p)
    for p in procs:
        p.join()
    t1 = time.time()
    print("Done", t1 - t0)
    # verify
    print(arr[0], arr[-1])
    shm.close()
    shm.unlink()

Explanation:

• Create a large NumPy array a.
• Create a SharedMemory block of appropriate size, attach a NumPy ndarray to it using the shared buffer.
• Spawn multiple Processes that attach to the same SharedMemory by name and operate in-place on non-overlapping slices.
• Close and unlink shared memory when done.

Edge cases:

• Ensure slices don't overlap to avoid race conditions.
• Must clean up shared memory (unlink) to prevent leaks.
• On Windows, object lifetime and spawn behavior require if __name__ == '__main__'.

Using functools.lru_cache to Reduce Workload

Before parallelizing, sometimes caching repeated expensive computations is enough. functools.lru_cache memoizes function calls.

from functools import lru_cache
@lru_cache(maxsize=1024)
def expensive_compute(x):
    # placeholder heavy compute
    return x * x  # replace with real work
When combined with multiprocessing, lru_cache is per-process; 
consider shared cache strategies (e.g., Redis) if cross-process caching is needed.

Notes:

• lru_cache runs in-process; it won't share cache across processes.
• For repeated calls within each worker, lru_cache helps reduce compute work.

Integrating with Web Apps and Event-Driven Systems

• Django: avoid heavy synchronous work during requests. Offload to background processes.

- Options: Celery (recommended) or invoke multiprocessing in a management command or separate worker service. If you attempt to spawn processes in a Django view you must ensure web server compatibility (Gunicorn workers, uWSGI, etc.) and avoid forking within process managers. - Example use-case: a Django CRUD (Post/Put/Delete) operation stores data to PostgreSQL and enqueues heavy bulk processing (e.g., analytics) handled by multiprocessing workers or a task queue. If you need a simple background job from Django, consider a management command or systemd service instead of forking from request threads. - Security/performance: avoid sharing Django DB connections across processes; always create new DB connections in the child process.

• Observer Pattern & Event-Driven Architecture:

- Use an Observer Pattern to decouple events (e.g., "file ingestion completed", "record updated") from processing logic. - Observers can enqueue messages to a multiprocessing Manager or a Queue consumed by worker processes. - If you need persistent or distributed event systems, consider Redis, RabbitMQ, or Kafka.

Integration example (conceptual):

• Django API writes record -> emits event to Manager/Queue -> worker processes pick events and run CPU-heavy transforms -> store results in PostgreSQL.
• This shows how Observer Pattern and web CRUD (Django + PostgreSQL) fit naturally with multiprocessing for decoupled, scalable pipelines.

Best Practices

• Always protect process-spawning code with if __name__ == '__main__': to avoid infinite process recursion on spawn platforms.
• Use Pools for simple map/filter workloads; use explicit Processes for custom lifecycles.
• Tune chunksize in Pool.map/imap to balance overhead and throughput.
• Avoid passing huge objects as args — use shared_memory or store data on disk and pass references.
• Be careful with pickling: top-level functions are safe; lambdas or nested functions are not picklable by default.
• Clean up resources (close and join pools, unlink shared memory).
• Profile before optimizing: use time.perf_counter(), cProfile, or line_profiler to find hotspots.
• Use process-safe logging (configure each process's logging handler or send logs through a QueueHandler).

Common Pitfalls and How to Avoid Them

• Forgetting if __name__ == '__main__': leads to recursion and many processes spawning.
• Memory explosion: copying large datasets to each process will exhaust RAM. Use shared memory or smaller chunks.
• Unpicklable objects: avoid sending DB connections, open file handles, or local functions into worker args.
• Not handling signals: use try/finally to close pool and join, and handle KeyboardInterrupt to terminate gracefully.
• Ignoring startup cost: processes have overhead; for very short tasks, multiplexing may be slower than sequential execution.

Graceful interrupt pattern (safe Pool termination):

from multiprocessing import Pool
import signal
def init_worker():
    signal.signal(signal.SIGINT, signal.SIG_IGN)
if __name__ == "__main__":
    pool = Pool(4, initializer=init_worker)
    try:
        results = pool.map(some_fn, data)
    except KeyboardInterrupt:
        print("Keyboard interrupt received, terminating workers")
        pool.terminate()
    else:
        pool.close()
    finally:
        pool.join()

Advanced Tips

• Use process pools hosted as worker services (long-lived) to avoid repeated startup costs for many small jobs.
• For massive parallelism and distributed workloads, consider Dask or Ray.
• When using numeric arrays, prefer shared_memory for zero-copy inter-process communication.
• To share complex Python objects or collections safely, use multiprocessing.Manager(), but be aware of performance costs since it proxies operations back through a server process.
• Combine lru_cache inside worker functions to avoid recomputing repeated sub-results within a worker.
• For Windows, pickling limitations and spawn semantics make testing multiprocess code more important.

Monitoring and Profiling

• Measure per-process CPU and memory (psutil is helpful).
• Use cProfile for CPU hotspots.
• Log throughput (items/sec) and latency per chunk.
• Benchmark different process counts; sometimes using N-1 cores leaves resources for other system tasks.

Conclusion

Multiprocessing is a powerful tool for accelerating CPU-bound and data-intensive Python workloads. Key takeaways:

• Use multiprocessing for compute-heavy tasks; threads and asyncio for I/O-bound workloads.
• Plan for data movement: avoid unnecessary copies, use shared memory, and chunk large datasets.
• Protect process creation with if __name__ == '__main__', handle signals, and avoid pickling pitfalls.
• Integrate multiprocessing mindfully in web systems (e.g., Django) — usually via background workers or task queues.
• Combine multiprocessing with caching (functools.lru_cache), the Observer Pattern, and robust CRUD workflows (e.g., Django + PostgreSQL) to build scalable, maintainable systems.

Try the examples in this post, profile them on your data, and iterate. Want next steps? Convert one of your heavy-processing Django management commands into a multiprocessing worker and benchmark the result.

Further Reading and Resources

• Python official docs — multiprocessing module
• multiprocessing.shared_memory documentation (Python 3.8+)
• functools.lru_cache documentation
• "Practical Guide to Implementing Observer Pattern in Python for Event-Driven Programming" (search this title for a focused tutorial)
• "Implementing a Simple CRUD Application with Django and PostgreSQL" (for integrating database-backed workflows with background processing)
• Books and tools: "High Performance Python", Dask, Ray, Celery

If you found this useful, try converting one of your slow scripts to use Pool.map or shared_memory and measure the speedup. Share your results or questions — I'd be happy to help debug or optimize a real-world workload!