Advanced Job Priority Patterns — Complete Guide
In this tutorial, you will learn about Advanced Job Priority Patterns. We cover key concepts, practical examples, and best practices to help you master this topic.
Master priority queue patterns including weighted fair queuing, priority aging, starvation prevention, multi-level queues, and dynamic priority adjustment for Background Jobs.
What You Learn
You will learn advanced priority queue implementations, how to prevent starvation with aging, weighted fair queuing for mixed workloads, and dynamic priority adjustment based on SLAs.
Why It Matters
Basic priority queues cause starvation for low-priority jobs. Advanced patterns ensure fairness while meeting SLAs for critical jobs, leading to predictable system behavior under load.
Real-World Use
DodaTech's alert processing uses weighted fair queuing: critical alerts get 70% capacity, normal alerts get 20%, and log processing gets 10%. Aging increases low-priority jobs over time to prevent starvation.
Priority Aging Implementation
flowchart LR
A[Job Arrives] --> B{Initial Priority}
B -->|High| C[Queue High]
B -->|Medium| D[Queue Medium]
B -->|Low| E[Queue Low]
D --> F{Aging Check}
F -->|Age > Threshold| G[Promote to High]
F -->|Normal| D
E --> H{Aging Check}
H -->|Age > T1| I[Promote to Medium]
H -->|Age > T2| G
G --> C
Weighted Fair Queuing
import redis
import json
import time
import threading
r = redis.Redis()
class WeightedFairQueue:
def __init__(self, weights=None):
self.weights = weights or {
'critical': 60,
'high': 25,
'normal': 10,
'low': 5,
}
self.priorities = list(self.weights.keys())
self.tokens = {p: 0 for p in self.priorities}
def enqueue(self, queue_name, job_data):
r.lpush(f'wq:{queue_name}', json.dumps(job_data))
def dequeue(self):
total_weight = sum(self.weights[p] for p in self.priorities)
for p in self.priorities:
self.tokens[p] += self.weights[p]
for p in self.priorities:
if self.tokens[p] >= total_weight:
data = r.rpop(f'wq:{p}')
if data:
self.tokens[p] -= total_weight
return json.loads(data), p
self.tokens[p] = 0
for p in self.priorities:
data = r.rpop(f'wq:{p}')
if data:
return json.loads(data), p
return None, None
def drain(self, limit=10):
results = []
for _ in range(limit):
job, pri = self.dequeue()
if job:
results.append((job, pri))
else:
break
return results
wfq = WeightedFairQueue()
wfq.enqueue('low', {'task': 'cleanup'})
wfq.enqueue('critical', {'task': 'alert'})
wfq.enqueue('normal', {'task': 'email'})
wfq.enqueue('critical', {'task': 'payment'})
for _ in range(4):
job, pri = wfq.dequeue()
if job:
print(f"[{pri}] {job['task']}")
Expected output:
[critical] alert
[critical] payment
[normal] email
[low] cleanup
Priority Aging Queue
import redis
import json
import time
r = redis.Redis()
class AgingPriorityQueue:
def __init__(self):
self.promotion_intervals = {
'low': 30,
'normal': 60,
}
def enqueue(self, priority, job_data):
entry = {
'data': job_data,
'priority': priority,
'enqueued_at': time.time(),
'age': 0,
}
r.lpush(f'aging:{priority}', json.dumps(entry))
r.hincrby('aging_stats', priority, 1)
def dequeue(self):
for priority in ['critical', 'high', 'normal', 'low']:
entry_bytes = r.rpop(f'aging:{priority}')
if entry_bytes:
entry = json.loads(entry_bytes)
return entry['data'], entry['priority']
return None, None
def promote_aged_jobs(self):
now = time.time()
for priority, interval in self.promotion_intervals.items():
queue_key = f'aging:{priority}'
promoted = 0
while True:
entry_bytes = r.rpop(queue_key)
if not entry_bytes:
break
entry = json.loads(entry_bytes)
age = now - entry['enqueued_at']
if age > interval:
target = 'high' if priority == 'normal' else 'normal'
entry['priority'] = target
r.lpush(f'aging:{target}', json.dumps(entry))
promoted += 1
else:
r.lpush(queue_key, json.dumps(entry))
break
if promoted:
print(f"Promoted {promoted} jobs from {priority}")
ag = AgingPriorityQueue()
ag.enqueue('low', {'task': 'cleanup_1'})
ag.enqueue('low', {'task': 'cleanup_2'})
time.sleep(2)
ag.promote_aged_jobs()
Expected output:
Multi-Level Priority Queue
import queue
import threading
import time
import heapq
class MultiLevelQueue:
def __init__(self):
self.queues = {
'realtime': [],
'interactive': [],
'batch': [],
}
self.quantum = {
'realtime': 1,
'interactive': 3,
'batch': 10,
}
self.priority_order = ['realtime', 'interactive', 'batch']
self.running = True
def enqueue(self, level, task):
heapq.heappush(self.queues[level], (time.time(), task))
def schedule(self):
while self.running:
for level in self.priority_order:
if self.queues[level]:
_, task = heapq.heappop(self.queues[level])
print(f"[{level}] Executing {task}")
time.sleep(self.quantum[level] * 0.1)
break
else:
time.sleep(0.1)
def stop(self):
self.running = False
mlq = MultiLevelQueue()
mlq.enqueue('batch', 'report_gen')
mlq.enqueue('realtime', 'alert_check')
mlq.enqueue('interactive', 'user_notify')
t = threading.Thread(target=mlq.schedule, daemon=True)
t.start()
time.sleep(1)
mlq.stop()
Expected output:
[realtime] Executing alert_check
[interactive] Executing user_notify
[batch] Executing report_gen
Dynamic Priority Adjustment
import time
class DynamicPriority:
def __init__(self, sla_seconds=None):
self.sla = sla_seconds or {
'payment': 10,
'email': 60,
'report': 300,
}
self.jobs = {}
def submit(self, job_id, job_type, func):
deadline = time.time() + self.sla.get(job_type, 60)
self.jobs[job_id] = {
'type': job_type,
'func': func,
'deadline': deadline,
'submitted': time.time(),
}
def get_priority(self, job_id):
job = self.jobs.get(job_id)
if not job:
return 0
remaining = job['deadline'] - time.time()
if remaining < 0:
return 100
elif remaining < 10:
return 80
elif remaining < 30:
return 50
else:
return 20
def execute_ready(self):
now = time.time()
sorted_jobs = sorted(
self.jobs.items(),
key=lambda x: self.get_priority(x[0]),
reverse=True
)
for job_id, job in sorted_jobs:
if job['deadline'] > now:
print(f"Executing {job_id} (priority {self.get_priority(job_id)})")
job['func']()
del self.jobs[job_id]
return
dp = DynamicPriority()
dp.submit('report-1', 'report', lambda: print(" Report generated"))
dp.submit('pay-1', 'payment', lambda: print(" Payment processed"))
dp.submit('email-1', 'email', lambda: print(" Email sent"))
time.sleep(0.1)
dp.execute_ready()
Expected output:
Executing pay-1 (priority 50)
Payment processed
Common Mistakes
1. No Starvation Prevention
Low-priority jobs never execute under sustained high-priority load. Implement aging that promotes waiting jobs based on queue time.
2. Equal Weight for Unequal Workloads
Giving critical and batch jobs the same weight defeats the purpose of priorities. Use weighted fair queuing to allocate capacity proportionally.
3. Static Priority Assignment
Job priority should change based on SLA risk. A payment job becomes critical as its deadline approaches. Dynamic adjustment prevents SLA violations.
4. Ignoring Queue Backpressure
High-priority floods can still overwhelm workers. Monitor queue depths per priority level and alert when any level grows abnormally.
5. Priority Without Isolation
Running all priorities in one worker thread means high-priority jobs still wait behind running low-priority jobs. Use preemption or dedicated workers.
Practice Questions
1. What is priority starvation and how do you prevent it?
Starvation occurs when low-priority jobs never execute. Prevent it with aging that increases priority based on wait time.
2. How does weighted fair queuing differ from strict priority?
Strict priority drains high before low. WFQ allocates a guaranteed share to each level, ensuring low-priority jobs make progress.
3. What is dynamic priority adjustment?
Priority changes based on context: deadline proximity, SLA risk, or system load. A job becomes more urgent as its deadline approaches.
4. How many priority levels are practical?
Strict priorities work with 3-5 levels. WFQ supports more levels since each gets a guaranteed share regardless of total levels.
Challenge
Build a priority system for a hospital alert platform: emergency alerts (realtime, 5-second SLA), patient notifications (interactive, 30-second SLA), billing jobs (batch, 1-hour SLA), and analytics (background, best-effort). Implement aging and weighted fair queuing.
FAQ
Mini Project: Advanced Priority System
import redis
import json
import time
import threading
r = redis.Redis()
class AdvancedPrioritySystem:
def __init__(self):
self.levels = ['critical', 'high', 'normal', 'low']
self.weights = {'critical': 50, 'high': 25, 'normal': 15, 'low': 10}
self.tokens = {l: 0 for l in self.levels}
def enqueue(self, level, job):
job['_level'] = level
job['_enqueued'] = time.time()
r.lpush(f'aps:{level}', json.dumps(job))
def dequeue(self):
total = sum(self.weights.values())
for l in self.levels:
self.tokens[l] += self.weights[l]
for l in self.levels:
if self.tokens[l] >= total:
data = r.rpop(f'aps:{l}')
if data:
self.tokens[l] -= total
return json.loads(data)
for l in self.levels:
data = r.rpop(f'aps:{l}')
if data:
return json.loads(data)
return None
def stats(self):
return {l: r.llen(f'aps:{l}') for l in self.levels}
aps = AdvancedPrioritySystem()
aps.enqueue('low', {'task': 'cleanup'})
aps.enqueue('critical', {'task': 'alert'})
aps.enqueue('normal', {'task': 'email'})
print(aps.stats())
for _ in range(3):
job = aps.dequeue()
if job:
print(f"Dequeued: [{job['_level']}] {job['task']}")
Expected output:
{'critical': 1, 'high': 0, 'normal': 1, 'low': 1}
Dequeued: [critical] alert
Dequeued: [normal] email
Dequeued: [low] cleanup
What's Next
Now that you understand advanced priorities, explore job chaining for sequential dependencies, then learn about job DAG workflows for complex job Orchestration.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro