The GIL is a mutex in CPython that allows only ONE thread to execute Python bytecode at a time, even on multi-core machines. It exists to make CPython's memory management simpler and thread-safe.

import threading
import time

def cpu_bound_task():
    count = 0
    for i in range(50_000_000):
        count += 1
    return count

# Running 2 threads for a CPU-bound task
start = time.time()
t1 = threading.Thread(target=cpu_bound_task)
t2 = threading.Thread(target=cpu_bound_task)
t1.start(); t2.start()
t1.join(); t2.join()
print(f'Threaded: {time.time() - start:.2f}s')
# Due to the GIL, this does NOT run twice as fast on a multi-core CPU -
# threads take turns holding the GIL, so CPU-bound work barely benefits

# Single-threaded equivalent runs at similar or even better speed
start = time.time()
cpu_bound_task()
cpu_bound_task()
print(f'Sequential: {time.time() - start:.2f}s')
# Often comparable to the threaded version, because of GIL contention overhead

import threading
import multiprocessing
import time

def cpu_task(n):
    return sum(i * i for i in range(n))

# Threading - limited by GIL for CPU-bound work
start = time.time()
threads = [threading.Thread(target=cpu_task, args=(10_000_000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f'Threading: {time.time() - start:.2f}s')

# Multiprocessing - true parallelism across CPU cores
if __name__ == '__main__':
    start = time.time()
    processes = [multiprocessing.Process(target=cpu_task, args=(10_000_000,)) for _ in range(4)]
    for p in processes: p.start()
    for p in processes: p.join()
    print(f'Multiprocessing: {time.time() - start:.2f}s')
    # Noticeably FASTER on a multi-core machine because each process
    # has its own Python interpreter and GIL - no contention between them

# Sharing data between processes requires special mechanisms (unlike threads)
from multiprocessing import Queue
q = Queue()
q.put('hello')
print(q.get())  # 'hello' - processes don't share memory, so need explicit IPC

asyncio provides concurrency within a SINGLE thread using cooperative multitasking - coroutines voluntarily yield control at await points, rather than being preemptively switched like OS threads.

import asyncio

async def fetch_data(name, delay):
    print(f'{name}: starting')
    await asyncio.sleep(delay)  # yields control here, doesn't block the thread
    print(f'{name}: done')
    return f'{name} result'

async def main():
    # Running concurrently - all three start almost immediately
    results = await asyncio.gather(
        fetch_data('Task A', 2),
        fetch_data('Task B', 1),
        fetch_data('Task C', 3)
    )
    print(results)

asyncio.run(main())
# Task A: starting, Task B: starting, Task C: starting (all immediately)
# Task B: done (after 1s), Task A: done (after 2s), Task C: done (after 3s)
# Total time ~3s, NOT 6s (2+1+3) - they ran concurrently, not sequentially

# CRITICAL: asyncio only helps if the I/O operation is 'awaitable'
# (async-aware). Blocking calls like time.sleep() or requests.get()
# inside an async function STILL block the entire event loop
async def bad_example():
    import time
    time.sleep(2)  # BLOCKS the whole event loop, defeats the purpose of asyncio

import asyncio

# Defining a coroutine function
async def greet(name):
    print(f'Hello, {name}')
    await asyncio.sleep(1)  # simulates async work (I/O wait)
    print(f'Goodbye, {name}')
    return f'{name} greeted'

# Calling a coroutine function WITHOUT await just creates a coroutine object,
# it does NOT run the code yet
coro = greet('John')
print(coro)  # <coroutine object greet at 0x...>
# Need to actually run it via asyncio.run() or await inside another coroutine

# Running the top-level coroutine
asyncio.run(greet('John'))

# Running multiple coroutines CONCURRENTLY with gather
async def main():
    results = await asyncio.gather(
        greet('Alice'),
        greet('Bob')
    )
    print(results)  # ['Alice greeted', 'Bob greeted']

asyncio.run(main())

# Creating tasks explicitly for finer control (starts running immediately)
async def main2():
    task1 = asyncio.create_task(greet('Task1'))
    task2 = asyncio.create_task(greet('Task2'))
    # both tasks are now scheduled and running concurrently
    await task1
    await task2

asyncio.run(main2())

# Common mistake: forgetting await
async def broken():
    greet('Oops')  # missing 'await' - coroutine created but never runs!
    # RuntimeWarning: coroutine 'greet' was never awaited

A race condition occurs when multiple threads access and modify shared data simultaneously, producing unpredictable results. Locks (threading.Lock) ensure only one thread can access a critical section at a time.

import threading

counter = 0

def increment_unsafe():
    global counter
    for _ in range(100000):
        counter += 1  # NOT atomic - read, add, write can be interrupted mid-operation

threads = [threading.Thread(target=increment_unsafe) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)  # often LESS than 400000 due to race conditions - unpredictable!

# FIX using a Lock
counter_safe = 0
lock = threading.Lock()

def increment_safe():
    global counter_safe
    for _ in range(100000):
        with lock:  # only one thread can be inside this block at a time
            counter_safe += 1

threads = [threading.Thread(target=increment_safe) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter_safe)  # always exactly 400000 - correct and predictable

# Other synchronization primitives:
# - RLock: reentrant lock, same thread can acquire it multiple times
# - Semaphore: allows a limited NUMBER of threads through at once
# - Event: threads wait until a flag is set
# - Condition: threads wait for a specific condition to become true

semaphore = threading.Semaphore(3)  # max 3 threads access a resource concurrently
def limited_access():
    with semaphore:
        print('Accessing limited resource')

# CPU-bound: multiprocessing wins
from multiprocessing import Pool

def square(n):
    return n * n

if __name__ == '__main__':
    with Pool(4) as pool:  # 4 worker processes
        results = pool.map(square, range(1000000))
    # Genuinely parallel across 4 CPU cores

# I/O-bound with MANY connections: asyncio wins
import asyncio
import aiohttp  # async-aware HTTP library

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        return await asyncio.gather(*[fetch(session, url) for url in urls])

# Can handle THOUSANDS of concurrent requests with minimal memory overhead,
# compared to thousands of OS threads which would be very resource-heavy

# Rule of thumb:
# - CPU-bound -> multiprocessing
# - I/O-bound, need scale (1000s of connections) -> asyncio
# - I/O-bound, simpler code, moderate scale, blocking libraries -> threading

A deadlock occurs when two or more threads are each waiting for a resource held by the other, so none can proceed - the program freezes indefinitely.

import threading
import time

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

def task1():
    with lock_a:
        print('Task1 acquired lock_a')
        time.sleep(0.1)  # gives task2 time to acquire lock_b
        print('Task1 waiting for lock_b')
        with lock_b:  # blocks here if task2 already holds lock_b
            print('Task1 acquired lock_b')

def task2():
    with lock_b:
        print('Task2 acquired lock_b')
        time.sleep(0.1)
        print('Task2 waiting for lock_a')
        with lock_a:  # blocks here - DEADLOCK: each waits on what the other holds
            print('Task2 acquired lock_a')

# t1 = threading.Thread(target=task1)
# t2 = threading.Thread(target=task2)
# t1.start(); t2.start()
# Program hangs forever - classic deadlock

# FIX 1: always acquire locks in the SAME consistent order across all threads
def task1_fixed():
    with lock_a:
        with lock_b:
            print('Task1 done')

def task2_fixed():
    with lock_a:  # same order as task1_fixed - lock_a first, then lock_b
        with lock_b:
            print('Task2 done')

# FIX 2: use a timeout when acquiring locks to avoid indefinite blocking
acquired = lock_a.acquire(timeout=2)
if acquired:
    try:
        pass  # do work
    finally:
        lock_a.release()
else:
    print('Could not acquire lock in time, avoiding deadlock')

concurrent.futures provides a high-level interface (ThreadPoolExecutor and ProcessPoolExecutor) for running tasks concurrently, abstracting away manual thread/process creation and management.

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

def fetch_url(url):
    time.sleep(1)  # simulating network delay
    return f'Data from {url}'

urls = ['url1', 'url2', 'url3', 'url4']

# ThreadPoolExecutor - good for I/O-bound tasks
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fetch_url, urls))
print(results)  # all 4 urls fetched concurrently, ~1s total instead of 4s

