Threading & Thread Pools

What is Threading?

Threading allows concurrent execution of code. Useful for I/O-bound tasks (file operations, network calls).

import threading
import time

# Function to run in thread
def download(url):
    print(f'Downloading {url}...')
    time.sleep(2)  # Simulate I/O
    print(f'Downloaded {url}')

# Create threads
thread1 = threading.Thread(target=download, args=('url1',))
thread2 = threading.Thread(target=download, args=('url2',))

# Start threads
thread1.start()
thread2.start()

# Wait for completion
thread1.join()
thread2.join()

print('All downloads complete')

Thread Class

import threading
import time

class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None
    
    def run(self):
        # Code to run in thread
        print(f'Downloading {self.url}...')
        time.sleep(2)
        self.result = f'Data from {self.url}'
        print(f'Downloaded {self.url}')

# Create and start
thread = DownloadThread('https://example.com')
thread.start()
thread.join()  # Wait for completion

print(thread.result)  # Data from https://example.com

The GIL (Global Interpreter Lock)

# GIL allows only ONE thread to execute Python bytecode at a time
# This means:
# - Threading is good for I/O-bound tasks (releases GIL during I/O)
# - Threading is BAD for CPU-bound tasks (GIL prevents true parallelism)

# For CPU-bound: use multiprocessing
# For I/O-bound: use threading or asyncio

Daemon Threads

import threading
import time

def background_task():
    while True:
        print('Background task running...')
        time.sleep(1)

# Daemon thread dies when main thread exits
thread = threading.Thread(target=background_task, daemon=True)
thread.start()

time.sleep(3)
print('Main thread exiting')
# Daemon thread is killed automatically

Threading Benefits and Limitations

Race Conditions

# ❌ Race condition
counter = 0

def increment():
    global counter
    for _ in range(100000):
        counter += 1  # NOT atomic!

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

print(counter)  # Expected: 500000, Actual: varies!

Lock

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:  # Acquire lock
            counter += 1
        # Lock released automatically

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

print(counter)  # Always 500000

RLock (Reentrant Lock)

import threading

rlock = threading.RLock()

def nested_function():
    with rlock:
        print('Nested lock acquired')
        # Can acquire same lock again (unlike Lock)
        with rlock:
            print('Deeply nested')

# Lock would deadlock here!
# rlock = threading.Lock()
# rlock.acquire()
# rlock.acquire()  # DEADLOCK!

Semaphore

import threading
import time

# Limit concurrent access
semaphore = threading.Semaphore(3)  # Max 3 concurrent

def access_resource(name):
    with semaphore:
        print(f'{name} accessing resource')
        time.sleep(1)
        print(f'{name} done')

threads = [threading.Thread(target=access_resource, args=(f'Thread-{i}',)) 
           for i in range(10)]
for t in threads:
    t.start()

Condition

import threading
import time

condition = threading.Condition()
items = []

def producer():
    with condition:
        for i in range(5):
            items.append(i)
            print(f'Produced {i}')
            condition.notify()  # Wake up consumer
            time.sleep(0.1)

def consumer():
    with condition:
        while not items:
            condition.wait()  # Wait for notification
        item = items.pop(0)
        print(f'Consumed {item}')

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

consumer_thread.start()
producer_thread.start()

producer_thread.join()
consumer_thread.join()

Event

import threading
import time

event = threading.Event()

def waiter():
    print('Waiting for event...')
    event.wait()  # Blocks until event is set
    print('Event received!')

def setter():
    time.sleep(2)
    print('Setting event...')
    event.set()  # Unblock all waiters

threading.Thread(target=waiter).start()
threading.Thread(target=setter).start()

ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor
import time

def download(url):
    print(f'Downloading {url}...')
    time.sleep(2)
    return f'Data from {url}'

# Create thread pool
urls = ['url1', 'url2', 'url3', 'url4', 'url5']

with ThreadPoolExecutor(max_workers=3) as executor:
    # Submit tasks
    futures = [executor.submit(download, url) for url in urls]
    
    # Get results
    for future in futures:
        result = future.result()
        print(result)

# Or using map (simpler)
with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(download, urls)
    for result in results:
        print(result)

Callbacks

from concurrent.futures import ThreadPoolExecutor

def process_result(future):
    print(f'Result: {future.result()}')

def task(n):
    return n * n

with ThreadPoolExecutor() as executor:
    future = executor.submit(task, 10)
    future.add_done_callback(process_result)

Error Handling

from concurrent.futures import ThreadPoolExecutor, as_completed

def risky_task(n):
    if n == 3:
        raise ValueError(f'Bad value: {n}')
    return n * n

with ThreadPoolExecutor() 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: {e}')

Best Practices

# ✅ Use ThreadPoolExecutor for I/O-bound tasks
# ✅ Use ProcessPoolExecutor for CPU-bound tasks
# ✅ Limit workers to number of cores (CPU) or 10-20x (I/O)
# ✅ Handle exceptions in futures
# ✅ Use context managers for cleanup

# ❌ Don't use threads for CPU-bound (use multiprocessing)
# ❌ Don't create too many threads (overhead)
# ❌ Don't forget to join threads
# ❌ Don't use global variables without locks