Distributed Workers and Multi-Node Processing
In this tutorial, you will learn about Distributed Workers and Multi. We cover key concepts, practical examples, and best practices to help you master this topic.
Distribute background job processing across multiple worker nodes using shared queues, worker coordination, and horizontal scaling strategies.
What You Learn
You will learn how to run multiple worker processes across different machines, coordinate work distribution with Redis, handle worker failures, and scale horizontally by adding more worker nodes.
Why It Matters
A single worker cannot handle all jobs for a growing application. Processing 10,000 image uploads per minute requires multiple workers running in parallel across multiple servers. Distributed workers provide the throughput your application needs.
Real-World Use
DodaTech's file scanning service runs 50 workers across 10 machines. When a user uploads a file, the job enters a shared Redis queue. Any available worker picks it up. If one machine fails, other workers continue processing. Adding more machines increases throughput linearly.
Multi-Worker Architecture
import redis
import json
import time
import threading
import random
r = redis.Redis()
class DistributedWorker:
def __init__(self, worker_id, queue='shared_jobs'):
self.worker_id = worker_id
self.queue = queue
self.running = True
def process_job(self, job):
print(f"Worker {self.worker_id} processing: {job['task']} ({job['id']})")
time.sleep(random.uniform(0.1, 0.5))
return True
def start(self):
print(f"Worker {self.worker_id} started")
while self.running:
job_data = r.brpop(self.queue, timeout=2)
if job_data:
queue_name, data = job_data
job = json.loads(data)
try:
self.process_job(job)
except Exception as e:
print(f"Worker {self.worker_id} error: {e}")
print(f"Worker {self.worker_id} stopped")
def stop(self):
self.running = False
def simulate_load():
for i in range(5):
r.lpush('shared_jobs', json.dumps({'id': f'job-{i}', 'task': 'scan'}))
simulate_load()
workers = []
for wid in ['A', 'B', 'C']:
w = DistributedWorker(wid)
t = threading.Thread(target=w.start, daemon=True)
t.start()
workers.append(w)
time.sleep(4)
for w in workers:
w.stop()
time.sleep(1)
Expected output:
Worker A started
Worker B started
Worker C started
Worker A processing: scan (job-0)
Worker B processing: scan (job-1)
Worker C processing: scan (job-2)
Worker A processing: scan (job-3)
Worker B processing: scan (job-4)
Worker Registration and Heartbeat
import redis
import time
import json
import threading
import uuid
r = redis.Redis()
class HeartbeatWorker:
def __init__(self, worker_id=None, ttl=10):
self.worker_id = worker_id or str(uuid.uuid4())[:8]
self.ttl = ttl
self.running = True
self.registry_key = 'workers:active'
def register(self):
worker_info = {
'id': self.worker_id,
'started_at': time.time(),
'status': 'active',
'jobs_processed': 0,
}
r.hset(self.registry_key, self.worker_id, json.dumps(worker_info))
r.expire(self.registry_key, self.ttl * 2)
def send_heartbeat(self):
while self.running:
self.register()
time.sleep(self.ttl // 2)
def update_job_count(self):
data = r.hget(self.registry_key, self.worker_id)
if data:
info = json.loads(data)
info['jobs_processed'] += 1
r.hset(self.registry_key, self.worker_id, json.dumps(info))
def start(self):
self.register()
t = threading.Thread(target=self.send_heartbeat, daemon=True)
t.start()
return self.worker_id
@staticmethod
def get_active_workers():
workers = r.hgetall('workers:active')
active = []
for wid, data in workers.items():
info = json.loads(data)
active.append(info)
return active
hb = HeartbeatWorker('worker-1')
hb.start()
time.sleep(2)
active = HeartbeatWorker.get_active_workers()
print(f"Active workers: {len(active)}")
for w in active:
print(f" {w['id']}: {w['status']}")
Work Stealing
import redis
import json
import time
import random
r = redis.Redis()
class WorkStealingPool:
def __init__(self, worker_id, queues):
self.worker_id = worker_id
self.primary_queue = queues[worker_id]
self.all_queues = list(queues.values())
def get_job(self, timeout=2):
job_data = r.brpoplpush(
self.primary_queue,
f"{self.primary_queue}:processing",
timeout=timeout
)
if job_data:
return json.loads(job_data)
for q in self.all_queues:
if q == self.primary_queue:
continue
job_data = r.rpoplpush(q, f"{q}:processing")
if job_data:
job = json.loads(job_data)
print(f"Worker {self.worker_id} stole job from {q}")
return job
return None
def process(self):
job = self.get_job(timeout=1)
if job:
print(f" Processing: {job['task']}")
return True
return False
queues = {
'W1': 'queue:w1',
'W2': 'queue:w2',
'W3': 'queue:w3',
}
r.lpush('queue:w1', json.dumps({'task': 'heavy_scan', 'id': 'j1'}))
r.lpush('queue:w1', json.dumps({'task': 'heavy_scan', 'id': 'j2'}))
r.lpush('queue:w1', json.dumps({'task': 'heavy_scan', 'id': 'j3'}))
pool = WorkStealingPool('W2', queues)
for _ in range(4):
pool.process()
time.sleep(0.3)
Expected output:
Worker W2 stole job from queue:w1
Processing: heavy_scan
Worker W2 stole job from queue:w1
Processing: heavy_scan
Worker W2 stole job from queue:w1
Processing: heavy_scan
Worker Failure Detection
import redis
import time
import json
r = redis.Redis()
class FailureDetector:
def __init__(self, timeout=5):
self.timeout = timeout
def check_worker(self, worker_id):
data = r.hget('workers:active', worker_id)
if not data:
return {'status': 'unknown'}
info = json.loads(data)
elapsed = time.time() - info.get('started_at', 0)
if elapsed > self.timeout and info['status'] == 'active':
return {'status': 'suspected', 'worker': worker_id, 'elapsed': elapsed}
return {'status': 'healthy', 'worker': worker_id}
def find_dead_workers(self):
workers = r.hgetall('workers:active')
dead = []
for wid, data in workers.items():
wid = wid.decode()
status = self.check_worker(wid)
if status['status'] == 'suspected':
dead.append(wid)
return dead
def rebalance(self, dead_worker_id):
queue = f'queue:{dead_worker_id}'
jobs = r.lrange(queue, 0, -1)
if jobs:
r.rpush('shared_jobs', *jobs)
r.delete(queue)
print(f"Rebalanced {len(jobs)} jobs from dead worker {dead_worker_id}")
detector = FailureDetector(timeout=3)
r.hset('workers:active', 'worker-1', json.dumps({
'id': 'worker-1', 'started_at': time.time() - 10, 'status': 'active'
}))
r.lpush('queue:worker-1', json.dumps({'task': 'orphaned_job'}))
dead = detector.find_dead_workers()
for wid in dead:
detector.rebalance(wid)
Expected output:
Rebalanced 1 jobs from dead worker worker-1
Scaling Workers Dynamically
import redis
import time
import json
import threading
import random
r = redis.Redis()
class AutoScaler:
def __init__(self, min_workers=2, max_workers=10, queue_length_threshold=20):
self.min_workers = min_workers
self.max_workers = max_workers
self.threshold = queue_length_threshold
self.active_workers = {}
def get_queue_length(self):
return r.llen('shared_jobs')
def spawn_worker(self, worker_id):
worker = DistributedWorker(worker_id, 'shared_jobs')
thread = threading.Thread(target=worker.start, daemon=True)
thread.start()
self.active_workers[worker_id] = {'worker': worker, 'thread': thread}
print(f"Spawned worker: {worker_id}")
def stop_worker(self, worker_id):
if worker_id in self.active_workers:
self.active_workers[worker_id]['worker'].stop()
del self.active_workers[worker_id]
print(f"Stopped worker: {worker_id}")
def scale(self):
queue_len = self.get_queue_length()
current = len(self.active_workers)
if queue_len > self.threshold and current < self.max_workers:
new_worker_id = f"auto-{len(self.active_workers) + 1}"
self.spawn_worker(new_worker_id)
elif queue_len < self.threshold // 2 and current > self.min_workers:
worker_id = list(self.active_workers.keys())[-1]
self.stop_worker(worker_id)
return current, queue_len
def run(self):
while True:
before, qlen = self.scale()
time.sleep(3)
scaler = AutoScaler(min_workers=1, max_workers=3, queue_length_threshold=3)
for i in range(10):
r.lpush('shared_jobs', json.dumps({'id': f'load-{i}', 'task': 'process'}))
scaler.scale()
time.sleep(1)
scaler.scale()
time.sleep(1)
r.delete('shared_jobs')
for wid in list(scaler.active_workers.keys()):
scaler.stop_worker(wid)
Common Mistakes
1. No Shared Queue
Each worker has its own queue, so idle workers cannot help busy ones. Use a shared queue so any worker can pick up any job.
2. Sticky Jobs Without Load Balancing
Assigning jobs to specific workers by hash leads to uneven distribution. Random or round-robin assignment is simpler and more balanced.
3. Ignoring Worker Heartbeats
Without heartbeats, you cannot detect failed workers. Dead workers leave jobs unprocessed. Implement heartbeats with timeouts.
4. No Rebalancing on Failure
When a worker dies, its in-flight and queued jobs become orphaned. Rebalance them to surviving workers.
5. Manual Scaling
Adding and removing workers requires manual intervention. Implement autoscaling based on queue depth for elasticity.
Practice Questions
1. How do workers coordinate in a distributed setup?
Through a shared queue (Redis, RabbitMQ). Workers independently poll the queue and pick up available jobs. No direct worker-to-worker communication is needed.
2. What is a worker heartbeat?
A periodic signal from the worker to a central registry indicating it is alive. If the heartbeat stops, the worker is presumed dead.
3. How do you handle a worker crash?
The shared queue system automatically reassigns unacknowledged jobs to other workers. Orphaned jobs in worker-local queues need rebalancing.
4. What is work stealing?
Idle workers take jobs from busy workers' queues, improving overall throughput when load distribution is uneven.
Challenge
Build a distributed worker system for video transcoding: 10 workers across 3 machines, shared Redis queue, heartbeat every 5 seconds with 15-second timeout, autoscale between 5-20 workers based on queue depth, rebalance orphaned jobs on worker failure, and track per-worker metrics (jobs processed, average processing time, failure rate).
FAQ
Mini Project: Distributed Worker Pool
import redis
import json
import time
import threading
import uuid
import random
r = redis.Redis()
class WorkerPool:
def __init__(self, pool_name='pool', queue='shared'):
self.pool_name = pool_name
self.queue = queue
self.workers = {}
def register_worker(self, worker_id=None):
wid = worker_id or str(uuid.uuid4())[:8]
info = {'id': wid, 'pool': self.pool_name, 'started': time.time(), 'jobs': 0}
r.hset(f'pool:{self.pool_name}', wid, json.dumps(info))
return wid
def unregister_worker(self, wid):
r.hdel(f'pool:{self.pool_name}', wid)
def get_queue_length(self):
return r.llen(self.queue)
def list_workers(self):
data = r.hgetall(f'pool:{self.pool_name}')
result = []
for wid, info in data.items():
result.append(json.loads(info))
return result
def worker_heartbeat(self, wid, ttl=10):
info = {'id': wid, 'pool': self.pool_name, 'started': time.time(), 'ttl': ttl}
r.hset(f'pool:{self.pool_name}', wid, json.dumps(info))
r.expire(f'pool:{self.pool_name}', ttl * 2)
def start_worker(self, wid=None, heartbeat_interval=5):
wid = wid or str(uuid.uuid4())[:8]
def run():
self.register_worker(wid)
while True:
job_data = r.brpop(self.queue, timeout=2)
if job_data:
_, data = job_data
job = json.loads(data)
print(f"[{wid}] Processing: {job.get('task', 'unknown')}")
time.sleep(random.uniform(0.1, 0.3))
self.worker_heartbeat(wid)
thread = threading.Thread(target=run, daemon=True)
thread.start()
self.workers[wid] = thread
return wid
def scale_to(self, target):
current = len(self.workers)
if target > current:
for _ in range(target - current):
self.start_worker()
return len(self.workers)
pool = WorkerPool('scan-pool', 'scan_jobs')
for i in range(6):
r.lpush('scan_jobs', json.dumps({'task': 'scan', 'id': f'job-{i}'}))
pool.scale_to(3)
time.sleep(4)
workers = pool.list_workers()
print(f"Active workers: {len(workers)}")
qlen = pool.get_queue_length()
print(f"Queue remaining: {qlen}")
Expected output:
[<id>] Processing: scan
[<id>] Processing: scan
[<id>] Processing: scan
[<id>] Processing: scan
[<id>] Processing: scan
[<id>] Processing: scan
Active workers: 3
Queue remaining: 0
What's Next
Now that you understand distributed workers, explore worker scalability for handling increased load, then build the background jobs mini project to apply everything you have learned.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro