Multiprocessing

Threading (shared memory, one GIL):
┌──────────────────────────────────┐
│          Process (1 GIL)         │
│  Thread 1 ──┐                    │
│  Thread 2 ──┤── shared memory    │
│  Thread 3 ──┘                    │
└──────────────────────────────────┘

Multiprocessing (separate memory, no GIL):
┌───────────────┐  ┌───────────────┐  ┌───────────────┐
│  Process 1    │  │  Process 2    │  │  Process 3    │
│  own GIL      │  │  own GIL      │  │  own GIL      │
│  own memory   │  │  own memory   │  │  own memory   │
└───────────────┘  └───────────────┘  └───────────────┘

import multiprocessing
import threading
import time

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

N = 5_000_000
WORKERS = 4

# Sequential baseline
start = time.perf_counter()
for _ in range(WORKERS):
    cpu_work(N)
seq = time.perf_counter() - start

# Threading — limited by GIL
start = time.perf_counter()
threads = [threading.Thread(target=cpu_work, args=(N,)) for _ in range(WORKERS)]
for t in threads: t.start()
for t in threads: t.join()
thr = time.perf_counter() - start

# Multiprocessing — true parallelism
start = time.perf_counter()
procs = [multiprocessing.Process(target=cpu_work, args=(N,)) for _ in range(WORKERS)]
for p in procs: p.start()
for p in procs: p.join()
mp = time.perf_counter() - start

print(f"Sequential:     {seq:.2f}s")
print(f"Threading:      {thr:.2f}s  (~same as sequential)")
print(f"Multiprocessing:{mp:.2f}s  (~{seq/mp:.1f}x faster)")

import multiprocessing
import os

def worker(name):
    print(f"Worker {name}: PID={os.getpid()}, parent={os.getppid()}")

if __name__ == "__main__":   # ← REQUIRED on Windows / macOS (spawn start method)
    p = multiprocessing.Process(target=worker, args=("Alpha",))
    p.start()
    p.join()
    print(f"Main PID={os.getpid()}")

# Each process has its own PID and memory — changes don't affect the parent

import multiprocessing

def task(x):
    return x ** 2

if __name__ == "__main__":
    p = multiprocessing.Process(
        target=task,
        args=(5,),
        name="squarer",
        daemon=False,    # daemon process exits when parent exits
    )
    p.start()
    p.join(timeout=10)   # wait at most 10 seconds

    if p.is_alive():
        p.terminate()    # SIGTERM — ask to stop
        p.join()

    print(f"Exit code: {p.exitcode}")   # 0 = clean exit

from concurrent.futures import ProcessPoolExecutor, as_completed
import math

def is_prime(n):
    """CPU-bound primality test."""
    if n < 2: return False
    if n == 2: return True
    if n % 2 == 0: return False
    for i in range(3, int(math.sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True

if __name__ == "__main__":
    candidates = range(10_000_000, 10_001_000)

    # map — submit all, get results in order
    with ProcessPoolExecutor() as executor:   # defaults to cpu_count() workers
        primes = [n for n, p in zip(candidates, executor.map(is_prime, candidates)) if p]
    print(f"Found {len(primes)} primes: {primes[:5]}…")

    # submit + as_completed — results arrive as they finish
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(is_prime, n): n for n in candidates[:20]}
        for future in as_completed(futures):
            n = futures[future]
            if future.result():
                print(f"{n} is prime")

import multiprocessing

def square(n):
    return n ** 2

if __name__ == "__main__":
    with multiprocessing.Pool(processes=4) as pool:
        # map — synchronous: blocks until all results are ready
        results = pool.map(square, range(10))
        print(results)   # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

        # imap — lazy iterator version (memory efficient)
        for result in pool.imap(square, range(10)):
            print(result, end=" ")

        # starmap — when each task takes multiple arguments
        pairs = [(2, 3), (4, 5), (6, 7)]
        results = pool.starmap(pow, pairs)
        print(results)   # [8, 1024, 279936]

        # apply_async — non-blocking single call
        ar = pool.apply_async(square, (42,))
        print(ar.get(timeout=5))   # 1764

import multiprocessing

def producer(q, items):
    for item in items:
        q.put(item)
    q.put(None)   # sentinel

def consumer(q, results):
    while True:
        item = q.get()
        if item is None:
            break
        results.put(item ** 2)
    results.put(None)

if __name__ == "__main__":
    work_q    = multiprocessing.Queue()
    results_q = multiprocessing.Queue()

    p1 = multiprocessing.Process(target=producer, args=(work_q, range(10)))
    p2 = multiprocessing.Process(target=consumer, args=(work_q, results_q))

    p1.start(); p2.start()
    p1.join();  p2.join()

    squares = []
    while True:
        item = results_q.get()
        if item is None: break
        squares.append(item)
    print(squares)

import multiprocessing

def child(conn):
    msg = conn.recv()          # receive from parent
    print(f"Child got: {msg}")
    conn.send(msg.upper())     # send back
    conn.close()

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

    p = multiprocessing.Process(target=child, args=(child_conn,))
    p.start()

    parent_conn.send("hello from parent")
    reply = parent_conn.recv()
    print(f"Parent got: {reply}")   # HELLO FROM PARENT

    p.join()

import multiprocessing

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

if __name__ == "__main__":
    # Value — single shared typed value
    counter = multiprocessing.Value("i", 0)   # "i" = signed int
    lock    = multiprocessing.Lock()

    procs = [
        multiprocessing.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"Counter: {counter.value}")   # 40000

    # Array — fixed-size shared array
    shared = multiprocessing.Array("d", [1.0, 2.0, 3.0, 4.0])   # "d" = double
    print(list(shared))   # [1.0, 2.0, 3.0, 4.0]

import multiprocessing

def worker(results, key, value):
    results[key] = value   # modify the shared dict

if __name__ == "__main__":
    with multiprocessing.Manager() as manager:
        # manager creates proxy objects backed by a server process
        shared_dict = manager.dict()
        shared_list = manager.list()

        procs = [
            multiprocessing.Process(target=worker, args=(shared_dict, i, i**2))
            for i in range(5)
        ]
        for p in procs: p.start()
        for p in procs: p.join()

        print(dict(shared_dict))   # {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}

import pickle

# ✓ Picklable
pickle.dumps(42)
pickle.dumps([1, 2, 3])
pickle.dumps({"a": 1})
pickle.dumps(lambda x: x)     # lambdas CAN be pickled in Python 3.8+
                               # (but NOT in all contexts — e.g. Pool.map)

# ❌ Common unpicklable things
# - Lambda functions passed to Pool.map (use top-level def instead)
# - Nested functions (defined inside another function)
# - File handles, sockets, database connections
# - Objects with non-picklable attributes

import multiprocessing

# ❌ Lambdas don't work with Pool.map
# pool.map(lambda x: x**2, range(10))  # PicklingError!

# ✓ Use top-level functions
def square(x):
    return x ** 2

if __name__ == "__main__":
    with multiprocessing.Pool(4) as pool:
        print(pool.map(square, range(10)))

import multiprocessing

if __name__ == "__main__":
    # Explicitly set start method (call once, before any Process creation)
    multiprocessing.set_start_method("spawn")

    # Or use a context to avoid changing global state
    ctx = multiprocessing.get_context("spawn")
    p   = ctx.Process(target=print, args=("hello from spawn",))
    p.start()
    p.join()

import multiprocessing

def process_batch(batch):
    return [x ** 2 for x in batch]

def chunk(lst, size):
    for i in range(0, len(lst), size):
        yield lst[i:i + size]

if __name__ == "__main__":
    data       = list(range(1_000_000))
    batch_size = 10_000

    with multiprocessing.Pool() as pool:
        # Each worker gets a batch of 10,000 items — fewer pickle round-trips
        batches = list(chunk(data, batch_size))
        nested  = pool.map(process_batch, batches)
        results = [item for batch in nested for item in batch]

    print(f"Processed {len(results):,} items")

💻 Exercises

import math
import time
from concurrent.futures import ProcessPoolExecutor

def is_prime(n):
    if n < 2: return False
    if n == 2: return True
    if n % 2 == 0: return False
    for i in range(3, int(math.sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True

def find_primes(start, end):
    return [n for n in range(start, end) if is_prime(n)]

def parallel_primes(end, workers=4):
    chunk_size = end // workers
    ranges = [
        (i * chunk_size, (i + 1) * chunk_size if i < workers - 1 else end)
        for i in range(workers)
    ]
    with ProcessPoolExecutor(max_workers=workers) as executor:
        results = executor.map(lambda r: find_primes(*r), ranges)
        # Note: lambda won't pickle — use a helper
    return sorted(p for chunk in results for p in chunk)

# Picklable helper
def find_primes_range(args):
    return find_primes(*args)

def parallel_primes(end, workers=4):
    chunk_size = (end + workers - 1) // workers
    ranges = [
        (i * chunk_size, min((i + 1) * chunk_size, end))
        for i in range(workers)
    ]
    with ProcessPoolExecutor(max_workers=workers) as executor:
        chunks = list(executor.map(find_primes_range, ranges))
    return sorted(p for chunk in chunks for p in chunk)

if __name__ == "__main__":
    END = 100_000

    t = time.perf_counter()
    seq = find_primes(0, END)
    print(f"Sequential: {time.perf_counter()-t:.3f}s — {len(seq)} primes")

    t = time.perf_counter()
    par = parallel_primes(END, workers=4)
    print(f"Parallel:   {time.perf_counter()-t:.3f}s — {len(par)} primes")

    print(f"Results match: {seq == par}")

from collections import Counter
from concurrent.futures import ProcessPoolExecutor
import re

def count_words_chunk(lines):
    """Count words in a list of lines (runs in a worker process)."""
    counter = Counter()
    for line in lines:
        words = re.findall(r"\b[a-z]+\b", line.lower())
        counter.update(words)
    return counter

def parallel_word_count(lines, top_n=10, workers=4):
    # Split into roughly equal chunks
    chunk_size = max(1, len(lines) // workers)
    chunks = [lines[i:i+chunk_size] for i in range(0, len(lines), chunk_size)]

    with ProcessPoolExecutor(max_workers=workers) as executor:
        partial_counts = list(executor.map(count_words_chunk, chunks))

    # Merge all partial counters in the main process
    total = Counter()
    for partial in partial_counts:
        total.update(partial)

    return total.most_common(top_n)

if __name__ == "__main__":
    # Generate synthetic text
    import random
    words = ["the", "quick", "brown", "fox", "jumps", "over",
             "lazy", "dog", "python", "is", "great"]
    lines = [" ".join(random.choices(words, k=20)) for _ in range(10_000)]

    top = parallel_word_count(lines, top_n=5)
    for word, count in top:
        print(f"  {word:<12} {count:>6}")

import multiprocessing
import time
import random

def worker(worker_id, counter, lock, total_tasks):
    """Perform tasks and increment the shared counter."""
    tasks = total_tasks // multiprocessing.cpu_count()
    for _ in range(tasks):
        time.sleep(random.uniform(0.01, 0.05))  # simulate work
        with lock:
            counter.value += 1

def progress_bar(current, total, width=40):
    pct   = current / total
    filled = int(width * pct)
    bar   = "█" * filled + "░" * (width - filled)
    return f"\r[{bar}] {current}/{total} ({pct:.0%})"

if __name__ == "__main__":
    NUM_WORKERS  = 4
    TOTAL_TASKS  = 100

    counter = multiprocessing.Value("i", 0)
    lock    = multiprocessing.Lock()

    workers = [
        multiprocessing.Process(
            target=worker,
            args=(i, counter, lock, TOTAL_TASKS)
        )
        for i in range(NUM_WORKERS)
    ]

    for w in workers:
        w.start()

    # Main process: print live progress
    while any(w.is_alive() for w in workers):
        print(progress_bar(counter.value, TOTAL_TASKS), end="", flush=True)
        time.sleep(0.1)

    for w in workers:
        w.join()

    print(progress_bar(counter.value, TOTAL_TASKS))
    print(f"\nDone! Processed {counter.value} tasks.")