Threading & the GIL

import threading
import time

def worker(name, delay):
    print(f"{name} starting")
    time.sleep(delay)      # simulate I/O work
    print(f"{name} done after {delay}s")

# Without threads — sequential: ~3 seconds total
worker("A", 1)
worker("B", 2)

# With threads — concurrent: ~2 seconds total
t1 = threading.Thread(target=worker, args=("A", 1))
t2 = threading.Thread(target=worker, args=("B", 2))

t1.start()
t2.start()

t1.join()   # wait for t1 to finish
t2.join()   # wait for t2 to finish
print("All done")

With the GIL — CPU-bound threads take turns (no speedup):

Thread 1: ██░░██░░██░░██     (GIL acquired → released → acquired…)
Thread 2: ░░██░░██░░██░░

Without the GIL (e.g. multiprocessing) — true parallel:

Process 1: ████████████████
Process 2: ████████████████

import threading
import time

def cpu_work(n):
    """Purely CPU-bound — count to n."""
    total = 0
    for i in range(n):
        total += i
    return total

N = 50_000_000

# Sequential — one thread
start = time.perf_counter()
cpu_work(N)
cpu_work(N)
seq_time = time.perf_counter() - start

# Two threads — NOT faster for CPU work (GIL)
start = time.perf_counter()
t1 = threading.Thread(target=cpu_work, args=(N,))
t2 = threading.Thread(target=cpu_work, args=(N,))
t1.start(); t2.start()
t1.join();  t2.join()
par_time = time.perf_counter() - start

print(f"Sequential: {seq_time:.2f}s")
print(f"Threaded:   {par_time:.2f}s  (similar or slower — GIL!)")

import threading
import time

# ── Subclassing Thread ──
class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__(daemon=True)  # daemon: dies when main thread exits
        self.url    = url
        self.result = None

    def run(self):
        """Overriding run() is the alternative to target=."""
        print(f"Downloading {self.url}")
        time.sleep(1)   # simulate network
        self.result = f"data from {self.url}"

