Multithreading vs Multiprocessing: Choosing the Right Model

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: Shared state without synchronization
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: Use lock
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: ~3-4x speedup with 4 processes

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