bulkhead-pattern

# Bulkhead Pattern ## What I do I provide a pattern that isolates system components to prevent failures in one part from affecting the entire system. Named after ship bulkheads that contain flooding, the bulkhead pattern allocates separate resources (threads, connections, capacity) to different components. If one component fails or becomes slow, it doesn't exhaust resources for other components, enabling graceful degradation and improved system resilience. ## When to use me Use bulkheads when you have multiple services or components that share common resources, when you need to prioritize certain operations over others, or when preventing cascade failures is critical. They're essential in high-traffic systems, microservices architectures, and applications with varying priority operations. Bulkheads are valuable when different operations have different resource requirements or failure tolerance. ## Core Concepts - **Isolation Levels**: Complete, pooled, or hybrid isolation - **Thread Pool**: Separate pools for different operations - **Connection Pooling**: Dedicated database connections per service - **Rate Limiting**: Controlling request rates per component - **Priority Queuing**: Processing high-priority first - **Resource Partitioning**: Dividing resources between components - **Semaphore**: Controlling concurrent access - **Failure Containment**: Containing failures within partitions - **Graceful Degradation**: Partial functionality during failures - **Resource Exhaustion**: Preventing total system failure ## Code Examples ### Thread Pool Bulkhead ```python import queue import threading import time from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Callable, Generic, TypeVar from uuid import uuid4 from enum import Enum T = TypeVar("T") class TaskPriority(Enum): HIGH = 0 MEDIUM = 1 LOW = 2 @dataclass class Task(Generic[T]): id: str func: Callable[..., T] args: tuple kwargs: dict priority: TaskPriority created_at: float = time.time() class PriorityThreadPoolExecutor: def __init__( self, max_workers: int, max_high_priority: int = 10, max_medium_priority: int = 20, max_low_priority: int = 50 ): self._max_workers = max_workers self._queues: dict[TaskPriority, queue.PriorityQueue] = { TaskPriority.HIGH: queue.PriorityQueue(maxsize=max_high_priority), TaskPriority.MEDIUM: queue.PriorityQueue(maxsize=max_medium_priority), TaskPriority.LOW: queue.PriorityQueue(maxsize=max_low_priority) } self._workers: list[threading.Thread] = [] self._running = True self._metrics = { "submitted": 0, "completed": 0, "rejected": 0 } def submit( self, func: Callable[..., T], *args, priority: TaskPriority = TaskPriority.MEDIUM, **kwargs ) -> str: task_id = str(uuid4()) task = Task( id=task_id, func=func, args=args, kwargs=kwargs, priority=priority ) q = self._queues[priority] if q.full(): self._metrics["rejected"] += 1 raise QueueFullError(f"Queue for {priority} is full") q.put((priority.value, task)) self._metrics["submitted"] += 1 return task_id def start(self) -> None: for _ in range(self._max_workers): worker = threading.Thread(target=self._worker_loop) worker.daemon = True worker.start() self._workers.append(worker) def _worker_loop(self) -> None: while self._running: for priority in TaskPriority: q = self._queues[priority] try: _, task = q.get(timeout=0.1) result = task.func(*task.args, **task.k kwargs) self._metrics["completed"] += 1 except queue.Empty: continue except Exception as e: print(f"Task failed: {e}") def shutdown(self) -> None: self._running = False for q in self._queues.values(): q.join() @property def metrics(self) -> dict: return self._metrics.copy() class QueueFullError(Exception): pass def process_order(order_id: str) -> dict: time.sleep(0.5) return {"order_id": order_id, "status": "processed"} def main(): executor = PriorityThreadPoolExecutor( max_workers=5, max_high_priority=5, max_medium_priority=10, max_low_priority=100 ) executor.start() try: high_id = executor.submit( process_order, "order-1", priority=TaskPriority.HIGH ) low_id = executor.submit( process_order, "order-2", priority=TaskPriority.LOW ) print(f"Submitted tasks: {high_id}, {low_id}") finally: time.sleep(2) executor.shutdown() print(f"Metrics: {executor.metrics}") ``` ### Connection Pool Bulkhead ```python import threading import queue from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Protocol, Generic, TypeVar from contextlib import contextmanager from uuid import uuid4 T = TypeVar("T", bound='PooledConnection') @dataclass class ConnectionConfig: max_connections: int min_idle: int = 0 connection_timeout: float = 30.0 idle_timeout: float = 600.0 class PooledConnection(Protocol): @property def id(self) -> str: pass def is_healthy(self) -> bool: pass def execute(self, query: str) -> dict: pass def close(self) -> None: pass class ConnectionPool(Generic[T]): def __init__(self, factory: Callable[[], T], config: ConnectionConfig): self._factory = factory self._config = config self._pool: queue.Queue[T] = queue.Queue() self._in_use: dict[str, T] = {} self._lock = threading.Lock() self._metrics = { "total_created": 0, "currently_in_use": 0, "total_errors": 0 } @contextmanager def acquire(self): conn = self._get_connection() self._in_use[conn.id] = conn self._metrics["currently_in_use"] += 1 try: yield conn finally: self._return_connection(conn) del self._in_use[conn.id] self._metrics["currently_in_use"] -= 1 def _get_connection(self) -> T: try: conn = self._pool.get_nowait() if conn.is_healthy(): return conn conn.close() except queue.Empty: pass with self._lock: if self._metrics["total_created"] < self._config.max_connections: conn = self._factory() self._metrics["total_created"] += 1 return conn conn = self._pool.get(timeout=self._config.connection_timeout) return conn def _return_connection(self, conn: T) -> None: if conn.is_healthy(): try: self._pool.put_nowait(conn) except queue.Full: conn.close() else: conn.close() def prewarm(self, count: int) -> None: for _ in range(min(count, self._config.max_connections)): conn = self._factory() self._pool.put(conn) self._metrics["total_created"] += 1 def close_all(self) -> None: while not self._pool.empty(): try: conn = self._pool.get_nowait() conn.close() except queue.Empty: break for conn in self._in_use.values(): conn.close() @property def metrics(self) -> dict: return { **self._metrics, "available_in_pool": self._pool.qsize(), "in_use": len(self._in_use) } class DatabaseConnection: def __init__(self, db_name: str): self.id = str(uuid4()) self.db_name = db_name self._healthy = True def is_healthy(self) -> bool: return self._healthy def execute(self, query: str) -> dict: return {"query": query, "result": "success"} def close(self) -> None: self._healthy = False pool = ConnectionPool( factory=lambda: DatabaseConnection("main"), config=ConnectionConfig(max_connections=10, min_idle=2) ) pool.prewarm(2) with pool.acquire() as conn: result = conn.execute("SELECT * FROM users") print(f"Result: {result}") print(f"Pool metrics: {pool.metrics}") ``` ### Semaphore-Based Bulkhead ```python import asyncio import threading from typing import Callable from dataclasses import dataclass from datetime import datetime @dataclass class BulkheadConfig: max_concurrent_calls: int max_waiting: int = 0 timeout_seconds: float = 30.0 class ThreadBulkhead: def __init__(self, config: BulkheadConfig): self._semaphore = threading.BoundedSemaphore(config.max_concurrent_calls) self._waiting = 0 self._lock = threading.Lock() self._metrics = { "total_calls": 0, "rejected_calls": 0, "active_calls": 0 } def execute(self, func: Callable, *args, **kwargs): with self._lock: self._metrics["total_calls"] acquired = self._semaphore.acquire(timeout=30) if not acquired: with self._lock: self._metrics["rejected_calls"] += 1 raise BulkheadRejectedError("Concurrent call limit reached") try: with self._lock: self._metrics["active_calls"] += 1 return func(*args, **kwargs) finally: with self._lock: self._metrics["active_calls"] -= 1 self._semaphore.release() @property def metrics(self) -> dict: return self._metrics.copy() class AsyncBulkhead: def __init__(self, config: BulkheadConfig): self._semaphore = asyncio.Semaphore(config.max_concurrent_calls) self._waiting = 0 self._metrics = { "total_calls": 0, "rejected_calls": 0, "active_calls": 0 } async def execute(self, func: Callable, *args, **kwargs): self._metrics["total_calls"] += 1 try: async with asyncio.timeout(30): await self._semaphore.acquire() self._metrics["active_calls"] += 1 try: return await func(*args, **kwargs) finally: self._metrics["active_calls"] -= 1 self._semaphore.release() except asyncio.TimeoutError: self._metrics["rejected_calls"] += 1 raise BulkheadRejectedError("Call timed out") class BulkheadRejectedError(Exception): pass def external_api_call(service_name: str) -> dict: import time time.sleep(0.1) return {"service": service_name, "status": "ok"} bulkhead = ThreadBulkhead(BulkheadConfig(max_concurrent_calls=3)) results = [] for i in range(10): try: result = bulkhead.execute(external_api_call, f"service-{i}") results.append(result) except BulkheadRejectedError: results.append(f"Rejected service-{i}") print(f"Results: {len(results)} calls, {bulkhead.metrics['rejected_calls']} rejected") ``` ### Service Partitioning ```python from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Protocol @dataclass class ServicePartition: name: str max_concurrent: int max_queue: int priority: int class ServiceRouter(ABC): @abstractmethod def get_partition(self, request: 'Request') -> 'ServicePartition': pass class TieredServiceBulkhead: def __init__(self, partitions: list[ServicePartition]): self._partitions = { p.name: { "config": p, "semaphore": asyncio.Semaphore(p.max_concurrent) if hasattr(asyncio, 'Semaphore') else None, "queue": queue.Queue(p.max_queue), "metrics": { "calls": 0, "rejected": 0, "executed": 0 } } for p in partitions } def get_partition_info(self, partition_name: str) -> dict: return self._partitions.get(partition_name, {}) def route_request( self, request: 'Request', router: ServiceRouter ) -> 'PartitionResult': partition = router.get_partition(request) partition_info = self._partitions.get(partition.name) if not partition_info: return PartitionResult(status="unknown_partition") try: partition_info["queue"].put_nowait(request) return PartitionResult(status="queued")