Multithreading vs Multiprocessing

Overview

Python offers two concurrency models with different trade-offs:

• Threading: Lightweight, shared memory, fast communication, but GIL limits CPU parallelism
• Multiprocessing: Heavy processes, isolated memory, true CPU parallelism, but slow communication

Choosing the right model is critical for system performance.

-

Threading: Lightweight Concurrency

ThreadPoolExecutor for I/O

from concurrent.futures import ThreadPoolExecutor
import time
import requests

def fetch_url(url):
 """Fetch URL (I/O-bound)."""
 response = requests.get(url, timeout=5)
 return len(response.content)

# Sequential (5 URLs, 1 second each = 5 seconds)
urls = ['https://example.com'] * 5

start = time.time()
for url in urls:
 fetch_url(url)
sequential_time = time.time() - start

# Parallel with threads (5 URLs in parallel = 1 second)
start = time.time()
with ThreadPoolExecutor(max_workers=5) as executor:
 sizes = list(executor.map(fetch_url, urls))
threaded_time = time.time() - start

print(f"Sequential: {sequential_time:.1f}s")
print(f"Threaded: {threaded_time:.1f}s")
print(f"Speedup: {sequential_time / threaded_time:.1f}x")

Thread-Safe Data Structures

import threading
from queue import Queue
import time

# Thread-safe queue
queue = Queue(maxsize=10)

def producer():
 """Put items in queue."""
 for i in range(20):
 queue.put(i)
 print(f"Produced {i}")
 time.sleep(0.1)

def consumer():
 """Get items from queue."""
 while True:
 item = queue.get()
 if item is None:
 break
 print(f"Consumed {item}")
 queue.task_done()

# Run producer and consumer in parallel
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)

t1.start()
t2.start()

t1.join()
queue.put(None) # Signal consumer to stop
t2.join()

Threading Caveats

import threading

# WRONG
counter = 0

def increment():
 global counter
 for _ in range(1000000):
 counter += 1 # RACE CONDITION!

threads = [threading.Thread(target=increment) for _ in range(2)]
for t in threads:
 t.start()
for t in threads:
 t.join()

print(counter) # Expected: 2000000, Actual: ~1200000 (random!)

# CORRECT
import threading

counter = 0
lock = threading.Lock()

def increment_safe():
 global counter
 for _ in range(1000000):
 with lock:
 counter += 1

threads = [threading.Thread(target=increment_safe) for _ in range(2)]
for t in threads:
 t.start()
for t in threads:
 t.join()

print(counter) # Correct: 2000000

-

Multiprocessing: True Parallelism

ProcessPoolExecutor for CPU

from concurrent.futures import ProcessPoolExecutor
import time
import math

def cpu_bound(n):
 """CPU-intensive calculation."""
 return sum(math.sqrt(i) for i in range(n))

# Sequential (slow)
start = time.time()
for _ in range(4):
 cpu_bound(10000000)
sequential_time = time.time() - start

# Parallel with processes (true parallelism, no GIL)
start = time.time()
with ProcessPoolExecutor(max_workers=4) as executor:
 results = list(executor.map(cpu_bound, [10000000] * 4))
parallel_time = time.time() - start

print(f"Sequential: {sequential_time:.1f}s")
print(f"Parallel: {parallel_time:.1f}s")
print(f"Speedup: {sequential_time / parallel_time:.1f}x")
# Result

Inter-Process Communication

from multiprocessing import Process, Queue, Pipe
import time

# Queue for one-way communication
def producer(queue):
 for i in range(10):
 queue.put(i)

def consumer(queue):
 for _ in range(10):
 item = queue.get()
 print(f"Received: {item}")

# Pipe for two-way communication
def process1(conn):
 for i in range(5):
 conn.send(f"Message {i}")
 response = conn.recv()
 print(f"Got response: {response}")

def process2(conn):
 for i in range(5):
 msg = conn.recv()
 print(f"Received: {msg}")
 conn.send(f"Response {i}")

# Queue example
if __name__ == '__main__':
 q = Queue()
 p1 = Process(target=producer, args=(q,))
 p2 = Process(target=consumer, args=(q,))

 p1.start()
 p2.start()

 p1.join()
 p2.join()

Shared Memory (Dangerous)

from multiprocessing import Process, Value, Array
import time

# Shared integer
def increment_shared(shared_counter):
 for _ in range(1000000):
 with shared_counter.get_lock():
 shared_counter.value += 1

if __name__ == '__main__':
 # Create shared value
 counter = Value('i', 0) # 'i' = integer

 # Create shared array
 data = Array('d', [1.0, 2.0, 3.0, 4.0]) # 'd' = double

 # Increment in parallel
 processes = [Process(target=increment_shared, args=(counter,)) 
 for _ in range(2)]

 for p in processes:
 p.start()

 for p in processes:
 p.join()

 print(f"Counter: {counter.value}") # Correct: 2000000

 # Note: Shared memory is slower than message passing!

---

Comparison Table

Aspect	Threading	Multiprocessing
CPU-Bound	No (GIL)	Yes (true parallelism)
I/O-Bound	Yes (fast)	Yes (overkill)
Memory	Shared, low overhead	Isolated, high overhead
Startup	Fast (~1ms)	Slow (~100ms)
Communication	Fast (shared memory)	Slow (IPC)
Synchronization	Complex (locks, etc)	Simpler (message passing)
Debugging	Easier	Harder

-

Hybrid Approach: Multi-GPU Inference Server

from multiprocessing import Process
from concurrent.futures import ThreadPoolExecutor
import torch
import queue

class MultiGPUServer:
 """Use processes for GPU isolation, threads for I/O."""

 def __init__(self, num_gpus=4):
 self.num_gpus = num_gpus
 self.request_queues = [queue.Queue() for _ in range(num_gpus)]
 self.result_queues = [queue.Queue() for _ in range(num_gpus)]

 def gpu_worker(self, gpu_id):
 """Run inference on specific GPU."""
 model = torch.load('model.pth')
 model = model.to(f'cuda:{gpu_id}')
 model.eval()

 while True:
 request_id, data = self.request_queues[gpu_id].get()

 if request_id is None: # Stop signal
 break

 with torch.no_grad():
 x = torch.tensor(data, device=f'cuda:{gpu_id}')
 result = model(x)

 self.result_queues[gpu_id].put((request_id, result))

 def start_workers(self):
 """Start GPU workers in separate processes."""
 self.processes = []
 for gpu_id in range(self.num_gpus):
 p = Process(target=self.gpu_worker, args=(gpu_id,))
 p.start()
 self.processes.append(p)

 def infer(self, request_id, data):
 """Submit inference request."""
 # Round-robin load balancing
 gpu_id = request_id % self.num_gpus
 self.request_queues[gpu_id].put((request_id, data))

 # Get result
 _, result = self.result_queues[gpu_id].get()
 return result

if __name__ == '__main__':
 server = MultiGPUServer(num_gpus=4)
 server.start_workers()

 # Submit requests (can use threads to fetch data)
 results = []
 with ThreadPoolExecutor(max_workers=10) as executor:
 futures = []
 for i in range(100):
 future = executor.submit(server.infer, i, data[i])
 futures.append(future)

 for future in futures:
 results.append(future.result())

-

Decision Framework

Use Threading When

• I/O-bound (network, disk)
• Need low latency
• Shared state is simple
• NumPy/PyTorch computation (GIL released)

Use Multiprocessing When

• CPU-bound pure Python
• Need true parallelism
• Can tolerate higher latency
• Processes should be isolated

Use Async When

• Very high concurrency (1000s of connections)
• Mostly I/O-bound
• Can rewrite code as async

-

Performance Checklist

# Threading
def use_threading():
 # Network requests
 # Disk I/O
 # Database queries
 # DataLoader with workers
 pass

# Multiprocessing
def use_multiprocessing():
 # CPU-intensive computation
 # Multiple GPUs
 # Long-running workers
 # Process isolation
 pass

# Async
def use_async():
 # Web server
 # WebSocket connections
 # Concurrent I/O
 # High concurrency
 pass

-

Related Topics

• 03 Global Interpreter Lock (Gil) - Why threading has limitations
• 05 Async Await - Async alternative
• 03 Distributed Training - Multi-process training