# Using submit() for more control, and as_completed() for results as they finish
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch_url, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
            print(f'{url}: {result}')
        except Exception as e:
            print(f'{url} failed: {e}')

# ProcessPoolExecutor - same API, but for CPU-bound tasks (true parallelism)
def cpu_intensive(n):
    return sum(i * i for i in range(n))

if __name__ == '__main__':
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_intensive, [10_000_000] * 4))

# Advantage over raw threading/multiprocessing: consistent, simple API
# (.submit, .map, .result) that works the same way for both thread
# and process pools - just swap the executor class

Exceptions raised inside a thread's target function don't automatically propagate to the main thread - they're silently printed but don't crash the calling code. Async tasks behave differently and require explicit handling too.

import threading

def risky_task():
    raise ValueError('Something went wrong')

t = threading.Thread(target=risky_task)
t.start()
t.join()
print('Main thread continues normally')
# The exception is printed to stderr by the threading module,
# but the main thread is NOT interrupted or notified automatically

# FIX: capture the exception yourself and store/communicate it
class ThreadWithException(threading.Thread):
    def __init__(self, target):
        super().__init__()
        self.target = target
        self.exception = None

    def run(self):
        try:
            self.target()
        except Exception as e:
            self.exception = e  # store it for the main thread to check

t = ThreadWithException(risky_task)
t.start()
t.join()
if t.exception:
    print(f'Thread raised: {t.exception}')

# Using concurrent.futures - exceptions ARE captured and re-raised via .result()
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
    future = executor.submit(risky_task)
    try:
        future.result()  # re-raises the original exception HERE
    except ValueError as e:
        print(f'Caught: {e}')

# Asyncio - exceptions in a task propagate when you await it
import asyncio
async def risky_async():
    raise ValueError('Async error')

async def main():
    try:
        await risky_async()
    except ValueError as e:
        print(f'Caught async: {e}')
asyncio.run(main())