Threading & the GIL

0/4 in this phase0/54 across the roadmap

📖 Concept

Python's threading module provides OS-level threads for concurrent execution, but understanding the Global Interpreter Lock (GIL) is essential before writing any threaded Python code.

The GIL explained: The GIL is a mutex in CPython that allows only one thread to execute Python bytecode at a time. It exists because CPython's memory management (reference counting) is not thread-safe. The GIL is released during I/O operations (file reads, network calls, time.sleep), which is why threading is still effective for I/O-bound workloads. For CPU-bound tasks, threads in CPython cannot achieve true parallelism — use multiprocessing or concurrent.futures.ProcessPoolExecutor instead.

Scenario	Threading effective?	Why
HTTP requests	Yes	GIL released during socket I/O
File I/O	Yes	GIL released during OS read/write
CPU computation	No	GIL prevents parallel bytecode execution
C extensions (NumPy)	Yes	Well-written C extensions release the GIL

Key synchronization primitives:

Lock — Mutual exclusion. Only one thread can acquire() at a time. Always use with lock: to guarantee release.
RLock (Reentrant Lock) — Same thread can acquire() multiple times without deadlocking. Must release() the same number of times.
Semaphore — Allows up to N threads to enter a section concurrently. Useful for rate-limiting or connection pooling.
Event — One thread signals, others wait. set() / clear() / wait().
Condition — Threads wait for a condition to become true. Supports notify() / notify_all() / wait(). Used in producer-consumer patterns.
Barrier — N threads block until all N arrive, then all proceed together.

Thread safety rule: Any mutable shared state accessed by multiple threads must be protected by a lock. Even simple operations like counter += 1 are not atomic in Python — they compile to LOAD, ADD, STORE bytecodes, and a context switch can happen between them.

💻 Code Example

codeTap to expand ⛶

1# ============================================================
2# Basic threading with Lock for shared state
3# ============================================================
4import threading
5import time
6import logging
7from typing import Optional
8
9logging.basicConfig(level=logging.INFO, format="%(threadName)s: %(message)s")
10logger = logging.getLogger(__name__)
11
12
13class ThreadSafeCounter:
14    """A counter safe for concurrent access from multiple threads."""
15
16    def __init__(self) -> None:
17        self._value = 0
18        self._lock = threading.Lock()
19
20    def increment(self, amount: int = 1) -> None:
21        with self._lock:  # Acquire and release automatically
22            self._value += amount
23
24    def decrement(self, amount: int = 1) -> None:
25        with self._lock:
26            self._value -= amount
27
28    @property
29    def value(self) -> int:
30        with self._lock:
31            return self._value
32
33
34counter = ThreadSafeCounter()
35
36
37def worker(n: int) -> None:
38    """Each worker increments the counter n times."""
39    for _ in range(n):
40        counter.increment()
41
42
43threads = [threading.Thread(target=worker, args=(100_000,)) for _ in range(10)]
44
45start = time.perf_counter()
46for t in threads:
47    t.start()
48for t in threads:
49    t.join()
50elapsed = time.perf_counter() - start
51
52print(f"Counter value: {counter.value}")  # Always 1_000_000
53print(f"Elapsed: {elapsed:.3f}s")
54
55
56# ============================================================
57# RLock for reentrant (nested) locking
58# ============================================================
59class CachedRepository:
60    """Repository that uses RLock so public methods can call each other."""
61
62    def __init__(self) -> None:
63        self._lock = threading.RLock()
64        self._cache: dict[str, str] = {}
65
66    def get(self, key: str) -> Optional[str]:
67        with self._lock:
68            return self._cache.get(key)
69
70    def set(self, key: str, value: str) -> None:
71        with self._lock:
72            self._cache[key] = value
73
74    def get_or_set(self, key: str, default: str) -> str:
75        """Calls get() and set() internally -- RLock prevents deadlock."""
76        with self._lock:
77            existing = self.get(key)   # Acquires _lock again (RLock OK)
78            if existing is None:
79                self.set(key, default)  # Acquires _lock again (RLock OK)
80                return default
81            return existing
82
83
84# ============================================================
85# Semaphore for connection pool / rate limiting
86# ============================================================
87import random
88
89MAX_CONCURRENT_CONNECTIONS = 3
90pool_semaphore = threading.Semaphore(MAX_CONCURRENT_CONNECTIONS)
91
92
93def fetch_url(url: str) -> str:
94    """Simulate fetching a URL with limited concurrency."""
95    with pool_semaphore:  # At most 3 threads here concurrently
96        logger.info(f"Fetching {url}")
97        time.sleep(random.uniform(0.1, 0.5))  # Simulate network I/O
98        logger.info(f"Done fetching {url}")
99        return f"Response from {url}"
100
101
102urls = [f"https://api.example.com/item/{i}" for i in range(10)]
103threads = [threading.Thread(target=fetch_url, args=(url,)) for url in urls]
104for t in threads:
105    t.start()
106for t in threads:
107    t.join()
108
109
110# ============================================================
111# Event for signaling between threads
112# ============================================================
113data_ready = threading.Event()
114shared_data: dict = {}
115
116
117def producer() -> None:
118    """Produce data and signal consumers."""
119    logger.info("Producing data...")
120    time.sleep(1)  # Simulate work
121    shared_data["result"] = [1, 2, 3, 4, 5]
122    data_ready.set()  # Signal consumers
123    logger.info("Data is ready")
124
125
126def consumer(name: str) -> None:
127    """Wait for data, then consume it."""
128    logger.info(f"{name} waiting for data...")
129    data_ready.wait()  # Blocks until set()
130    logger.info(f"{name} got data: {shared_data['result']}")
131
132
133prod = threading.Thread(target=producer)
134cons1 = threading.Thread(target=consumer, args=("Consumer-1",))
135cons2 = threading.Thread(target=consumer, args=("Consumer-2",))
136
137for t in [cons1, cons2, prod]:
138    t.start()
139for t in [prod, cons1, cons2]:
140    t.join()
141
142
143# ============================================================
144# Condition variable for producer-consumer queue
145# ============================================================
146class BoundedBuffer:
147    """Thread-safe bounded buffer using Condition variables."""
148
149    def __init__(self, capacity: int = 10) -> None:
150        self._buffer: list = []
151        self._capacity = capacity
152        self._condition = threading.Condition()
153
154    def put(self, item) -> None:
155        with self._condition:
156            while len(self._buffer) >= self._capacity:
157                self._condition.wait()  # Wait until space available
158            self._buffer.append(item)
159            self._condition.notify()  # Notify waiting consumers
160
161    def get(self):
162        with self._condition:
163            while len(self._buffer) == 0:
164                self._condition.wait()  # Wait until item available
165            item = self._buffer.pop(0)
166            self._condition.notify()  # Notify waiting producers
167            return item
168
169
170buffer = BoundedBuffer(capacity=5)
171
172
173def buffer_producer(n: int) -> None:
174    for i in range(n):
175        buffer.put(i)
176        logger.info(f"Produced {i}")
177        time.sleep(0.05)
178
179
180def buffer_consumer(n: int) -> None:
181    for _ in range(n):
182        item = buffer.get()
183        logger.info(f"Consumed {item}")
184        time.sleep(0.1)
185
186
187p = threading.Thread(target=buffer_producer, args=(20,))
188c = threading.Thread(target=buffer_consumer, args=(20,))
189p.start()
190c.start()
191p.join()
192c.join()
193
194
195# ============================================================
196# Daemon threads and graceful shutdown
197# ============================================================
198shutdown_event = threading.Event()
199
200
201def background_monitor(interval: float = 2.0) -> None:
202    """Background daemon that runs until shutdown is signaled."""
203    while not shutdown_event.is_set():
204        logger.info("Monitor heartbeat")
205        shutdown_event.wait(timeout=interval)  # Sleep but wake on shutdown
206    logger.info("Monitor shutting down")
207
208
209monitor = threading.Thread(target=background_monitor, daemon=True)
210monitor.start()
211
212# ... do work ...
213
214shutdown_event.set()  # Signal graceful shutdown
215monitor.join(timeout=5)