threads = [DownloadThread(f"https://example.com/page/{i}") for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join(timeout=5)   # wait at most 5s

results = [t.result for t in threads]
print(results)

import threading

counter = 0

def increment(n):
    global counter
    for _ in range(n):
        counter += 1   # NOT atomic: read → add 1 → write (3 steps!)

threads = [threading.Thread(target=increment, args=(100_000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()

print(counter)  # Should be 1,000,000 but is often less — race condition!

import threading

counter = 0
lock    = threading.Lock()

def increment(n):
    global counter
    for _ in range(n):
        with lock:          # acquire lock → do work → release lock
            counter += 1   # now atomic

threads = [threading.Thread(target=increment, args=(100_000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()

print(counter)   # Always 1,000,000 ✓

import threading

# RLock can be acquired multiple times by the SAME thread
# (regular Lock would deadlock on second acquire from same thread)
rlock = threading.RLock()

def outer():
    with rlock:
        print("outer acquired")
        inner()   # same thread acquires again — OK with RLock

def inner():
    with rlock:   # second acquire by same thread
        print("inner acquired")

outer()
# outer acquired
# inner acquired

import threading
import time

ready = threading.Event()

def producer():
    print("Producer: preparing data…")
    time.sleep(2)
    print("Producer: data ready!")
    ready.set()   # signal the event

def consumer():
    print("Consumer: waiting for data…")
    ready.wait()  # block until event is set
    print("Consumer: processing data!")

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t2.start(); t1.start()
t1.join();  t2.join()
# Consumer: waiting for data…
# Producer: preparing data…
# Producer: data ready!
# Consumer: processing data!

import threading
import time
import random

# Allow at most 3 threads to run simultaneously (connection pool pattern)
pool = threading.Semaphore(3)

def worker(name):
    with pool:   # acquire one slot; release on exit
        print(f"{name} working")
        time.sleep(random.uniform(0.5, 1.5))
        print(f"{name} done")

threads = [threading.Thread(target=worker, args=(f"T{i}",)) for i in range(8)]
for t in threads: t.start()
for t in threads: t.join()

import threading

queue  = []
cond   = threading.Condition()
MAX    = 3

def producer():
    for i in range(10):
        with cond:
            while len(queue) >= MAX:
                cond.wait()    # release lock + wait for notify
            queue.append(i)
            print(f"Produced {i}, queue={queue}")
            cond.notify_all()

def consumer():
    received = 0
    while received < 10:
        with cond:
            while not queue:
                cond.wait()
            item = queue.pop(0)
            received += 1
            print(f"Consumed {item}, queue={queue}")
            cond.notify_all()

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()

import threading

# Each thread gets its own 'local.value' — they don't share
local = threading.local()

def worker(name, value):
    local.value = value          # set in THIS thread only
    import time; time.sleep(0.1)
    print(f"{name}: local.value = {local.value}")   # sees its own value

threads = [
    threading.Thread(target=worker, args=(f"Thread-{i}", i))
    for i in range(5)
]
for t in threads: t.start()
for t in threads: t.join()
# Thread-0: local.value = 0
# Thread-1: local.value = 1  (etc. — no mixing)

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch(url):
    """Simulate fetching a URL."""
    time.sleep(0.5)
    return f"Response from {url}"

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

# ── map: submit all, collect results in order ──
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fetch, urls))

print(results[:3])

# ── submit + as_completed: results arrive as they finish ──
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}

    for future in as_completed(futures):
        url    = futures[future]
        result = future.result()   # blocks until this future is done
        print(f"Done: {url} → {result[:30]}")

from concurrent.futures import ThreadPoolExecutor
import time

# ── Exception handling ──
def risky(n):
    if n == 3:
        raise ValueError(f"Bad value: {n}")
    return n * 2

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(risky, i) for i in range(6)]

for i, future in enumerate(futures):
    try:
        print(f"Result {i}: {future.result()}")
    except ValueError as e:
        print(f"Error  {i}: {e}")

import threading
import queue
import time

work_queue = queue.Queue(maxsize=5)  # bounded — blocks producers when full

def producer(q, items):
    for item in items:
        q.put(item)          # blocks if queue is full
        print(f"Produced: {item}")
    q.put(None)              # sentinel: signal "no more work"

def consumer(q):
    while True:
        item = q.get()       # blocks until an item is available
        if item is None:
            break
        print(f"Consumed: {item}")
        time.sleep(0.2)
        q.task_done()        # signal that this item is processed

q   = queue.Queue(maxsize=5)
t_p = threading.Thread(target=producer, args=(q, range(10)))
t_c = threading.Thread(target=consumer, args=(q,))

t_p.start(); t_c.start()
t_p.join();  t_c.join()

print("Done")

import threading

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

def thread1():
    with lock_a:
        print("T1 acquired A, waiting for B…")
        import time; time.sleep(0.1)
        with lock_b:   # waits forever — T2 holds B
            print("T1 acquired B")

def thread2():
    with lock_b:
        print("T2 acquired B, waiting for A…")
        import time; time.sleep(0.1)
        with lock_a:   # waits forever — T1 holds A
            print("T2 acquired A")

# t1 = threading.Thread(target=thread1)
# t2 = threading.Thread(target=thread2)
# t1.start(); t2.start()  # DEADLOCK!

# ── Fix: always acquire locks in the same order ──
def thread1_safe():
    with lock_a:   # always acquire A before B
        with lock_b:
            print("T1 done")

def thread2_safe():
    with lock_a:   # same order — no deadlock
        with lock_b:
            print("T2 done")

💻 Exercises

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

def fake_fetch(url):
    """Simulate an HTTP request."""
    time.sleep(random.uniform(0.2, 1.0))
    if "error" in url:
        raise ConnectionError(f"Failed to connect to {url}")
    return f"<html>Content of {url}</html>"

def download_all(urls, max_workers=4):
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(fake_fetch, url): url for url in urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                data = future.result()
                results[url] = data
                print(f"✓ {url} ({len(data)} bytes)")
            except Exception as e:
                results[url] = f"ERROR: {e}"
                print(f"✗ {url}: {e}")
    return results

urls = [
    "https://example.com/page/1",
    "https://example.com/page/2",
    "https://example.com/error/3",
    "https://example.com/page/4",
    "https://example.com/page/5",
]

results = download_all(urls)
print(f"\nCompleted {len(results)} requests")

import threading

class ThreadSafeCounter:
    def __init__(self, initial=0):
        self._value = initial
        self._lock  = threading.Lock()

    @property
    def value(self):
        with self._lock:
            return self._value

    def increment(self, amount=1):
        with self._lock:
            self._value += amount

    def decrement(self, amount=1):
        with self._lock:
            self._value -= amount

    def reset(self):
        with self._lock:
            self._value = 0

    def __repr__(self):
        return f"ThreadSafeCounter({self._value})"

# Verification
counter = ThreadSafeCounter()
THREADS = 100
INC_PER_THREAD = 1000

threads = [
    threading.Thread(target=lambda: [counter.increment() for _ in range(INC_PER_THREAD)])
    for _ in range(THREADS)
]
for t in threads: t.start()
for t in threads: t.join()

expected = THREADS * INC_PER_THREAD
print(f"Result:   {counter.value}")
print(f"Expected: {expected}")
print(f"Correct:  {counter.value == expected}")

import threading
import queue
import time

NUM_WORKERS = 3
work_q    = queue.Queue()
results_q = queue.Queue()

def producer(items):
    for item in items:
        time.sleep(0.05)
        work_q.put(item)
    # Send one sentinel per worker
    for _ in range(NUM_WORKERS):
        work_q.put(None)
    print("Producer done")

def worker(worker_id):
    while True:
        item = work_q.get()
        if item is None:
            results_q.put(None)   # forward sentinel to collector
            break
        result = item ** 2
        results_q.put((item, result))
        work_q.task_done()
        print(f"Worker-{worker_id}: {item}² = {result}")

def collector(num_workers):
    finished = 0
    results  = {}
    while finished < num_workers:
        item = results_q.get()
        if item is None:
            finished += 1
        else:
            n, sq = item
            results[n] = sq
    print(f"\nCollector: gathered {len(results)} results")
    print("Squares:", dict(sorted(results.items())))

# Launch all threads
threads = []
threads.append(threading.Thread(target=producer, args=(range(1, 21),)))
for i in range(NUM_WORKERS):
    threads.append(threading.Thread(target=worker, args=(i,)))
threads.append(threading.Thread(target=collector, args=(NUM_WORKERS,)))

for t in threads: t.start()
for t in threads: t.join()
print("Pipeline complete")