# Distributed Jobs: Run Tasks Across Servers # Distributed Jobs: Run Tasks Across Servers ## Problem Modern applications need to process large volumes of tasks efficiently. Single-server job processing creates bottlenecks: - **Scalability Issues**: A single server can only handle so many concurrent jobs - **Reliability Concerns**: Server failure means job loss and service interruption - **Resource Constraints**: CPU-intensive or long-running tasks block other operations - **Uneven Load Distribution**: Some servers become overloaded while others sit idle - **Lack of Visibility**: Difficult to monitor job status across infrastructure - **No Fault Tolerance**: Failed jobs disappear without retry mechanisms Consider an e-commerce platform processing 10,000 orders daily. Each order requires image processing, payment verification, and email notifications. A single server would struggle, causing delays and customer dissatisfaction. ## Solution Implement a **distributed job queue system** that: 1. **Decouples Job Production from Execution**: Producers enqueue jobs; workers process them independently 2. **Horizontal Scaling**: Add more workers to handle increased load 3. **Load Balancing**: Distribute jobs evenly across available workers 4. **Fault Tolerance**: Retry failed jobs automatically 5. **Persistence**: Store jobs in a message broker (Redis, RabbitMQ, AWS SQS) 6. **Monitoring**: Track job status, metrics, and worker health 7. **Priority Queues**: Process critical jobs first 8. **Idempotency**: Safely retry jobs without side effects ### Architecture Components ``` ┌─────────────────────────────────────────────────────┐ │ Job Producers │ │ (Web Servers, APIs, Scheduled Tasks) │ └────────────────┬────────────────────────────────────┘ │ Enqueue ▼ ┌─────────────────────────────────────────────────────┐ │ Message Broker / Queue │ │ (Redis, RabbitMQ, AWS SQS, Kafka) │ │ ┌──────────────┬──────────────┬──────────────┐ │ │ │ High Priority│ Normal Queue │ Low Priority │ │ │ └──────────────┴──────────────┴──────────────┘ │ └────────────────┬────────────────────────────────────┘ │ Dequeue ▼ ┌─────────────────────────────────────────────────────┐ │ Worker Pool (Scalable) │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Worker 1 │ │ Worker 2 │ │ Worker N │ │ │ └──────────┘ └──────────┘ └──────────┘ │ └────────────────┬────────────────────────────────────┘ │ Results ▼ ┌─────────────────────────────────────────────────────┐ │ Result Storage / Callbacks │ │ (Database, Cache, Webhooks) │ └─────────────────────────────────────────────────────┘ ``` ## Code Implementation ### 1. Job Queue Manager (Python with Redis) ```python import json import uuid import time from datetime import datetime, timedelta from typing import Any, Dict, Optional, List from enum import Enum import redis import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class JobStatus(Enum): PENDING = "pending" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" RETRYING = "retrying" class JobPriority(Enum): LOW = 3 NORMAL = 2 HIGH = 1 class DistributedJobQueue: """Manages distributed job processing across multiple workers""" def __init__(self, redis_host: str = "localhost", redis_port: int = 6379): self.redis_client = redis.Redis( host=redis_host, port=redis_port, decode_responses=True ) self.queue_prefix = "job:queue:" self.job_prefix = "job:data:" self.result_prefix = "job:result:" self.worker_prefix = "worker:" def enqueue_job( self, job_type: str, payload: Dict[str, Any], priority: JobPriority = JobPriority.NORMAL, max_retries: int = 3, timeout: int = 3600 ) -> str: """Enqueue a new job""" job_id = str(uuid.uuid4()) job_data = { "id": job_id, "type": job_type, "payload": json.dumps(payload), "status": JobStatus.PENDING.value, "priority": priority.value, "created_at": datetime.utcnow().isoformat(), "max_retries": max_retries, "retry_count": 0, "timeout": timeout, "attempts": [] } # Store job metadata self.redis_client.hset( f"{self.job_prefix}{job_id}", mapping=job_data ) # Add to priority queue (lower value = higher priority) queue_key = f"{self.queue_prefix}{job_type}" self.redis_client.zadd( queue_key, {job_id: priority.value} ) logger.info(f"Job enqueued: {job_id} (type: {job_type}, priority: {priority.name})") return job_id def dequeue_job(self, job_type: str, worker_id: str) -> Optional[Dict[str, Any]]: """Dequeue a job for processing""" queue_key = f"{self.queue_prefix}{job_type}" # Get highest priority job (lowest score) job_ids = self.redis_client.zrange(queue_key, 0, 0) if not job_ids: return None job_id = job_ids[0] # Remove from queue self.redis_client.zrem(queue_key, job_id) # Get job data job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}") if not job_data: return None # Update status and assign to worker job_data["status"] = JobStatus.PROCESSING.value job_data["worker_id"] = worker_id job_data["started_at"] = datetime.utcnow().isoformat() self.redis_client.hset( f"{self.job_prefix}{job_id}", mapping=job_data ) # Track worker activity self.redis_client.hset( f"{self.worker_prefix}{worker_id}", mapping={ "current_job": job_id, "last_heartbeat": datetime.utcnow().isoformat(), "jobs_processed": self.redis_client.hget(f"{self.worker_prefix}{worker_id}", "jobs_processed") or 0 } ) logger.info(f"Job dequeued: {job_id} by worker {worker_id}") return job_data def complete_job(self, job_id: str, result: Dict[str, Any]) -> bool: """Mark job as completed""" job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}") if not job_data: logger.warning(f"Job not found: {job_id}") return False # Store result result_data = { "job_id": job_id, "status": JobStatus.COMPLETED.value, "result": json.dumps(result), "completed_at": datetime.utcnow().isoformat() } self.redis_client.hset( f"{self.result_prefix}{job_id}", mapping=result_data ) # Update job status job_data["status"] = JobStatus.COMPLETED.value job_data["completed_at"] = datetime.utcnow().isoformat() self.redis_client.hset( f"{self.job_prefix}{job_id}", mapping=job_data ) # Increment worker stats worker_id = job_data.get("worker_id") if worker_id: self.redis_client.hincrby( f"{self.worker_prefix}{worker_id}", "jobs_processed", 1 ) logger.info(f"Job completed: {job_id}") return True def fail_job(self, job_id: str, error: str) -> bool: """Handle job failure with retry logic""" job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}") if not job_data: logger.warning(f"Job not found: {job_id}") return False retry_count = int(job_data.get("retry_count", 0)) max_retries = int(job_data.get("max_retries", 3)) # Record attempt attempt = { "timestamp": datetime.utcnow().isoformat(), "error": error, "retry_count": retry_count } attempts = json.loads(job_data.get("attempts", "[]")) attempts.append(attempt) if retry_count < max_retries: # Retry with exponential backoff backoff_seconds = min(2 ** retry_count * 60, 3600) # Max 1 hour retry_at = datetime.utcnow() + timedelta(seconds=backoff_seconds) job_data["status"] = JobStatus.RETRYING.value job_data["retry_count"] = retry_count + 1 job_data["retry_at"] = retry_at.isoformat() job_data["attempts"] = json.dumps(attempts) self.redis_client.hset( f"{self.job_prefix}{job_id}", mapping=job_data ) # Re-queue with delay queue_key = f"{self.queue_prefix}{job_data['type']}" self.redis_client.zadd( queue_key, {job_id: int(retry_at.timestamp())} ) logger.info(f"Job retrying: {job_id} (attempt {retry_count + 1}/{max_retries})") else: # Max retries exceeded job_data["status"] = JobStatus.FAILED.value job_data["failed_at"] = datetime.utcnow().isoformat() job_data["attempts"] = json.dumps(attempts) self.redis_client.hset( f"{self.job_prefix}{job_id}", mapping=job_data ) logger.error(f"Job failed permanently: {job_id} after {max_retries} retries") return True def get_job_status(self, job_id: str) -> Optional[Dict[str, Any]]: """Get current job status""" job_data = self.redis_client.hgetall(f"{self.job_prefix}{job_id}") if not job_data: return None result_data = self.redis_client.hgetall(f"{self.result_prefix}{job_id}") return { "job": job_data, "result": result_data if result_data else None } def get_queue_stats(self, job_type: str) -> Dict[str, Any]: """Get queue statistics""" queue_key = f"{self.queue_prefix}{job_type}" pending_count = self.redis_client.zcard(queue_key) return { "job_type": job_type, "pending_jobs": pending_count, "timestamp": datetime.utcnow().isoformat() } def get_worker_stats(self, worker_id: str) -> Optional[Dict[str, Any]]: """Get worker statistics""" worker_data = self.redis_client.hgetall(f"{self.worker_prefix}{worker_id}") if not worker_data: return None return worker_data def cleanup_stale_jobs(self, timeout_seconds: int = 3600) -> int: """Clean up jobs that have timed out""" cursor = 0 cleaned = 0 while True: cursor, keys = self.redis_client.scan( cursor, match=f"{self.job_prefix}*", count=100 ) for key in keys: job_data = self.redis_client.hgetall(key) if job_data.get("status") == JobStatus.PROCESSING.value: started_at = datetime.fromisoformat(job_data.get("started_at", "")) if datetime.utcnow() - started_at > timedelta(seconds=timeout_seconds): job_id = job_data["id"] self.fail_job(job_id, "Job timeout") cleaned += 1 if cursor == 0: break return cleaned ``` ### 2. Worker Implementation ```python import time import signal import sys from typing import Callable, Dict, Any import logging logger = logging.getLogger(__name__) class JobWorker: """Worker process that executes distributed jobs""" def __init__(self, worker_id: str, job_queue: DistributedJobQueue): self.worker_id = worker_id self.job_queue = job_queue self.running = True self.job_handlers: Dict[str, Callable] = {} # Register signal handlers signal.signal(signal.SIGTERM, self._handle_shutdown) signal.signal(signal.SIGINT, self._handle_shutdown) def register_handler(self, job_type: str, handler: Callable) -> None: """Register a handler for a job type""" self.job_handlers[job_type] = handler logger.info(f"Handler registered for job type: {job_type}") def start(self, job_types: list, poll_interval: int = 5) -> None: """Start processing jobs""" logger.info(f"Worker {self.worker_id} started") while self.running: job_processed = False for job_type in job_types: job_data = self.job_queue.dequeue_job(job_type, self.worker_id) if job_data: self._process_job(job_data) job_processed = True break if not job_processed: time.sleep(poll_interval) def _process_job(self, job_data: Dict[str, Any]) -> None: """Process a single job""" job_id = job_data["id"] job_type = job_data["type"] try: logger.info(f"Processing job {job_id} (type: {job_type})") # Get handler handler = self.job_handlers.get(job_type) if not handler: raise ValueError(f"No handler for job type: {job_type}") # Parse payload payload = json.loads(job_data["payload"]) # Execute job with timeout result = self._execute_with_timeout( handler, payload, int(job_data.get("timeout", 3600)) ) # Mark as completed self.job_queue.complete_job(job_id, result) except Exception as e: logger.error(f"Job failed: {job_id}, Error: {str(e)}") self.job_queue.fail_job(job_id, str(e)) def _execute_with_timeout( self, handler: Callable, payload: Dict[str, Any], timeout: int ) -> Dict[str, Any]: """Execute handler with timeout""" import signal def timeout_handler(signum, frame): raise TimeoutError(f"Job execution exceeded {timeout} seconds") # Set timeout signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: result = handler(payload) signal.alarm(0) # Cancel alarm return result except Exception as e: signal.alarm(0) raise def _handle_shutdown(self, signum, frame) -> None: """Handle graceful shutdown""" logger.info(f"Worker {self.worker_id} shutting down...") self.running = False sys.exit(0) ``` ### 3. Example Job Handlers ```python import requests from PIL import Image from io import BytesIO class OrderProcessingHandlers: """Example job handlers for order processing""" @staticmethod def process_payment(payload: Dict[str, Any]) -> Dict[str, Any]: """Process payment for an order""" order_id = payload["order_id"] amount = payload["amount"] logger.info(f"Processing payment for order {order_id}: ${amount}") # Simulate payment processing time.sleep(2) return { "order_id": order_id, "status": "paid", "transaction_id": f"TXN-{order_id}-{int(time.time())}" } @staticmethod def process_images(payload: Dict[str, Any]) -> Dict[str, Any]: """Process product images""" image_urls = payload["image_urls"] order_id = payload["order_id"] logger.info(f"Processing {len(image_urls)} images for order {order_id}") processed_images = [] for url in image_urls: try: response = requests.get(url, timeout=10) img = Image.open(BytesIO(response.content)) # Resize and optimize img.thumbnail((800, 800)) processed_images.append({ "original_url": url, "status": "processed" }) except Exception as e: logger.error(f"Failed to process image {url}: {str(e)}") processed_images.append({ "original_url": url, "status": "failed", "error": str(e) }) return { "order_id": order_id, "processed_count": len(processed_images), "images": processed_images } @staticmethod def send_notification(payload: Dict[str, Any]) -> Dict[str, Any]: """Send email notification""" order_id = payload["order_id"]