Job Timeout Configuration and Handling
In this tutorial, you will learn about Job Timeout Configuration and Handling. We cover key concepts, practical examples, and best practices to help you master this topic.
Configure and handle job timeouts with hard and soft limits, timeout exceptions, cleanup handlers, and prevention of runaway jobs in background processing systems.
What You Learn
You will learn how to set hard and soft time limits, handle timeout exceptions gracefully, implement timeout with Redis keys, and design jobs that respect time budgets.
Why It Matters
Runaway jobs consume worker resources indefinitely, blocking other jobs and degrading system performance. Timeouts ensure that no job runs longer than expected, protecting overall system stability.
Real-World Use
DodaTech's malware scanner enforces a 30-second soft timeout and 60-second hard timeout per file. If a scan exceeds 30 seconds, a warning is logged. At 60 seconds, the worker kills the scan and moves on.
Timeout Architecture
flowchart TD
J[Job Starts] --> S{Soft Timeout?}
S -->|No| C{Continue}
C -->|Not done| S
C -->|Done| E[Complete]
S -->|Yes| W[Log Warning]
W --> H{Hard Timeout?}
H -->|No| C
H -->|Yes| K[Kill Job]
K --> L[Cleanup Handler]
L --> N[Notify Failure]
E --> R[Return Result]
Hard Timeout Implementation
import signal
import time
class HardTimeout:
def __init__(self, seconds=10):
self.seconds = seconds
def __enter__(self):
signal.signal(signal.SIGALRM, self._handler)
signal.alarm(self.seconds)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
signal.alarm(0)
if exc_type == TimeoutError:
print(f"Job timed out after {self.seconds}s")
return True
return False
def _handler(self, signum, frame):
raise TimeoutError("Job execution timed out")
def expensive_scan():
time.sleep(5)
return "Scan complete"
try:
with HardTimeout(seconds=3):
result = expensive_scan()
print(f"Result: {result}")
except TimeoutError:
print("Timeout occurred, job killed")
Expected output:
Job timed out after 3s
Timeout occurred, job killed
Soft and Hard Time Limits
import threading
import time
class SoftHardTimeout:
def __init__(self, soft_limit=5, hard_limit=10):
self.soft_limit = soft_limit
self.hard_limit = hard_limit
self._soft_reached = False
def execute(self, func, *args, **kwargs):
soft_timer = threading.Timer(self.soft_limit, self._soft_timeout)
hard_timer = threading.Timer(self.hard_limit, self._hard_timeout)
soft_timer.start()
hard_timer.start()
result_container = []
exception_container = []
def target():
try:
result = func(*args, **kwargs)
result_container.append(result)
except Exception as e:
exception_container.append(e)
t = threading.Thread(target=target, daemon=True)
t.start()
t.join(timeout=self.hard_limit + 1)
soft_timer.cancel()
hard_timer.cancel()
if exception_container:
raise exception_container[0]
if t.is_alive():
raise TimeoutError(f"Hard timeout after {self.hard_limit}s")
return result_container[0] if result_container else None
def _soft_timeout(self):
self._soft_reached = True
print(f"Soft timeout reached at {self.soft_limit}s")
def _hard_timeout(self):
pass
def slow_task(duration):
for i in range(duration):
time.sleep(1)
print(f" Working... {i+1}s")
return "Done"
timeout_mgr = SoftHardTimeout(soft_limit=2, hard_limit=4)
try:
result = timeout_mgr.execute(slow_task, 5)
print(f"Result: {result}")
except TimeoutError as e:
print(f"Error: {e}")
Expected output:
Working... 1s
Working... 2s
Soft timeout reached at 2s
Working... 3s
Working... 4s
Error: Hard timeout after 4s
Redis-Based Job Timeout
import redis
import json
import time
import threading
r = redis.Redis()
class RedisTimeoutTracker:
def __init__(self, default_timeout=30):
self.default_timeout = default_timeout
def start_job(self, job_id, timeout=None):
timeout = timeout or self.default_timeout
deadline = time.time() + timeout
r.hset(f'timeout:{job_id}', 'deadline', deadline)
r.hset(f'timeout:{job_id}', 'timeout', timeout)
r.expire(f'timeout:{job_id}', timeout + 60)
def check_timeout(self, job_id):
deadline = r.hget(f'timeout:{job_id}', 'deadline')
if deadline and time.time() > float(deadline):
return True
return False
def extend_timeout(self, job_id, extra_seconds):
r.hincrbyfloat(f'timeout:{job_id}', 'deadline', extra_seconds)
def cleanup(self, job_id):
r.delete(f'timeout:{job_id}')
def get_remaining(self, job_id):
deadline = r.hget(f'timeout:{job_id}', 'deadline')
if deadline:
remaining = float(deadline) - time.time()
return max(0, remaining)
return None
tracker = RedisTimeoutTracker()
tracker.start_job('scan-heavy', 5)
def monitored_work(job_id):
for i in range(10):
if tracker.check_timeout(job_id):
print(f"Job {job_id} timed out at iteration {i}")
return False
print(f" Working iteration {i}")
time.sleep(1)
print("Job completed within timeout")
return True
t = threading.Thread(target=monitored_work, args=('scan-heavy',), daemon=True)
t.start()
t.join(timeout=8)
remaining = tracker.get_remaining('scan-heavy')
print(f"Remaining time: {remaining:.1f}s")
tracker.cleanup('scan-heavy')
Expected output:
Working iteration 0
Working iteration 1
Working iteration 2
Working iteration 3
Working iteration 4
Job scan-heavy timed out at iteration 5
Remaining time: 0.0s
Graceful Timeout with Cleanup
import threading
import time
class GracefulTimeoutJob:
def __init__(self, timeout=10):
self.timeout = timeout
self._cancelled = False
def run(self, func, *args, **kwargs):
def worker():
try:
result = func(*args, **kwargs)
self._result = result
self._completed = True
except Exception as e:
self._error = e
self._completed = False
self._result = None
self._error = None
t = threading.Thread(target=worker, daemon=True)
t.start()
t.join(timeout=self.timeout)
if t.is_alive():
self._cancelled = True
self._cleanup()
raise TimeoutError(f"Job exceeded {self.timeout}s timeout")
if self._error:
raise self._error
return self._result
def _cleanup(self):
print(f" Cleaning up partial work...")
time.sleep(0.5)
print(f" Resources released")
def process_with_cleanup():
print(" Worker: Opening file...")
time.sleep(2)
print(" Worker: Processing data...")
time.sleep(4)
print(" Worker: Writing results...")
return "processed"
job = GracefulTimeoutJob(timeout=3)
try:
result = job.run(process_with_cleanup)
print(f"Result: {result}")
except TimeoutError as e:
print(f"Error: {e}")
Expected output:
Worker: Opening file...
Worker: Processing data...
Cleaning up partial work...
Resources released
Error: Job exceeded 3s timeout
Common Mistakes
1. No Timeout at All
Jobs that hang due to network issues or infinite loops run forever, blocking workers. Always set timeouts for every job type.
2. Timeout Too Generous
Setting 30-minute timeouts for quick jobs means slow failures take 30 minutes to surface. Match timeout to expected job duration with reasonable margin.
3. No Cleanup on Timeout
When a job times out, it may leave locks, open files, or partial data. Implement cleanup handlers that run on timeout.
4. Killing Without Warning
Hard-killing a job without soft timeout means no chance for graceful shutdown. Always implement soft timeout with warning before hard kill.
5. Timeout Only on Total Duration
For long-running iterative jobs, timeout per iteration is more useful than total timeout. An iteration timeout catches individual slow steps.
Practice Questions
1. What is the difference between soft and hard timeout?
Soft timeout logs a warning and allows the job to continue. Hard timeout forcefully terminates the job after the hard limit is reached.
2. How do you implement timeout in Python?
Use signal.alarm for Unix (hard kill) or threading.Timer for cross-platform. The signal approach interrupts the running thread immediately.
3. Why should timeout cleanup be implemented?
Timed-out jobs may have acquired locks, opened files, or started transactions. Cleanup releases these resources to prevent leaks.
4. How do you choose timeout values?
Measure P95 and P99 execution times from production data. Set soft timeout at P99 and hard timeout at P99 + 50% margin.
Challenge
Build a timeout system that supports per-job timeout configuration, soft warning before hard kill, cleanup handler registration, Redis-based timeout tracking for distributed workers, and timeout metrics collection.
FAQ
Mini Project: Timeout System
import time
import threading
class TimeoutSystem:
def __init__(self):
self.jobs = {}
def submit(self, job_id, func, timeout=10, cleanup=None):
self.jobs[job_id] = {
'status': 'running',
'started': time.time(),
'timeout': timeout,
'cleanup': cleanup,
}
def wrapper():
try:
result = func()
self.jobs[job_id]['status'] = 'completed'
self.jobs[job_id]['result'] = result
return result
except Exception as e:
self.jobs[job_id]['status'] = 'failed'
self.jobs[job_id]['error'] = str(e)
t = threading.Thread(target=wrapper, daemon=True)
t.start()
monitor = threading.Thread(
target=self._monitor, args=(job_id, t), daemon=True
)
monitor.start()
return job_id
def _monitor(self, job_id, thread):
timeout = self.jobs[job_id]['timeout']
thread.join(timeout=timeout)
if thread.is_alive():
print(f"TIMEOUT: {job_id} exceeded {timeout}s")
if self.jobs[job_id].get('cleanup'):
self.jobs[job_id]['cleanup']()
self.jobs[job_id]['status'] = 'timed_out'
def status(self, job_id):
return self.jobs.get(job_id, {}).get('status', 'not_found')
ts = TimeoutSystem()
ts.submit('slow-job', lambda: time.sleep(5), timeout=2,
cleanup=lambda: print(" Cleaned up"))
time.sleep(3)
print(f"Status: {ts.status('slow-job')}")
Expected output:
TIMEOUT: slow-job exceeded 2s
Cleaned up
Status: timed_out
What's Next
Now that you understand timeouts, explore job middleware hooks for cross-cutting concerns, then learn about job lifecycle events for comprehensive monitoring.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro