Threading & Concurrency

Master threads, locks, the GIL, thread pools, and concurrent.futures

Threading in Python

Threading allows multiple tasks to run concurrently within a single process. While Python's Global Interpreter Lock (GIL) prevents true parallel execution of Python code, threading is excellent for I/O-bound operations like network requests, file I/O, and database queries where threads spend most of their time waiting.

When to Use Threading:

✓ I/O-bound tasks: Network calls, file operations, database queries
✓ Concurrent operations: Handling multiple connections simultaneously
✓ Responsive UIs: Keep interface responsive during background work
✗ CPU-bound tasks: Use multiprocessing instead (covered in next lesson)

Understanding the Global Interpreter Lock (GIL)

The GIL is a mutex that protects access to Python objects, preventing multiple threads from executing Python bytecode simultaneously. This is Python's most misunderstood feature.

❌ GIL Hurts CPU-Bound Tasks

Only one thread executes Python code at a time

Thread 1: [████░░░░░░]
Thread 2: [░░░░████░░]
Thread 3: [░░░░░░░░██]

Threads take turns, no speedup for pure Python computation

✓ GIL Released During I/O

GIL automatically released during blocking I/O

Thread 1: [I/O........]
Thread 2: [..I/O.....]
Thread 3: [.....I/O..]

All threads can wait for I/O simultaneously!

Key insight: The GIL is released during I/O operations (network, disk, sleep), allowing other threads to run. This makes threading perfect for I/O-bound workloads but ineffective for CPU-bound pure Python code.

Basic Threading

Creating and starting threads with the threading module.

import threading
import time

def worker():
    """Function to run in a thread"""
    print(f"Worker started in thread: {threading.current_thread().name}")
    time.sleep(2)
    print(f"Worker finished in thread: {threading.current_thread().name}")

# Create thread
thread = threading.Thread(target=worker)

# Start thread (non-blocking)
thread.start()
print("Main thread continues...")

# Wait for thread to complete
thread.join()
print("Thread completed!")

# Multiple threads
threads = []
for i in range(5):
    t = threading.Thread(target=worker, name=f"Worker-{i}")
    threads.append(t)
    t.start()

# Wait for all threads
for t in threads:
    t.join()

print("All threads completed!")

Thread Synchronization: Locks and More

When multiple threads access shared data, you need synchronization primitives to prevent race conditions and data corruption.

Lock: Basic Mutual Exclusion

import threading
import time

# Shared resource
counter = 0
lock = threading.Lock()

def increment_without_lock():
    """Race condition - unsafe!"""
    global counter
    for _ in range(100000):
        temp = counter
        temp += 1
        counter = temp
    # Final counter will be unpredictable!

def increment_with_lock():
    """Thread-safe with lock"""
    global counter
    for _ in range(100000):
        with lock:  # Acquire lock
            temp = counter
            temp += 1
            counter = temp
        # Lock automatically released

