From 7eb6c2ab9f74bf209439a9228447daf3cfda19b4 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 4 Apr 2014 17:16:33 +0700 Subject: [PATCH] Job timeouts are now handled by "worker.death_penalty_class". --- rq/timeouts.py | 15 +++++++++++++-- rq/worker.py | 6 ++++-- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/rq/timeouts.py b/rq/timeouts.py index f1e1848f..a96d8919 100644 --- a/rq/timeouts.py +++ b/rq/timeouts.py @@ -8,7 +8,9 @@ class JobTimeoutException(Exception): pass -class death_penalty_after(object): +class BaseDeathPenalty(object): + """Base class to setup job timeouts.""" + def __init__(self, timeout): self._timeout = timeout @@ -31,6 +33,15 @@ class death_penalty_after(object): # invoking context. return False + def setup_death_penalty(self): + raise NotImplementedError() + + def cancel_death_penalty(self): + raise NotImplementedError() + + +class UnixSignalDeathPenalty(BaseDeathPenalty): + def handle_death_penalty(self, signum, frame): raise JobTimeoutException('Job exceeded maximum timeout ' 'value (%d seconds).' % self._timeout) @@ -48,4 +59,4 @@ class death_penalty_after(object): default signal handling. """ signal.alarm(0) - signal.signal(signal.SIGALRM, signal.SIG_DFL) + signal.signal(signal.SIGALRM, signal.SIG_DFL) \ No newline at end of file diff --git a/rq/worker.py b/rq/worker.py index 688b1a44..05b4fa0d 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -18,7 +18,7 @@ from .job import Job, Status from .utils import make_colorizer, utcnow, utcformat from .logutils import setup_loghandlers from .exceptions import NoQueueError, DequeueTimeout -from .timeouts import death_penalty_after +from .timeouts import UnixSignalDeathPenalty from .version import VERSION from rq.compat import text_type, as_text @@ -59,6 +59,8 @@ def signal_name(signum): class Worker(object): redis_worker_namespace_prefix = 'rq:worker:' redis_workers_keys = 'rq:workers' + death_penalty_class = UnixSignalDeathPenalty + @classmethod def all(cls, connection=None): @@ -462,7 +464,7 @@ class Worker(object): with self.connection._pipeline() as pipeline: try: - with death_penalty_after(job.timeout or Queue.DEFAULT_TIMEOUT): + with self.death_penalty_class(job.timeout or Queue.DEFAULT_TIMEOUT): rv = job.perform() # Pickle the result in the same try-except block since we need to