|
|
|
@ -12,18 +12,22 @@ import sys
|
|
|
|
import time
|
|
|
|
import time
|
|
|
|
import traceback
|
|
|
|
import traceback
|
|
|
|
import warnings
|
|
|
|
import warnings
|
|
|
|
|
|
|
|
from datetime import timedelta
|
|
|
|
|
|
|
|
|
|
|
|
from rq.compat import as_text, string_types, text_type
|
|
|
|
from rq.compat import as_text, string_types, text_type
|
|
|
|
|
|
|
|
|
|
|
|
from .connections import get_current_connection
|
|
|
|
from .connections import get_current_connection
|
|
|
|
from .exceptions import DequeueTimeout, NoQueueError
|
|
|
|
from .defaults import DEFAULT_RESULT_TTL, DEFAULT_WORKER_TTL
|
|
|
|
from .job import Job, Status
|
|
|
|
from .exceptions import DequeueTimeout
|
|
|
|
|
|
|
|
from .job import Job, JobStatus
|
|
|
|
from .logutils import setup_loghandlers
|
|
|
|
from .logutils import setup_loghandlers
|
|
|
|
from .queue import get_failed_queue, Queue
|
|
|
|
from .queue import Queue, get_failed_queue
|
|
|
|
|
|
|
|
from .registry import FinishedJobRegistry, StartedJobRegistry, clean_registries
|
|
|
|
|
|
|
|
from .suspension import is_suspended
|
|
|
|
from .timeouts import UnixSignalDeathPenalty
|
|
|
|
from .timeouts import UnixSignalDeathPenalty
|
|
|
|
from .utils import import_attribute, make_colorizer, utcformat, utcnow
|
|
|
|
from .utils import (ensure_list, enum, import_attribute, make_colorizer,
|
|
|
|
|
|
|
|
utcformat, utcnow, utcparse)
|
|
|
|
from .version import VERSION
|
|
|
|
from .version import VERSION
|
|
|
|
from .registry import FinishedJobRegistry, StartedJobRegistry
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
from procname import setprocname
|
|
|
|
from procname import setprocname
|
|
|
|
@ -35,8 +39,7 @@ green = make_colorizer('darkgreen')
|
|
|
|
yellow = make_colorizer('darkyellow')
|
|
|
|
yellow = make_colorizer('darkyellow')
|
|
|
|
blue = make_colorizer('darkblue')
|
|
|
|
blue = make_colorizer('darkblue')
|
|
|
|
|
|
|
|
|
|
|
|
DEFAULT_WORKER_TTL = 420
|
|
|
|
|
|
|
|
DEFAULT_RESULT_TTL = 500
|
|
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@ -65,6 +68,15 @@ def signal_name(signum):
|
|
|
|
return 'SIG_UNKNOWN'
|
|
|
|
return 'SIG_UNKNOWN'
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
WorkerStatus = enum(
|
|
|
|
|
|
|
|
'WorkerStatus',
|
|
|
|
|
|
|
|
STARTED='started',
|
|
|
|
|
|
|
|
SUSPENDED='suspended',
|
|
|
|
|
|
|
|
BUSY='busy',
|
|
|
|
|
|
|
|
IDLE='idle'
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class Worker(object):
|
|
|
|
class Worker(object):
|
|
|
|
redis_worker_namespace_prefix = 'rq:worker:'
|
|
|
|
redis_worker_namespace_prefix = 'rq:worker:'
|
|
|
|
redis_workers_keys = 'rq:workers'
|
|
|
|
redis_workers_keys = 'rq:workers'
|
|
|
|
@ -91,7 +103,7 @@ class Worker(object):
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
prefix = cls.redis_worker_namespace_prefix
|
|
|
|
prefix = cls.redis_worker_namespace_prefix
|
|
|
|
if not worker_key.startswith(prefix):
|
|
|
|
if not worker_key.startswith(prefix):
|
|
|
|
raise ValueError('Not a valid RQ worker key: %s' % (worker_key,))
|
|
|
|
raise ValueError('Not a valid RQ worker key: {0}'.format(worker_key))
|
|
|
|
|
|
|
|
|
|
|
|
if connection is None:
|
|
|
|
if connection is None:
|
|
|
|
connection = get_current_connection()
|
|
|
|
connection = get_current_connection()
|
|
|
|
@ -110,13 +122,14 @@ class Worker(object):
|
|
|
|
return worker
|
|
|
|
return worker
|
|
|
|
|
|
|
|
|
|
|
|
def __init__(self, queues, name=None,
|
|
|
|
def __init__(self, queues, name=None,
|
|
|
|
default_result_ttl=None, connection=None,
|
|
|
|
default_result_ttl=None, connection=None, exc_handler=None,
|
|
|
|
exc_handler=None, default_worker_ttl=None, job_class=None): # noqa
|
|
|
|
exception_handlers=None, default_worker_ttl=None, job_class=None): # noqa
|
|
|
|
if connection is None:
|
|
|
|
if connection is None:
|
|
|
|
connection = get_current_connection()
|
|
|
|
connection = get_current_connection()
|
|
|
|
self.connection = connection
|
|
|
|
self.connection = connection
|
|
|
|
if isinstance(queues, self.queue_class):
|
|
|
|
|
|
|
|
queues = [queues]
|
|
|
|
queues = [self.queue_class(name=q) if isinstance(q, text_type) else q
|
|
|
|
|
|
|
|
for q in ensure_list(queues)]
|
|
|
|
self._name = name
|
|
|
|
self._name = name
|
|
|
|
self.queues = queues
|
|
|
|
self.queues = queues
|
|
|
|
self.validate_queues()
|
|
|
|
self.validate_queues()
|
|
|
|
@ -133,15 +146,26 @@ class Worker(object):
|
|
|
|
self._state = 'starting'
|
|
|
|
self._state = 'starting'
|
|
|
|
self._is_horse = False
|
|
|
|
self._is_horse = False
|
|
|
|
self._horse_pid = 0
|
|
|
|
self._horse_pid = 0
|
|
|
|
self._stopped = False
|
|
|
|
self._stop_requested = False
|
|
|
|
self.log = logger
|
|
|
|
self.log = logger
|
|
|
|
self.failed_queue = get_failed_queue(connection=self.connection)
|
|
|
|
self.failed_queue = get_failed_queue(connection=self.connection)
|
|
|
|
|
|
|
|
self.last_cleaned_at = None
|
|
|
|
|
|
|
|
|
|
|
|
# By default, push the "move-to-failed-queue" exception handler onto
|
|
|
|
# By default, push the "move-to-failed-queue" exception handler onto
|
|
|
|
# the stack
|
|
|
|
# the stack
|
|
|
|
self.push_exc_handler(self.move_to_failed_queue)
|
|
|
|
if exception_handlers is None:
|
|
|
|
if exc_handler is not None:
|
|
|
|
self.push_exc_handler(self.move_to_failed_queue)
|
|
|
|
self.push_exc_handler(exc_handler)
|
|
|
|
if exc_handler is not None:
|
|
|
|
|
|
|
|
self.push_exc_handler(exc_handler)
|
|
|
|
|
|
|
|
warnings.warn(
|
|
|
|
|
|
|
|
"use of exc_handler is deprecated, pass a list to exception_handlers instead.",
|
|
|
|
|
|
|
|
DeprecationWarning
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
elif isinstance(exception_handlers, list):
|
|
|
|
|
|
|
|
for h in exception_handlers:
|
|
|
|
|
|
|
|
self.push_exc_handler(h)
|
|
|
|
|
|
|
|
elif exception_handlers is not None:
|
|
|
|
|
|
|
|
self.push_exc_handler(exception_handlers)
|
|
|
|
|
|
|
|
|
|
|
|
if job_class is not None:
|
|
|
|
if job_class is not None:
|
|
|
|
if isinstance(job_class, string_types):
|
|
|
|
if isinstance(job_class, string_types):
|
|
|
|
@ -150,19 +174,17 @@ class Worker(object):
|
|
|
|
|
|
|
|
|
|
|
|
def validate_queues(self):
|
|
|
|
def validate_queues(self):
|
|
|
|
"""Sanity check for the given queues."""
|
|
|
|
"""Sanity check for the given queues."""
|
|
|
|
if not iterable(self.queues):
|
|
|
|
|
|
|
|
raise ValueError('Argument queues not iterable.')
|
|
|
|
|
|
|
|
for queue in self.queues:
|
|
|
|
for queue in self.queues:
|
|
|
|
if not isinstance(queue, self.queue_class):
|
|
|
|
if not isinstance(queue, self.queue_class):
|
|
|
|
raise NoQueueError('Give each worker at least one Queue.')
|
|
|
|
raise TypeError('{0} is not of type {1} or text type'.format(queue, self.queue_class))
|
|
|
|
|
|
|
|
|
|
|
|
def queue_names(self):
|
|
|
|
def queue_names(self):
|
|
|
|
"""Returns the queue names of this worker's queues."""
|
|
|
|
"""Returns the queue names of this worker's queues."""
|
|
|
|
return map(lambda q: q.name, self.queues)
|
|
|
|
return list(map(lambda q: q.name, self.queues))
|
|
|
|
|
|
|
|
|
|
|
|
def queue_keys(self):
|
|
|
|
def queue_keys(self):
|
|
|
|
"""Returns the Redis keys representing this worker's queues."""
|
|
|
|
"""Returns the Redis keys representing this worker's queues."""
|
|
|
|
return map(lambda q: q.key, self.queues)
|
|
|
|
return list(map(lambda q: q.key, self.queues))
|
|
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
@property
|
|
|
|
def name(self):
|
|
|
|
def name(self):
|
|
|
|
@ -175,7 +197,7 @@ class Worker(object):
|
|
|
|
if self._name is None:
|
|
|
|
if self._name is None:
|
|
|
|
hostname = socket.gethostname()
|
|
|
|
hostname = socket.gethostname()
|
|
|
|
shortname, _, _ = hostname.partition('.')
|
|
|
|
shortname, _, _ = hostname.partition('.')
|
|
|
|
self._name = '%s.%s' % (shortname, self.pid)
|
|
|
|
self._name = '{0}.{1}'.format(shortname, self.pid)
|
|
|
|
return self._name
|
|
|
|
return self._name
|
|
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
@property
|
|
|
|
@ -205,15 +227,15 @@ class Worker(object):
|
|
|
|
|
|
|
|
|
|
|
|
This can be used to make `ps -ef` output more readable.
|
|
|
|
This can be used to make `ps -ef` output more readable.
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
setprocname('rq: %s' % (message,))
|
|
|
|
setprocname('rq: {0}'.format(message))
|
|
|
|
|
|
|
|
|
|
|
|
def register_birth(self):
|
|
|
|
def register_birth(self):
|
|
|
|
"""Registers its own birth."""
|
|
|
|
"""Registers its own birth."""
|
|
|
|
self.log.debug('Registering birth of worker %s' % (self.name,))
|
|
|
|
self.log.debug('Registering birth of worker {0}'.format(self.name))
|
|
|
|
if self.connection.exists(self.key) and \
|
|
|
|
if self.connection.exists(self.key) and \
|
|
|
|
not self.connection.hexists(self.key, 'death'):
|
|
|
|
not self.connection.hexists(self.key, 'death'):
|
|
|
|
raise ValueError('There exists an active worker named \'%s\' '
|
|
|
|
msg = 'There exists an active worker named {0!r} already'
|
|
|
|
'already.' % (self.name,))
|
|
|
|
raise ValueError(msg.format(self.name))
|
|
|
|
key = self.key
|
|
|
|
key = self.key
|
|
|
|
queues = ','.join(self.queue_names())
|
|
|
|
queues = ','.join(self.queue_names())
|
|
|
|
with self.connection._pipeline() as p:
|
|
|
|
with self.connection._pipeline() as p:
|
|
|
|
@ -235,6 +257,20 @@ class Worker(object):
|
|
|
|
p.expire(self.key, 60)
|
|
|
|
p.expire(self.key, 60)
|
|
|
|
p.execute()
|
|
|
|
p.execute()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
|
|
|
def birth_date(self):
|
|
|
|
|
|
|
|
"""Fetches birth date from Redis."""
|
|
|
|
|
|
|
|
birth_timestamp = self.connection.hget(self.key, 'birth')
|
|
|
|
|
|
|
|
if birth_timestamp is not None:
|
|
|
|
|
|
|
|
return utcparse(as_text(birth_timestamp))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
|
|
|
def death_date(self):
|
|
|
|
|
|
|
|
"""Fetches death date from Redis."""
|
|
|
|
|
|
|
|
death_timestamp = self.connection.hget(self.key, 'death')
|
|
|
|
|
|
|
|
if death_timestamp is not None:
|
|
|
|
|
|
|
|
return utcparse(as_text(death_timestamp))
|
|
|
|
|
|
|
|
|
|
|
|
def set_state(self, state, pipeline=None):
|
|
|
|
def set_state(self, state, pipeline=None):
|
|
|
|
self._state = state
|
|
|
|
self._state = state
|
|
|
|
connection = pipeline if pipeline is not None else self.connection
|
|
|
|
connection = pipeline if pipeline is not None else self.connection
|
|
|
|
@ -282,56 +318,75 @@ class Worker(object):
|
|
|
|
|
|
|
|
|
|
|
|
return self.job_class.fetch(job_id, self.connection)
|
|
|
|
return self.job_class.fetch(job_id, self.connection)
|
|
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
|
|
|
def stopped(self):
|
|
|
|
|
|
|
|
return self._stopped
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _install_signal_handlers(self):
|
|
|
|
def _install_signal_handlers(self):
|
|
|
|
"""Installs signal handlers for handling SIGINT and SIGTERM
|
|
|
|
"""Installs signal handlers for handling SIGINT and SIGTERM
|
|
|
|
gracefully.
|
|
|
|
gracefully.
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
|
|
def request_force_stop(signum, frame):
|
|
|
|
signal.signal(signal.SIGINT, self.request_stop)
|
|
|
|
"""Terminates the application (cold shutdown).
|
|
|
|
signal.signal(signal.SIGTERM, self.request_stop)
|
|
|
|
"""
|
|
|
|
|
|
|
|
self.log.warning('Cold shut down.')
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Take down the horse with the worker
|
|
|
|
def request_force_stop(self, signum, frame):
|
|
|
|
if self.horse_pid:
|
|
|
|
"""Terminates the application (cold shutdown).
|
|
|
|
msg = 'Taking down horse %d with me.' % self.horse_pid
|
|
|
|
"""
|
|
|
|
self.log.debug(msg)
|
|
|
|
self.log.warning('Cold shut down')
|
|
|
|
try:
|
|
|
|
|
|
|
|
os.kill(self.horse_pid, signal.SIGKILL)
|
|
|
|
# Take down the horse with the worker
|
|
|
|
except OSError as e:
|
|
|
|
if self.horse_pid:
|
|
|
|
# ESRCH ("No such process") is fine with us
|
|
|
|
msg = 'Taking down horse {0} with me'.format(self.horse_pid)
|
|
|
|
if e.errno != errno.ESRCH:
|
|
|
|
self.log.debug(msg)
|
|
|
|
self.log.debug('Horse already down.')
|
|
|
|
try:
|
|
|
|
raise
|
|
|
|
os.kill(self.horse_pid, signal.SIGKILL)
|
|
|
|
raise SystemExit()
|
|
|
|
except OSError as e:
|
|
|
|
|
|
|
|
# ESRCH ("No such process") is fine with us
|
|
|
|
|
|
|
|
if e.errno != errno.ESRCH:
|
|
|
|
|
|
|
|
self.log.debug('Horse already down')
|
|
|
|
|
|
|
|
raise
|
|
|
|
|
|
|
|
raise SystemExit()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def request_stop(self, signum, frame):
|
|
|
|
|
|
|
|
"""Stops the current worker loop but waits for child processes to
|
|
|
|
|
|
|
|
end gracefully (warm shutdown).
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
self.log.debug('Got signal {0}'.format(signal_name(signum)))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
signal.signal(signal.SIGINT, self.request_force_stop)
|
|
|
|
|
|
|
|
signal.signal(signal.SIGTERM, self.request_force_stop)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
msg = 'Warm shut down requested'
|
|
|
|
|
|
|
|
self.log.warning(msg)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# If shutdown is requested in the middle of a job, wait until
|
|
|
|
|
|
|
|
# finish before shutting down
|
|
|
|
|
|
|
|
if self.get_state() == 'busy':
|
|
|
|
|
|
|
|
self._stop_requested = True
|
|
|
|
|
|
|
|
self.log.debug('Stopping after current horse is finished. '
|
|
|
|
|
|
|
|
'Press Ctrl+C again for a cold shutdown.')
|
|
|
|
|
|
|
|
else:
|
|
|
|
|
|
|
|
raise StopRequested()
|
|
|
|
|
|
|
|
|
|
|
|
def request_stop(signum, frame):
|
|
|
|
def check_for_suspension(self, burst):
|
|
|
|
"""Stops the current worker loop but waits for child processes to
|
|
|
|
"""Check to see if workers have been suspended by `rq suspend`"""
|
|
|
|
end gracefully (warm shutdown).
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
self.log.debug('Got signal %s.' % signal_name(signum))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
signal.signal(signal.SIGINT, request_force_stop)
|
|
|
|
before_state = None
|
|
|
|
signal.signal(signal.SIGTERM, request_force_stop)
|
|
|
|
notified = False
|
|
|
|
|
|
|
|
|
|
|
|
msg = 'Warm shut down requested.'
|
|
|
|
while not self._stop_requested and is_suspended(self.connection):
|
|
|
|
self.log.warning(msg)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# If shutdown is requested in the middle of a job, wait until
|
|
|
|
if burst:
|
|
|
|
# finish before shutting down
|
|
|
|
self.log.info('Suspended in burst mode, exiting')
|
|
|
|
if self.get_state() == 'busy':
|
|
|
|
self.log.info('Note: There could still be unfinished jobs on the queue')
|
|
|
|
self._stopped = True
|
|
|
|
raise StopRequested
|
|
|
|
self.log.debug('Stopping after current horse is finished. '
|
|
|
|
|
|
|
|
'Press Ctrl+C again for a cold shutdown.')
|
|
|
|
|
|
|
|
else:
|
|
|
|
|
|
|
|
raise StopRequested()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
signal.signal(signal.SIGINT, request_stop)
|
|
|
|
if not notified:
|
|
|
|
signal.signal(signal.SIGTERM, request_stop)
|
|
|
|
self.log.info('Worker suspended, run `rq resume` to resume')
|
|
|
|
|
|
|
|
before_state = self.get_state()
|
|
|
|
|
|
|
|
self.set_state(WorkerStatus.SUSPENDED)
|
|
|
|
|
|
|
|
notified = True
|
|
|
|
|
|
|
|
time.sleep(1)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if before_state:
|
|
|
|
|
|
|
|
self.set_state(before_state)
|
|
|
|
|
|
|
|
|
|
|
|
def work(self, burst=False):
|
|
|
|
def work(self, burst=False):
|
|
|
|
"""Starts the work loop.
|
|
|
|
"""Starts the work loop.
|
|
|
|
@ -347,18 +402,27 @@ class Worker(object):
|
|
|
|
|
|
|
|
|
|
|
|
did_perform_work = False
|
|
|
|
did_perform_work = False
|
|
|
|
self.register_birth()
|
|
|
|
self.register_birth()
|
|
|
|
self.log.info('RQ worker started, version %s' % VERSION)
|
|
|
|
self.log.info("RQ worker {0!r} started, version {1}".format(self.key, VERSION))
|
|
|
|
self.set_state('starting')
|
|
|
|
self.set_state(WorkerStatus.STARTED)
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
while True:
|
|
|
|
while True:
|
|
|
|
if self.stopped:
|
|
|
|
|
|
|
|
self.log.info('Stopping on request.')
|
|
|
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
timeout = None if burst else max(1, self.default_worker_ttl - 60)
|
|
|
|
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
|
|
|
|
self.check_for_suspension(burst)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if self.should_run_maintenance_tasks:
|
|
|
|
|
|
|
|
self.clean_registries()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if self._stop_requested:
|
|
|
|
|
|
|
|
self.log.info('Stopping on request')
|
|
|
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
timeout = None if burst else max(1, self.default_worker_ttl - 60)
|
|
|
|
|
|
|
|
|
|
|
|
result = self.dequeue_job_and_maintain_ttl(timeout)
|
|
|
|
result = self.dequeue_job_and_maintain_ttl(timeout)
|
|
|
|
if result is None:
|
|
|
|
if result is None:
|
|
|
|
|
|
|
|
if burst:
|
|
|
|
|
|
|
|
self.log.info("RQ worker {0!r} done, quitting".format(self.key))
|
|
|
|
break
|
|
|
|
break
|
|
|
|
except StopRequested:
|
|
|
|
except StopRequested:
|
|
|
|
break
|
|
|
|
break
|
|
|
|
@ -367,10 +431,11 @@ class Worker(object):
|
|
|
|
self.execute_job(job)
|
|
|
|
self.execute_job(job)
|
|
|
|
self.heartbeat()
|
|
|
|
self.heartbeat()
|
|
|
|
|
|
|
|
|
|
|
|
if job.get_status() == Status.FINISHED:
|
|
|
|
if job.get_status() == JobStatus.FINISHED:
|
|
|
|
queue.enqueue_dependents(job)
|
|
|
|
queue.enqueue_dependents(job)
|
|
|
|
|
|
|
|
|
|
|
|
did_perform_work = True
|
|
|
|
did_perform_work = True
|
|
|
|
|
|
|
|
|
|
|
|
finally:
|
|
|
|
finally:
|
|
|
|
if not self.is_horse:
|
|
|
|
if not self.is_horse:
|
|
|
|
self.register_death()
|
|
|
|
self.register_death()
|
|
|
|
@ -380,11 +445,10 @@ class Worker(object):
|
|
|
|
result = None
|
|
|
|
result = None
|
|
|
|
qnames = self.queue_names()
|
|
|
|
qnames = self.queue_names()
|
|
|
|
|
|
|
|
|
|
|
|
self.set_state('idle')
|
|
|
|
self.set_state(WorkerStatus.IDLE)
|
|
|
|
self.procline('Listening on %s' % ','.join(qnames))
|
|
|
|
self.procline('Listening on {0}'.format(','.join(qnames)))
|
|
|
|
self.log.info('')
|
|
|
|
self.log.info('')
|
|
|
|
self.log.info('*** Listening on %s...' %
|
|
|
|
self.log.info('*** Listening on {0}...'.format(green(', '.join(qnames))))
|
|
|
|
green(', '.join(qnames)))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
while True:
|
|
|
|
while True:
|
|
|
|
self.heartbeat()
|
|
|
|
self.heartbeat()
|
|
|
|
@ -394,8 +458,8 @@ class Worker(object):
|
|
|
|
connection=self.connection)
|
|
|
|
connection=self.connection)
|
|
|
|
if result is not None:
|
|
|
|
if result is not None:
|
|
|
|
job, queue = result
|
|
|
|
job, queue = result
|
|
|
|
self.log.info('%s: %s (%s)' % (green(queue.name),
|
|
|
|
self.log.info('{0}: {1} ({2})'.format(green(queue.name),
|
|
|
|
blue(job.description), job.id))
|
|
|
|
blue(job.description), job.id))
|
|
|
|
|
|
|
|
|
|
|
|
break
|
|
|
|
break
|
|
|
|
except DequeueTimeout:
|
|
|
|
except DequeueTimeout:
|
|
|
|
@ -427,15 +491,17 @@ class Worker(object):
|
|
|
|
within the given timeout bounds, or will end the work horse with
|
|
|
|
within the given timeout bounds, or will end the work horse with
|
|
|
|
SIGALRM.
|
|
|
|
SIGALRM.
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
|
|
|
|
self.set_state('busy')
|
|
|
|
child_pid = os.fork()
|
|
|
|
child_pid = os.fork()
|
|
|
|
if child_pid == 0:
|
|
|
|
if child_pid == 0:
|
|
|
|
self.main_work_horse(job)
|
|
|
|
self.main_work_horse(job)
|
|
|
|
else:
|
|
|
|
else:
|
|
|
|
self._horse_pid = child_pid
|
|
|
|
self._horse_pid = child_pid
|
|
|
|
self.procline('Forked %d at %d' % (child_pid, time.time()))
|
|
|
|
self.procline('Forked {0} at {0}'.format(child_pid, time.time()))
|
|
|
|
while True:
|
|
|
|
while True:
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
os.waitpid(child_pid, 0)
|
|
|
|
os.waitpid(child_pid, 0)
|
|
|
|
|
|
|
|
self.set_state('idle')
|
|
|
|
break
|
|
|
|
break
|
|
|
|
except OSError as e:
|
|
|
|
except OSError as e:
|
|
|
|
# In case we encountered an OSError due to EINTR (which is
|
|
|
|
# In case we encountered an OSError due to EINTR (which is
|
|
|
|
@ -477,17 +543,16 @@ class Worker(object):
|
|
|
|
timeout = (job.timeout or 180) + 60
|
|
|
|
timeout = (job.timeout or 180) + 60
|
|
|
|
|
|
|
|
|
|
|
|
with self.connection._pipeline() as pipeline:
|
|
|
|
with self.connection._pipeline() as pipeline:
|
|
|
|
self.set_state('busy', pipeline=pipeline)
|
|
|
|
self.set_state(WorkerStatus.BUSY, pipeline=pipeline)
|
|
|
|
self.set_current_job_id(job.id, pipeline=pipeline)
|
|
|
|
self.set_current_job_id(job.id, pipeline=pipeline)
|
|
|
|
self.heartbeat(timeout, pipeline=pipeline)
|
|
|
|
self.heartbeat(timeout, pipeline=pipeline)
|
|
|
|
registry = StartedJobRegistry(job.origin, self.connection)
|
|
|
|
registry = StartedJobRegistry(job.origin, self.connection)
|
|
|
|
registry.add(job, timeout, pipeline=pipeline)
|
|
|
|
registry.add(job, timeout, pipeline=pipeline)
|
|
|
|
job.set_status(Status.STARTED, pipeline=pipeline)
|
|
|
|
job.set_status(JobStatus.STARTED, pipeline=pipeline)
|
|
|
|
pipeline.execute()
|
|
|
|
pipeline.execute()
|
|
|
|
|
|
|
|
|
|
|
|
self.procline('Processing %s from %s since %s' % (
|
|
|
|
msg = 'Processing {0} from {1} since {2}'
|
|
|
|
job.func_name,
|
|
|
|
self.procline(msg.format(job.func_name, job.origin, time.time()))
|
|
|
|
job.origin, time.time()))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def perform_job(self, job):
|
|
|
|
def perform_job(self, job):
|
|
|
|
"""Performs the actual work of a job. Will/should only be called
|
|
|
|
"""Performs the actual work of a job. Will/should only be called
|
|
|
|
@ -508,10 +573,10 @@ class Worker(object):
|
|
|
|
|
|
|
|
|
|
|
|
self.set_current_job_id(None, pipeline=pipeline)
|
|
|
|
self.set_current_job_id(None, pipeline=pipeline)
|
|
|
|
|
|
|
|
|
|
|
|
result_ttl = job.get_ttl(self.default_result_ttl)
|
|
|
|
result_ttl = job.get_result_ttl(self.default_result_ttl)
|
|
|
|
if result_ttl != 0:
|
|
|
|
if result_ttl != 0:
|
|
|
|
job.ended_at = utcnow()
|
|
|
|
job.ended_at = utcnow()
|
|
|
|
job._status = Status.FINISHED
|
|
|
|
job._status = JobStatus.FINISHED
|
|
|
|
job.save(pipeline=pipeline)
|
|
|
|
job.save(pipeline=pipeline)
|
|
|
|
|
|
|
|
|
|
|
|
finished_job_registry = FinishedJobRegistry(job.origin, self.connection)
|
|
|
|
finished_job_registry = FinishedJobRegistry(job.origin, self.connection)
|
|
|
|
@ -523,23 +588,28 @@ class Worker(object):
|
|
|
|
pipeline.execute()
|
|
|
|
pipeline.execute()
|
|
|
|
|
|
|
|
|
|
|
|
except Exception:
|
|
|
|
except Exception:
|
|
|
|
job.set_status(Status.FAILED, pipeline=pipeline)
|
|
|
|
job.set_status(JobStatus.FAILED, pipeline=pipeline)
|
|
|
|
started_job_registry.remove(job, pipeline=pipeline)
|
|
|
|
started_job_registry.remove(job, pipeline=pipeline)
|
|
|
|
pipeline.execute()
|
|
|
|
try:
|
|
|
|
|
|
|
|
pipeline.execute()
|
|
|
|
|
|
|
|
except Exception:
|
|
|
|
|
|
|
|
# Ensure that custom exception handlers are called
|
|
|
|
|
|
|
|
# even if Redis is down
|
|
|
|
|
|
|
|
pass
|
|
|
|
self.handle_exception(job, *sys.exc_info())
|
|
|
|
self.handle_exception(job, *sys.exc_info())
|
|
|
|
return False
|
|
|
|
return False
|
|
|
|
|
|
|
|
|
|
|
|
if rv is None:
|
|
|
|
self.log.info('{0}: {1} ({2})'.format(green(job.origin), blue('Job OK'), job.id))
|
|
|
|
self.log.info('Job OK')
|
|
|
|
if rv:
|
|
|
|
else:
|
|
|
|
log_result = "{0!r}".format(as_text(text_type(rv)))
|
|
|
|
self.log.info('Job OK, result = %s' % (yellow(text_type(rv)),))
|
|
|
|
self.log.debug('Result: {0}'.format(yellow(log_result)))
|
|
|
|
|
|
|
|
|
|
|
|
if result_ttl == 0:
|
|
|
|
if result_ttl == 0:
|
|
|
|
self.log.info('Result discarded immediately.')
|
|
|
|
self.log.info('Result discarded immediately')
|
|
|
|
elif result_ttl > 0:
|
|
|
|
elif result_ttl > 0:
|
|
|
|
self.log.info('Result is kept for %d seconds.' % result_ttl)
|
|
|
|
self.log.info('Result is kept for {0} seconds'.format(result_ttl))
|
|
|
|
else:
|
|
|
|
else:
|
|
|
|
self.log.warning('Result will never expire, clean up result key manually.')
|
|
|
|
self.log.warning('Result will never expire, clean up result key manually')
|
|
|
|
|
|
|
|
|
|
|
|
return True
|
|
|
|
return True
|
|
|
|
|
|
|
|
|
|
|
|
@ -555,7 +625,7 @@ class Worker(object):
|
|
|
|
})
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
|
|
for handler in reversed(self._exc_handlers):
|
|
|
|
for handler in reversed(self._exc_handlers):
|
|
|
|
self.log.debug('Invoking exception handler %s' % (handler,))
|
|
|
|
self.log.debug('Invoking exception handler {0}'.format(handler))
|
|
|
|
fallthrough = handler(job, *exc_info)
|
|
|
|
fallthrough = handler(job, *exc_info)
|
|
|
|
|
|
|
|
|
|
|
|
# Only handlers with explicit return values should disable further
|
|
|
|
# Only handlers with explicit return values should disable further
|
|
|
|
@ -569,7 +639,7 @@ class Worker(object):
|
|
|
|
def move_to_failed_queue(self, job, *exc_info):
|
|
|
|
def move_to_failed_queue(self, job, *exc_info):
|
|
|
|
"""Default exception handler: move the job to the failed queue."""
|
|
|
|
"""Default exception handler: move the job to the failed queue."""
|
|
|
|
exc_string = ''.join(traceback.format_exception(*exc_info))
|
|
|
|
exc_string = ''.join(traceback.format_exception(*exc_info))
|
|
|
|
self.log.warning('Moving job to %s queue.' % self.failed_queue.name)
|
|
|
|
self.log.warning('Moving job to {0!r} queue'.format(self.failed_queue.name))
|
|
|
|
self.failed_queue.quarantine(job, exc_info=exc_string)
|
|
|
|
self.failed_queue.quarantine(job, exc_info=exc_string)
|
|
|
|
|
|
|
|
|
|
|
|
def push_exc_handler(self, handler_func):
|
|
|
|
def push_exc_handler(self, handler_func):
|
|
|
|
@ -580,13 +650,33 @@ class Worker(object):
|
|
|
|
"""Pops the latest exception handler off of the exc handler stack."""
|
|
|
|
"""Pops the latest exception handler off of the exc handler stack."""
|
|
|
|
return self._exc_handlers.pop()
|
|
|
|
return self._exc_handlers.pop()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def __eq__(self, other):
|
|
|
|
|
|
|
|
"""Equality does not take the database/connection into account"""
|
|
|
|
|
|
|
|
if not isinstance(other, self.__class__):
|
|
|
|
|
|
|
|
raise TypeError('Cannot compare workers to other types (of workers)')
|
|
|
|
|
|
|
|
return self.name == other.name
|
|
|
|
|
|
|
|
|
|
|
|
class SimpleWorker(Worker):
|
|
|
|
def __hash__(self):
|
|
|
|
def _install_signal_handlers(self, *args, **kwargs):
|
|
|
|
"""The hash does not take the database/connection into account"""
|
|
|
|
"""Signal handlers are useless for test worker, as it
|
|
|
|
return hash(self.name)
|
|
|
|
does not have fork() ability"""
|
|
|
|
|
|
|
|
pass
|
|
|
|
def clean_registries(self):
|
|
|
|
|
|
|
|
"""Runs maintenance jobs on each Queue's registries."""
|
|
|
|
|
|
|
|
for queue in self.queues:
|
|
|
|
|
|
|
|
clean_registries(queue)
|
|
|
|
|
|
|
|
self.last_cleaned_at = utcnow()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
|
|
|
def should_run_maintenance_tasks(self):
|
|
|
|
|
|
|
|
"""Maintenance tasks should run on first startup or every hour."""
|
|
|
|
|
|
|
|
if self.last_cleaned_at is None:
|
|
|
|
|
|
|
|
return True
|
|
|
|
|
|
|
|
if (utcnow() - self.last_cleaned_at) > timedelta(hours=1):
|
|
|
|
|
|
|
|
return True
|
|
|
|
|
|
|
|
return False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class SimpleWorker(Worker):
|
|
|
|
def main_work_horse(self, *args, **kwargs):
|
|
|
|
def main_work_horse(self, *args, **kwargs):
|
|
|
|
raise NotImplementedError("Test worker does not implement this method")
|
|
|
|
raise NotImplementedError("Test worker does not implement this method")
|
|
|
|
|
|
|
|
|
|
|
|
|
glaslos
9 years ago