Parallelization Engine

Production-grade distributed task execution engine built with Docker, Celery, RabbitMQ, and Redis. Designed for scalable, fault-tolerant parallel processing of any workload.

Architecture

                         ┌─────────────────────────────────────┐
                         │           RabbitMQ Broker            │
  CLI / Python API ────► │  Queues: default | cpu | io | dlq   │ ◄── Management UI :15672
                         └──────────────┬──────────────────────┘
                                        │
                    ┌───────────────────┼───────────────────┐
                    ▼                   ▼                   ▼
              ┌───────────┐      ┌───────────┐      ┌───────────┐
              │  Worker    │      │  Worker   │      │  Worker   │
              │ (default)  │      │  (cpu)    │      │  (io)     │
              │ conc: 4    │      │ conc: 2   │      │ conc: 8   │
              └─────┬──────┘      └─────┬─────┘      └─────┬─────┘
                    └───────────────────┼───────────────────┘
                                        ▼
                              ┌─────────────────┐
                              │  Redis Backend   │  ◄── Persistent results
                              └─────────────────┘
                                        ▲
                              ┌─────────────────┐
                              │  Flower :5555    │  ◄── Monitoring dashboard
                              └─────────────────┘

Features

• Generalized task system — Plugin-based architecture: drop a .py file in src/tasks/ to register new task types
• Named queue routing — CPU-intensive, I/O-bound, and default queues with independent concurrency tuning
• Fault tolerance — Exponential backoff retries, dead-letter queues, acks_late for crash recovery
• Dynamic scaling — Scale any worker type independently with docker compose up --scale worker=N
• Monitoring — Flower dashboard (:5555) + RabbitMQ management UI (:15672)
• CLI — Submit tasks, batch process files, and check results from the terminal
• Production Docker — Multi-stage builds, non-root user, health checks, resource limits

Quick Start

Prerequisites

• Docker & Docker Compose
• Python 3.11+ (for CLI and tests)

Setup

# Clone and enter the project
cd parallelization-engine

# First-time setup (copies .env, builds images, starts services)
make init

# Or manually:
cp .env.example .env
docker compose build
docker compose up -d

Verify

# Check all services are running
docker compose ps

# RabbitMQ Management UI
open http://localhost:15672  # engine / engine_secret

# Flower Monitoring
open http://localhost:5555   # admin / flower_secret

Submit Tasks

# Install dependencies locally (for CLI)
pip install -r requirements.txt

# CPU-intensive: find primes up to 100,000
python -m cli.submit task src.tasks.cpu_intensive.prime_sieve --args '{"limit": 100000}'

# I/O-bound: fetch a URL
python -m cli.submit task src.tasks.io_bound.fetch_url --args '{"url": "https://httpbin.org/get"}'

# Check result
python -m cli.submit status <task-id>

# Batch: process URLs from file
python -m cli.submit file examples/sample_urls.json --task src.tasks.io_bound.fetch_url

Project Structure

parallelization-engine/
├── src/
│   ├── celery_app.py         # Celery app factory with production config
│   ├── config.py             # Pydantic Settings (broker, backend, worker)
│   ├── logging_config.py     # Structured logging (structlog)
│   ├── callbacks.py          # Task signal handlers (success/failure/retry)
│   ├── routing.py            # Queue declarations, DLQ, task routing
│   ├── tasks/
│   │   ├── base.py           # BaseTask: retries, backoff, timeouts
│   │   ├── cpu_intensive.py  # Prime sieve, matrix multiply
│   │   ├── io_bound.py       # URL fetcher, file downloader
│   │   ├── chained.py        # ETL pipeline (extract → transform → load)
│   │   └── batch.py          # CSV/JSON batch processing with chord
│   └── utils/
│       └── serialization.py  # JSON encoding helpers
├── cli/
│   ├── submit.py             # CLI entrypoint (task, file, status)
│   ├── from_file.py          # Batch submission from CSV/JSON
│   └── status.py             # Task/group result querying
├── tests/                    # Unit tests (no Docker required)
├── examples/                 # Sample data and usage scripts
├── scripts/                  # Health checks, wait-for-it
├── Dockerfile                # Multi-stage, non-root user
├── docker-compose.yml        # Dev: all services
├── docker-compose.prod.yml   # Prod: resource limits, log rotation
└── Makefile                  # Convenience targets

