Threading & Concurrency

# gil_bench.py - measure the three claims instead of assuming them
import os
import threading
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

def cpu_work(n=6_000_000):
    """Pure Python arithmetic: needs the GIL the whole way through"""
    total = 0
    for i in range(n):
        total += i
    return total

def io_work(_=None):
    """Sleeping releases the GIL, exactly like waiting on a socket"""
    time.sleep(1)

def timed(fn):
    start = time.perf_counter()
    fn()
    return time.perf_counter() - start

def cpu_sequential(workers=4):
    for _ in range(workers):
        cpu_work()

def cpu_threaded(workers=4):
    ts = [threading.Thread(target=cpu_work) for _ in range(workers)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()

def cpu_processes(workers=4):
    with ProcessPoolExecutor(max_workers=workers) as ex:
        list(ex.map(cpu_work, [6_000_000] * workers))

def io_sequential(n=5):
    for _ in range(n):
        io_work()

def io_threaded(n=5):
    with ThreadPoolExecutor(max_workers=n) as ex:
        list(ex.map(io_work, range(n)))

if __name__ == "__main__":
    print(f"cores: {os.cpu_count()}")

    seq = timed(cpu_sequential)
    thr = timed(cpu_threaded)
    proc = timed(cpu_processes)
    print("CPU-bound, 4 tasks")
    print(f"  sequential      {seq:5.2f}s")
    print(f"  4 threads       {thr:5.2f}s   ({seq / thr:.2f}x vs sequential)")
    print(f"  4 processes     {proc:5.2f}s   ({seq / proc:.2f}x vs sequential)")

    io_seq = timed(io_sequential)
    io_thr = timed(io_threaded)
    print("I/O-bound, 5 x 1s sleep")
    print(f"  sequential      {io_seq:5.2f}s")
    print(f"  5 threads       {io_thr:5.2f}s   ({io_seq / io_thr:.2f}x vs sequential)")

import threading
import time

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

def increment_without_lock():
    """Race condition - unsafe!"""
    global counter
    for _ in range(20000):
        temp = counter
        time.sleep(0)   # hands the GIL to another thread, mid-update
        temp += 1
        counter = temp
        # While this thread is paused above, another one reads the same
        # 'counter'. Both then write back the same value, so one increment
        # is silently lost.

def increment_with_lock():
    """Thread-safe with lock"""
    global counter
    for _ in range(20000):
        with lock:  # Acquire lock
            temp = counter
            time.sleep(0)   # other threads still cannot enter: lock is held
            temp += 1
            counter = temp
        # Lock automatically released

# Demonstrate race condition
counter = 0
threads = [threading.Thread(target=increment_without_lock) for _ in range(8)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Without lock: {counter}")

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

# Remove the time.sleep(0) and the unlocked version will very likely print
# 160000 too. That does not mean it is safe. Python only switches threads at
# certain points, and this short read-modify-write usually finishes before
# one arrives. Nothing guarantees it: a slower expression, an I/O call, or a
# different Python version moves the switch into the middle and increments
# start disappearing.
#
# Decide whether you need a lock by asking whether the operation is atomic.
# Reading a shared variable, changing it, and writing it back is three
# separate steps, so it is not atomic, so it needs a 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

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()

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

from concurrent.futures import ThreadPoolExecutor, as_completed
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

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!")

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")

    # Only a task still sitting in the queue can be cancelled. There are
    # three workers and three tasks, so this one is already running and
    # cancel() returns False. Submit a fourth task and that one cancels.
    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)

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()

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")

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()

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:
        return 0

# Real URLs that return different amounts of data. Pointing 50 requests at
# example.com/fileN.jpg would "work", but every response is the same 559-byte
# placeholder page, so the sizes below would tell you nothing.
urls = [
    f"https://jsonplaceholder.typicode.com/posts/{i}"
    for i in range(1, 11)
]

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")

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. Note the print order below:
# the limiter controls the rate, not which worker goes next, so the items
# come out shuffled. results is still in order, because map() puts it back.

import threading
import time

# BAD: Can cause deadlock
lock_a = threading.Lock()
lock_b = threading.Lock()

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")

# This really does deadlock. Daemon threads plus a join timeout let the
# demo report the hang rather than become it.
t1 = threading.Thread(target=thread1_bad, daemon=True)
t2 = threading.Thread(target=thread2_bad, daemon=True)
t1.start()
t2.start()
t1.join(timeout=2)
t2.join(timeout=2)
print(f"Opposite order  -> still stuck after 2s: {t1.is_alive() and t2.is_alive()}")

# GOOD: Always acquire locks in same order. Fresh locks, because the two
# above are still held by the wedged threads and are never coming back.
lock_x = threading.Lock()
lock_y = threading.Lock()

def thread1_good():
    with lock_x:
        time.sleep(0.1)
        with lock_y:
            print("Thread 1")

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

t3 = threading.Thread(target=thread1_good)
t4 = threading.Thread(target=thread2_good)
t3.start()
t4.start()
t3.join(timeout=2)
t4.join(timeout=2)
print(f"Same order      -> both finished: {not (t3.is_alive() or t4.is_alive())}")

# BETTER: Use timeout, so a thread that cannot get the lock gives up
# instead of waiting forever
lock_held = threading.Lock()
lock_held.acquire()  # pretend someone else is holding it

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

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

t5 = threading.Thread(target=thread_with_timeout)
t5.start()
t5.join()

# BEST for re-entry: RLock can be acquired again by the same thread
rlock = threading.RLock()

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

recursive_function(3)
print("RLock recursion returned without deadlocking")

# 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.)