# Demonstrate race condition
counter = 0
threads = [threading.Thread(target=increment_without_lock) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Without lock: {counter}")  # Wrong! Should be 1,000,000

# With proper locking
counter = 0
threads = [threading.Thread(target=increment_with_lock) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"With lock: {counter}")  # Correct: 1,000,000

RLock: Reentrant Lock

import threading

# RLock can be acquired multiple times by same thread
rlock = threading.RLock()

def recursive_function(n):
    """Can acquire same lock multiple times"""
    with rlock:
        if n > 0:
            print(f"Level {n}")
            recursive_function(n - 1)  # Reacquires same lock
        return n

# This works with RLock
thread = threading.Thread(target=recursive_function, args=(5,))
thread.start()
thread.join()

# Would deadlock with regular Lock!
# lock = threading.Lock()
# with lock:
#     with lock:  # Deadlock! Same thread trying to acquire same lock
#         pass

Semaphore: Limit Concurrent Access

import threading
import time

# Allow maximum 3 threads to access resource simultaneously
semaphore = threading.Semaphore(3)

def access_limited_resource(worker_id):
    """Only 3 workers can be here at once"""
    print(f"Worker {worker_id} waiting...")

    with semaphore:
        print(f"Worker {worker_id} accessing resource")
        time.sleep(2)  # Simulate work
        print(f"Worker {worker_id} done")

# Start 10 workers, but only 3 can access resource at once
threads = []
for i in range(10):
    t = threading.Thread(target=access_limited_resource, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# Use case: Database connection pool
class ConnectionPool:
    def __init__(self, max_connections=5):
        self.semaphore = threading.Semaphore(max_connections)
        self.connections = []

    def acquire(self):
        self.semaphore.acquire()
        # Get or create connection
        return "connection"

    def release(self, conn):
        # Return connection to pool
        self.semaphore.release()

Event: Thread Signaling

import threading
import time

event = threading.Event()

def waiter():
    """Wait for signal"""
    print("Waiter: Waiting for event...")
    event.wait()  # Blocks until event is set
    print("Waiter: Event received! Proceeding...")

def setter():
    """Send signal after delay"""
    print("Setter: Working...")
    time.sleep(3)
    print("Setter: Setting event")
    event.set()  # Wake up all waiting threads

# Start waiter first
waiter_thread = threading.Thread(target=waiter)
waiter_thread.start()

# Start setter
setter_thread = threading.Thread(target=setter)
setter_thread.start()

waiter_thread.join()
setter_thread.join()

# Use cases:
# - Coordinate startup sequence
# - Signal completion of initialization
# - Implement barriers/checkpoints

ThreadPoolExecutor: High-Level Threading

concurrent.futures provides a high-level interface for managing thread pools, making it much easier to work with multiple threads.

Basic ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import requests

def download_url(url):
    """Download a URL"""
    response = requests.get(url)
    return len(response.content)

urls = [
    "https://example.com",
    "https://python.org",
    "https://github.com",
] * 5  # 15 URLs total

# Method 1: map() - simple and ordered
with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(download_url, urls)
    for url, size in zip(urls, results):
        print(f"{url}: {size} bytes")

# Method 2: submit() - more control
with ThreadPoolExecutor(max_workers=5) as executor:
    # Submit all tasks
    futures = [executor.submit(download_url, url) for url in urls]

    # Get results as they complete
    for future in as_completed(futures):
        try:
            size = future.result()
            print(f"Downloaded {size} bytes")
        except Exception as e:
            print(f"Error: {e}")

# Automatic cleanup - threads are joined when exiting with block

Real-World: Concurrent API Requests

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import time

def fetch_user_data(user_id):
    """Fetch user data from API"""
    try:
        response = requests.get(
            f"https://jsonplaceholder.typicode.com/users/{user_id}",
            timeout=5
        )
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"Error fetching user {user_id}: {e}")
        return None

# Sequential (slow)
start = time.time()
results = []
for user_id in range(1, 11):
    results.append(fetch_user_data(user_id))
sequential_time = time.time() - start
print(f"Sequential: {sequential_time:.2f}s")

# Concurrent with ThreadPoolExecutor (fast!)
start = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
    # Submit all requests
    future_to_id = {
        executor.submit(fetch_user_data, user_id): user_id
        for user_id in range(1, 11)
    }

    results = []
    for future in as_completed(future_to_id):
        user_id = future_to_id[future]
        try:
            data = future.result()
            if data:
                results.append(data)
                print(f"✓ User {user_id}: {data['name']}")
        except Exception as e:
            print(f"✗ User {user_id} failed: {e}")

concurrent_time = time.time() - start
print(f"Concurrent: {concurrent_time:.2f}s")
print(f"Speedup: {sequential_time / concurrent_time:.1f}x faster!")

Advanced: Timeout and Cancellation

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import time

def slow_task(n):
    time.sleep(n)
    return f"Task {n} completed"

with ThreadPoolExecutor(max_workers=3) as executor:
    # Submit tasks
    future1 = executor.submit(slow_task, 1)
    future2 = executor.submit(slow_task, 5)
    future3 = executor.submit(slow_task, 10)

    # Wait with timeout
    try:
        result = future1.result(timeout=2)
        print(result)
    except TimeoutError:
        print("Task 1 timed out")

    # Cancel task that hasn't started
    if future3.cancel():
        print("Task 3 cancelled successfully")
    else:
        print("Task 3 already running, can't cancel")

    # Check task status
    print(f"Future2 done: {future2.done()}")
    print(f"Future2 running: {future2.running()}")
    print(f"Future2 cancelled: {future2.cancelled()}")

    # Wait for result
    result = future2.result()
    print(result)

Thread-Safe Data Structures

Python provides thread-safe data structures that handle synchronization internally.

import threading
import queue
import time

# Queue: Thread-safe FIFO queue
task_queue = queue.Queue(maxsize=10)

def producer(producer_id):
    """Add items to queue"""
    for i in range(5):
        item = f"Item-{producer_id}-{i}"
        task_queue.put(item)  # Thread-safe
        print(f"Producer {producer_id} added {item}")
        time.sleep(0.1)

def consumer(consumer_id):
    """Process items from queue"""
    while True:
        try:
            item = task_queue.get(timeout=1)  # Thread-safe
            print(f"Consumer {consumer_id} processing {item}")
            time.sleep(0.2)
            task_queue.task_done()
        except queue.Empty:
            break

# Start producers and consumers
producers = [threading.Thread(target=producer, args=(i,)) for i in range(2)]
consumers = [threading.Thread(target=consumer, args=(i,)) for i in range(3)]

for p in producers:
    p.start()
for c in consumers:
    c.start()

for p in producers:
    p.join()
for c in consumers:
    c.join()

# Other queue types
priority_queue = queue.PriorityQueue()  # Items sorted by priority
lifo_queue = queue.LifoQueue()          # Stack (LIFO)

# PriorityQueue example
priority_queue.put((1, "High priority"))
priority_queue.put((3, "Low priority"))
priority_queue.put((2, "Medium priority"))

while not priority_queue.empty():
    priority, item = priority_queue.get()
    print(f"Priority {priority}: {item}")

# Thread-local storage
thread_local = threading.local()

def worker():
    # Each thread has its own copy
    thread_local.value = threading.current_thread().name
    print(f"Thread local value: {thread_local.value}")

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

Threading Best Practices

✓ Do This

Use ThreadPoolExecutor for most cases
Use threading for I/O-bound operations
Always use locks when accessing shared data
Use context managers (with) for locks
Use thread-safe Queue for communication
Set daemon=True for background threads
Handle exceptions in thread functions
Use concurrent.futures for cleaner code
Join threads to wait for completion
Use Event for signaling between threads

✗ Avoid This

Using threads for CPU-bound tasks
Accessing shared data without locks
Creating too many threads (use pool)
Forgetting to join threads
Using global variables without locks
Nested locks (can cause deadlock)
Ignoring the GIL's limitations
Manual thread management when pool works
Daemon threads for critical tasks
Expecting parallel CPU computation

Common Threading Patterns

Pattern 1: Worker Pool

from concurrent.futures import ThreadPoolExecutor
import time

def process_item(item):
    """Process a single item"""
    time.sleep(0.1)
    return item * 2

items = list(range(100))

# Process all items using thread pool
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(process_item, items))

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

Pattern 2: Producer-Consumer

import threading
import queue
import time

def producer_consumer_pattern():
    work_queue = queue.Queue()

    def producer():
        for i in range(20):
            work_queue.put(i)
            time.sleep(0.1)
        # Signal completion
        work_queue.put(None)

    def consumer():
        while True:
            item = work_queue.get()
            if item is None:
                break
            # Process item
            print(f"Processing {item}")
            time.sleep(0.2)
            work_queue.task_done()

    prod_thread = threading.Thread(target=producer)
    cons_thread = threading.Thread(target=consumer)

    prod_thread.start()
    cons_thread.start()

    prod_thread.join()
    cons_thread.join()

producer_consumer_pattern()

Pattern 3: Parallel Downloads

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def download_file(url):
    """Download and return file size"""
    try:
        response = requests.get(url, timeout=10)
        return len(response.content)
    except Exception as e:
        return 0

urls = [f"https://example.com/file{i}.jpg" for i in range(50)]

with ThreadPoolExecutor(max_workers=10) as executor:
    future_to_url = {executor.submit(download_file, url): url for url in urls}

    for future in as_completed(future_to_url):
        url = future_to_url[future]
        size = future.result()
        print(f"{url}: {size} bytes")

Pattern 4: Rate-Limited Requests

from concurrent.futures import ThreadPoolExecutor
import threading
import time

class RateLimiter:
    """Limit operations to N per second"""
    def __init__(self, max_per_second):
        self.max_per_second = max_per_second
        self.lock = threading.Lock()
        self.last_call = 0

    def wait(self):
        with self.lock:
            now = time.time()
            time_since_last = now - self.last_call
            min_interval = 1.0 / self.max_per_second

            if time_since_last < min_interval:
                sleep_time = min_interval - time_since_last
                time.sleep(sleep_time)

            self.last_call = time.time()

rate_limiter = RateLimiter(max_per_second=5)

def api_call(item):
    rate_limiter.wait()
    # Make actual API call
    print(f"Processing {item}")
    return item

items = range(20)
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(api_call, items))

# Only 5 requests per second despite 10 workers

Deadlock Prevention

Deadlocks occur when threads wait for each other indefinitely. Here's how to avoid them.

import threading
import time

lock_a = threading.Lock()
lock_b = threading.Lock()

# BAD: Can cause deadlock
def thread1_bad():
    with lock_a:
        time.sleep(0.1)
        with lock_b:  # Waiting for lock_b
            print("Thread 1")

def thread2_bad():
    with lock_b:
        time.sleep(0.1)
        with lock_a:  # Waiting for lock_a - DEADLOCK!
            print("Thread 2")

# GOOD: Always acquire locks in same order
def thread1_good():
    with lock_a:
        time.sleep(0.1)
        with lock_b:
            print("Thread 1")

def thread2_good():
    with lock_a:  # Same order as thread1
        time.sleep(0.1)
        with lock_b:
            print("Thread 2")

# BETTER: Use timeout
def thread_with_timeout():
    acquired = lock_a.acquire(timeout=1)
    if not acquired:
        print("Could not acquire lock, aborting")
        return

    try:
        # Do work
        pass
    finally:
        lock_a.release()

# BEST: Use RLock for recursive acquisition
rlock = threading.RLock()

def recursive_function(n):
    with rlock:  # Can acquire multiple times
        if n > 0:
            recursive_function(n - 1)

# Deadlock prevention strategies:
# 1. Always acquire locks in same order
# 2. Use timeout on lock acquisition
# 3. Use RLock when needed
# 4. Minimize critical sections
# 5. Use higher-level primitives (Queue, etc.)

Key Takeaways

Threading is perfect for I/O-bound tasks - network, files, databases
The GIL prevents parallel Python execution - but releases during I/O
Use ThreadPoolExecutor for simplicity - high-level, easy to use
Always synchronize shared data - use locks, queues, or thread-safe structures
Avoid deadlocks - acquire locks in consistent order
Use daemon threads for background work - they terminate with main program
Queue is your friend - thread-safe communication between threads
Don't use threading for CPU-bound tasks - use multiprocessing instead

Practice Exercises

Exercise 1: Web Scraper

Build a multi-threaded web scraper using ThreadPoolExecutor that fetches 50 web pages concurrently. Track success/failure rates, handle timeouts, and measure the speedup compared to sequential fetching. Use a semaphore to limit concurrent connections to 10.

Exercise 2: Thread-Safe Counter

Create a thread-safe counter class with increment, decrement, and get methods. Test it with 10 threads each performing 100,000 operations. Verify the final count is correct. Compare performance with and without locks.

Exercise 3: Producer-Consumer System

Implement a producer-consumer system using Queue where 3 producer threads generate random tasks and 5 consumer threads process them. Add metrics tracking (tasks produced, consumed, queue size) and implement graceful shutdown.

Additional Resources

Official Docs: docs.python.org/3/library/threading.html
concurrent.futures: docs.python.org/3/library/concurrent.futures.html
Understanding the GIL: realpython.com/python-gil
Book: "Python Concurrency with asyncio" by Matthew Fowler
Threading vs Multiprocessing: Know when to use each
David Beazley's talks: Understanding the Python GIL

What's Next?

You've learned threading! Now let's explore multiprocessing to achieve true parallelism and bypass the GIL for CPU-bound tasks.

Process Pools - Manage multiple processes for parallel execution
Shared Memory - Share data efficiently between processes
IPC - Inter-process communication with queues and pipes

Back to "Async Patterns"

Python AdvancedLesson 3

Next: "Multiprocessing & Parallelism"