Configuration

All configuration is via environment variables (.env file). See .env.example for all options.

Variable	Default	Description
RABBITMQ_DEFAULT_USER	engine	RabbitMQ username
RABBITMQ_DEFAULT_PASS	engine_secret	RabbitMQ password
REDIS_HOST	redis	Redis hostname
CELERY_WORKER_CONCURRENCY	4	Processes per default worker
CELERY_TASK_SOFT_TIME_LIMIT	300	Soft time limit (seconds)
CELERY_TASK_TIME_LIMIT	600	Hard time limit (seconds)
CELERY_TASK_MAX_RETRIES	3	Max retry attempts
FLOWER_BASIC_AUTH	admin:flower_secret	Flower dashboard credentials

Queue Architecture

Queue	Purpose	Worker Concurrency	Prefetch
default	General-purpose tasks	4	1
cpu_intensive	Compute-heavy tasks (math, ML)	2	1
io_bound	Network/file I/O tasks	8	4
dead_letter	Failed tasks (after all retries)	—	—

Tasks are auto-routed by module path. Override with --queue flag in CLI.

Writing Custom Tasks

1. Create a new file in src/tasks/:

# src/tasks/my_task.py
from src.celery_app import app
from src.tasks.base import BaseTask

@app.task(base=BaseTask, bind=True)
def my_custom_task(self, param1: str, param2: int) -> dict:
    # Your logic here
    return {"result": param1 * param2}

2. Add routing in src/routing.py:

TASK_ROUTES = {
    ...
    "src.tasks.my_task.*": {"queue": "default"},
}

3. Submit:

python -m cli.submit task src.tasks.my_task.my_custom_task --args '{"param1": "hello", "param2": 3}'

Scaling

# Scale default workers to 5 instances
make scale N=5

# Or directly:
docker compose up -d --scale worker=5

# Scale specific worker types
docker compose up -d --scale worker-io=10
docker compose up -d --scale worker-cpu=3

Fault Tolerance

• Automatic retries with exponential backoff + jitter (prevents thundering herd)
• acks_late — tasks are acknowledged only after completion; if a worker crashes, the task is requeued
• Dead-letter queue — tasks that exhaust all retries are routed to dead_letter for inspection
• Task time limits — soft limit raises SoftTimeLimitExceeded; hard limit kills the worker process
• Worker restart — restart: unless-stopped in Docker Compose
• worker_max_tasks_per_child=1000 — workers recycle after 1000 tasks to prevent memory leaks

Monitoring

• Flower at http://localhost:5555 — real-time worker status, task history, active/reserved/revoked counts
• RabbitMQ Management at http://localhost:15672 — queue depths, message rates, connection status
• Structured logs — JSON or console format via LOG_FORMAT env var, written to logs/engine.log

Production Deployment

docker compose -f docker-compose.yml -f docker-compose.prod.yml up -d

Production overrides include:

• Resource limits (CPU, memory) per service
• Log rotation (json-file driver, 10MB max, 3 files)
• No source volume mounts (code baked into image)
• restart: always policy
• Internal-only Redis (no exposed port)

Testing

Tests run in eager mode with an in-memory broker — no Docker required:

pip install -r requirements-dev.txt
make test

# Or directly:
pytest tests/ -v

Built With

• Celery 5.4+ — Distributed task queue
• RabbitMQ 3.13 — Message broker with management UI
• Redis 7 — Result backend with AOF persistence
• Flower 2.0+ — Real-time Celery monitoring
• Pydantic Settings — Type-safe configuration
• structlog — Structured logging
• Docker — Containerized deployment

License

MIT