🏋️ Practice Exercise

Exercises:

Write a program that spawns 5 threads, each incrementing a shared counter 1,000,000 times. First run it without a lock and observe the race condition. Then add a Lock and confirm the final value is always 5,000,000.
Implement a thread-safe LRUCache class using threading.Lock that supports get(key) and put(key, value) with a configurable max size. Write a stress test with 10 threads doing random reads/writes.
Build a producer-consumer pipeline using threading.Condition: 3 producer threads generate random numbers, 2 consumer threads compute their squares. Use a bounded buffer of size 10. Print the throughput (items/sec) at the end.
Create a ConnectionPool class using Semaphore(max_size). Threads call pool.acquire() to get a connection and pool.release(conn) to return it. Add a timeout parameter that raises TimeoutError if no connection is available within the limit.
Write a benchmark that compares threading vs sequential execution for (a) downloading 20 web pages (I/O-bound) and (b) computing SHA-256 hashes of 20 large strings (CPU-bound). Measure and explain the results in terms of the GIL.
Implement a ReadWriteLock from scratch that allows unlimited concurrent readers but exclusive writer access. Test it with 10 reader threads and 2 writer threads accessing a shared dictionary.

⚠️ Common Mistakes

Using threading for CPU-bound work and expecting a speedup. The GIL prevents parallel bytecode execution in CPython, so CPU-bound threads actually run slower than sequential code due to lock contention and context-switch overhead. Use multiprocessing for CPU parallelism.
Forgetting to call thread.join() and letting the main thread exit. If non-daemon threads are still running, the process hangs. If daemon threads are running, they are killed abruptly without cleanup. Always join() threads you care about.
Assuming simple operations like list.append() or dict[key] = value are thread-safe because they are 'atomic.' While CPython's GIL makes some bytecode operations accidentally safe, this is an implementation detail, not a language guarantee. Always use explicit locks for shared mutable state.
Acquiring multiple locks in inconsistent order across threads, causing deadlocks. Thread A holds Lock1 and waits for Lock2, while Thread B holds Lock2 and waits for Lock1. Always acquire locks in a globally consistent order, or use threading.RLock for self-reentrant cases.
Not using with lock: context-manager syntax and forgetting to release the lock in an exception path. Manual lock.acquire() / lock.release() without try/finally will leak locks if the critical section raises.

💼 Interview Questions

🎤 Mock Interview

Practice a live interview for Threading & the GIL

Was this topic helpful?

← Previouscollections Module Next →Multiprocessing