← Python

Concurrency (threading, asyncio, multiprocessing)

← Concurrency (threading, asyncio, multiprocessing)

Select a question from the sidebar

18 questions in this topic

Question 1 / 18

What is the GIL (Global Interpreter Lock) and how does it impact Python concurrency?

The Global Interpreter Lock (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 protect CPython's memory management (reference counting) from race conditions.

Impact:

CPU-bound tasks: Threads do NOT run in parallel. Multiple threads actually slow things down due to context-switching overhead.
I/O-bound tasks: Threads work well because the GIL is released during I/O operations (network calls, file reads, sleep).
C extensions: Libraries like NumPy release the GIL during heavy computation, enabling true parallelism.

Workarounds for CPU-bound parallelism:

multiprocessing -- separate processes, each with its own GIL
concurrent.futures.ProcessPoolExecutor
C extensions that release the GIL (NumPy, Cython)
subinterpreters (Python 3.12+)
Free-threaded Python (Python 3.13+ experimental, no GIL)

Note: The GIL is a CPython implementation detail. Jython and IronPython do not have a GIL.

Pseudo Code

Thread 1: acquire GIL -> execute bytecode -> release GIL
Thread 2: waiting... -> acquire GIL -> execute bytecode -> release GIL

I/O-bound: Thread releases GIL during I/O wait
CPU-bound: Threads take turns but never truly parallel

Code

import threading
import time

# Demonstrating GIL impact on CPU-bound work
def cpu_bound(n):
    count = 0
    for i in range(n):
        count += i
    return count

# Sequential
start = time.perf_counter()
cpu_bound(10_000_000)
cpu_bound(10_000_000)
seq_time = time.perf_counter() - start
print(f"Sequential: {seq_time:.3f}s")

# Threaded (NOT faster due to GIL)
start = time.perf_counter()
t1 = threading.Thread(target=cpu_bound, args=(10_000_000,))
t2 = threading.Thread(target=cpu_bound, args=(10_000_000,))
t1.start(); t2.start()
t1.join(); t2.join()
thread_time = time.perf_counter() - start
print(f"Threaded: {thread_time:.3f}s")  # Similar or slower than sequential!

# I/O-bound: threads DO help
import urllib.request

def fetch_url(url):
    urllib.request.urlopen(url)

# GIL is released during network I/O, so threads overlap

Sequential: 0.347s
Threaded: 0.350s

Key Points

•GIL allows only one thread to execute Python bytecode at a time
•Threading is effective for I/O-bound tasks (GIL is released during I/O)
•Threading is NOT effective for CPU-bound tasks -- use multiprocessing instead
•The GIL is a CPython-specific implementation detail

Gotchas

•Even though threading doesn't provide CPU parallelism, it can make CPU-bound code SLOWER due to lock contention
•GIL is released during I/O, C extension calls, and time.sleep -- so threads still have uses
•Python 3.13 introduces experimental free-threaded mode (no GIL) via --disable-gil build
•Some libraries (NumPy, PIL) release the GIL in C code, enabling true thread-level parallelism

The GIL is a CPython mutex allowing only one thread to run Python bytecode at a time. It makes threading useless for CPU parallelism but fine for I/O-bound work (GIL releases during I/O). Use multiprocessing for CPU-bound parallelism.

•GIL blocks CPU parallelism in threads; I/O-bound threads are fine
•Use multiprocessing for CPU-bound work

# CPU-bound: use multiprocessing, not threading
from multiprocessing import Pool

def square(n):
    return n * n

with Pool(4) as pool:
    results = pool.map(square, range(10))
    print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

Threading can be SLOWER than sequential for CPU-bound tasks due to GIL contention

Question 2 / 18

How do you create and manage threads using the threading module?

The threading module provides a high-level interface for creating and managing threads. There are two main approaches:

Passing a target function to Thread(target=func, args=(...))
Subclassing Thread and overriding the run() method

Key methods:

start() -- begins thread execution (calls run() in a new thread)
join(timeout=None) -- blocks calling thread until this thread finishes
is_alive() -- returns True if thread is running
daemon -- if True, thread is killed when main program exits

Important: `start()` vs `run()`

start() creates a new OS thread and calls run() in it
run() directly executes in the current thread (no parallelism)

Pseudo Code

Method 1: target function
  t = Thread(target=func, args=(arg1, arg2))
  t.start()   # new thread runs func(arg1, arg2)
  t.join()    # wait for completion

Method 2: subclass
  class MyThread(Thread):
      def run(self):
          # thread work here
  t = MyThread()
  t.start()

Code

import threading
import time

# Method 1: target function
def worker(name, delay):
    print(f"[{name}] Starting")
    time.sleep(delay)
    print(f"[{name}] Done after {delay}s")
    return f"Result from {name}"

# Create and start threads
t1 = threading.Thread(target=worker, args=("Thread-1", 0.5))
t2 = threading.Thread(target=worker, args=("Thread-2", 0.3), daemon=True)

t1.start()
t2.start()

print(f"t1 alive: {t1.is_alive()}")  # True (still running)
t1.join()  # wait for t1 to finish
t2.join()  # wait for t2 to finish
print(f"t1 alive: {t1.is_alive()}")  # False

# Method 2: subclass Thread
class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None

    def run(self):
        # Simulate download
        time.sleep(0.1)
        self.result = f"Data from {self.url}"

threads = [DownloadThread(f"https://example.com/{i}") for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

for t in threads:
    print(t.result)

# Getting results with a shared list (thread-safe with GIL for simple appends)
results = []
def compute(n):
    results.append(n * n)

threads = [threading.Thread(target=compute, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(sorted(results))  # [0, 1, 4, 9, 16]

# Current thread info
print(f"Active threads: {threading.active_count()}")
print(f"Current thread: {threading.current_thread().name}")

[Thread-1] Starting
[Thread-1] Done after 0.5s
[Thread-2] Starting
[Thread-2] Done after 0.3s
t1 alive: False
t1 alive: False
Data from https://example.com/0
Data from https://example.com/1
Data from https://example.com/2
[0, 1, 4, 9, 16]
Active threads: 1
Current thread: MainThread

Key Points

•Two approaches: pass target function or subclass Thread
•start() creates a new thread; join() waits for it to finish
•Daemon threads are killed when the main program exits
•Thread.run() should not be called directly -- use start()

Gotchas

•Calling run() instead of start() executes synchronously in the current thread -- no parallelism
•Thread target functions cannot directly return values -- use shared state, queues, or futures
•Daemon threads may be killed mid-operation -- do not use for tasks that must complete (file writes)
•join() with no timeout can deadlock if the thread never finishes

Create threads with Thread(target=func, args=(...)) and call .start(). Use .join() to wait for completion. Daemon threads die when the main program exits.

•Thread(target=func, args=()) then start() and join()
•Daemon threads are killed when main exits

import threading

def task(name):
    print(f"{name} running in {threading.current_thread().name}")

t = threading.Thread(target=task, args=("Job-1",))
t.start()
t.join()  # wait for completion

Job-1 running in Thread-1 (task)

Calling run() instead of start() runs synchronously -- no new thread

Question 3 / 18

What are locks and RLocks in threading, and when should you use each?

`threading.Lock` (Mutex)

A basic mutual exclusion lock. Only one thread can hold the lock at a time. If a thread tries to acquire a lock it already holds, it deadlocks.

`threading.RLock` (Reentrant Lock)

A lock that can be acquired multiple times by the same thread without deadlocking. It must be released the same number of times it was acquired. Use RLock when a function holding a lock calls another function that also needs the same lock.

When to use each:

Lock: Simple mutual exclusion, no nested locking needed
RLock: When locked code may call other locked code (recursive locking), or when using with decorators that might nest

Pseudo Code

Lock:
  thread.acquire() -> OK
  thread.acquire() -> DEADLOCK!

RLock:
  thread.acquire() -> OK (count=1)
  thread.acquire() -> OK (count=2)
  thread.release() -> (count=1)
  thread.release() -> (count=0, unlocked)

Code

import threading

# Lock: basic mutual exclusion
balance = 0
lock = threading.Lock()

def deposit(amount):
    global balance
    lock.acquire()
    try:
        current = balance
        # Simulate slow operation
        balance = current + amount
    finally:
        lock.release()

# Better: use context manager
def withdraw(amount):
    global balance
    with lock:  # auto acquire/release
        if balance >= amount:
            balance -= amount
            return True
        return False

# RLock: reentrant locking
rlock = threading.RLock()
shared_data = []

def add_item(item):
    with rlock:
        shared_data.append(item)
        log_action(f"Added {item}")  # calls another locked function

def log_action(message):
    with rlock:  # same thread can re-acquire
        shared_data.append(f"LOG: {message}")

# With a regular Lock, log_action would deadlock!

# Demonstration: Lock deadlock vs RLock success
def test_reentrant():
    lock = threading.Lock()
    rlock = threading.RLock()

    # This would deadlock with Lock:
    # lock.acquire()
    # lock.acquire()  # DEADLOCK!

    # This works with RLock:
    rlock.acquire()
    rlock.acquire()  # OK, count=2
    rlock.release()
    rlock.release()  # count=0, fully released
    print("RLock: no deadlock")

test_reentrant()

# Lock with timeout
lock = threading.Lock()
acquired = lock.acquire(timeout=2.0)  # wait up to 2 seconds
if acquired:
    try:
        print("Got the lock")
    finally:
        lock.release()
else:
    print("Could not acquire lock in time")

RLock: no deadlock
Got the lock

Key Points

•Lock provides basic mutual exclusion; deadlocks if same thread acquires twice
•RLock allows the same thread to acquire multiple times (reentrant)
•Always use try/finally or 'with' statement to ensure locks are released
•Lock.acquire(timeout=N) prevents indefinite blocking

Gotchas

•Forgetting to release a lock (no try/finally) can deadlock the entire program
•RLock must be released exactly as many times as acquired
•Locks do not protect against deadlocks between multiple locks (lock ordering matters)
•Using a Lock where an RLock is needed causes hard-to-debug deadlocks in nested calls

Lock: basic mutex, deadlocks on double-acquire by same thread. RLock: reentrant, same thread can acquire multiple times. Always use with lock: pattern for safety.

•Lock: simple mutual exclusion, no re-entry
•RLock: allows same thread to re-acquire

import threading

lock = threading.Lock()
with lock:
    print("Critical section")

rlock = threading.RLock()
rlock.acquire()
rlock.acquire()  # OK with RLock, deadlock with Lock
rlock.release()
rlock.release()

Critical section

Always use with-statement or try/finally to guarantee release

Question 4 / 18

What is threading.Event and how is it used for thread coordination?

threading.Event is a simple synchronization primitive for communication between threads. An Event manages an internal flag that can be set to True with set(), reset to False with clear(), and waited on with wait().

Methods:

set() -- sets the flag to True, waking all waiting threads
clear() -- resets the flag to False
wait(timeout=None) -- blocks until the flag is True (or timeout)
is_set() -- returns True if the flag is set

Use cases:

Signal that initialization is complete
Coordinate start of multiple threads
Implement graceful shutdown
Producer signaling consumer that data is ready

Pseudo Code

Event flow:
  Main thread: event = Event()  (flag = False)
  Worker:      event.wait()     (blocks)
  Main thread: event.set()      (flag = True)
  Worker:      (unblocked, continues execution)

Code

import threading
import time

# Example 1: Signal thread to start
start_event = threading.Event()

def worker(event, name):
    print(f"{name}: waiting for signal...")
    event.wait()  # blocks until event is set
    print(f"{name}: got signal, starting work!")
    time.sleep(0.1)
    print(f"{name}: done")

threads = [threading.Thread(target=worker, args=(start_event, f"Worker-{i}"))
           for i in range(3)]
for t in threads:
    t.start()

time.sleep(0.5)  # simulate setup
print("Main: sending start signal")
start_event.set()  # all workers wake up simultaneously

for t in threads:
    t.join()

# Example 2: Graceful shutdown
shutdown_event = threading.Event()

def background_task(stop_event):
    while not stop_event.is_set():
        print("Working...", end=" ")
        stop_event.wait(timeout=0.2)  # sleep but wake on shutdown
    print("\nShutting down gracefully")

t = threading.Thread(target=background_task, args=(shutdown_event,))
t.start()
time.sleep(0.5)
shutdown_event.set()  # signal shutdown
t.join()

# Example 3: wait with timeout
event = threading.Event()
success = event.wait(timeout=0.1)  # returns False after timeout
print(f"Event received: {success}")  # False
event.set()
success = event.wait(timeout=0.1)  # returns True immediately
print(f"Event received: {success}")  # True

Worker-0: waiting for signal...
Worker-1: waiting for signal...
Worker-2: waiting for signal...
Main: sending start signal
Worker-2: got signal, starting work!
Worker-2: done
Worker-0: got signal, starting work!
Worker-0: done
Worker-1: got signal, starting work!
Worker-1: done
Working...
Shutting down gracefully
Event received: False
Event received: True

Key Points

•Event.wait() blocks until the flag is True or timeout expires
•Event.set() wakes ALL waiting threads simultaneously
•Event.wait(timeout=N) returns True if set, False if timed out
•Great for startup signals and graceful shutdown patterns

Gotchas

•Event.wait() without timeout can block forever if no thread calls set()
•After set(), the flag stays True -- new wait() calls return immediately until clear() is called
•Event is NOT for passing data between threads -- use Queue for that
•If you need to signal repeatedly, you must call clear() between set() calls

threading.Event is a flag-based synchronization primitive. wait() blocks until set() is called. Use it for signaling between threads (startup, shutdown, readiness).

•wait() blocks, set() unblocks all waiters
•Use for startup signals and graceful shutdown

import threading

event = threading.Event()

def worker():
    event.wait()  # blocks until set
    print("Signal received!")

t = threading.Thread(target=worker)
t.start()
event.set()  # unblock worker
t.join()

Signal received!

Flag stays True after set() -- call clear() to reset for reuse

Question 5 / 18

How does the multiprocessing module differ from threading?

Key Differences:

Feature	`threading`	`multiprocessing`
Parallelism	Limited by GIL	True parallelism
Memory	Shared memory space	Separate memory per process
Communication	Direct variable sharing	IPC (Queue, Pipe, shared memory)
Overhead	Low (lightweight threads)	High (process creation)
Best for	I/O-bound tasks	CPU-bound tasks
Crash impact	Thread crash can corrupt process	Process crash is isolated

multiprocessing API mirrors threading:

Process (like Thread)
Lock, Event, Semaphore (same API)
Queue (process-safe)
Pool for worker pool patterns

Data sharing:

Since processes have separate memory, data must be explicitly shared via:

multiprocessing.Queue / Pipe for message passing
multiprocessing.Value / Array for shared memory
multiprocessing.Manager for shared objects (dicts, lists)

Pseudo Code

Threading:
  All threads share same memory
  GIL prevents CPU parallelism
  Low overhead

Multiprocessing:
  Each process has own memory + GIL
  True CPU parallelism
  Higher overhead, needs IPC

Code

from multiprocessing import Process, Queue, Value, Lock
import os
import time

# Basic Process
def worker(name):
    print(f"{name} running in PID {os.getpid()}")
    return name

if __name__ == '__main__':
    processes = []
    for i in range(3):
        p = Process(target=worker, args=(f"Worker-{i}",))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    # Communication via Queue
    def producer(q):
        for i in range(5):
            q.put(i * i)
        q.put(None)  # sentinel

    def consumer(q):
        while True:
            item = q.get()
            if item is None:
                break
            print(f"Got: {item}")

    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

    # Shared memory with Value
    counter = Value('i', 0)  # 'i' = integer
    lock = Lock()

    def increment(counter, lock, n):
        for _ in range(n):
            with lock:
                counter.value += 1

    procs = [Process(target=increment, args=(counter, lock, 1000))
             for _ in range(4)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()
    print(f"Counter: {counter.value}")  # 4000

Counter: 0

Key Points

•multiprocessing bypasses the GIL with separate processes
•API mirrors threading (Process, Lock, Event, Queue)
•Data sharing requires explicit IPC (Queue, Pipe, Value, Array)
•Best for CPU-bound tasks; threading is better for I/O-bound

Gotchas

•On Windows/macOS, multiprocessing uses 'spawn' by default -- all code must be inside if __name__ == '__main__'
•Arguments to Process must be picklable (no lambdas, no open files)
•Process creation overhead is much higher than thread creation
•Shared Value/Array without proper locking still causes race conditions

multiprocessing creates separate OS processes, each with its own GIL, enabling true CPU parallelism. threading shares memory (easy) but is GIL-limited. Use multiprocessing for CPU-bound, threading for I/O-bound.

•Separate processes bypass GIL for true parallelism
•Data sharing requires Queue, Pipe, or shared memory

from multiprocessing import Process
import os

def task():
    print(f"PID: {os.getpid()}")

if __name__ == '__main__':
    p = Process(target=task)
    p.start()
    p.join()

PID: 12345  (actual PID varies)

Must guard with if __name__ == '__main__' on Windows/macOS

Question 6 / 18

How does multiprocessing.Pool work for parallel computation?

multiprocessing.Pool creates a pool of worker processes that can execute tasks in parallel. It provides several methods:

Blocking methods:

map(func, iterable) -- parallel map, returns results in order
starmap(func, iterable) -- like map, but unpacks argument tuples
apply(func, args) -- single function call, blocks until done

Non-blocking methods:

map_async(func, iterable) -- returns AsyncResult
starmap_async(func, iterable) -- returns AsyncResult
apply_async(func, args) -- returns AsyncResult

Lifecycle:

pool.close() -- no more tasks accepted
pool.join() -- wait for all workers to finish
pool.terminate() -- kill all workers immediately

Use with Pool() as pool: for automatic cleanup.

Pseudo Code

Pool(processes=4):
  Creates 4 worker processes
  pool.map(f, [a,b,c,d]) -> distributes to workers
  Returns [f(a), f(b), f(c), f(d)] in order

Code

from multiprocessing import Pool, cpu_count
import time

def heavy_computation(n):
    """Simulate CPU-bound work."""
    total = sum(i * i for i in range(n))
    return total

if __name__ == '__main__':
    numbers = [5_000_000] * 8

    # Sequential
    start = time.perf_counter()
    results_seq = [heavy_computation(n) for n in numbers]
    seq_time = time.perf_counter() - start
    print(f"Sequential: {seq_time:.3f}s")

    # Parallel with Pool
    start = time.perf_counter()
    with Pool(processes=cpu_count()) as pool:
        results_par = pool.map(heavy_computation, numbers)
    par_time = time.perf_counter() - start
    print(f"Parallel ({cpu_count()} cores): {par_time:.3f}s")
    print(f"Speedup: {seq_time / par_time:.1f}x")

    # starmap: multiple arguments
    def power(base, exp):
        return base ** exp

    with Pool(4) as pool:
        args = [(2, 10), (3, 5), (4, 3), (5, 2)]
        results = pool.starmap(power, args)
        print(results)  # [1024, 243, 64, 25]

    # apply_async: non-blocking with callback
    def on_complete(result):
        print(f"Callback got: {result}")

    with Pool(2) as pool:
        future = pool.apply_async(heavy_computation, args=(1_000_000,),
                                   callback=on_complete)
        # Do other work while waiting...
        result = future.get(timeout=10)  # blocks until ready
        print(f"Result: {result}")

    # imap: lazy parallel map (results yielded as completed in order)
    with Pool(4) as pool:
        for result in pool.imap(heavy_computation, [1_000_000] * 4):
            print(f"Got result: {result}")

    # imap_unordered: fastest results first
    with Pool(4) as pool:
        for result in pool.imap_unordered(heavy_computation, [1_000_000] * 4):
            print(f"Got result: {result}")

Sequential: 3.456s  (varies by machine)
Parallel (8 cores): 0.567s  (varies by machine)
Speedup: 6.1x  (varies by core count)
[1024, 243, 64, 25]
Callback got: 333332833333500000
Result: 333332833333500000
Ordered imap results: [0, 1, 4, 9, 16, 25, 36, 49, 64]

Key Points

•Pool manages a fixed set of worker processes for parallel execution
•map/starmap return results in order; imap_unordered returns as completed
•apply_async returns AsyncResult for non-blocking execution
•Use 'with Pool() as pool:' for automatic cleanup

Gotchas

•Arguments and return values must be picklable (serialized between processes)
•Creating too many processes can waste resources -- use cpu_count() as guide
•Pool overhead makes it inefficient for very small/fast tasks
•Exceptions in workers are re-raised in the main process when .get() is called

Pool(N) creates N worker processes. pool.map(func, iterable) distributes work in parallel and returns ordered results. Use with Pool() as pool: for cleanup.

•pool.map() for ordered parallel execution
•Arguments must be picklable

from multiprocessing import Pool

def square(n):
    return n * n

if __name__ == '__main__':
    with Pool(4) as pool:
        print(pool.map(square, range(10)))  # [0, 1, 4, 9, ...81]

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

Overhead makes it inefficient for trivially fast functions

Question 7 / 18

What are the basics of asyncio and the async/await syntax?

asyncio is Python's framework for cooperative multitasking using coroutines. Unlike threads (preemptive), asyncio coroutines voluntarily yield control at await points.

Key concepts:

Coroutine: A function defined with async def. Calling it returns a coroutine object (does not execute).
await: Suspends the coroutine until the awaited operation completes, allowing other coroutines to run.
Event loop: The central scheduler that runs coroutines, handles I/O callbacks, and manages tasks.

When to use asyncio:

High-concurrency I/O-bound operations (thousands of simultaneous network connections)
Web servers, WebSocket handlers, API clients
NOT for CPU-bound work (use multiprocessing)

asyncio vs threading:

asyncio: single-threaded, cooperative, no race conditions from shared state
threading: multi-threaded, preemptive, needs locks for shared state

Pseudo Code

async def main():
    result = await some_io_operation()  # yields control here
    return result

asyncio.run(main())  # creates event loop, runs coroutine

Code

import asyncio
import time

# Basic coroutine
async def greet(name, delay):
    await asyncio.sleep(delay)  # non-blocking sleep
    print(f"Hello, {name}!")
    return f"Greeted {name}"

# Running a single coroutine
async def main():
    result = await greet("World", 0.1)
    print(result)

asyncio.run(main())

# Running coroutines concurrently
async def fetch_data(url, delay):
    print(f"Fetching {url}...")
    await asyncio.sleep(delay)  # simulates I/O
    return f"Data from {url}"

async def main_concurrent():
    start = time.perf_counter()

    # Sequential (slow)
    r1 = await fetch_data("url1", 0.2)
    r2 = await fetch_data("url2", 0.2)
    print(f"Sequential: {time.perf_counter() - start:.2f}s")

    start = time.perf_counter()

    # Concurrent (fast)
    r1, r2 = await asyncio.gather(
        fetch_data("url1", 0.2),
        fetch_data("url2", 0.2)
    )
    print(f"Concurrent: {time.perf_counter() - start:.2f}s")
    print(r1, r2)

asyncio.run(main_concurrent())

# Async context managers and iterators
async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.01)
        yield i

async def demo_async_for():
    async for value in async_range(5):
        print(value, end=' ')  # 0 1 2 3 4
    print()

asyncio.run(demo_async_for())

Hello, World!
Greeted World
Fetching url1...
Fetching url2...
Sequential: 0.40s
Fetching url1...
Fetching url2...
Concurrent: 0.20s
Data from url1 Data from url2
0 1 2 3 4

Key Points

•async def creates a coroutine; calling it returns a coroutine object
•await suspends execution and yields control to the event loop
•asyncio.run() is the entry point that creates and runs the event loop
•Coroutines run concurrently only if gathered/tasked -- sequential await is serial

Gotchas

•Calling a coroutine function without await just creates the object -- it never executes
•You cannot use await outside of an async function
•asyncio.sleep() is non-blocking; time.sleep() blocks the entire event loop
•CPU-bound code in a coroutine blocks all other coroutines since asyncio is single-threaded

async def creates coroutines, await suspends them at I/O points. asyncio.run(main()) starts the event loop. Use asyncio.gather() to run coroutines concurrently.

•async def + await for cooperative multitasking
•asyncio.gather() for concurrent execution

import asyncio

async def fetch(url):
    await asyncio.sleep(0.1)  # simulate I/O
    return f"data from {url}"

async def main():
    results = await asyncio.gather(fetch("a"), fetch("b"))
    print(results)  # ['data from a', 'data from b']

asyncio.run(main())

['data from a', 'data from b']

time.sleep() blocks the entire event loop -- use asyncio.sleep()

Question 8 / 18

What is the asyncio event loop and how does it work?

The event loop is the core of asyncio. It is a single-threaded scheduler that:

Runs coroutines, switching between them at await points
Manages I/O callbacks (sockets, file descriptors)
Schedules delayed calls (call_later, call_at)
Handles signals

How it works:

The loop maintains a queue of ready tasks. On each iteration:

Executes ready tasks until they hit an await
Checks for completed I/O operations
Runs scheduled callbacks
Goes to step 1

Key APIs:

asyncio.run(coro) -- creates a new event loop, runs the coroutine, then closes the loop
asyncio.get_event_loop() -- get the running loop (deprecated in 3.10+)
asyncio.get_running_loop() -- get the currently running loop (preferred)
loop.run_until_complete(coro) -- low-level way to run a coroutine
loop.run_forever() -- run until stop() is called

Pseudo Code

Event Loop Cycle:
  while True:
    run all ready callbacks
    poll for I/O events (with timeout)
    schedule callbacks for completed I/O
    run scheduled (timed) callbacks
    if nothing to do: exit or wait

Code

import asyncio

# asyncio.run() -- the standard way (Python 3.7+)
async def main():
    loop = asyncio.get_running_loop()
    print(f"Loop running: {loop.is_running()}")
    print(f"Loop type: {type(loop).__name__}")
    return "done"

result = asyncio.run(main())
print(result)  # 'done'

# Scheduling callbacks on the loop
async def schedule_demo():
    loop = asyncio.get_running_loop()

    # Schedule a callback
    def my_callback(message):
        print(f"Callback: {message}")

    loop.call_soon(my_callback, "immediate")
    loop.call_later(0.1, my_callback, "delayed 0.1s")

    await asyncio.sleep(0.2)  # let callbacks execute

asyncio.run(schedule_demo())

# Running blocking code in executor
import time

def blocking_io():
    time.sleep(0.1)  # simulates blocking I/O
    return "blocking result"

async def main_with_executor():
    loop = asyncio.get_running_loop()

    # Run blocking function in thread pool (does not block event loop)
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

    # Or with a custom executor
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=4) as executor:
        result = await loop.run_in_executor(executor, blocking_io)
        print(result)

asyncio.run(main_with_executor())

# Debug mode
async def debug_demo():
    await asyncio.sleep(0.01)

asyncio.run(debug_demo(), debug=True)  # enables slow callback warnings

Loop running: True
Loop type: _UnixSelectorEventLoop
done
Callback: immediate
Callback: delayed 0.1s
blocking result
blocking result

Key Points

•Event loop is a single-threaded scheduler managing coroutines and I/O
•asyncio.run() creates, runs, and closes the loop (preferred entry point)
•loop.run_in_executor() bridges blocking code with asyncio
•get_running_loop() is preferred over get_event_loop() in Python 3.10+

Gotchas

•There can be only one running event loop per thread
•asyncio.run() cannot be called from within a running event loop (use nest_asyncio in Jupyter)
•Blocking calls (time.sleep, requests.get) freeze the entire event loop
•get_event_loop() is deprecated in Python 3.10+ -- use get_running_loop() inside coroutines

The event loop is asyncio's single-threaded scheduler. It runs coroutines, handles I/O, and manages callbacks. asyncio.run(coro) is the standard entry point. Use run_in_executor() for blocking calls.

•Single-threaded scheduler for coroutines and I/O
•asyncio.run() is the standard entry point

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    # Run blocking code without freezing the loop
    result = await loop.run_in_executor(None, lambda: sum(range(1000000)))
    print(result)

asyncio.run(main())

499999500000

Blocking calls freeze the entire event loop -- use run_in_executor()

Question 9 / 18

How do asyncio.gather() and asyncio.create_task() differ?

`asyncio.create_task(coro)`

Schedules a coroutine to run on the event loop and returns a Task object immediately. The task starts running as soon as the event loop gets a chance (at the next await). You can await the task later to get its result.

`asyncio.gather(*coros_or_futures)`

Runs multiple coroutines concurrently and waits for all of them to complete. Returns results as a list in the same order as the input coroutines.

Key differences:

create_task schedules ONE coroutine, returns immediately
gather runs MULTIPLE coroutines and awaits all results
create_task gives you a Task handle for cancellation/inspection
gather has return_exceptions=True to collect errors instead of raising

Pseudo Code

create_task:
  task = create_task(coro())  # schedules immediately
  # do other work...
  result = await task  # get result when needed

gather:
  results = await gather(coro1(), coro2(), coro3())
  # waits for ALL to complete, returns [r1, r2, r3]

Code

import asyncio
import time

async def fetch(name, delay):
    await asyncio.sleep(delay)
    print(f"{name} completed")
    return f"result_{name}"

# create_task: schedule and control individual tasks
async def demo_create_task():
    start = time.perf_counter()

    # Schedule tasks (they start running immediately)
    task1 = asyncio.create_task(fetch("A", 0.2))
    task2 = asyncio.create_task(fetch("B", 0.1))

    # Do other work while tasks run in background
    print("Tasks scheduled, doing other work...")
    await asyncio.sleep(0.05)

    # Now wait for results
    result1 = await task1
    result2 = await task2
    print(f"Results: {result1}, {result2}")
    print(f"Time: {time.perf_counter() - start:.2f}s")  # ~0.2s (concurrent)

asyncio.run(demo_create_task())

# gather: run all and collect results
async def demo_gather():
    start = time.perf_counter()

    results = await asyncio.gather(
        fetch("A", 0.2),
        fetch("B", 0.1),
        fetch("C", 0.15)
    )
    print(f"All results: {results}")  # ['result_A', 'result_B', 'result_C']
    print(f"Time: {time.perf_counter() - start:.2f}s")  # ~0.2s (concurrent)

asyncio.run(demo_gather())

# gather with return_exceptions
async def might_fail(name, should_fail=False):
    if should_fail:
        raise ValueError(f"{name} failed")
    return f"{name} ok"

async def demo_error_handling():
    results = await asyncio.gather(
        might_fail("A"),
        might_fail("B", should_fail=True),
        might_fail("C"),
        return_exceptions=True  # collect errors instead of raising
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"Success: {r}")

asyncio.run(demo_error_handling())

# Task cancellation
async def demo_cancel():
    async def long_task():
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            print("Task was cancelled")
            raise  # re-raise to properly mark as cancelled

    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Confirmed cancelled")

asyncio.run(demo_cancel())

Tasks scheduled, doing other work...
B completed
A completed
Results: result_A, result_B
Time: 0.20s
B completed
C completed
A completed
All results: ['result_A', 'result_B', 'result_C']
Time: 0.20s
Success: A ok
Error: B failed
Success: C ok
Task was cancelled
Confirmed cancelled

Key Points

•create_task schedules one coroutine, returns a Task handle immediately
•gather runs multiple coroutines concurrently and returns all results in order
•create_task allows individual task cancellation and status checking
•gather with return_exceptions=True collects errors instead of raising the first one

Gotchas

•create_task requires a running event loop -- cannot be called outside async context
•Tasks from create_task are garbage collected and cancelled if you don't keep a reference
•gather without return_exceptions raises the first exception and may leave other tasks running
•In Python 3.11+, use asyncio.TaskGroup as a more structured alternative to gather

create_task(coro) schedules one coroutine and returns a Task handle. gather(*coros) runs multiple coroutines concurrently and returns all results as a list.

•create_task: one task, returns handle for control
•gather: multiple tasks, returns all results in order

import asyncio

async def job(n):
    await asyncio.sleep(0.1)
    return n * 2

async def main():
    task = asyncio.create_task(job(5))  # schedule one
    results = await asyncio.gather(job(1), job(2), job(3))  # run multiple
    print(await task)  # 10
    print(results)     # [2, 4, 6]

asyncio.run(main())

10
[2, 4, 6]

Unreferenced tasks may be garbage collected and cancelled

Question 10 / 18

What are concurrent.futures ThreadPoolExecutor and ProcessPoolExecutor?

concurrent.futures provides a high-level interface for asynchronously executing callables using thread or process pools. It abstracts away the details of thread/process management.

`ThreadPoolExecutor`

Uses threads. Best for I/O-bound tasks.
Shares memory, lower overhead.

`ProcessPoolExecutor`

Uses processes. Best for CPU-bound tasks.
Bypasses GIL, separate memory.

Key API:

executor.submit(fn, *args) -- returns a Future object
executor.map(fn, *iterables) -- parallel map, returns iterator of results
Future.result(timeout=None) -- blocks until result is ready
Future.done() -- returns True if completed
as_completed(futures) -- yields futures as they complete

Pseudo Code

with ThreadPoolExecutor(max_workers=5) as executor:
    future = executor.submit(func, arg)  # returns Future
    result = future.result()  # blocks until done

# Or use map for batch processing:
    results = executor.map(func, items)  # parallel map

Code

from concurrent.futures import (
    ThreadPoolExecutor, ProcessPoolExecutor,
    as_completed, wait, FIRST_COMPLETED
)
import time
import urllib.request

# ThreadPoolExecutor for I/O-bound tasks
def fetch_url(url):
    response = urllib.request.urlopen(url)
    return f"{url}: {len(response.read())} bytes"

urls = ['https://httpbin.org/delay/0'] * 3  # example URLs

with ThreadPoolExecutor(max_workers=5) as executor:
    # Method 1: submit + as_completed (results as they finish)
    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(result)
        except Exception as e:
            print(f"{url} failed: {e}")

# ProcessPoolExecutor for CPU-bound tasks
def compute(n):
    return sum(i * i for i in range(n))

if __name__ == '__main__':
    with ProcessPoolExecutor(max_workers=4) as executor:
        # Method 2: map (results in order)
        results = list(executor.map(compute, [5_000_000] * 4))
        print(f"Results: {len(results)} items computed")

    # Future inspection
    with ThreadPoolExecutor(max_workers=2) as executor:
        future = executor.submit(time.sleep, 0.1)
        print(f"Done: {future.done()}")      # False
        print(f"Running: {future.running()}")  # True
        future.result()  # wait for completion
        print(f"Done: {future.done()}")      # True

    # wait() for batch completion
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(time.sleep, i * 0.1) for i in range(3)]
        done, not_done = wait(futures, return_when=FIRST_COMPLETED)
        print(f"First completed: {len(done)}, still running: {len(not_done)}")

    # Exception handling
    def risky_task(n):
        if n == 0:
            raise ValueError("Cannot be zero")
        return 100 / n

    with ThreadPoolExecutor() as executor:
        future = executor.submit(risky_task, 0)
        try:
            result = future.result()
        except ValueError as e:
            print(f"Caught: {e}")

https://httpbin.org/delay/0: 460 bytes  (varies)
https://httpbin.org/delay/0: 460 bytes  (varies)
https://httpbin.org/delay/0: 460 bytes  (varies)
Results: 4 items computed
Done: False
Running: True
Done: True
First completed: 1, still running: 2
Caught: Cannot be zero

Key Points

•ThreadPoolExecutor for I/O-bound; ProcessPoolExecutor for CPU-bound
•submit() returns a Future; map() returns results in order
•as_completed() yields futures as they finish (unordered)
•Exceptions are stored in Future and re-raised on .result()

Gotchas

•ProcessPoolExecutor arguments must be picklable
•executor.map() hides exceptions until you iterate -- they raise on access
•Default max_workers for ThreadPoolExecutor is min(32, os.cpu_count() + 4) in Python 3.8+
•Futures hold references to results in memory -- clear them when done

concurrent.futures provides ThreadPoolExecutor (I/O-bound) and ProcessPoolExecutor (CPU-bound). Use submit() for individual tasks returning Futures, or map() for batch processing.

•Thread pool for I/O, process pool for CPU
•submit() for Futures, map() for batch results

from concurrent.futures import ThreadPoolExecutor, as_completed

def double(n):
    return n * 2

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(double, range(10)))
    print(results)  # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

ProcessPoolExecutor requires picklable arguments

Question 11 / 18

How does the queue module work for thread-safe communication?

The queue module provides thread-safe queue implementations for passing data between threads. All queue operations are thread-safe (use internal locks).

Queue types:

queue.Queue(maxsize=0) -- FIFO (First-In, First-Out)
queue.LifoQueue(maxsize=0) -- LIFO (stack)
queue.PriorityQueue(maxsize=0) -- lowest value first

Key methods:

put(item, block=True, timeout=None) -- add item (blocks if full)
get(block=True, timeout=None) -- remove and return item (blocks if empty)
task_done() -- signal that a gotten item is processed
join() -- blocks until all items are processed (all task_done() called)
qsize(), empty(), full() -- approximate size info

maxsize=0 means unlimited queue size.

Pseudo Code

Producer-Consumer Pattern:
  q = Queue()
  Producer: q.put(item)   # blocks if full
  Consumer: item = q.get() # blocks if empty
            q.task_done()  # signal completion
  Main:     q.join()       # wait for all items processed

Code

import queue
import threading
import time

# Basic producer-consumer
def producer(q, items):
    for item in items:
        print(f"Producing: {item}")
        q.put(item)
        time.sleep(0.05)
    q.put(None)  # sentinel to signal done

def consumer(q):
    while True:
        item = q.get()  # blocks until item available
        if item is None:
            q.task_done()
            break
        print(f"Consuming: {item}")
        time.sleep(0.1)  # simulate processing
        q.task_done()  # signal this item is processed

q = queue.Queue(maxsize=5)  # bounded queue
t1 = threading.Thread(target=producer, args=(q, range(5)))
t2 = threading.Thread(target=consumer, args=(q,))
t1.start()
t2.start()
t1.join()
t2.join()

# Multiple consumers
def multi_consumer_demo():
    q = queue.Queue()
    results = []
    lock = threading.Lock()

    def worker():
        while True:
            item = q.get()
            if item is None:
                break
            with lock:
                results.append(item * 2)
            q.task_done()

    # Start 3 consumers
    threads = []
    for _ in range(3):
        t = threading.Thread(target=worker)
        t.start()
        threads.append(t)

    # Produce items
    for i in range(10):
        q.put(i)

    q.join()  # wait for all items to be processed

    # Stop workers
    for _ in range(3):
        q.put(None)
    for t in threads:
        t.join()

    print(f"Results: {sorted(results)}")

multi_consumer_demo()

# PriorityQueue
pq = queue.PriorityQueue()
pq.put((3, "low priority"))
pq.put((1, "high priority"))
pq.put((2, "medium priority"))

while not pq.empty():
    priority, item = pq.get()
    print(f"Priority {priority}: {item}")
# high priority, medium priority, low priority

# Non-blocking operations
q = queue.Queue()
try:
    item = q.get_nowait()  # same as q.get(block=False)
except queue.Empty:
    print("Queue is empty")

try:
    q.put_nowait("item")  # same as q.put(block=False)
except queue.Full:
    print("Queue is full")

Producing: 0
Producing: 1
Producing: 2
Producing: 3
Producing: 4
Consuming: 0
Consuming: 1
Consuming: 2
Consuming: 3
Consuming: 4
Results: [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
Priority 1: high priority
Priority 2: medium priority
Priority 3: low priority
Queue is empty

Key Points

•Queue, LifoQueue, PriorityQueue are all thread-safe
•put() blocks when full; get() blocks when empty
•task_done() + join() ensures all items are fully processed
•Use sentinels (None) to signal workers to stop

Gotchas

•qsize(), empty(), full() are approximate -- not reliable for synchronization
•Forgetting task_done() causes join() to block forever
•PriorityQueue items must be comparable -- wrap in (priority, index, item) for non-comparable items
•queue.Queue is for threads; multiprocessing.Queue is for processes

queue.Queue is a thread-safe FIFO queue. put() adds items (blocks if full), get() removes (blocks if empty). Use task_done() + join() for completion tracking.

•Thread-safe queue with blocking put/get
•task_done() + join() tracks completion

import queue
import threading

q = queue.Queue()

def worker():
    while (item := q.get()) is not None:
        print(f"Processing {item}")
        q.task_done()

t = threading.Thread(target=worker)
t.start()
for i in range(5):
    q.put(i)
q.join()  # wait for all processed
q.put(None)  # stop worker
t.join()

Processing 0
Processing 1
Processing 2
Processing 3
Processing 4

Forgetting task_done() causes join() to hang forever

Question 12 / 18

What is a race condition and how do you prevent it in Python?

A race condition occurs when the behavior of code depends on the timing/ordering of thread execution. It happens when multiple threads read and write shared data without proper synchronization, leading to unpredictable results.

Common pattern:

Thread A reads value (e.g., balance = 100)
Thread B reads value (balance = 100)
Thread A writes modified value (balance = 100 + 50 = 150)
Thread B writes modified value (balance = 100 - 30 = 70)
Result: 70 instead of expected 120

Prevention methods:

Locks (threading.Lock, RLock)
Thread-safe data structures (queue.Queue)
Atomic operations (some GIL-protected operations)
Thread-local storage (threading.local())
Immutable data (no shared mutable state)

Pseudo Code

Race condition:
  Thread A: read X=0 -> compute X+1 -> write X=1
  Thread B: read X=0 -> compute X+1 -> write X=1
  Result: X=1 (expected X=2)

Fix with lock:
  Thread A: acquire lock -> read X=0 -> write X=1 -> release lock
  Thread B: wait... -> acquire lock -> read X=1 -> write X=2 -> release lock
  Result: X=2 (correct)

Code

import threading
import time

# BROKEN: Race condition
counter = 0

def increment_unsafe(n):
    global counter
    for _ in range(n):
        temp = counter      # read
        temp += 1            # modify
        counter = temp       # write

counter = 0
threads = [threading.Thread(target=increment_unsafe, args=(100_000,))
           for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Unsafe counter: {counter}")  # Often less than 500000!

# FIXED: Using a lock
counter = 0
lock = threading.Lock()

def increment_safe(n):
    global counter
    for _ in range(n):
        with lock:
            counter += 1

counter = 0
threads = [threading.Thread(target=increment_safe, args=(100_000,))
           for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Safe counter: {counter}")  # Always 500000

# FIXED: Using thread-local storage
local_data = threading.local()

def process_request(request_id):
    local_data.request_id = request_id  # each thread has its own copy
    time.sleep(0.01)
    print(f"Thread {threading.current_thread().name}: request {local_data.request_id}")

threads = [threading.Thread(target=process_request, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# FIXED: Using Queue instead of shared state
import queue

def safe_accumulator():
    q = queue.Queue()
    total = 0

    def producer(n):
        for i in range(n):
            q.put(1)

    threads = [threading.Thread(target=producer, args=(1000,)) for _ in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    while not q.empty():
        total += q.get()
    print(f"Queue-based total: {total}")  # Always 5000

safe_accumulator()

Unsafe counter: 500000
Safe counter: 500000
Thread Thread-11 (process_request): request 0
Thread Thread-12 (process_request): request 1
Thread Thread-13 (process_request): request 2
Queue-based total: 5000

Key Points

•Race conditions occur when threads read/modify/write shared data without synchronization
•Use locks (with statement) to protect critical sections
•threading.local() gives each thread its own copy of data
•Queue is inherently thread-safe and avoids shared mutable state

Gotchas

•Even simple operations like counter += 1 are NOT atomic in Python (read-modify-write)
•The GIL does not prevent race conditions -- it only ensures bytecode atomicity, not operation atomicity
•Race conditions are non-deterministic -- code may pass tests and fail in production
•Over-locking (too many locks, too broad) causes performance issues and deadlocks

A race condition occurs when threads access shared data without synchronization, producing unpredictable results. Fix with locks, thread-safe queues, or thread-local storage.

•Shared mutable state + multiple threads = race condition
•Use locks or thread-safe structures to fix

import threading

counter = 0
lock = threading.Lock()

def safe_increment():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1

threads = [threading.Thread(target=safe_increment) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)  # Always 500000

counter += 1 is NOT atomic -- it's read, modify, write

Question 13 / 18

What is a deadlock and how do you prevent it?

A deadlock occurs when two or more threads are blocked forever, each waiting for the other to release a resource.

Four conditions for deadlock (Coffman conditions):

Mutual exclusion: Resources can only be held by one thread
Hold and wait: Thread holds one resource while waiting for another
No preemption: Resources cannot be forcibly taken from a thread
Circular wait: Thread A waits for B, B waits for A

Prevention strategies:

Lock ordering: Always acquire locks in the same global order
Timeouts: Use lock.acquire(timeout=N) to detect deadlocks
Lock-free algorithms: Use atomic operations or queues
Context managers: Acquire all locks atomically
Avoid nested locks: Restructure code to minimize lock nesting

Pseudo Code

Deadlock:
  Thread A: acquire(lock1) -> waiting for lock2
  Thread B: acquire(lock2) -> waiting for lock1
  Both threads wait forever!

Fix (lock ordering):
  Thread A: acquire(lock1) -> acquire(lock2) -> release both
  Thread B: acquire(lock1) -> acquire(lock2) -> release both
  Always same order = no deadlock

Code

import threading
import time
from contextlib import contextmanager

# DEADLOCK EXAMPLE (DO NOT USE)
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread_a_deadlock():
    with lock1:
        time.sleep(0.1)  # gives thread_b time to acquire lock2
        with lock2:  # DEADLOCK: waiting for lock2 held by thread_b
            print("Thread A: got both locks")

def thread_b_deadlock():
    with lock2:
        time.sleep(0.1)  # gives thread_a time to acquire lock1
        with lock1:  # DEADLOCK: waiting for lock1 held by thread_a
            print("Thread B: got both locks")

# FIX 1: Lock ordering (always acquire in same order)
def thread_a_fixed():
    with lock1:
        with lock2:
            print("Thread A: got both locks")

def thread_b_fixed():
    with lock1:  # same order as thread_a
        with lock2:
            print("Thread B: got both locks")

t1 = threading.Thread(target=thread_a_fixed)
t2 = threading.Thread(target=thread_b_fixed)
t1.start(); t2.start()
t1.join(); t2.join()

# FIX 2: Timeout-based detection
def thread_with_timeout():
    acquired1 = lock1.acquire(timeout=1.0)
    if not acquired1:
        print("Could not acquire lock1, backing off")
        return
    try:
        acquired2 = lock2.acquire(timeout=1.0)
        if not acquired2:
            print("Could not acquire lock2, backing off")
            return
        try:
            print("Got both locks")
        finally:
            lock2.release()
    finally:
        lock1.release()

# FIX 3: Context manager that acquires multiple locks in order
@contextmanager
def acquire_locks(*locks):
    # Sort by id to ensure consistent ordering
    sorted_locks = sorted(locks, key=id)
    acquired = []
    try:
        for lock in sorted_locks:
            lock.acquire()
            acquired.append(lock)
        yield
    finally:
        for lock in reversed(acquired):
            lock.release()

def safe_thread():
    with acquire_locks(lock1, lock2):
        print("Got both locks safely")

t1 = threading.Thread(target=safe_thread)
t2 = threading.Thread(target=safe_thread)
t1.start(); t2.start()
t1.join(); t2.join()

Thread A: got both locks
Thread B: got both locks
Got both locks safely
Got both locks safely

Key Points

•Deadlock: threads wait forever for each other's locks
•Lock ordering (always same global order) prevents circular wait
•Timeouts detect deadlocks and allow recovery
•Sort locks by id() for automatic consistent ordering

Gotchas

•Deadlocks are hard to reproduce -- they depend on exact thread timing
•Lock ordering must be consistent across the ENTIRE codebase, not just one function
•RLock deadlocks if acquired by thread A and thread B tries to acquire it (only same-thread re-entry works)
•Deadlocks can involve resources beyond locks: database connections, file handles, etc.

A deadlock occurs when threads wait for each other's locks forever. Prevent by: always acquiring locks in the same order, using timeouts, or using a single lock.

•Always acquire multiple locks in the same global order
•Use timeouts to detect and recover from deadlocks

import threading

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

# Always acquire in same order to prevent deadlock
def safe_task():
    with lock_a:       # always lock_a first
        with lock_b:   # then lock_b
            print("Safe")

# Or use timeout to detect deadlocks
if lock_a.acquire(timeout=5):
    try:
        pass
    finally:
        lock_a.release()

Deadlocks are timing-dependent and hard to reproduce in tests

Question 14 / 18

How does multiprocessing shared memory work?

Since each process has its own memory space, multiprocessing provides several mechanisms for sharing data between processes:

1. `Value(typecode, value)` and `Array(typecode, size_or_initializer)`

Shared ctypes objects in shared memory
Typecodes: 'i' (int), 'd' (double), 'c' (char), etc.
Have an internal lock (.value property is synchronized)

2. `multiprocessing.shared_memory.SharedMemory` (Python 3.8+)

Raw shared memory blocks
Can be used with NumPy arrays for high-performance sharing
Manual lifecycle management (close + unlink)

3. `Manager()`

Creates a server process that manages shared objects
Supports dicts, lists, queues, namespaces, etc.
Slower (uses IPC) but more flexible

Pseudo Code

Value: single shared value with built-in lock
Array: shared array of same-type elements
SharedMemory: raw shared memory block
Manager: server-managed shared Python objects

Code

from multiprocessing import Process, Value, Array, Lock, Manager

# Value and Array
def increment_counter(counter, lock, n):
    for _ in range(n):
        with lock:
            counter.value += 1

def modify_array(arr):
    for i in range(len(arr)):
        arr[i] = arr[i] ** 2

if __name__ == '__main__':
    # Shared Value
    counter = Value('i', 0)  # integer, initial value 0
    lock = Lock()
    procs = [Process(target=increment_counter, args=(counter, lock, 10000))
             for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"Counter: {counter.value}")  # 40000

    # Shared Array
    arr = Array('d', [1.0, 2.0, 3.0, 4.0, 5.0])  # double array
    p = Process(target=modify_array, args=(arr,))
    p.start()
    p.join()
    print(f"Array: {list(arr)}")  # [1.0, 4.0, 9.0, 16.0, 25.0]

    # Manager for complex shared objects
    def worker_with_manager(shared_dict, shared_list, key, value):
        shared_dict[key] = value
        shared_list.append(value)

    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()

        procs = []
        for i in range(4):
            p = Process(target=worker_with_manager,
                        args=(shared_dict, shared_list, f"key_{i}", i * 10))
            procs.append(p)
            p.start()
        for p in procs:
            p.join()

        print(f"Dict: {dict(shared_dict)}")  # {'key_0': 0, 'key_1': 10, ...}
        print(f"List: {list(shared_list)}")   # [0, 10, 20, 30]

    # SharedMemory (Python 3.8+)
    from multiprocessing import shared_memory
    import struct

    shm = shared_memory.SharedMemory(create=True, size=100)
    # Write data
    struct.pack_into('5i', shm.buf, 0, 10, 20, 30, 40, 50)
    # Read data
    values = struct.unpack_from('5i', shm.buf, 0)
    print(f"SharedMemory values: {values}")  # (10, 20, 30, 40, 50)
    shm.close()
    shm.unlink()  # free the shared memory block

Counter: 0
Array: [1.0, 2.0, 3.0, 4.0, 5.0]
Dict: {}
List: []
SharedMemory values: (10, 20, 30, 40, 50)

Key Points

•Value/Array are shared ctypes objects with built-in synchronization
•Manager creates server-managed shared Python objects (dict, list, etc.)
•SharedMemory (3.8+) provides raw shared memory blocks
•All shared data still needs explicit synchronization (locks) for safe access

Gotchas

•Manager objects are slower than Value/Array due to IPC overhead
•SharedMemory must be explicitly unlinked or it leaks (persists until system reboot)
•Value.value property access is atomic, but read-modify-write (value += 1) is NOT
•Array type codes are C types ('i' for int, 'd' for double) -- not Python types

Value and Array share simple typed data. Manager() shares complex objects (dicts, lists) via a server process. SharedMemory (3.8+) gives raw shared memory blocks.

•Value/Array for simple shared data with built-in sync
•Manager for complex Python objects

from multiprocessing import Process, Value, Lock

def inc(counter, lock):
    for _ in range(1000):
        with lock:
            counter.value += 1

if __name__ == '__main__':
    counter = Value('i', 0)
    lock = Lock()
    procs = [Process(target=inc, args=(counter, lock)) for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(counter.value)  # 4000

counter.value += 1 is NOT atomic -- still needs a lock

Question 15 / 18

How do you run blocking code inside an asyncio event loop?

Blocking code (CPU-bound computation, synchronous I/O, third-party libraries that don't support asyncio) will freeze the event loop if called directly in a coroutine. The solution is loop.run_in_executor().

`await loop.run_in_executor(executor, func, *args)`

Runs func in a thread pool or process pool
Returns an awaitable (does not block the event loop)
executor=None uses the default ThreadPoolExecutor

`asyncio.to_thread(func, *args, **kwargs)` (Python 3.9+)

Simplified wrapper around run_in_executor(None, ...)
Supports keyword arguments (run_in_executor does not)

When to use which executor:

ThreadPoolExecutor: blocking I/O (file reads, HTTP requests, database queries)
ProcessPoolExecutor: CPU-bound computation (data processing, encryption)

Pseudo Code

async def main():
    loop = asyncio.get_running_loop()
    
    # Thread pool (blocking I/O)
    result = await loop.run_in_executor(None, blocking_io_func)
    
    # Process pool (CPU-bound)
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_bound_func)

Code

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

# Blocking function
def blocking_io(url):
    """Simulates slow blocking I/O."""
    time.sleep(0.2)
    return f"Data from {url}"

def cpu_heavy(n):
    """CPU-bound computation."""
    return sum(i * i for i in range(n))

# Using run_in_executor
async def main_executor():
    loop = asyncio.get_running_loop()

    # Default thread pool for blocking I/O
    start = time.perf_counter()
    results = await asyncio.gather(
        loop.run_in_executor(None, blocking_io, "url1"),
        loop.run_in_executor(None, blocking_io, "url2"),
        loop.run_in_executor(None, blocking_io, "url3"),
    )
    print(f"3 I/O calls in {time.perf_counter() - start:.2f}s")  # ~0.2s
    print(results)

    # Custom thread pool
    with ThreadPoolExecutor(max_workers=10) as pool:
        result = await loop.run_in_executor(pool, blocking_io, "custom")
        print(result)

    # Process pool for CPU-bound work
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = await asyncio.gather(
            loop.run_in_executor(pool, cpu_heavy, 1_000_000),
            loop.run_in_executor(pool, cpu_heavy, 1_000_000),
        )
        print(f"CPU results: {results}")

asyncio.run(main_executor())

# Python 3.9+: asyncio.to_thread (simpler syntax)
async def main_to_thread():
    start = time.perf_counter()
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io, "url1"),
        asyncio.to_thread(blocking_io, "url2"),
    )
    print(f"to_thread: {time.perf_counter() - start:.2f}s")
    print(results)

asyncio.run(main_to_thread())

# Wrapping a sync library for async use
def sync_db_query(query):
    time.sleep(0.1)  # simulates DB call
    return [{"id": 1, "name": "result"}]

async def async_db_query(query):
    """Async wrapper for sync DB library."""
    return await asyncio.to_thread(sync_db_query, query)

async def main_wrapper():
    data = await async_db_query("SELECT * FROM users")
    print(data)

asyncio.run(main_wrapper())

3 I/O calls in 0.20s  (varies)
['Data from url1', 'Data from url2', 'Data from url3']
Data from custom
CPU results: [333332833333500000, 333332833333500000]
to_thread: 0.20s  (varies)
['Data from url1', 'Data from url2', 'Data from url3']

Key Points

•run_in_executor() runs blocking code in a thread/process pool without blocking the loop
•asyncio.to_thread() (3.9+) is a simpler wrapper for threading
•Use ThreadPoolExecutor for blocking I/O, ProcessPoolExecutor for CPU-bound
•executor=None uses the default thread pool

Gotchas

•run_in_executor does not support keyword arguments -- use functools.partial or to_thread
•The default thread pool has a limited size -- heavy usage may need a custom pool
•ProcessPoolExecutor with run_in_executor requires picklable functions and arguments
•to_thread copies the current context -- be aware of context variable behavior

Use await loop.run_in_executor(None, blocking_func, *args) or await asyncio.to_thread(func, *args) (3.9+) to run blocking code without freezing the event loop.

•run_in_executor for thread/process pool execution
•to_thread (3.9+) for simpler syntax

import asyncio
import time

def blocking_work():
    time.sleep(0.2)
    return "done"

async def main():
    result = await asyncio.to_thread(blocking_work)  # Python 3.9+
    print(result)

asyncio.run(main())

done

run_in_executor does not support keyword arguments

Question 16 / 18

What is asyncio.TaskGroup and how does it improve over asyncio.gather?

asyncio.TaskGroup (Python 3.11+) provides structured concurrency -- a pattern where the lifetime of concurrent tasks is bounded by a scope (the async with block). It improves over gather() in several ways:

Advantages over gather():

Automatic cancellation: If any task fails, ALL remaining tasks are cancelled
ExceptionGroup: Collects ALL exceptions (not just the first)
Scope-bound lifecycle: Tasks cannot outlive the async with block
Cleaner error handling: Uses except* syntax for exception groups

gather() limitations:

With return_exceptions=False: first exception cancels, others may keep running
With return_exceptions=True: all exceptions collected but tasks always run to completion
No guaranteed cleanup if an exception escapes

Pseudo Code

async with TaskGroup() as tg:
    tg.create_task(coro1())
    tg.create_task(coro2())
    tg.create_task(coro3())
# All tasks guaranteed to be done or cancelled here
# If any failed, ExceptionGroup is raised

Code

import asyncio

async def fetch(name, delay, should_fail=False):
    await asyncio.sleep(delay)
    if should_fail:
        raise ValueError(f"{name} failed!")
    return f"{name} result"

# TaskGroup: basic usage
async def demo_taskgroup():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch("A", 0.1))
        task2 = tg.create_task(fetch("B", 0.2))
        task3 = tg.create_task(fetch("C", 0.15))
    # All tasks complete when we exit the block
    print(task1.result(), task2.result(), task3.result())

asyncio.run(demo_taskgroup())

# TaskGroup: error handling with except*
async def demo_taskgroup_errors():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("A", 0.1))
            tg.create_task(fetch("B", 0.05, should_fail=True))
            tg.create_task(fetch("C", 0.1, should_fail=True))
    except* ValueError as eg:
        # ExceptionGroup contains ALL ValueError exceptions
        for exc in eg.exceptions:
            print(f"Caught: {exc}")
    # All tasks are cancelled/complete here

asyncio.run(demo_taskgroup_errors())

# Compare with gather
async def demo_gather_comparison():
    # gather with return_exceptions=True
    results = await asyncio.gather(
        fetch("A", 0.1),
        fetch("B", 0.05, should_fail=True),
        fetch("C", 0.1),
        return_exceptions=True
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"Success: {r}")

asyncio.run(demo_gather_comparison())

# Dynamic task creation with TaskGroup
async def process_urls(urls):
    results = {}

    async def fetch_one(url):
        await asyncio.sleep(0.1)
        results[url] = f"data from {url}"

    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(fetch_one(url))

    return results

urls = [f"https://example.com/{i}" for i in range(5)]
result = asyncio.run(process_urls(urls))
print(f"Fetched {len(result)} URLs")

A result B result C result
Caught: B failed!
Success: A result
Error: B failed!
Success: C result
Fetched 5 URLs

Key Points

•TaskGroup provides structured concurrency with scope-bound task lifetimes
•If any task fails, all remaining tasks are automatically cancelled
•Raises ExceptionGroup containing ALL exceptions (use except* to handle)
•Tasks cannot outlive the async with block -- guaranteed cleanup

Gotchas

•Requires Python 3.11+ (use gather for older versions)
•except* syntax is needed to handle ExceptionGroup (regular except catches ExceptionGroup as a whole)
•You cannot use return values directly from create_task inside the block -- access task.result() after the block
•If the body of 'async with' raises, all tasks are cancelled even if they would succeed

asyncio.TaskGroup (3.11+) is structured concurrency: all tasks are scope-bound, auto-cancelled on failure, and exceptions are collected into ExceptionGroup. Replaces gather() for most use cases.

•Scope-bound tasks with automatic cancellation on failure
•ExceptionGroup collects all errors

import asyncio

async def job(n):
    await asyncio.sleep(0.1)
    return n * 2

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(job(1))
        t2 = tg.create_task(job(2))
    print(t1.result(), t2.result())  # 2 4

asyncio.run(main())

2 4

Requires Python 3.11+

Question 17 / 18

When should you use threading vs multiprocessing vs asyncio?

Decision matrix:

Scenario	Best choice	Why
Many network requests	asyncio	Single-threaded, handles thousands of connections
File I/O with moderate concurrency	threading	Simple, GIL released during I/O
CPU-heavy computation	multiprocessing	True parallelism, bypasses GIL
Mixed I/O and CPU	multiprocessing + asyncio	asyncio for I/O, processes for CPU
Simple script, few tasks	threading	Lowest complexity
Web scraping (many URLs)	asyncio (aiohttp)	Best scalability for network I/O
Image processing	multiprocessing	CPU-bound, benefits from multiple cores
GUI application	threading	Keep UI responsive during I/O

Scalability:

Threading: 100s of threads (OS limit, memory overhead ~8MB stack per thread)
Asyncio: 10,000s of concurrent coroutines (minimal memory per coroutine)
Multiprocessing: 10s of processes (limited by CPU cores)

Complexity:

Threading: Shared state requires locks, prone to race conditions
Asyncio: No locks needed (single-threaded), but requires async ecosystem
Multiprocessing: Process isolation is safer but IPC is complex

Pseudo Code

Decision tree:
  Is it CPU-bound?
    Yes -> multiprocessing
    No -> Is it I/O-bound?
      Yes -> Need 1000+ concurrent connections?
        Yes -> asyncio
        No -> threading (simpler) or asyncio
      No -> single-threaded is fine

Code

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

# I/O-bound: asyncio is best
async def async_io_example():
    async def fetch(url):
        await asyncio.sleep(0.1)  # simulate network I/O
        return url

    start = time.perf_counter()
    tasks = [fetch(f"url_{i}") for i in range(100)]
    results = await asyncio.gather(*tasks)
    print(f"asyncio: 100 I/O tasks in {time.perf_counter() - start:.2f}s")

asyncio.run(async_io_example())

# I/O-bound with simple threading
def threading_io_example():
    import urllib.request

    def fetch(url):
        time.sleep(0.1)  # simulate I/O
        return url

    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=20) as executor:
        results = list(executor.map(fetch, [f"url_{i}" for i in range(100)]))
    print(f"threading: 100 I/O tasks in {time.perf_counter() - start:.2f}s")

threading_io_example()

# CPU-bound: multiprocessing is best
def cpu_task(n):
    return sum(i * i for i in range(n))

def multiprocessing_cpu_example():
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_task, [2_000_000] * 8))
    print(f"multiprocessing: 8 CPU tasks in {time.perf_counter() - start:.2f}s")

if __name__ == '__main__':
    multiprocessing_cpu_example()

# Hybrid: asyncio + ProcessPoolExecutor for mixed workloads
async def hybrid_example():
    loop = asyncio.get_running_loop()

    # I/O-bound in coroutines
    async def io_work():
        await asyncio.sleep(0.1)
        return "io_result"

    # CPU-bound in process pool
    with ProcessPoolExecutor(max_workers=2) as pool:
        io_result, cpu_result = await asyncio.gather(
            io_work(),
            loop.run_in_executor(pool, cpu_task, 1_000_000)
        )
    print(f"Hybrid: {io_result}, CPU={cpu_result}")

asyncio.run(hybrid_example())

asyncio: 100 I/O tasks in 0.10s  (varies)
threading: 100 I/O tasks in 0.51s  (varies)
multiprocessing: 8 CPU tasks in 1.23s  (varies)

Key Points

•asyncio: best for high-concurrency I/O (1000s of connections)
•threading: simplest for moderate I/O concurrency
•multiprocessing: only option for true CPU parallelism
•Combine them for mixed workloads (asyncio loop + process pool)

Gotchas

•Don't use threading for CPU-bound work -- it will be slower than sequential
•asyncio requires the entire call chain to be async -- one blocking call freezes everything
•multiprocessing has high overhead for small tasks (process creation + serialization)
•Premature optimization: many tasks are fast enough single-threaded -- benchmark first

Use asyncio for many I/O connections (1000s), threading for simple I/O concurrency, multiprocessing for CPU-bound parallelism. Combine as needed.

•asyncio: many I/O connections; threading: simple I/O; multiprocessing: CPU
•Benchmark before adding concurrency

# I/O-bound: asyncio or threading
import asyncio
async def io_example():
    await asyncio.gather(*[asyncio.sleep(0.1) for _ in range(100)])

# CPU-bound: multiprocessing
from concurrent.futures import ProcessPoolExecutor
def cpu_example():
    with ProcessPoolExecutor(4) as pool:
        list(pool.map(sum, [range(1000000)] * 4))

Threading is slower than sequential for CPU-bound work

Question 18 / 18

How do you safely share data between threads using threading.local()?

threading.local() creates an object where each thread sees its own independent copy of attributes. Setting an attribute in one thread does not affect the same attribute in another thread.

Use cases:

Per-thread database connections
Per-request context in web servers (Flask uses this for request and g)
Thread-safe logging context
Avoiding lock contention by giving each thread its own data

How it works:

Internally, threading.local() uses a dictionary keyed by thread ID. When you access local.attr, Python looks up the value for the current thread's ID.

For asyncio, use contextvars.ContextVar instead (Python 3.7+), which works with both threads and async tasks.

Pseudo Code

local = threading.local()

Thread-1: local.x = 10  -> stored as {thread1_id: {'x': 10}}
Thread-2: local.x = 20  -> stored as {thread2_id: {'x': 20}}

Thread-1: local.x -> 10 (thread1 sees its own value)
Thread-2: local.x -> 20 (thread2 sees its own value)

Code

import threading
import time

# Basic thread-local storage
local_data = threading.local()

def worker(name, value):
    local_data.name = name
    local_data.value = value
    time.sleep(0.1)  # simulate work, other threads run
    # Each thread still sees its own values
    print(f"Thread {local_data.name}: value = {local_data.value}")

threads = [
    threading.Thread(target=worker, args=("A", 100)),
    threading.Thread(target=worker, args=("B", 200)),
    threading.Thread(target=worker, args=("C", 300)),
]
for t in threads:
    t.start()
for t in threads:
    t.join()
# Each thread prints its own value, no cross-contamination

# Practical: per-thread database connection
class DatabasePool:
    def __init__(self):
        self._local = threading.local()

    def get_connection(self):
        if not hasattr(self._local, 'connection'):
            self._local.connection = self._create_connection()
        return self._local.connection

    def _create_connection(self):
        thread_name = threading.current_thread().name
        print(f"Creating connection for {thread_name}")
        return f"conn_{thread_name}"

pool = DatabasePool()

def db_worker():
    conn = pool.get_connection()  # created once per thread
    conn2 = pool.get_connection()  # reuses same connection
    assert conn is conn2
    print(f"{threading.current_thread().name} using {conn}")

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

# For asyncio: use contextvars instead
import contextvars
import asyncio

request_id = contextvars.ContextVar('request_id', default=None)

async def handle_request(rid):
    request_id.set(rid)
    await asyncio.sleep(0.1)
    print(f"Handling request: {request_id.get()}")

async def main():
    await asyncio.gather(
        handle_request("req_1"),
        handle_request("req_2"),
        handle_request("req_3"),
    )

asyncio.run(main())

Thread A: value = 100
Thread B: value = 200
Thread C: value = 300
Creating connection for Thread-4 (db_worker)
Thread-4 (db_worker) using conn_Thread-4 (db_worker)
Creating connection for Thread-5 (db_worker)
Thread-5 (db_worker) using conn_Thread-5 (db_worker)
Creating connection for Thread-6 (db_worker)
Thread-6 (db_worker) using conn_Thread-6 (db_worker)
Handling request: req_1
Handling request: req_2
Handling request: req_3

Key Points

•threading.local() gives each thread its own copy of attributes
•Perfect for per-thread resources (DB connections, request context)
•No locks needed since data is inherently isolated
•Use contextvars.ContextVar for asyncio (works with both threads and tasks)

Gotchas

•Accessing an attribute that hasn't been set in the current thread raises AttributeError
•threading.local() does NOT work with asyncio coroutines (multiple coroutines share the same thread)
•Thread-local data is not inherited by child threads -- each starts with a clean slate
•Memory is not automatically cleaned up when a thread ends -- can leak if threads are reused (thread pools)

threading.local() gives each thread its own isolated copy of attributes. No locks needed. Use contextvars.ContextVar for asyncio instead.

•Each thread gets its own copy of attributes
•No locks needed -- inherently isolated

import threading

local = threading.local()

def worker(value):
    local.x = value
    print(f"{threading.current_thread().name}: x = {local.x}")

t1 = threading.Thread(target=worker, args=(10,))
t2 = threading.Thread(target=worker, args=(20,))
t1.start(); t2.start()
t1.join(); t2.join()  # each sees its own x

Thread-1 (worker): x = 10
Thread-2 (worker): x = 20

Does NOT work with asyncio -- use contextvars instead

Built by onllm.dev

Home GitHub Support awesomePrep

What is the GIL (Global Interpreter Lock) and how does it impact Python concurrency?

Impact:

Workarounds for CPU-bound parallelism:

How do you create and manage threads using the threading module?

Key methods:

Important: start() vs run()

What are locks and RLocks in threading, and when should you use each?

threading.Lock (Mutex)

threading.RLock (Reentrant Lock)

When to use each:

What is threading.Event and how is it used for thread coordination?

Methods:

Use cases:

How does the multiprocessing module differ from threading?

Key Differences:

multiprocessing API mirrors threading:

Data sharing:

How does multiprocessing.Pool work for parallel computation?

Blocking methods:

Non-blocking methods:

Lifecycle:

What are the basics of asyncio and the async/await syntax?

Key concepts:

When to use asyncio:

asyncio vs threading:

What is the asyncio event loop and how does it work?

How it works:

Key APIs:

How do asyncio.gather() and asyncio.create_task() differ?

asyncio.create_task(coro)

asyncio.gather(*coros_or_futures)

Key differences:

What are concurrent.futures ThreadPoolExecutor and ProcessPoolExecutor?

ThreadPoolExecutor

ProcessPoolExecutor

Key API:

How does the queue module work for thread-safe communication?

Queue types:

Key methods:

maxsize=0 means unlimited queue size.

What is a race condition and how do you prevent it in Python?

Common pattern:

Prevention methods:

What is a deadlock and how do you prevent it?

Four conditions for deadlock (Coffman conditions):

Prevention strategies:

How does multiprocessing shared memory work?

1. Value(typecode, value) and Array(typecode, size_or_initializer)

2. multiprocessing.shared_memory.SharedMemory (Python 3.8+)

3. Manager()

How do you run blocking code inside an asyncio event loop?

await loop.run_in_executor(executor, func, *args)

asyncio.to_thread(func, *args, **kwargs) (Python 3.9+)

When to use which executor:

What is asyncio.TaskGroup and how does it improve over asyncio.gather?

Advantages over gather():

gather() limitations:

When should you use threading vs multiprocessing vs asyncio?

Decision matrix:

Scalability:

Complexity:

How do you safely share data between threads using threading.local()?

Use cases:

How it works:

Concurrency (threading, asyncio, multiprocessing)

Important: `start()` vs `run()`

`threading.Lock` (Mutex)

`threading.RLock` (Reentrant Lock)

`asyncio.create_task(coro)`

`asyncio.gather(*coros_or_futures)`

`ThreadPoolExecutor`

`ProcessPoolExecutor`

1. `Value(typecode, value)` and `Array(typecode, size_or_initializer)`

2. `multiprocessing.shared_memory.SharedMemory` (Python 3.8+)

3. `Manager()`

`await loop.run_in_executor(executor, func, *args)`

`asyncio.to_thread(func, *args, **kwargs)` (Python 3.9+)