Concurrency and parallelism are essential for building efficient applications that can handle multiple tasks simultaneously. This lesson covers threading, multiprocessing, and asyncio for different types of concurrent programming in Python.
Understanding Concurrency vs Parallelism
Key Concepts
- Concurrency: Managing multiple tasks that make progress over the same period (may not run simultaneously)
- Parallelism: Executing multiple tasks simultaneously on different CPU cores
- Threading: Lightweight concurrency for I/O-bound tasks
- Multiprocessing: True parallelism for CPU-bound tasks
- Asyncio: Asynchronous programming for I/O-bound tasks with better performance than threading
Threading
Basic Threading
import threading
import time
import requests
def simple_task(name, duration):
"""Simple task that simulates work."""
print(f"Task {name} starting")
time.sleep(duration)
print(f"Task {name} completed")
def download_file(url, filename):
"""Download a file from URL."""
print(f"Downloading {filename} from {url}")
try:
response = requests.get(url, timeout=10)
with open(filename, 'wb') as f:
f.write(response.content)
print(f"Downloaded {filename} ({len(response.content)} bytes)")
except Exception as e:
print(f"Error downloading {filename}: {e}")
# Sequential execution
print("=== Sequential Execution ===")
start_time = time.time()
simple_task("A", 1)
simple_task("B", 1)
simple_task("C", 1)
sequential_time = time.time() - start_time
print(f"Sequential execution took: {sequential_time:.2f} seconds")
# Threaded execution
print("\n=== Threaded Execution ===")
start_time = time.time()
threads = []
for i in range(3):
thread = threading.Thread(target=simple_task, args=(f"Thread-{i}", 1))
threads.append(thread)
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()
threaded_time = time.time() - start_time
print(f"Threaded execution took: {threaded_time:.2f} seconds")
# Thread-safe counter example
class ThreadSafeCounter:
"""Thread-safe counter using locks."""
def __init__(self):
self._value = 0
self._lock = threading.Lock()
def increment(self):
"""Increment counter safely."""
with self._lock:
self._value += 1
def get_value(self):
"""Get current counter value."""
with self._lock:
return self._value
def increment_by(self, amount):
"""Increment counter by specified amount."""
with self._lock:
self._value += amount
def worker(counter, iterations):
"""Worker function that increments counter."""
for _ in range(iterations):
counter.increment()
# Thread-safe counter demo
print("\n=== Thread-Safe Counter Demo ===")
counter = ThreadSafeCounter()
threads = []
# Create multiple threads that increment the counter
for i in range(5):
thread = threading.Thread(target=worker, args=(counter, 1000))
threads.append(thread)
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()
print(f"Final counter value: {counter.get_value()}")
print(f"Expected value: {5 * 1000}")
Advanced Threading
import threading
import time
import queue
from concurrent.futures import ThreadPoolExecutor, as_completed
# Producer-Consumer Pattern
class ProducerConsumer:
"""Producer-consumer pattern using threads and queues."""
def __init__(self, max_size=5):
self.queue = queue.Queue(maxsize=max_size)
self.results = []
self.lock = threading.Lock()
def producer(self, items):
"""Produce items and put them in queue."""
for item in items:
print(f"Producing: {item}")
self.queue.put(item)
time.sleep(0.1) # Simulate work
print("Producer finished")
def consumer(self, consumer_id):
"""Consume items from queue."""
while True:
try:
item = self.queue.get(timeout=1)
print(f"Consumer {consumer_id} processing: {item}")
time.sleep(0.2) # Simulate work
with self.lock:
self.results.append(f"Processed by {consumer_id}: {item}")
self.queue.task_done()
except queue.Empty:
print(f"Consumer {consumer_id} finished")
break
# Using producer-consumer
print("\n=== Producer-Consumer Demo ===")
pc = ProducerConsumer()
# Create producer thread
producer_thread = threading.Thread(
target=pc.producer,
args=([f"Item-{i}" for i in range(10)],)
)
# Create consumer threads
consumer_threads = []
for i in range(3):
consumer_thread = threading.Thread(
target=pc.consumer,
args=(i,)
)
consumer_threads.append(consumer_thread)
# Start all threads
producer_thread.start()
for thread in consumer_threads:
thread.start()
# Wait for completion
producer_thread.join()
for thread in consumer_threads:
thread.join()
print(f"Results: {pc.results}")
# ThreadPoolExecutor
def fetch_url(url):
"""Fetch URL content."""
try:
response = requests.get(url, timeout=5)
return {
'url': url,
'status': response.status_code,
'length': len(response.content)
}
except Exception as e:
return {
'url': url,
'status': 'error',
'error': str(e)
}
# ThreadPoolExecutor demo
print("\n=== ThreadPoolExecutor Demo ===")
urls = [
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/2',
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/3',
'https://httpbin.org/delay/1'
]
start_time = time.time()
with ThreadPoolExecutor(max_workers=3) as executor:
# Submit all tasks
future_to_url = {executor.submit(fetch_url, url): url for url in urls}
# Process completed tasks
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
result = future.result()
print(f"URL: {result['url']}, Status: {result['status']}")
except Exception as e:
print(f"URL: {url}, Error: {e}")
executor_time = time.time() - start_time
print(f"ThreadPoolExecutor took: {executor_time:.2f} seconds")
Multiprocessing
Basic Multiprocessing
import multiprocessing
import time
import os
def cpu_intensive_task(n):
"""CPU-intensive task."""
print(f"Process {os.getpid()} processing {n}")
# Simulate CPU-intensive work
result = sum(i * i for i in range(n))
return result
def worker_process(name, queue, results):
"""Worker process that processes items from queue."""
while True:
try:
item = queue.get(timeout=1)
print(f"Process {name} (PID: {os.getpid()}) processing: {item}")
# Simulate work
time.sleep(0.5)
result = item * 2
results.append(result)
queue.task_done()
except:
print(f"Process {name} finished")
break
# Basic multiprocessing
print("=== Basic Multiprocessing ===")
# Sequential execution
start_time = time.time()
sequential_results = [cpu_intensive_task(100000) for _ in range(4)]
sequential_time = time.time() - start_time
print(f"Sequential execution took: {sequential_time:.2f} seconds")
# Multiprocessing execution
start_time = time.time()
with multiprocessing.Pool(processes=4) as pool:
multiprocessing_results = pool.map(cpu_intensive_task, [100000] * 4)
multiprocessing_time = time.time() - start_time
print(f"Multiprocessing execution took: {multiprocessing_time:.2f} seconds")
print(f"Speedup: {sequential_time / multiprocessing_time:.2f}x")
# Process communication with Queue
print("\n=== Process Communication ===")
manager = multiprocessing.Manager()
queue = manager.Queue()
results = manager.list()
# Add items to queue
for i in range(10):
queue.put(i)
# Create and start processes
processes = []
for i in range(3):
process = multiprocessing.Process(
target=worker_process,
args=(f"Worker-{i}", queue, results)
)
processes.append(process)
process.start()
# Wait for all processes to complete
for process in processes:
process.join()
print(f"Results: {list(results)}")
# Shared memory
def modify_shared_array(shared_array, start, end):
"""Modify a portion of shared array."""
for i in range(start, end):
shared_array[i] = shared_array[i] * 2
# Shared memory demo
print("\n=== Shared Memory Demo ===")
shared_array = multiprocessing.Array('i', range(10))
print(f"Original array: {list(shared_array)}")
# Create processes to modify different portions
processes = []
chunk_size = len(shared_array) // 2
for i in range(2):
start = i * chunk_size
end = start + chunk_size
process = multiprocessing.Process(
target=modify_shared_array,
args=(shared_array, start, end)
)
processes.append(process)
process.start()
# Wait for processes to complete
for process in processes:
process.join()
print(f"Modified array: {list(shared_array)}")
ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor, as_completed
import math
def is_prime(n):
"""Check if a number is prime."""
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
def find_primes_in_range(start, end):
"""Find all primes in a range."""
primes = []
for num in range(start, end):
if is_prime(num):
primes.append(num)
return primes
# ProcessPoolExecutor demo
print("\n=== ProcessPoolExecutor Demo ===")
# Sequential prime finding
start_time = time.time()
sequential_primes = find_primes_in_range(2, 10000)
sequential_time = time.time() - start_time
print(f"Sequential found {len(sequential_primes)} primes in {sequential_time:.2f} seconds")
# Parallel prime finding
start_time = time.time()
# Divide work into chunks
chunk_size = 2500
ranges = [(i, i + chunk_size) for i in range(2, 10000, chunk_size)]
with ProcessPoolExecutor(max_workers=4) as executor:
futures = [executor.submit(find_primes_in_range, start, end)
for start, end in ranges]
parallel_primes = []
for future in as_completed(futures):
primes = future.result()
parallel_primes.extend(primes)
parallel_time = time.time() - start_time
print(f"Parallel found {len(parallel_primes)} primes in {parallel_time:.2f} seconds")
print(f"Speedup: {sequential_time / parallel_time:.2f}x")
Asyncio
Basic Asyncio
import asyncio
import aiohttp
import time
async def simple_async_task(name, duration):
"""Simple async task."""
print(f"Task {name} starting")
await asyncio.sleep(duration)
print(f"Task {name} completed")
return f"Result from {name}"
async def fetch_url_async(session, url):
"""Fetch URL content asynchronously."""
try:
async with session.get(url) as response:
content = await response.text()
return {
'url': url,
'status': response.status,
'length': len(content)
}
except Exception as e:
return {
'url': url,
'status': 'error',
'error': str(e)
}
# Basic asyncio demo
async def basic_asyncio_demo():
"""Demonstrate basic asyncio functionality."""
print("=== Basic Asyncio Demo ===")
start_time = time.time()
# Run tasks concurrently
tasks = [
simple_async_task("A", 1),
simple_async_task("B", 1),
simple_async_task("C", 1)
]
results = await asyncio.gather(*tasks)
asyncio_time = time.time() - start_time
print(f"Async execution took: {asyncio_time:.2f} seconds")
print(f"Results: {results}")
# Run the basic demo
asyncio.run(basic_asyncio_demo())
# Async context manager
class AsyncDatabaseConnection:
"""Async database connection context manager."""
async def __aenter__(self):
print("Connecting to database...")
await asyncio.sleep(0.1) # Simulate connection time
self.connected = True
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print("Closing database connection...")
await asyncio.sleep(0.1) # Simulate disconnection time
self.connected = False
async def query(self, sql):
"""Execute a query."""
if not self.connected:
raise RuntimeError("Not connected to database")
await asyncio.sleep(0.1) # Simulate query time
return f"Result of: {sql}"
# Async context manager demo
async def async_context_demo():
"""Demonstrate async context manager."""
print("\n=== Async Context Manager Demo ===")
async with AsyncDatabaseConnection() as db:
result1 = await db.query("SELECT * FROM users")
print(result1)
result2 = await db.query("INSERT INTO users VALUES (1, 'John')")
print(result2)
asyncio.run(async_context_demo())
Advanced Asyncio
import asyncio
import random
# Async producer-consumer
class AsyncProducerConsumer:
"""Async producer-consumer using asyncio."""
def __init__(self, max_size=5):
self.queue = asyncio.Queue(maxsize=max_size)
self.results = []
async def producer(self, items):
"""Produce items and put them in queue."""
for item in items:
print(f"Producing: {item}")
await self.queue.put(item)
await asyncio.sleep(0.1) # Simulate work
print("Producer finished")
async def consumer(self, consumer_id):
"""Consume items from queue."""
while True:
try:
item = await asyncio.wait_for(self.queue.get(), timeout=1.0)
print(f"Consumer {consumer_id} processing: {item}")
await asyncio.sleep(0.2) # Simulate work
self.results.append(f"Processed by {consumer_id}: {item}")
self.queue.task_done()
except asyncio.TimeoutError:
print(f"Consumer {consumer_id} finished")
break
# Async producer-consumer demo
async def async_producer_consumer_demo():
"""Demonstrate async producer-consumer."""
print("\n=== Async Producer-Consumer Demo ===")
pc = AsyncProducerConsumer()
# Create tasks
producer_task = asyncio.create_task(
pc.producer([f"Item-{i}" for i in range(10)])
)
consumer_tasks = [
asyncio.create_task(pc.consumer(i))
for i in range(3)
]
# Wait for producer to finish
await producer_task
# Wait for consumers to finish
await asyncio.gather(*consumer_tasks)
print(f"Results: {pc.results}")
asyncio.run(async_producer_consumer_demo())
# Async semaphore for rate limiting
class AsyncRateLimiter:
"""Async rate limiter using semaphore."""
def __init__(self, rate_limit):
self.semaphore = asyncio.Semaphore(rate_limit)
async def acquire(self):
"""Acquire a permit."""
await self.semaphore.acquire()
def release(self):
"""Release a permit."""
self.semaphore.release()
async def __aenter__(self):
await self.acquire()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
self.release()
async def rate_limited_task(task_id, rate_limiter):
"""Task that respects rate limiting."""
async with rate_limiter:
print(f"Task {task_id} starting")
await asyncio.sleep(random.uniform(0.5, 1.5))
print(f"Task {task_id} completed")
# Rate limiting demo
async def rate_limiting_demo():
"""Demonstrate rate limiting."""
print("\n=== Rate Limiting Demo ===")
rate_limiter = AsyncRateLimiter(2) # Allow 2 concurrent tasks
tasks = [
asyncio.create_task(rate_limited_task(i, rate_limiter))
for i in range(10)
]
await asyncio.gather(*tasks)
asyncio.run(rate_limiting_demo())
# Async web scraping
async def async_web_scraping_demo():
"""Demonstrate async web scraping."""
print("\n=== Async Web Scraping Demo ===")
urls = [
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/2',
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/3',
'https://httpbin.org/delay/1'
]
start_time = time.time()
async with aiohttp.ClientSession() as session:
tasks = [fetch_url_async(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
async_time = time.time() - start_time
print(f"Async web scraping took: {async_time:.2f} seconds")
for result in results:
if isinstance(result, dict):
print(f"URL: {result['url']}, Status: {result['status']}")
# Run web scraping demo
asyncio.run(async_web_scraping_demo())
Performance Comparison
import time
import threading
import multiprocessing
import asyncio
import requests
import aiohttp
def sync_task(url):
"""Synchronous task."""
try:
response = requests.get(url, timeout=5)
return {'url': url, 'status': response.status_code}
except:
return {'url': url, 'status': 'error'}
def threading_task(urls):
"""Threading implementation."""
with ThreadPoolExecutor(max_workers=5) as executor:
results = list(executor.map(sync_task, urls))
return results
def multiprocessing_task(urls):
"""Multiprocessing implementation."""
with ProcessPoolExecutor(max_workers=5) as executor:
results = list(executor.map(sync_task, urls))
return results
async def async_task(session, url):
"""Async task."""
try:
async with session.get(url) as response:
return {'url': url, 'status': response.status}
except:
return {'url': url, 'status': 'error'}
async def asyncio_task(urls):
"""Asyncio implementation."""
async with aiohttp.ClientSession() as session:
tasks = [async_task(session, url) for url in urls]
results = await asyncio.gather(*tasks)
return results
# Performance comparison
async def performance_comparison():
"""Compare different concurrency approaches."""
print("\n=== Performance Comparison ===")
urls = [
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/1',
'https://httpbin.org/delay/1'
]
# Synchronous
start_time = time.time()
sync_results = [sync_task(url) for url in urls]
sync_time = time.time() - start_time
print(f"Synchronous: {sync_time:.2f} seconds")
# Threading
start_time = time.time()
threading_results = threading_task(urls)
threading_time = time.time() - start_time
print(f"Threading: {threading_time:.2f} seconds")
# Multiprocessing
start_time = time.time()
multiprocessing_results = multiprocessing_task(urls)
multiprocessing_time = time.time() - start_time
print(f"Multiprocessing: {multiprocessing_time:.2f} seconds")
# Asyncio
start_time = time.time()
asyncio_results = await asyncio_task(urls)
asyncio_time = time.time() - start_time
print(f"Asyncio: {asyncio_time:.2f} seconds")
# Speedup comparison
print(f"\nSpeedup vs Synchronous:")
print(f"Threading: {sync_time / threading_time:.2f}x")
print(f"Multiprocessing: {sync_time / multiprocessing_time:.2f}x")
print(f"Asyncio: {sync_time / asyncio_time:.2f}x")
# Run performance comparison
asyncio.run(performance_comparison())
Key Takeaways
- Threading is best for I/O-bound tasks with shared data
- Multiprocessing is best for CPU-bound tasks requiring true parallelism
- Asyncio is best for I/O-bound tasks with high concurrency requirements
- Thread-safe programming requires careful use of locks and synchronization
- Process communication uses queues, pipes, and shared memory
- Async/await provides elegant syntax for asynchronous programming
- Context managers ensure proper resource cleanup
- Performance varies significantly based on task type and implementation
Next Steps
In the next lesson, we'll explore Data Handling with Pandas and NumPy - powerful libraries for data analysis, manipulation, and scientific computing.