Multithreading & Multiprocessing

Python provides powerful tools for running code concurrently. Understanding threads vs processes — and Python's GIL — is essential for writing high-performance applications.

Concurrency vs Parallelism

The Global Interpreter Lock (GIL)

The GIL is Python's internal mutex that allows only one thread to execute Python bytecode at a time — even on multi-core machines.

Key insight: The GIL means Python threads are great for I/O-bound tasks (where threads release the GIL during I/O waits) but cannot truly parallelize CPU-bound work. Use multiprocessing to bypass the GIL.

---

Part 1 — Threading

Creating Threads

import threading
import time

def download_file(filename, duration):
    """Simulate downloading a file."""
    print(f"Starting download: {filename}")
    time.sleep(duration)  # Simulate network wait
    print(f"Finished download: {filename}")

# Sequential (slow)
start = time.time()
download_file("video.mp4", 2)
download_file("music.mp3", 1)
print(f"Sequential: {time.time() - start:.2f}s\n")

# Threaded (fast)
start = time.time()
t1 = threading.Thread(target=download_file, args=("video.mp4", 2))
t2 = threading.Thread(target=download_file, args=("music.mp3", 1))

t1.start()
t2.start()

t1.join()  # Wait for t1 to finish
t2.join()  # Wait for t2 to finish
print(f"Threaded: {time.time() - start:.2f}s")

Output:

Starting download: video.mp4
Starting download: music.mp3
Finished download: music.mp3
Finished download: video.mp4
Sequential: 3.00s
Threaded: 2.00s

Thread Subclassing

import threading
import time

class WorkerThread(threading.Thread):
    """A reusable worker thread."""

    def __init__(self, name, task_count):
        super().__init__()
        self.name = name
        self.task_count = task_count
        self.results = []

    def run(self):
        """Called when thread starts."""
        for i in range(self.task_count):
            time.sleep(0.1)  # Simulate work
            self.results.append(f"{self.name}-task-{i}")
        print(f"{self.name} completed {self.task_count} tasks")

# Create and start workers
workers = [
    WorkerThread("Worker-A", 3),
    WorkerThread("Worker-B", 2),
    WorkerThread("Worker-C", 4),
]

for w in workers:
    w.start()

for w in workers:
    w.join()

# Collect all results
all_results = [r for w in workers for r in w.results]
print(f"\nTotal tasks completed: {len(all_results)}")
print("Results:", all_results)

Output:

Worker-B completed 2 tasks
Worker-A completed 3 tasks
Worker-C completed 4 tasks

Total tasks completed: 9
Results: ['Worker-A-task-0', 'Worker-A-task-1', 'Worker-A-task-2', ...]

Thread Synchronization — Locks

Without synchronization, threads accessing shared data cause race conditions.

import threading

# UNSAFE: Race condition
counter_unsafe = 0

def increment_unsafe():
    global counter_unsafe
    for _ in range(10000):
        counter_unsafe += 1  # Read-modify-write: NOT atomic!

threads = [threading.Thread(target=increment_unsafe) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Unsafe counter: {counter_unsafe} (expected 50000)")
# May print: Unsafe counter: 43271  <- data corruption!

# SAFE: Using a Lock
counter_safe = 0
lock = threading.Lock()

def increment_safe():
    global counter_safe
    for _ in range(10000):
        with lock:              # Acquire lock
            counter_safe += 1  # Only one thread at a time
            # Lock auto-released on exit

threads = [threading.Thread(target=increment_safe) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Safe counter: {counter_safe} (expected 50000)")

Output:

Unsafe counter: 43271  (data race - unpredictable!)
Safe counter: 50000    (always correct)

Semaphore & Event

import threading
import time

# Semaphore: Limit concurrent access to a resource
pool_semaphore = threading.Semaphore(3)  # Max 3 threads at once

def limited_resource(thread_id):
    with pool_semaphore:
        print(f"Thread {thread_id} acquired resource")
        time.sleep(0.5)
        print(f"Thread {thread_id} released resource")

threads = [threading.Thread(target=limited_resource, args=(i,)) for i in range(7)]
for t in threads: t.start()
for t in threads: t.join()

# Event: Signal between threads
ready_event = threading.Event()

def producer():
    print("Producer: Preparing data...")
    time.sleep(1)
    ready_event.set()   # Signal all waiting consumers
    print("Producer: Signal sent.")

def consumer(cid):
    print(f"Consumer {cid}: Waiting for data...")
    ready_event.wait()  # Block until event is set
    print(f"Consumer {cid}: Got signal, processing!")

threads = [
    threading.Thread(target=producer),
    threading.Thread(target=consumer, args=(1,)),
    threading.Thread(target=consumer, args=(2,)),
]
for t in threads: t.start()
for t in threads: t.join()

Output:

Thread 0 acquired resource
Thread 1 acquired resource
Thread 2 acquired resource
Thread 3 acquired resource  <- waits until one releases
...
Producer: Preparing data...
Consumer 1: Waiting for data...
Consumer 2: Waiting for data...
Producer: Signal sent.
Consumer 1: Got signal, processing!
Consumer 2: Got signal, processing!

Thread-safe Queue

import threading
import queue
import time

# Producer-Consumer pattern using Queue
task_queue = queue.Queue(maxsize=5)

def producer(n_tasks):
    """Adds tasks to the queue."""
    for i in range(n_tasks):
        task_queue.put(f"task-{i}")
        print(f"Produced: task-{i}")
        time.sleep(0.2)
    # Send sentinel values to stop consumers
    for _ in range(3):
        task_queue.put(None)

def consumer(worker_id):
    """Processes tasks from the queue."""
    while True:
        task = task_queue.get()
        if task is None:
            print(f"Worker {worker_id}: No more tasks.")
            task_queue.task_done()
            break
        time.sleep(0.3)  # Simulate processing
        print(f"Worker {worker_id} processed: {task}")
        task_queue.task_done()

prod = threading.Thread(target=producer, args=(6,))
workers = [threading.Thread(target=consumer, args=(i,)) for i in range(3)]

for w in workers: w.start()
prod.start()

prod.join()
for w in workers: w.join()
print("\nAll tasks completed!")

Output:

Produced: task-0
Worker 0 processed: task-0
Produced: task-1
Worker 1 processed: task-1
...
All tasks completed!

ThreadPoolExecutor

The modern, high-level way to manage a pool of threads.

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch_data(url_id):
    """Simulate fetching data from a URL."""
    time.sleep(0.5)  # Simulate network delay
    return f"data_from_url_{url_id}"

urls = list(range(10))  # 10 URLs to fetch

start = time.time()
results = []

with ThreadPoolExecutor(max_workers=4) as executor:
    # Submit all tasks and map futures to their url
    futures = {executor.submit(fetch_data, url): url for url in urls}

    # Process results as they complete (not in submission order)
    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
            results.append(result)
            print(f"Completed URL {url}: {result}")
        except Exception as e:
            print(f"URL {url} failed: {e}")

print(f"\nFetched {len(results)} items in {time.time()-start:.2f}s")

Output:

Completed URL 2: data_from_url_2
Completed URL 0: data_from_url_0
Completed URL 3: data_from_url_3
...
Fetched 10 items in 1.51s   <- 4 workers x ~0.5s each

---

Part 2 — Multiprocessing

Creating Processes

import multiprocessing
import time
import os

def cpu_intensive(n):
    """CPU-bound: compute sum of squares."""
    pid = os.getpid()
    result = sum(i * i for i in range(n))
    print(f"PID {pid}: sum of squares up to {n} = {result}")
    return result

if __name__ == "__main__":
    numbers = [500_000, 1_000_000, 1_500_000, 2_000_000]

    # Sequential
    start = time.time()
    for n in numbers:
        cpu_intensive(n)
    print(f"Sequential: {time.time()-start:.2f}s\n")

    # Parallel processes
    start = time.time()
    processes = [multiprocessing.Process(target=cpu_intensive, args=(n,)) for n in numbers]
    for p in processes: p.start()
    for p in processes: p.join()
    print(f"Multiprocessing: {time.time()-start:.2f}s")

Output:

PID 12301: sum of squares up to 500000 = 41666...
PID 12302: sum of squares up to 1000000 = 333333...
Sequential: 4.21s

PID 12401: ...   <- all run simultaneously on separate CPU cores
PID 12402: ...
Multiprocessing: 1.38s

Sharing Data Between Processes

import multiprocessing

def square_worker(numbers, results, index):
    """Compute square and store in shared array."""
    results[index] = numbers[index] ** 2

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]

    # Shared Array - accessible across processes
    shared_results = multiprocessing.Array("i", len(numbers))  # "i" = int

    processes = [
        multiprocessing.Process(
            target=square_worker,
            args=(numbers, shared_results, i)
        )
        for i in range(len(numbers))
    ]
    for p in processes: p.start()
    for p in processes: p.join()

    print("Squares:", list(shared_results))

    # Shared Value with Lock for process safety
    counter = multiprocessing.Value("i", 0)
    lock = multiprocessing.Lock()

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

    procs = [multiprocessing.Process(target=safe_increment, args=(counter, lock, 1000))
             for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"Counter (expected 4000): {counter.value}")

Output:

Squares: [1, 4, 9, 16, 25]
Counter (expected 4000): 4000

Process Pools

from multiprocessing import Pool
import time

def heavy_computation(n):
    """Simulate CPU-heavy work."""
    return sum(i * i for i in range(n))

def power(base, exp):
    return base ** exp

if __name__ == "__main__":
    tasks = [500_000, 1_000_000, 1_500_000, 2_000_000,
             500_000, 1_000_000, 1_500_000, 2_000_000]

    # pool.map: Apply function to every item in the list
    start = time.time()
    with Pool(processes=4) as pool:
        results = pool.map(heavy_computation, tasks)
    print(f"pool.map first 3 results: {results[:3]}")
    print(f"Time: {time.time()-start:.2f}s\n")

    # pool.starmap: Pass multiple arguments per call
    args_list = [(2, 10), (3, 5), (4, 4), (5, 3)]
    with Pool() as pool:
        powered = pool.starmap(power, args_list)
    print("pool.starmap results:", powered)

Output:

pool.map first 3 results: [41666583333..., 333333833333..., 1125000750000...]
Time: 1.92s

pool.starmap results: [1024, 243, 256, 125]

ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor, as_completed
import time

def analyze_chunk(chunk_id, size):
    """Simulate analyzing a data chunk."""
    result = sum(i ** 2 for i in range(size))
    return chunk_id, result

if __name__ == "__main__":
    chunks = [(i, 300_000) for i in range(8)]  # 8 chunks

    start = time.time()
    results = {}

    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = {
            executor.submit(analyze_chunk, cid, size): cid
            for cid, size in chunks
        }

        for future in as_completed(futures):
            try:
                chunk_id, result = future.result()
                results[chunk_id] = result
                print(f"Chunk {chunk_id} done: {result}")
            except Exception as e:
                print(f"Error: {e}")

    print(f"\nProcessed {len(results)} chunks in {time.time()-start:.2f}s")

Output:

Chunk 1 done: 9000049...
Chunk 0 done: 9000049...
Chunk 3 done: 9000049...
...
Processed 8 chunks in 0.74s

Inter-Process Communication — Pipe & Queue

import multiprocessing
import time

def pipe_sender(conn):
    messages = ["hello", "world", "from", "process"]
    for msg in messages:
        conn.send(msg)
        time.sleep(0.1)
    conn.send(None)  # Sentinel to signal end
    conn.close()

def pipe_receiver(conn):
    while True:
        msg = conn.recv()
        if msg is None:
            break
        print(f"Received via Pipe: {msg}")
    conn.close()

if __name__ == "__main__":
    # Pipe: bidirectional channel between two processes
    parent_conn, child_conn = multiprocessing.Pipe()

    sender = multiprocessing.Process(target=pipe_sender, args=(child_conn,))
    receiver = multiprocessing.Process(target=pipe_receiver, args=(parent_conn,))

    sender.start(); receiver.start()
    sender.join(); receiver.join()

    print("\n--- Queue Demo ---")

    # Queue: Many-producer, many-consumer
    q = multiprocessing.Queue()

    def producer(q, items):
        for item in items:
            q.put(item)
            print(f"Queued: {item}")

    def consumer_proc(q, worker_id):
        while True:
            item = q.get()
            if item is None: break
            print(f"Worker-{worker_id} consumed: {item}")

    p1 = multiprocessing.Process(target=producer, args=(q, [1, 2, 3, 4, 5, 6]))
    c1 = multiprocessing.Process(target=consumer_proc, args=(q, 1))
    c2 = multiprocessing.Process(target=consumer_proc, args=(q, 2))

    p1.start(); c1.start(); c2.start()
    p1.join()
    q.put(None); q.put(None)  # Stop signals for each consumer
    c1.join(); c2.join()

Output:

Received via Pipe: hello
Received via Pipe: world
Received via Pipe: from
Received via Pipe: process

--- Queue Demo ---
Queued: 1
Worker-1 consumed: 1
Worker-2 consumed: 2
Queued: 2
...

---

When to Use What?

Coding Challenge: Parallel Image Processor

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time

def apply_filter(image_id):
    """CPU-bound: Apply a complex filter to an image."""
    result = sum(i ** 0.5 for i in range(100_000))
    return f"filtered_image_{image_id}"

def upload_image(image_id):
    """I/O-bound: Upload processed image to server."""
    time.sleep(0.3)  # Simulate network
    return f"uploaded_image_{image_id}"

image_ids = list(range(8))

# Step 1: CPU-bound processing — use multiprocessing
print("Step 1: Applying filters (CPU-bound)...")
start = time.time()
with ProcessPoolExecutor(max_workers=4) as executor:
    filtered = list(executor.map(apply_filter, image_ids))
print(f"Filtered {len(filtered)} images in {time.time()-start:.2f}s")

# Step 2: I/O-bound uploading — use threads
print("\nStep 2: Uploading images (I/O-bound)...")
start = time.time()
with ThreadPoolExecutor(max_workers=8) as executor:
    uploaded = list(executor.map(upload_image, range(len(filtered))))
print(f"Uploaded {len(uploaded)} images in {time.time()-start:.2f}s")
print("\nDone! First 3:", uploaded[:3])

Output:

Step 1: Applying filters (CPU-bound)...
Filtered 8 images in 0.62s

Step 2: Uploading images (I/O-bound)...
Uploaded 8 images in 0.30s

Done! First 3: ['uploaded_image_0', 'uploaded_image_1', 'uploaded_image_2']

Quiz

Key Takeaways

✅ Use threading for I/O-bound tasks (network, file, DB).
✅ Use multiprocessing for CPU-bound tasks (computations, image processing).
✅ The GIL prevents true thread parallelism in CPython.
✅ Always use locks / semaphores to protect shared mutable state.
✅ ThreadPoolExecutor / ProcessPoolExecutor are the modern high-level APIs.
✅ Use Queue and Pipe for safe inter-process communication.

What's Next?

In the next lesson, we'll master Async/Await — Python's most elegant concurrency model for handling thousands of I/O operations efficiently.

Keep coding! 🚀