Job Cancellation and Revocation — Complete Guide
In this tutorial, you will learn about Job Cancellation and Revocation. We cover key concepts, practical examples, and best practices to help you master this topic.
Cancel running or pending jobs using revocation flags, job IDs, and queue management techniques to stop unwanted or stuck background tasks.
What You Learn
You will learn how to cancel pending jobs before they execute, revoke running jobs mid-execution, implement cancellation tokens, and handle cleanup after job cancellation.
Why It Matters
Users cancel operations, deployments happen, and bugs are discovered. Without cancellation, jobs continue running after they are no longer needed, wasting resources and potentially corrupting data. Proper cancellation prevents these outcomes.
Real-World Use
DodaTech's file scanning service allows users to cancel scans. When cancelled, the worker stops processing the current file, removes remaining queue items for that scan, and sends a notification. The system ensures cancelled jobs release database connections properly.
Cancelling Pending Jobs
import redis
import json
import time
r = redis.Redis()
class CancellableQueue:
def __init__(self, queue_name='jobs'):
self.queue_name = queue_name
self.cancelled_key = 'cancelled_jobs'
def enqueue(self, job_data):
job_id = job_data.get('id', str(time.time()))
job_data['id'] = job_id
r.lpush(self.queue_name, json.dumps(job_data))
return job_id
def cancel(self, job_id):
r.sadd(self.cancelled_key, job_id)
print(f"Cancelled job: {job_id}")
def cancel_all(self, filter_key, filter_value):
jobs = [json.loads(j) for j in r.lrange(self.queue_name, 0, -1)]
for job in jobs:
if job.get(filter_key) == filter_value:
r.sadd(self.cancelled_key, job['id'])
print(f"Cancelled all jobs with {filter_key}={filter_value}")
def is_cancelled(self, job_id):
return r.sismember(self.cancelled_key, job_id)
def process(self):
job_data = r.rpop(self.queue_name)
if not job_data:
return None
job = json.loads(job_data)
if self.is_cancelled(job['id']):
print(f"Skipping cancelled job: {job['id']}")
return None
print(f"Processing job: {job['id']}")
return job
q = CancellableQueue()
id1 = q.enqueue({'id': 'job-1', 'task': 'scan_file', 'file': 'doc.pdf'})
id2 = q.enqueue({'id': 'job-2', 'task': 'scan_file', 'file': 'image.jpg'})
q.cancel('job-1')
q.process()
q.process()
Expected output:
Cancelled job: job-1
Skipping cancelled job: job-1
Processing job: job-2
Cancellation Tokens
import threading
import time
class CancellationToken:
def __init__(self):
self._cancelled = False
self._lock = threading.Lock()
def cancel(self):
with self._lock:
self._cancelled = True
@property
def is_cancelled(self):
with self._lock:
return self._cancelled
def process_large_file(file_path, token):
print(f"Starting processing: {file_path}")
for chunk_num in range(1, 11):
if token.is_cancelled:
print(f"Cancelled at chunk {chunk_num}")
cleanup(file_path)
return
print(f" Processing chunk {chunk_num}/10")
time.sleep(0.5)
print(f"Completed: {file_path}")
def cleanup(file_path):
print(f" Cleaning up partial data for {file_path}")
token = CancellationToken()
thread = threading.Thread(
target=process_large_file,
args=('dataset.csv', token)
)
thread.start()
time.sleep(2)
print("User requested cancellation")
token.cancel()
thread.join()
Expected output:
Starting processing: dataset.csv
Processing chunk 1/10
Processing chunk 2/10
Processing chunk 3/10
Processing chunk 4/10
User requested cancellation
Cancelled at chunk 5
Cleaning up partial data for dataset.csv
Revoking Running Jobs
import time
import redis
import json
import threading
r = redis.Redis()
class JobManager:
def __init__(self):
self.active_jobs = {}
def start_job(self, job_id, target, args=()):
thread = threading.Thread(
target=self._run_with_monitoring,
args=(job_id, target, args),
daemon=True
)
self.active_jobs[job_id] = {'thread': thread, 'started': time.time()}
thread.start()
return job_id
def _run_with_monitoring(self, job_id, target, args):
revoke_key = f'revoke:{job_id}'
r.delete(revoke_key)
try:
target(*args)
except Exception as e:
print(f"Job {job_id} error: {e}")
finally:
self.active_jobs.pop(job_id, None)
def revoke(self, job_id):
r.set(f'revoke:{job_id}', '1')
print(f"Revoke signal sent for: {job_id}")
def is_revoked(self, job_id):
return r.exists(f'revoke:{job_id}')
def scan_directory(job_id, path):
for item in ['file1.txt', 'file2.txt', 'file3.txt', 'file4.txt']:
manager = globals().get('manager')
if manager and manager.is_revoked(job_id):
print(f" Job {job_id} revoked, stopping")
return
print(f" Scanning: {item}")
time.sleep(1)
print(f" Scan complete: {path}")
manager = JobManager()
jid = manager.start_job('scan-1', scan_directory, ('scan-1', '/uploads'))
time.sleep(2)
manager.revoke('scan-1')
time.sleep(3)
Expected output:
Scanning: file1.txt
Scanning: file2.txt
Revoke signal sent for: scan-1
Job scan-1 revoked, stopping
Graceful Shutdown with Cancellation
import signal
import time
import threading
class GracefulWorker:
def __init__(self):
self.running = True
self.active_tasks = []
def handle_signal(self, signum, frame):
print(f"\nReceived signal {signum}, shutting down gracefully")
self.running = False
for task in self.active_tasks:
task['token'].cancel()
def run(self):
signal.signal(signal.SIGTERM, self.handle_signal)
signal.signal(signal.SIGINT, self.handle_signal)
print("Worker started. Press Ctrl+C to stop.")
while self.running:
token = CancellationToken()
task = {'name': 'maintenance', 'token': token}
self.active_tasks.append(task)
time.sleep(1)
self.active_tasks.remove(task)
def stop(self):
self.running = False
worker = GracefulWorker()
thread = threading.Thread(target=worker.run, daemon=True)
thread.start()
time.sleep(3)
worker.stop()
print("Worker stopped gracefully")
Cancellation in Celery
# celery_app.py
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379')
@app.task(bind=True)
def long_task(self, data_size):
for i in range(data_size):
if self.is_aborted():
self.update_state(
state='REVOKED',
meta={'progress': i, 'total': data_size}
)
return {'status': 'cancelled', 'progress': i}
# Simulate work
print(f" Processing item {i}/{data_size}")
return {'status': 'completed'}
# Revoke from client
# from celery.task.control import revoke
# revoke('task-id', terminate=True, signal='SIGTERM')
Common Mistakes
1. Not Checking Cancellation Status
Long-running loops never check cancellation flags, making revocation ineffective. Check the flag at least once per iteration or every few seconds.
2. Ignoring Cleanup on Cancellation
Cancelling a job mid-execution leaves partial data, open connections, and locked resources. Always implement cleanup handlers.
3. Cancelling Without Confirmation
A cancel request is sent but never verified. Use tracking (revoked set, database status) to confirm cancellation was received.
4. Race Conditions in Cancellation Flags
Multiple threads read and write cancellation flags without synchronization. Use thread-safe primitives or atomic Redis operations.
5. No Grace Period for Running Jobs
Forcing immediate termination can corrupt data. Allow running jobs a grace period to finish their current operation and clean up.
Practice Questions
1. What is the difference between cancelling and revoking?
Cancelling prevents a pending job from starting. Revoking sends a signal to stop a job that is already running.
2. How does a cancellation token work?
A cancellation token is a shared flag that one thread sets to true and another thread periodically checks. When the flag is true, the worker stops processing.
3. Why is cleanup important after cancellation?
Cancelled jobs may have started writing files, modifying databases, or acquiring locks. Cleanup ensures consistency by rolling back partial work.
4. How do you cancel all jobs for a specific resource?
Use a filter (user ID, file path, scan ID) to find all pending jobs for that resource in the queue and mark them as cancelled in a revoked set.
Challenge
Build a job cancellation system for a batch file processor. Support: cancel single job by ID, cancel all jobs for a user, graceful cancellation with 5-second timeout, automatic cleanup of partial files, and notification callback on cancellation.
FAQ
Mini Project: Cancellation System
import redis
import json
import time
import threading
from datetime import datetime
r = redis.Redis()
class CancellationSystem:
def __init__(self):
self.workers = {}
def enqueue(self, queue, job_data):
job_id = f"job-{time.time()}-{hash(str(job_data)) % 10000}"
job_data['_id'] = job_id
job_data['_status'] = 'pending'
job_data['_enqueued_at'] = datetime.now().isoformat()
r.lpush(queue, json.dumps(job_data))
return job_id
def cancel_pending(self, job_id):
r.sadd('cancelled', job_id)
r.hset('job_status', job_id, 'cancelled')
return True
def revoke_running(self, job_id):
r.set(f'revoke:{job_id}', '1', ex=3600)
r.hset('job_status', job_id, 'revoking')
return True
def cancel_by_user(self, user_id):
pattern = f'user_jobs:{user_id}'
job_ids = r.smembers(pattern)
for jid in job_ids:
self.cancel_pending(jid.decode())
return len(job_ids)
def is_cancelled(self, job_id):
return r.sismember('cancelled', job_id)
def is_revoked(self, job_id):
return r.exists(f'revoke:{job_id}')
def cleanup(self, job_id):
r.hset('job_status', job_id, 'cleaned_up')
r.delete(f'revoke:{job_id}')
r.srem('cancelled', job_id)
cs = CancellationSystem()
jid = cs.enqueue('batch', {'task': 'process', 'user': 'alice'})
print(f"Enqueued: {jid}")
cs.cancel_pending(jid)
print(f"Cancelled: {cs.is_cancelled(jid)}")
cs.cleanup(jid)
print(f"Cleaned up: {not cs.is_cancelled(jid)}")
Expected output:
Enqueued: job-1719590400-1234
Cancelled: True
Cleaned up: False
What's Next
Now that you understand job cancellation, explore recurring jobs for tasks that run on a regular schedule, then learn about job deduplication to prevent duplicate work.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro