Distributed Experiment Project

A lightweight distributed job queue system with worker heartbeat monitoring.

Quick Start

CLI (Typer):

distributed-experiment --jobs 100 --batch-size 10

Server:

from distributed_experiment.server import create_app
import uvicorn

job_ids = list(range(1, 101))
app = create_app(job_ids)
uvicorn.run(app, host="0.0.0.0", port=8000)

Worker:

from distributed_experiment.worker import Worker

def process_job(job_id: int):
    # Your job processing logic here
    result = job_id * 2
    # Save result to disk/database/etc

with Worker() as worker:
    while True:
        job_ids = worker.request_job()  # Returns [1, 2, 3, ...] or []
        if not job_ids:
            break
        for job_id in job_ids:
            process_job(job_id)
        worker.submit_jobs(job_ids)

How It Works

1. Server initializes with a list of job IDs and persists progress to disk
2. Workers request available jobs in batches, process them, and submit results
3. Dead worker detection automatically requeues jobs if a worker stops sending heartbeats
4. Restart safe — only incomplete jobs are requeued on server restart
5. Progress tracking — tqdm progress bars show job completion on server and workers

API

POST /request_jobs - Get available jobs

• Request: {"machine_id": int}
• Response: {"status": "ok|empty", "job_ids": [...]}

POST /submit_jobs - Submit completed jobs

• Request: {"machine_id": int, "job_ids": [...]}
• Response: {"status": "ok", "completed": N}

POST /update_heartbeat - Keep worker alive

• Request: {"machine_id": int|null}
• Response: {"status": "ok", "machine_id": int}

Configuration

CLI command:

distributed-experiment \
    --jobs 10 \
    --host 0.0.0.0 \
    --port 8000 \
    --db-path store.db \
    --worker-timeout-seconds 60 \
    --batch-size 10

CLI description:

• distributed-experiment: Starts the FastAPI server with job IDs generated by list(range(jobs)).
• --jobs: Number of IDs to generate.
• --batch-size: Maximum jobs handed out per /request_jobs call.

Server:

create_app(
    job_ids=[1, 2, 3, ...],      # Jobs to process
    db_path="store.db",          # Persistence file
    worker_timeout_seconds=60,   # Dead worker timeout
    batch_size=10                # Jobs per worker request
)

Worker:

Worker(
    server_url="http://localhost:8000",
    machine_id=None,                    # Auto-generated if None
    heartbeat_interval_seconds=5.0,
    request_timeout_seconds=10.0,
    show_progress=True                  # Display tqdm progress bar
)

Exposing with Cloudflare Quick Tunnel

If you want the easiest setup from a terminal, use Cloudflare Quick Tunnel. This does not require tunnel create, credentials JSON, or a config file.

1. Start your API server locally:

distributed-experiment --jobs 10 --host 0.0.0.0 --port 8000

2. In another terminal, start a Quick Tunnel:

cloudflared tunnel --url http://localhost:8000

cloudflared will print a public URL like https://random-name.trycloudflare.com.

3. Connect workers using that URL:

with Worker(server_url="https://random-name.trycloudflare.com") as worker:
    while True:
        job_ids = worker.request_job()
        if not job_ids:
            break
        for job_id in job_ids:
            process_job(job_id)
        worker.submit_jobs(job_ids)

Notes:

• Keep the cloudflared tunnel --url ... process running while workers are active.
• The Quick Tunnel URL is temporary and may change each time you restart it.
• If cloudflared is missing, install it first with your package manager.

Architecture

• Store: Thread-safe shelve-based persistent storage
• JobQueue: Manages job distribution and dead worker detection
• Worker: Simple client that auto-maintains background heartbeat
• CLI: Typer-based commands to run the server with generated job IDs