13/09/2025, 16:07 Python Concurrency Guide

Python Concurrency:
Complete Guide
Threading, Asyncio & Multiprocessing

Version 1.0 | Comprehensive Reference Guide

Table of Contents
1. Introduction to Concurrency
2. Threading Module
2.1 Basic Threading
2.2 Synchronization Primitives
2.3 Advanced Threading
3. Asyncio Module
3.1 Basic Asyncio
3.2 Advanced Asyncio
4. Multiprocessing Module
4.1 Basic Multiprocessing
4.2 Synchronization & Communication
4.3 Advanced Multiprocessing
5. Comparison & Best Practices

[Link] 1/36
13/09/2025, 16:07 Python Concurrency Guide

1. Introduction to Concurrency in Python

Key Concepts

Concept Description Use Case

Multiple threads in single

Threading I/O-bound tasks
process, shared memory

Single-threaded cooperative I/O-bound with many

Asyncio
multitasking connections

Multiple processes, separate

Multiprocessing CPU-bound tasks
memory

Note: Python's Global Interpreter Lock (GIL) prevents true parallel execution of
threads for CPU-bound tasks. Use multiprocessing for CPU-intensive operations.

[Link] 2/36
13/09/2025, 16:07 Python Concurrency Guide

2. Threading Module

2.1 Basic Threading

Creating and Starting Threads

import threading
import time

# Method 1: Using Thread class with target function

def worker(name, delay):
"""Simple worker function"""
print(f"Worker {name} starting")
[Link](delay)
print(f"Worker {name} finished")

# Create threads
thread1 = [Link](target=worker, args=("A", 2))
thread2 = [Link](target=worker, args=("B", 1))

# Start threads
[Link]()
[Link]()

# Wait for threads to complete

[Link]()
[Link]()

print("All threads completed")

Thread Subclassing

import threading
import time

class WorkerThread([Link]):
def __init__(self, name, delay):
super().__init__()
[Link] = name
[Link] = delay
[Link] = False # Set True for daemon threads

def run(self):
"""Method called when [Link]() is invoked"""

[Link] 3/36
13/09/2025, 16:07 Python Concurrency Guide
print(f"Thread {[Link]} starting")
[Link]([Link])
print(f"Thread {[Link]} finished")

# Create and start threads

workers = []
for i in range(3):
worker = WorkerThread(f"Worker-{i}", i + 1)
[Link]()
[Link](worker)

# Wait for all threads

for worker in workers:
[Link]()

Thread Functions Reference

Function/Method Description Example

Thread(target,
Create new t = Thread(target=func,
args, kwargs,
thread args=(1,))
daemon)

Start thread
start() [Link]()
execution

Wait for
join(timeout) thread to [Link](5.0)
finish

Check if
is_alive() thread is if thread.is_alive():
running

Get current
current_thread() threading.current_thread()
thread object

List all active

enumerate() [Link]()
threads

active_count() Number of threading.active_count()

active

[Link] 4/36
13/09/2025, 16:07 Python Concurrency Guide

threads

Get main
main_thread() threading.main_thread()
thread object

[Link] 5/36
13/09/2025, 16:07 Python Concurrency Guide

2.2 Synchronization Primitives

Lock

import threading
import time

# Shared resource
counter = 0
lock = [Link]()

def increment(n):
global counter
for _ in range(n):
# Method 1: Manual lock management
[Link]()
try:
temp = counter
[Link](0.0001) # Simulate work
counter = temp + 1
finally:
[Link]()

def increment_with_context(n):
global counter
for _ in range(n):
# Method 2: Using context manager (recommended)
with lock:
temp = counter
[Link](0.0001)
counter = temp + 1

# Create threads
threads = []
for i in range(5):
t = [Link](target=increment_with_context, args=(100,))
[Link](t)
[Link]()

for t in threads:
[Link]()

print(f"Counter value: {counter}") # Should be 500

RLock (Reentrant Lock)

[Link] 6/36
13/09/2025, 16:07 Python Concurrency Guide

import threading

rlock = [Link]()

def recursive_function(n):
with rlock:
print(f"Acquired lock, n={n}")
if n > 0:
recursive_function(n - 1) # Can acquire same lock again
print(f"Releasing lock, n={n}")

thread = [Link](target=recursive_function, args=(3,))

[Link]()
[Link]()

Semaphore

import threading
import time

# Limit concurrent access to 3 threads

semaphore = [Link](3)

def access_resource(id):
with semaphore:
print(f"Thread {id} accessing resource")
[Link](2)
print(f"Thread {id} releasing resource")

threads = []
for i in range(10):
t = [Link](target=access_resource, args=(i,))
[Link](t)
[Link]()

for t in threads:
[Link]()

Event

import threading
import time

# Create event
event = [Link]()

[Link] 7/36
13/09/2025, 16:07 Python Concurrency Guide
def waiter(name):
print(f"{name} waiting for event")
[Link]() # Block until event is set
print(f"{name} proceeding after event")

def setter():
[Link](3)
print("Setting event")
[Link]()

# Create threads
waiters = [[Link](target=waiter, args=(f"Waiter-{i}",))
for i in range(3)]
setter_thread = [Link](target=setter)

# Start all threads

for w in waiters:
[Link]()
setter_thread.start()

# Wait for completion

for w in waiters:
[Link]()
setter_thread.join()

# Event methods
# [Link]() - Set event to true
# [Link]() - Reset event to false
# event.is_set() - Check if event is set
# [Link](timeout) - Wait for event

Condition

import threading
import time
import random

# Shared resource and condition

items = []
condition = [Link]()

def consumer(name):
with condition:
while len(items) == 0:
print(f"{name} waiting for items")
[Link]() # Release lock and wait
item = [Link](0)
print(f"{name} consumed {item}")

[Link] 8/36
13/09/2025, 16:07 Python Concurrency Guide
def producer():
for i in range(5):
[Link]([Link]())
with condition:
item = f"item-{i}"
[Link](item)
print(f"Produced {item}")
condition.notify_all() # Wake up all waiting threads

