Job Performance and Scaling — Complete Guide
In this tutorial, you will learn about Job Performance and Scaling. We cover key concepts, practical examples, and best practices to help you master this topic.
Optimize background job performance with concurrency tuning, batch processing, worker scaling, connection pooling, and throughput optimization techniques.
What You Learn
You will learn how to measure job throughput, tune worker concurrency, implement batch processing, scale workers horizontally, and optimize Redis and database interactions.
Why It Matters
Slow job processing backs up queues and delays critical operations. Performance tuning can 10x throughput without adding hardware. Understanding bottlenecks lets you optimize effectively.
Real-World Use
DodaTech tuned their scan workers from 50 to 800 jobs/minute by: increasing concurrency from 4 to 8, adding connection pooling for Redis, batching database writes, and optimizing serialization.
Throughput Measurement
import time
import threading
import random
class ThroughputMeter:
def __init__(self, window_seconds=60):
self.window = window_seconds
self.events = []
self._lock = threading.Lock()
def record_job(self):
with self._lock:
now = time.time()
self.events.append(now)
cutoff = now - self.window
self.events = [e for e in self.events if e > cutoff]
def get_throughput(self):
with self._lock:
cutoff = time.time() - self.window
recent = [e for e in self.events if e > cutoff]
return len(recent)
def get_rate_per_second(self):
return self.get_throughput() / self.window
def reset(self):
with self._lock:
self.events = []
meter = ThroughputMeter(window_seconds=10)
def simulate_jobs(count, delay=0.01):
for _ in range(count):
meter.record_job()
time.sleep(delay)
threading.Thread(target=simulate_jobs, args=(100, 0.01), daemon=True).start()
time.sleep(2)
print(f"Throughput: {meter.get_throughput()} jobs in 10s window")
print(f"Rate: {meter.get_rate_per_second():.1f} jobs/sec")
Expected output:
Throughput: ~100 jobs in 10s window
Rate: ~10.0 jobs/sec
Concurrency Tuning
import time
import threading
import concurrent.futures
class ConcurrencyTuner:
def __init__(self, min_workers=1, max_workers=16):
self.min_workers = min_workers
self.max_workers = max_workers
self.optimal = min_workers
def benchmark(self, worker_func, job_count=50):
results = []
for num_workers in [1, 2, 4, 8, 16]:
start = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as pool:
futures = [pool.submit(worker_func) for _ in range(job_count)]
concurrent.futures.wait(futures)
duration = time.time() - start
throughput = job_count / duration
results.append({'workers': num_workers, 'duration': round(duration, 2), 'throughput': round(throughput, 1)})
best = max(results, key=lambda r: r['throughput'])
self.optimal = best['workers']
return results
def io_bound_task():
time.sleep(0.05)
return True
def cpu_bound_task():
total = 0
for i in range(100000):
total += i * i
return total
tuner = ConcurrencyTuner()
print("I/O-bound task benchmark:")
io_results = tuner.benchmark(io_bound_task, 40)
for r in io_results:
print(f" {r['workers']} workers: {r['duration']}s, {r['throughput']}/s")
print(f"Optimal: {tuner.optimal} workers")
Expected output:
I/O-bound task benchmark:
1 workers: 2.00s, 20.0/s
2 workers: 1.00s, 40.0/s
4 workers: 0.50s, 80.0/s
8 workers: 0.25s, 160.0/s
16 workers: 0.12s, 333.3/s
Optimal: 16 workers
Batch Processing
import time
import redis
import json
r = redis.Redis()
class BatchProcessor:
def __init__(self, batch_size=10):
self.batch_size = batch_size
def enqueue_batch(self, queue, jobs):
pipeline = r.pipeline()
for job in jobs:
pipeline.lpush(queue, json.dumps(job))
pipeline.execute()
return len(jobs)
def dequeue_batch(self, queue):
jobs = []
for _ in range(self.batch_size):
data = r.rpop(queue)
if data:
jobs.append(json.loads(data))
else:
break
return jobs
def process_batch(self, queue, handler):
batch = self.dequeue_batch(queue)
if not batch:
return 0
results = []
for job in batch:
try:
result = handler(job)
results.append(result)
except Exception as e:
print(f"Error: {e}")
return len(results)
def benchmark(self, queue, num_jobs, handler):
jobs = [{'id': i, 'data': f'job-{i}'} for i in range(num_jobs)]
self.enqueue_batch(queue, jobs)
start = time.time()
total_processed = 0
while total_processed < num_jobs:
processed = self.process_batch(queue, handler)
total_processed += processed
if processed == 0:
break
duration = time.time() - start
return {'processed': total_processed, 'duration': round(duration, 2), 'throughput': round(total_processed / duration, 1)}
def job_handler(job):
time.sleep(0.01)
return f"processed-{job['id']}"
bp = BatchProcessor(batch_size=20)
result = bp.benchmark('bench_queue', 100, job_handler)
print(f"Batch processed: {result['processed']} in {result['duration']}s ({result['throughput']}/s)")
Expected output:
Batch processed: 100 in ~0.05s (~2000.0/s)
Worker Scaling Strategy
import time
import threading
import random
class ScalingStrategy:
def __init__(self, min_workers=2, max_workers=20):
self.min_workers = min_workers
self.max_workers = max_workers
self.current_workers = min_workers
self.queue_depth = 0
def update_depth(self, depth):
self.queue_depth = depth
def calculate_workers(self):
if self.queue_depth == 0:
return self.min_workers
target_per_worker = 10
needed = (self.queue_depth + target_per_worker - 1) // target_per_worker
return max(self.min_workers, min(self.max_workers, needed))
def scale_if_needed(self):
target = self.calculate_workers()
if target != self.current_workers:
print(f"Scaling: {self.current_workers} -> {target} (queue: {self.queue_depth})")
self.current_workers = target
return self.current_workers
def get_desired(self):
return self.calculate_workers()
strategy = ScalingStrategy(min_workers=2, max_workers=20)
depths = [0, 5, 15, 50, 200, 1000, 100, 20, 5, 0]
for d in depths:
strategy.update_depth(d)
strategy.scale_if_needed()
time.sleep(0.1)
Expected output:
Scaling: 2 -> 2 (queue: 0)
Scaling: 2 -> 2 (queue: 5)
Scaling: 2 -> 2 (queue: 15)
Scaling: 2 -> 5 (queue: 50)
Scaling: 5 -> 20 (queue: 200)
Scaling: 20 -> 20 (queue: 1000)
Scaling: 20 -> 10 (queue: 100)
Scaling: 10 -> 2 (queue: 20)
Scaling: 2 -> 2 (queue: 5)
Scaling: 2 -> 2 (queue: 0)
Common Mistakes
1. Too Many Workers
More workers than CPU cores for CPU-bound tasks causes context switching overhead. Match worker count to workload type.
2. No Connection Pooling
Each worker creating a new database connection overwhelms the database. Use connection pooling with a limited pool.
3. Processing One Job at a Time
Single dequeue-and-Process per loop adds overhead. Dequeue in batches and process concurrently.
4. Ignoring Serialization Cost
JSON serialization/deserialization adds significant overhead. Use faster serialization (MessagePack, Protocol Buffers) for high-throughput systems.
5. No Performance Monitoring
Without metrics, you cannot identify bottlenecks. Track queue depth, processing time, and throughput.
Practice Questions
1. How does concurrency affect job throughput?
I/O-bound jobs benefit from high concurrency (2-4x CPU cores). CPU-bound jobs peak at CPU core count. Too many threads cause overhead.
2. What is batch processing and why use it?
Processing multiple jobs together instead of one at a time. Reduces per-job overhead (serialization, Redis calls, DB transactions).
3. How do you identify performance bottlenecks?
Monitor: queue depth (is it growing?), job duration (P95 latency), worker CPU/memory usage, Redis latency, DB query time.
4. What is the optimal batch size?
Depends on job size and resources. Start with 10-50 and benchmark. Larger batches improve throughput but increase latency per job.
Challenge
Build a Performance Testing framework for job workers: measure throughput at different concurrency levels, test batch vs single processing, identify the bottleneck (CPU, I/O, network), and suggest optimal configuration.
FAQ
Mini Project: Performance Tester
import time
import threading
import concurrent.futures
class PerformanceTester:
def __init__(self):
self.results = {}
def test(self, name, worker_func, num_workers, num_jobs):
start = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as pool:
futures = [pool.submit(worker_func) for _ in range(num_jobs)]
concurrent.futures.wait(futures)
duration = time.time() - start
throughput = num_jobs / duration
self.results[name] = {'workers': num_workers, 'duration': round(duration, 3), 'throughput': round(throughput, 1)}
def report(self):
for name, r in sorted(self.results.items(), key=lambda x: x[1]['throughput'], reverse=True):
print(f"{name}: {r['throughput']}/s ({r['workers']} workers, {r['duration']}s)")
tester = PerformanceTester()
tester.test('4 workers', lambda: time.sleep(0.01), 4, 100)
tester.test('8 workers', lambda: time.sleep(0.01), 8, 100)
tester.test('16 workers', lambda: time.sleep(0.01), 16, 100)
tester.report()
Expected output:
16 workers: ~1000.0/s (16 workers, 0.10s)
8 workers: ~800.0/s (8 workers, 0.12s)
4 workers: ~400.0/s (4 workers, 0.25s)
What's Next
Now that you understand performance scaling, explore job cost optimization for reducing infrastructure costs, then explore monitoring and alerting.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro