Async Programming Patterns
Concurrent Task Execution
# Pattern 1: Gather with timeout and error handling
async def process_concurrent_tasks(
tasks: list[Coroutine[Any, Any, T]],
timeout: float = 10.0
) -> list[T | Exception]:
"""Process tasks concurrently with timeout and exception handling."""
try:
async with asyncio.timeout(timeout): # Python 3.11+
# return_exceptions=True prevents one failure from cancelling others
return await asyncio.gather(*tasks, return_exceptions=True)
except asyncio.TimeoutError:
logger.warning("Tasks timed out after %s seconds", timeout)
raise
Worker Pool with Concurrency Control
# Pattern 2: Semaphore-based worker pool
async def worker_pool(
tasks: list[Callable[[], Coroutine[Any, Any, T]]],
max_workers: int = 10
) -> list[T]:
"""Execute tasks with bounded concurrency using semaphore."""
semaphore = asyncio.Semaphore(max_workers)
async def bounded_task(task: Callable) -> T:
async with semaphore:
return await task()
return await asyncio.gather(*[bounded_task(t) for t in tasks])
Retry with Exponential Backoff
# Pattern 3: Resilient async operations with retries
async def retry_with_backoff(
coro: Callable[[], Coroutine[Any, Any, T]],
max_retries: int = 3,
backoff_factor: float = 2.0,
exceptions: tuple[type[Exception], ...] = (Exception,)
) -> T:
"""Retry async operation with exponential backoff."""
for attempt in range(max_retries):
try:
return await coro()
except exceptions as e:
if attempt == max_retries - 1:
raise
delay = backoff_factor ** attempt
logger.warning("Attempt %d failed, retrying in %s seconds", attempt + 1, delay)
await asyncio.sleep(delay)
Task Cancellation and Cleanup
# Pattern 4: Graceful task cancellation
async def cancelable_task_group(
tasks: list[Coroutine[Any, Any, T]]
) -> list[T]:
"""Run tasks with automatic cancellation on first exception."""
async with asyncio.TaskGroup() as tg: # Python 3.11+
results = [tg.create_task(task) for task in tasks]
return [r.result() for r in results]
Production-Ready AsyncWorkerPool
# Pattern 5: Async Worker Pool with Retries and Exponential Backoff
import asyncio
from typing import Callable, Any, Optional
from dataclasses import dataclass
import time
import logging
logger = logging.getLogger(__name__)
@dataclass
class TaskResult:
"""Result of task execution with retry metadata."""
success: bool
result: Any = None
error: Optional[Exception] = None
attempts: int = 0
total_time: float = 0.0
class AsyncWorkerPool:
"""Worker pool with configurable retry logic and exponential backoff.
Features:
- Fixed number of worker tasks
- Task queue with asyncio.Queue
- Retry logic with exponential backoff
- Graceful shutdown with drain semantics
- Per-task retry tracking
Example:
pool = AsyncWorkerPool(num_workers=5, max_retries=3)
result = await pool.submit(my_async_task)
await pool.shutdown()
"""
def __init__(self, num_workers: int, max_retries: int):
"""Initialize worker pool.
Args:
num_workers: Number of concurrent worker tasks
max_retries: Maximum retry attempts per task (0 = no retries)
"""
self.num_workers = num_workers
self.max_retries = max_retries
self.task_queue: asyncio.Queue = asyncio.Queue()
self.workers: list[asyncio.Task] = []
self.shutdown_event = asyncio.Event()
self._start_workers()
def _start_workers(self) -> None:
"""Start worker tasks that process from queue."""
for i in range(self.num_workers):
worker = asyncio.create_task(self._worker(i))
self.workers.append(worker)
async def _worker(self, worker_id: int) -> None:
"""Worker coroutine that processes tasks from queue.
Continues until shutdown_event is set AND queue is empty.
"""
while not self.shutdown_event.is_set() or not self.task_queue.empty():
try:
# Wait for task with timeout to check shutdown periodically
task_data = await asyncio.wait_for(
self.task_queue.get(),
timeout=0.1
)
# Process task with retries
await self._execute_with_retry(task_data)
self.task_queue.task_done()
except asyncio.TimeoutError:
# No task available, continue to check shutdown
continue
except Exception as e:
logger.error(f"Worker {worker_id} error: {e}")
async def _execute_with_retry(
self,
task_data: dict[str, Any]
) -> None:
"""Execute task with exponential backoff retry logic.
Args:
task_data: Dict with 'task' (callable) and 'future' (to set result)
"""
task: Callable = task_data['task']
future: asyncio.Future = task_data['future']
last_error: Optional[Exception] = None
start_time = time.time()
for attempt in range(self.max_retries + 1):
try:
# Execute the task
result = await task()
# Success! Set result and return
if not future.done():
future.set_result(TaskResult(
success=True,
result=result,
attempts=attempt + 1,
total_time=time.time() - start_time
))
return
except Exception as e:
last_error = e
# If we've exhausted retries, fail
if attempt >= self.max_retries:
break
# Exponential backoff: 0.1s, 0.2s, 0.4s, 0.8s, ...
backoff_time = 0.1 * (2 ** attempt)
logger.warning(
f"Task failed (attempt {attempt + 1}/{self.max_retries + 1}), "
f"retrying in {backoff_time}s: {e}"
)
await asyncio.sleep(backoff_time)
# All retries exhausted, set failure result
if not future.done():
future.set_result(TaskResult(
success=False,
error=last_error,
attempts=self.max_retries + 1,
total_time=time.time() - start_time
))
async def submit(self, task: Callable) -> Any:
"""Submit task to worker pool and wait for result.
Args:
task: Async callable to execute
Returns:
TaskResult with execution metadata
Raises:
RuntimeError: If pool is shutting down
"""
if self.shutdown_event.is_set():
raise RuntimeError("Cannot submit to shutdown pool")
# Create future to receive result
future: asyncio.Future = asyncio.Future()
# Add task to queue
await self.task_queue.put({'task': task, 'future': future})
# Wait for result
return await future
async def shutdown(self, timeout: Optional[float] = None) -> None:
"""Gracefully shutdown worker pool.
Drains queue, then cancels workers after timeout.
Args:
timeout: Max time to wait for queue drain (None = wait forever)
"""
# Signal shutdown
self.shutdown_event.set()
# Wait for queue to drain
try:
if timeout:
await asyncio.wait_for(
self.task_queue.join(),
timeout=timeout
)
else:
await self.task_queue.join()
except asyncio.TimeoutError:
logger.warning("Shutdown timeout, forcing worker cancellation")
# Cancel all workers
for worker in self.workers:
worker.cancel()
# Wait for workers to finish
await asyncio.gather(*self.workers, return_exceptions=True)
# Usage Example:
async def example_usage():
# Create pool with 5 workers, max 3 retries
pool = AsyncWorkerPool(num_workers=5, max_retries=3)
# Define task that might fail
async def flaky_task():
import random
if random.random() < 0.5:
raise ValueError("Random failure")
return "success"
# Submit task
result = await pool.submit(flaky_task)
if result.success:
print(f"Task succeeded: {result.result} (attempts: {result.attempts})")
else:
print(f"Task failed after {result.attempts} attempts: {result.error}")
# Graceful shutdown
await pool.shutdown(timeout=5.0)
# Key Concepts:
# - Worker pool: Fixed workers processing from shared queue
# - Exponential backoff: 0.1 * (2 ** attempt) seconds
# - Graceful shutdown: Drain queue, then cancel workers
# - Future pattern: Submit returns future, worker sets result
# - TaskResult dataclass: Track attempts, time, success/failure
When to Use Each Pattern
- Gather with timeout: Multiple independent operations (API calls, DB queries)
- Worker pool (simple): Rate-limited operations (API with rate limits, DB connection pool)
- Retry with backoff: Unreliable external services (network calls, third-party APIs)
- TaskGroup: Related operations where failure of one should cancel others
- AsyncWorkerPool (production): Production systems needing retry logic, graceful shutdown, task tracking
bobmatnyc