# Create threads
consumers = [[Link](target=consumer, args=(f"Consumer-{i}",
for i in range(3)]
producer_thread = [Link](target=producer)

# Start threads
for c in consumers:
[Link]()
producer_thread.start()

# Wait for completion

producer_thread.join()
for c in consumers:
[Link]()

Barrier

import threading
import time
import random

# Create barrier for 3 threads

barrier = [Link](3)

def worker(name):
print(f"{name} working...")
[Link]([Link]() * 3)
print(f"{name} waiting at barrier")

try:
index = [Link]() # Wait for all threads
if index == 0:
print("All threads reached barrier!")
except [Link]:
print(f"{name}: Barrier is broken")

print(f"{name} continuing after barrier")

# Create and start threads

threads = []

[Link] 9/36
13/09/2025, 16:07 Python Concurrency Guide
for i in range(3):
t = [Link](target=worker, args=(f"Worker-{i}",))
[Link](t)
[Link]()

for t in threads:
[Link]()

[Link] 10/36
13/09/2025, 16:07 Python Concurrency Guide

2.3 Advanced Threading

Thread Pool Executor

from [Link] import ThreadPoolExecutor, as_completed

import time

def process_item(item):
"""Simulate processing"""
[Link](1)
return item * 2

# Method 1: Using submit()

with ThreadPoolExecutor(max_workers=4) as executor:
# Submit tasks
futures = []
for i in range(10):
future = [Link](process_item, i)
[Link](future)

# Get results as they complete

for future in as_completed(futures):
result = [Link]()
print(f"Result: {result}")

# Method 2: Using map()

with ThreadPoolExecutor(max_workers=4) as executor:
items = range(10)
results = [Link](process_item, items)
for result in results:
print(f"Result: {result}")

# Method 3: With exception handling

def risky_operation(x):
if x == 5:
raise ValueError(f"Error with {x}")
return x * 2

with ThreadPoolExecutor(max_workers=4) as executor:

futures = {[Link](risky_operation, i): i
for i in range(10)}

for future in as_completed(futures):

i = futures[future]
try:
result = [Link]()
print(f"Item {i}: {result}")
except Exception as e:

[Link] 11/36
13/09/2025, 16:07 Python Concurrency Guide
print(f"Item {i} raised: {e}")

Thread Local Storage

import threading
import time

# Create thread-local storage

local_data = [Link]()

def process_with_context(name):
# Set thread-local data
local_data.name = name
local_data.counter = 0

for i in range(3):
local_data.counter += 1
print(f"Thread {local_data.name}: {local_data.counter}")
[Link](0.5)

threads = []
for i in range(3):
t = [Link](target=process_with_context, args=(f"Worker-
[Link](t)
[Link]()

for t in threads:
[Link]()

Timer Threads

import threading

def delayed_action():
print("Timer expired! Action executed.")

# Create timer that waits 5 seconds

timer = [Link](5.0, delayed_action)
[Link]()

print("Timer started, waiting...")

# Cancel timer if needed (uncomment to test)

# [Link]()

[Link] 12/36
13/09/2025, 16:07 Python Concurrency Guide
[Link]()

[Link] 13/36
13/09/2025, 16:07 Python Concurrency Guide

3. Asyncio Module

3.1 Basic Asyncio

Coroutines and Tasks

import asyncio
import time

# Basic coroutine
async def hello_world():
print("Hello")
await [Link](1)
print("World")
return "Done"

# Running a coroutine
# Method 1: [Link]() (Python 3.7+)
result = [Link](hello_world())
print(f"Result: {result}")

# Multiple coroutines
async def task(name, delay):
print(f"Task {name} starting")
await [Link](delay)
print(f"Task {name} completed")
return f"Result-{name}"

async def main():

# Create tasks
task1 = asyncio.create_task(task("A", 2))
task2 = asyncio.create_task(task("B", 1))
task3 = asyncio.create_task(task("C", 3))

# Wait for all tasks

results = await [Link](task1, task2, task3)
print(f"All results: {results}")

[Link](main())

Asyncio Functions Reference

Function Description Example

[Link] 14/36
13/09/2025, 16:07 Python Concurrency Guide

Run async
[Link]() [Link](main())
function

Schedule task =
asyncio.create_task()
coroutine asyncio.create_task(coro())

Run multiple await [Link](c1(),

[Link]()
coroutines c2())

Wait for
done, pending = await
[Link]() multiple
[Link](tasks)
tasks

[Link]() Async sleep await [Link](1)

await
Wait with
asyncio.wait_for() asyncio.wait_for(coro(),
timeout
timeout=5)

Protect from
[Link]() await [Link](task)
cancellation

Iterate as
for coro in
asyncio.as_completed() tasks
asyncio.as_completed(tasks):
complete

Different Ways to Run Coroutines

import asyncio

async def fetch_data(url):

print(f"Fetching {url}")
await [Link](1) # Simulate network delay
return f"Data from {url}"

async def main():

urls = ["url1", "url2", "url3"]

# Method 1: Sequential execution

print("Sequential:")
for url in urls:
result = await fetch_data(url)
[Link] 15/36
13/09/2025, 16:07 Python Concurrency Guide
print(result)

# Method 2: Concurrent with gather

print("\nConcurrent with gather:")
results = await [Link](
fetch_data("url1"),
fetch_data("url2"),
fetch_data("url3")
)
print(results)

# Method 3: Concurrent with tasks

print("\nConcurrent with tasks:")
tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
results = await [Link](*tasks)
print(results)

# Method 4: As completed
print("\nAs completed:")
tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
for coro in asyncio.as_completed(tasks):
result = await coro
print(result)

# Method 5: With wait()

print("\nWith wait():")
tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
done, pending = await [Link](tasks, return_when=[Link]
for task in done:
print([Link]())

[Link](main())

Exception Handling in Asyncio

import asyncio

async def risky_operation(n):

await [Link](1)
if n == 2:
raise ValueError(f"Error with {n}")
return n * 2

async def main():

# Method 1: Try-except with single coroutine
try:
result = await risky_operation(2)
except ValueError as e:
print(f"Caught: {e}")

[Link] 16/36
13/09/2025, 16:07 Python Concurrency Guide

# Method 2: Handle exceptions in gather

tasks = [risky_operation(i) for i in range(5)]
results = await [Link](*tasks, return_exceptions=True)

for i, result in enumerate(results):

if isinstance(result, Exception):
print(f"Task {i} failed: {result}")
else:
print(f"Task {i} result: {result}")

# Method 3: Handle with wait()

tasks = [asyncio.create_task(risky_operation(i)) for i in range(5
done, pending = await [Link](tasks, return_when=[Link]

for task in done:

try:
result = [Link]()
print(f"Result: {result}")
except Exception as e:
print(f"Error: {e}")

[Link](main())

[Link] 17/36
13/09/2025, 16:07 Python Concurrency Guide

3.2 Advanced Asyncio

Asyncio Queue

import asyncio
import random

async def producer(queue, name):

for i in range(5):
await [Link]([Link]())
item = f"{name}-item-{i}"
await [Link](item)
print(f"Producer {name} added {item}")

async def consumer(queue, name):

while True:
item = await [Link]()
if item is None: # Poison pill
break
print(f"Consumer {name} processing {item}")
await [Link](0.5)
queue.task_done()

async def main():

# Create queue with max size
queue = [Link](maxsize=10)

# Create producers and consumers

producers = [
asyncio.create_task(producer(queue, f"P{i}"))
for i in range(2)
]

consumers = [
asyncio.create_task(consumer(queue, f"C{i}"))
for i in range(3)
]

# Wait for producers to finish

await [Link](*producers)

# Wait for queue to be processed

await [Link]()

# Stop consumers
for _ in consumers:
await [Link](None)

await [Link](*consumers)

[Link] 18/36
13/09/2025, 16:07 Python Concurrency Guide

[Link](main())

Asyncio Lock and Semaphore

import asyncio

# Shared resource
counter = 0

async def increment_with_lock(lock, name):

global counter
async with lock:
temp = counter
await [Link](0.1) # Simulate work
counter = temp + 1
print(f"{name}: counter = {counter}")

async def access_limited_resource(semaphore, name):

async with semaphore:
print(f"{name} accessing resource")
await [Link](2)
print(f"{name} releasing resource")

async def main():

# Using Lock
lock = [Link]()
tasks = [
asyncio.create_task(increment_with_lock(lock, f"Task-{i}"))
for i in range(5)
]
await [Link](*tasks)
print(f"Final counter: {counter}")

# Using Semaphore (limit to 2 concurrent)

semaphore = [Link](2)
tasks = [
asyncio.create_task(access_limited_resource(semaphore, f"Work
for i in range(5)
]
await [Link](*tasks)

[Link](main())

Asyncio Event and Condition

[Link] 19/36
13/09/2025, 16:07 Python Concurrency Guide

import asyncio

async def waiter(event, name):

print(f"{name} waiting for event")
await [Link]()
print(f"{name} got event!")

async def setter(event):

await [Link](2)
print("Setting event")
[Link]()

async def consumer_with_condition(condition, shared_data, name):

async with condition:
while len(shared_data) == 0:
print(f"{name} waiting for data")
await [Link]()
item = shared_data.pop(0)
print(f"{name} consumed {item}")

async def producer_with_condition(condition, shared_data):

for i in range(5):
await [Link](0.5)
async with condition:
item = f"item-{i}"
shared_data.append(item)
print(f"Produced {item}")
condition.notify_all()

async def main():

# Using Event
event = [Link]()
await [Link](
waiter(event, "W1"),
waiter(event, "W2"),
setter(event)
)

# Using Condition
condition = [Link]()
shared_data = []

await [Link](
consumer_with_condition(condition, shared_data, "C1"),
consumer_with_condition(condition, shared_data, "C2"),
producer_with_condition(condition, shared_data)
)

[Link](main())

[Link] 20/36
13/09/2025, 16:07 Python Concurrency Guide

Asyncio Timeouts and Cancellation

import asyncio

async def long_operation():

try:
print("Starting long operation")
await [Link](10)
print("Long operation completed")
return "Success"
except [Link]:
print("Operation was cancelled")
raise

async def main():

# Method 1: Using wait_for with timeout
try:
result = await asyncio.wait_for(long_operation(), timeout=2.0
print(f"Result: {result}")
except [Link]:
print("Operation timed out")

# Method 2: Manual cancellation

task = asyncio.create_task(long_operation())
await [Link](1)
[Link]()

try:
await task
except [Link]:
print("Task was cancelled")

# Method 3: Shield from cancellation

task = asyncio.create_task(long_operation())
shielded = [Link](task)

try:
await asyncio.wait_for(shielded, timeout=1.0)
except [Link]:
print("Timed out, but task continues")
# Task continues running despite timeout

[Link](main())

[Link] 21/36
13/09/2025, 16:07 Python Concurrency Guide

4. Multiprocessing Module

4.1 Basic Multiprocessing

Creating Processes

import multiprocessing
import os
import time

# Basic process function

def worker(name):
print(f"Worker {name} started, PID: {[Link]()}")
[Link](2)
print(f"Worker {name} finished")
return f"Result from {name}"

# Method 1: Using Process class

if __name__ == "__main__":
# Create processes
p1 = [Link](target=worker, args=("A",))
p2 = [Link](target=worker, args=("B",))

# Start processes
[Link]()
[Link]()

# Wait for processes

[Link]()
[Link]()

print("All processes completed")

Process Subclassing

import multiprocessing
import os
import time

class WorkerProcess([Link]):
def __init__(self, name, delay):
super().__init__()
[Link] = name
[Link] = delay

[Link] 22/36
13/09/2025, 16:07 Python Concurrency Guide

def run(self):
print(f"Process {[Link]} started, PID: {[Link]()}")
[Link]([Link])
print(f"Process {[Link]} finished")

if __name__ == "__main__":
processes = []
for i in range(3):
p = WorkerProcess(f"Worker-{i}", i + 1)
[Link]()
[Link](p)

for p in processes:
[Link]()
print(f"Process {[Link]} exited with code {[Link]}")

Process Pool

import multiprocessing
import time

def cpu_bound_task(n):
"""CPU-intensive task"""
total = 0
for i in range(n * 1000000):
total += i
return total

if __name__ == "__main__":
# Create pool with 4 worker processes
with [Link](processes=4) as pool:
# Method 1: map()
inputs = [10, 20, 30, 40, 50]
results = [Link](cpu_bound_task, inputs)
print(f"Map results: {results}")

# Method 2: map_async()
async_result = pool.map_async(cpu_bound_task, inputs)
results = async_result.get()
print(f"Map async results: {results}")

# Method 3: apply()
result = [Link](cpu_bound_task, args=(25,))
print(f"Apply result: {result}")

# Method 4: apply_async()
async_results = []
for i in inputs:

[Link] 23/36
13/09/2025, 16:07 Python Concurrency Guide
async_result = pool.apply_async(cpu_bound_task, args=(i,)
async_results.append(async_result)

results = [[Link]() for ar in async_results]

print(f"Apply async results: {results}")

# Method 5: starmap() for multiple arguments

def multiply(x, y):
return x * y

pairs = [(2, 3), (4, 5), (6, 7)]

results = [Link](multiply, pairs)
print(f"Starmap results: {results}")

Multiprocessing Functions Reference

Function/Method Description Example

Process(target, Create new p = Process(target=func, args=

args) process (1,))

Start
start() [Link]()
process

Wait for
join(timeout) [Link](5)
process

Terminate
terminate() [Link]()
process

kill() Kill process [Link]()

Check if
is_alive() if process.is_alive():
running

Get current
current_process() multiprocessing.current_process()
process

Number of
cpu_count() multiprocessing.cpu_count()
CPUs

[Link] 24/36
13/09/2025, 16:07 Python Concurrency Guide

Create
Pool(processes) Pool(processes=4)
process pool

[Link] 25/36
13/09/2025, 16:07 Python Concurrency Guide

4.2 Synchronization & Communication

Queue

import multiprocessing
import time

def producer(queue, name):

for i in range(5):
item = f"{name}-item-{i}"
[Link](item)
print(f"Producer {name} added {item}")
[Link](0.5)

def consumer(queue, name):

while True:
try:
item = [Link](timeout=2)
print(f"Consumer {name} got {item}")
[Link](0.3)
except:
print(f"Consumer {name} timed out")
break

if __name__ == "__main__":
# Create queue
queue = [Link]()

# Create processes
producers = [
[Link](target=producer, args=(queue, f"P{i}"
for i in range(2)
]

consumers = [
[Link](target=consumer, args=(queue, f"C{i}"
for i in range(3)
]

# Start all processes

for p in producers + consumers:
[Link]()

# Wait for producers to finish

for p in producers:
[Link]()

# Wait for consumers

[Link](3)

[Link] 26/36
13/09/2025, 16:07 Python Concurrency Guide

# Terminate consumers
for c in consumers:
[Link]()
[Link]()

Pipe

import multiprocessing

def sender(conn):
messages = ["Hello", "World", {"key": "value"}, [1, 2, 3]]
for msg in messages:
[Link](msg)
print(f"Sent: {msg}")
[Link]()

def receiver(conn):
while True:
try:
msg = [Link]()
print(f"Received: {msg}")
except EOFError:
break
[Link]()

if __name__ == "__main__":
# Create pipe (returns two connection objects)
parent_conn, child_conn = [Link]()

# Create processes
p1 = [Link](target=sender, args=(child_conn,))
p2 = [Link](target=receiver, args=(parent_conn,)

# Start processes
[Link]()
[Link]()

# Wait for completion

[Link]()
[Link]()

Shared Memory

import multiprocessing
import ctypes

[Link] 27/36
13/09/2025, 16:07 Python Concurrency Guide
def worker_with_value(shared_value, lock, name):
for _ in range(100):
with lock:
temp = shared_value.value
shared_value.value = temp + 1
print(f"{name} finished")

def worker_with_array(shared_array, index):

shared_array[index] = index * 2
print(f"Worker {index} set array[{index}] = {shared_array[index]}

if __name__ == "__main__":
# Shared Value
shared_value = [Link]('i', 0) # 'i' for integer
lock = [Link]()

processes = []
for i in range(4):
p = [Link](
target=worker_with_value,
args=(shared_value, lock, f"Worker-{i}")
)
[Link]()
[Link](p)

for p in processes:
[Link]()

print(f"Final value: {shared_value.value}")

# Shared Array
shared_array = [Link]('d', 10) # 'd' for double

processes = []
for i in range(10):
p = [Link](
target=worker_with_array,
args=(shared_array, i)
)
[Link]()
[Link](p)

for p in processes:
[Link]()

print(f"Final array: {list(shared_array)}")

Manager

[Link] 28/36
13/09/2025, 16:07 Python Concurrency Guide

import multiprocessing

def worker_with_shared_dict(shared_dict, key, value):

shared_dict[key] = value
print(f"Set {key} = {value}")

def worker_with_shared_list(shared_list, item):

shared_list.append(item)
print(f"Added {item} to list")

if __name__ == "__main__":
# Create manager
manager = [Link]()

# Create managed objects

shared_dict = [Link]()
shared_list = [Link]()
shared_lock = [Link]()
shared_event = [Link]()

# Work with shared dictionary

processes = []
for i in range(5):
p = [Link](
target=worker_with_shared_dict,
args=(shared_dict, f"key{i}", i * 10)
)
[Link]()
[Link](p)

# Work with shared list

for i in range(5):
p = [Link](
target=worker_with_shared_list,
args=(shared_list, f"item-{i}")
)
[Link]()
[Link](p)

# Wait for all processes

for p in processes:
[Link]()

print(f"Final dictionary: {dict(shared_dict)}")

print(f"Final list: {list(shared_list)}")

[Link] 29/36
13/09/2025, 16:07 Python Concurrency Guide

4.3 Advanced Multiprocessing

Process Pool Executor

from [Link] import ProcessPoolExecutor, as_completed

import time

def cpu_intensive_task(n):
"""Simulate CPU-intensive work"""
total = sum(i * i for i in range(n * 1000000))
return f"Task {n}: {total}"

if __name__ == "__main__":
# Using ProcessPoolExecutor
with ProcessPoolExecutor(max_workers=4) as executor:
# Submit tasks
futures = []
for i in range(10):
future = [Link](cpu_intensive_task, i)
[Link](future)

# Get results as they complete

for future in as_completed(futures):
result = [Link]()
print(result)

# Using map
with ProcessPoolExecutor(max_workers=4) as executor:
inputs = range(10)
results = [Link](cpu_intensive_task, inputs)
for result in results:
print(result)

Multiprocessing with Logging

import multiprocessing
import logging
import sys

def configure_logging():
"""Configure logging for multiprocessing"""
logger = multiprocessing.get_logger()
[Link]([Link])

handler = [Link]([Link])
formatter = [Link](

[Link] 30/36
13/09/2025, 16:07 Python Concurrency Guide
'[%(levelname)s/%(processName)s] %(message)s'
)
[Link](formatter)
[Link](handler)

return logger

def worker_with_logging(name):
logger = multiprocessing.get_logger()
[Link](f"Worker {name} starting")
# Do work
[Link](f"Worker {name} finished")

if __name__ == "__main__":
configure_logging()

processes = []
for i in range(3):
p = [Link](
target=worker_with_logging,
args=(f"Worker-{i}",)
)
[Link]()
[Link](p)

for p in processes:
[Link]()

Handling Process Exceptions

import multiprocessing
import traceback

def risky_worker(n, result_queue):

try:
if n == 3:
raise ValueError(f"Error with value {n}")
result = n * 2
result_queue.put(("success", n, result))
except Exception as e:
error_info = {
"error": str(e),
"traceback": traceback.format_exc(),
"input": n
}
result_queue.put(("error", n, error_info))

if __name__ == "__main__":
result_queue = [Link]()

[Link] 31/36
13/09/2025, 16:07 Python Concurrency Guide

processes = []
for i in range(5):
p = [Link](
target=risky_worker,
args=(i, result_queue)
)
[Link]()
[Link](p)

for p in processes:
[Link]()

# Collect results
while not result_queue.empty():
status, n, data = result_queue.get()
if status == "success":
print(f"Success for {n}: {data}")
else:
print(f"Error for {n}: {data['error']}")
print(f"Traceback: {data['traceback']}")

[Link] 32/36
13/09/2025, 16:07 Python Concurrency Guide

5. Comparison & Best Practices

When to Use Each Approach

Scenario Best Choice Reason

Web scraping multiple I/O-bound with many

Asyncio
URLs connections

Reading/writing multiple I/O-bound, simpler than

Threading
files asyncio

True parallelism, bypasses

CPU-intensive calculations Multiprocessing
GIL

Handle many concurrent

REST API server Asyncio
requests

Image/video processing Multiprocessing CPU-intensive operations

Database operations Threading/Asyncio I/O-bound operations

Real-time chat application Asyncio Many concurrent connections

Machine learning training Multiprocessing CPU-intensive computations

Performance Comparison Example

import time
import threading
import asyncio
import multiprocessing
from [Link] import ThreadPoolExecutor, ProcessPoolExecuto

# CPU-bound task
def cpu_bound(n):
return sum(i * i for i in range(n))

# I/O-bound task
def io_bound(n):

[Link] 33/36
13/09/2025, 16:07 Python Concurrency Guide
[Link](0.1)
return n

async def async_io_bound(n):

await [Link](0.1)
return n

def benchmark_threading(func, inputs):

start = [Link]()
with ThreadPoolExecutor(max_workers=4) as executor:
results = list([Link](func, inputs))
return [Link]() - start

def benchmark_multiprocessing(func, inputs):

start = [Link]()
with ProcessPoolExecutor(max_workers=4) as executor:
results = list([Link](func, inputs))
return [Link]() - start

async def benchmark_asyncio(inputs):

start = [Link]()
tasks = [async_io_bound(n) for n in inputs]
results = await [Link](*tasks)
return [Link]() - start

if __name__ == "__main__":
# Test data
cpu_inputs = [1000000] * 10
io_inputs = list(range(20))

print("CPU-bound task comparison:")

print(f"Threading: {benchmark_threading(cpu_bound, cpu_inputs):.2
print(f"Multiprocessing: {benchmark_multiprocessing(cpu_bound, cp

print("\nI/O-bound task comparison:")

print(f"Threading: {benchmark_threading(io_bound, io_inputs):.2f}
print(f"Asyncio: {[Link](benchmark_asyncio(io_inputs)):.2f}s

Best Practices

Threading Best Practices:

Always use locks when accessing shared data

Prefer [Link]() for thread-specific data
Use ThreadPoolExecutor for managing thread pools
Avoid creating too many threads (overhead)
Use daemon threads for background tasks

[Link] 34/36
13/09/2025, 16:07 Python Concurrency Guide

Asyncio Best Practices:

Never use blocking I/O in async functions

Use asyncio.create_task() for concurrent execution
Handle exceptions properly with gather(return_exceptions=True)
Use [Link]() for the main entry point
Prefer async context managers (async with)

Multiprocessing Best Practices:

Always use if __name__ == "__main__": guard

Minimize data transfer between processes
Use Pool for multiple similar tasks
Consider memory usage (each process has overhead)
Use Manager for complex shared state

Common Pitfalls to Avoid

Common Mistakes:

GIL Misconception: Using threading for CPU-bound tasks

Shared State: Modifying shared data without synchronization
Deadlocks: Improper lock acquisition order
Resource Leaks: Not properly closing pools/executors
Blocking Asyncio: Using [Link]() instead of [Link]()
Pickling Issues: Passing non-picklable objects to processes

Conclusion

Understanding when and how to use threading, asyncio, and multiprocessing is

crucial for writing efficient Python applications. Remember:

Threading: Best for I/O-bound tasks with moderate concurrency

Asyncio: Ideal for I/O-bound tasks with high concurrency
Multiprocessing: Essential for CPU-bound tasks requiring true parallelism

Choose the right tool based on your specific use case, and always consider the trade-
offs between complexity and performance gains.

[Link] 35/36
13/09/2025, 16:07 Python Concurrency Guide

Final Tips:
• Profile your code to identify bottlenecks before optimizing
• Start simple and add concurrency only when needed
• Test thoroughly, concurrent code can have subtle bugs
• Consider using higher-level libraries built on these primitives

import asyncio

async def run_command(cmd):

proc = await asyncio.create_subprocess_shell(
cmd,
stdout=[Link],
stderr=[Link]
)

stdout, stderr = await [Link]()

print(f"Command: {cmd}")
print(f"Return code: {[Link]}")
print(f"Output: {[Link]()}")
if stderr:
print(f"Error: {[Link]()}")

async def run_python_code():

code = 'print("Hello from subprocess")'
proc = await asyncio.create_subprocess_exec(
'python', '-c', code,
stdout=asyncio.

[Link] 36/36