Multiprocessing Basics
What is Multiprocessing?
Multiprocessing creates separate processes, each with its own memory space. Bypasses the GIL for true parallelism.
import multiprocessing
import time
def cpu_bound_task(n):
return sum(i * i for i in range(n))
if __name__ == '__main__':
# Create processes
p1 = multiprocessing.Process(target=cpu_bound_task, args=(10**7,))
p2 = multiprocessing.Process(target=cpu_bound_task, args=(10**7,))
# Start
p1.start()
p2.start()
# Wait
p1.join()
p2.join()
print('Done')
Process Class
import multiprocessing
import time
class Worker(multiprocessing.Process):
def __init__(self, name, data):
super().__init__()
self.name = name
self.data = data
self.result = None
def run(self):
# Code to run in process
self.result = sum(x * x for x in self.data)
print(f'{self.name} finished: {self.result}')
if __name__ == '__main__':
data1 = list(range(10**6))
data2 = list(range(10**6, 2*10**6))
p1 = Worker('Process-1', data1)
p2 = Worker('Process-2', data2)
p1.start()
p2.start()
p1.join()
p2.join()
print(f'Results: {p1.result}, {p2.result}')
Threading vs Multiprocessing
# Threading:
# - Shared memory (faster communication)
# - GIL limits to one thread executing Python
# - Good for I/O-bound tasks
# - Lower overhead
# Multiprocessing:
# - Separate memory (no shared state)
# - Bypasses GIL (true parallelism)
# - Good for CPU-bound tasks
# - Higher overhead (process creation)
# When to use what:
# - I/O-bound (network, file, database) -> threading or asyncio
# - CPU-bound (math, data processing) -> multiprocessing
ProcessPoolExecutor
Basic Usage
from concurrent.futures import ProcessPoolExecutor
import math
def is_prime(n):
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
# Find primes in parallel
numbers = list(range(10**6, 10**6 + 10000))
with ProcessPoolExecutor(max_workers=4) as executor:
results = list(executor.map(is_prime, numbers))
primes = [n for n, is_p in zip(numbers, results) if is_p]
print(f'Found {len(primes)} primes')
Submit and As Completed
from concurrent.futures import ProcessPoolExecutor, as_completed
def heavy_computation(n):
return sum(i * i for i in range(n))
with ProcessPoolExecutor() as executor:
# Submit tasks
futures = [executor.submit(heavy_computation, 10**6) for _ in range(4)]
# Process as completed
for future in as_completed(futures):
result = future.result()
print(f'Result: {result}')
Error Handling
from concurrent.futures import ProcessPoolExecutor, as_completed
def risky_task(n):
if n == 0:
raise ValueError('Cannot process zero')
return 100 / n
with ProcessPoolExecutor() as executor:
futures = {executor.submit(risky_task, i): 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'Error processing {futures[future]}: {e}')
When to Use ProcessPoolExecutor
# ✅ Use for:
# - CPU-bound tasks (math, data processing)
# - Parallel processing of independent items
# - Tasks that benefit from multiple cores
# ❌ Don't use for:
# - I/O-bound tasks (use threading/asyncio)
# - Tasks needing shared state (use threading)
# - Very short tasks (overhead > benefit)
# Rule of thumb:
# - max_workers = number of CPU cores for CPU-bound
# - max_workers = 10-20x cores for I/O-bound
Inter-Process Communication
Shared Memory
import multiprocessing
import time
# Shared memory (fast, limited types)
def worker(shared_array, lock, index):
with lock:
shared_array[index] = index * 2
if __name__ == '__main__':
# Create shared array
shared_array = multiprocessing.Array('i', [0] * 10)
lock = multiprocessing.Lock()
processes = []
for i in range(10):
p = multiprocessing.Process(target=worker, args=(shared_array, lock, i))
processes.append(p)
p.start()
for p in processes:
p.join()
print(list(shared_array)) # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
Queue
import multiprocessing
import time
def producer(queue):
for i in range(5):
queue.put(f'Item {i}')
time.sleep(0.1)
queue.put(None) # Sentinel
def consumer(queue):
while True:
item = queue.get()
if item is None:
break
print(f'Processing {item}')
if __name__ == '__main__':
queue = multiprocessing.Queue()
producer_proc = multiprocessing.Process(target=producer, args=(queue,))
consumer_proc = multiprocessing.Process(target=consumer, args=(queue,))
producer_proc.start()
consumer_proc.start()
producer_proc.join()
consumer_proc.join()
Pipe
import multiprocessing
def sender(conn):
conn.send('Hello from sender')
conn.send(42)
conn.close()
def receiver(conn):
while True:
try:
msg = conn.recv()
print(f'Received: {msg}')
except EOFError:
break
if __name__ == '__main__':
parent_conn, child_conn = multiprocessing.Pipe()
sender_proc = multiprocessing.Process(target=sender, args=(child_conn,))
receiver_proc = multiprocessing.Process(target=receiver, args=(parent_conn,))
sender_proc.start()
receiver_proc.start()
sender_proc.join()
receiver_proc.join()
Manager
import multiprocessing
def worker(shared_dict, key, value):
shared_dict[key] = value
if __name__ == '__main__':
manager = multiprocessing.Manager()
shared_dict = manager.dict()
processes = []
for i in range(5):
p = multiprocessing.Process(
target=worker,
args=(shared_dict, f'key{i}', i * 10)
)
processes.append(p)
p.start()
for p in processes:
p.join()
print(dict(shared_dict))
# {'key0': 0, 'key1': 10, 'key2': 20, 'key3': 30, 'key4': 40}
IPC Comparison
# Shared Memory:
# - Fastest
# - Limited to simple types (Array, Value)
# - Need synchronization (Lock)
# Queue:
# - Thread-safe and process-safe
# - Good for producer-consumer
# - Pickles data (slower)
# Pipe:
# - Two-way communication
# - Faster than Queue for two processes
# - Can only connect two processes
# Manager:
# - Supports complex types (dict, list)
# - Slowest (proxy objects)
# - Good for shared state