From 38dc20043c90fe8facf6c3a6105e41881c2087e3 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Tue, 7 Jul 2015 15:48:13 +0200 Subject: [PATCH 1/6] Stabilize the stop procedure after forced termination of the monitor causes an unstable queue. * added timeout to final stop * ony terminate on timeouts * renamed done_queue to result_queue --- django_q/cluster.py | 57 ++++++++++++++++++++++++++++----------------- django_q/monitor.py | 8 +++---- 2 files changed, 39 insertions(+), 26 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 0e45695..f037449 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -129,10 +129,10 @@ class Sentinel(object): self.pool = [] self.timeout = timeout self.task_queue = Queue() - self.done_queue = Queue() + self.result_queue = Queue() self.event_out = Event() - self.monitor = None - self.pusher = None + self.monitor = Process() + self.pusher = Process() if start: self.start() @@ -144,7 +144,7 @@ class Sentinel(object): if not self.start_event.is_set() and not self.stop_event.is_set(): return Conf.STARTING elif self.start_event.is_set() and not self.stop_event.is_set(): - if self.done_queue.qsize() == 0 and self.task_queue.qsize() == 0: + if self.result_queue.qsize() == 0 and self.task_queue.qsize() == 0: return Conf.IDLE return Conf.WORKING elif self.stop_event.is_set() and self.start_event.is_set(): @@ -174,17 +174,16 @@ class Sentinel(object): return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key, self.r) def spawn_worker(self): - self.spawn_process(worker, self.task_queue, self.done_queue, Value('b', -1)) + self.spawn_process(worker, self.task_queue, self.result_queue, Value('b', -1)) def spawn_monitor(self): - return self.spawn_process(monitor, self.done_queue) + return self.spawn_process(monitor, self.result_queue) def reincarnate(self, process): """ :param process: the process to reincarnate - :type process: Process or None + :type process: Process """ - process.terminate() if process == self.monitor: self.monitor = self.spawn_monitor() logger.error(_("reincarnated monitor {} after sudden death").format(process.name)) @@ -195,11 +194,14 @@ class Sentinel(object): self.pool.remove(process) self.spawn_worker() if int(process.timer.value) >= self.timeout: + # only need to terminate on timeout, otherwise we risk destabilizing the queues + process.terminate() logger.warn(_("reincarnated worker {} after timeout").format(process.name)) elif int(process.timer.value) == -2: logger.info(_("recycled worker {}").format(process.name)) else: logger.error(_("reincarnated worker {} after death").format(process.name)) + self.reincarnations += 1 def spawn_cluster(self): @@ -235,12 +237,12 @@ class Sentinel(object): self.reincarnate(self.pusher) # Call scheduler once a minute (or so) counter += 1 - if counter > 60: + if counter > 120: counter = 0 scheduler(list_key=self.list_key) # Save current status Stat(self).save() - sleep(1) + sleep(0.5) self.stop() def stop(self): @@ -254,21 +256,32 @@ class Sentinel(object): sleep(0.2) Stat(self).save() # Put poison pills in the queue - for _ in range(self.pool_size): + for _ in range(len(self.pool)): self.task_queue.put('STOP') + self.task_queue.close() + # wait for the task queue to empty + self.task_queue.join_thread() # Wait for all the workers to exit - while len(self.pool) > 0: + while len(self.pool): for p in self.pool: if not p.is_alive(): - logger.debug('{} stopped gracefully'.format(p.pid)) self.pool.remove(p) sleep(0.2) Stat(self).save() # Finally stop the monitor - self.done_queue.put('STOP') - while self.status() != Conf.STOPPED: + self.result_queue.put('STOP') + self.result_queue.close() + # Wait for the result queue to empty + self.result_queue.join_thread() + logger.info('{} waiting for the monitor.'.format(name)) + count = 0 + # Wait for everything to close or time out + while self.status() == Conf.STOPPING and count < self.timeout * 5: sleep(0.2) Stat(self).save() + count += 1 + # Final status + Stat(self).save() self.pool = [] @@ -291,27 +304,27 @@ def pusher(task_queue, e, list_key=Conf.Q_LIST, r=redis_client): logger.info(_("{} stopped pushing tasks").format(current_process().name)) -def monitor(done_queue): +def monitor(result_queue): """ Gets finished tasks from the result queue and saves them to Django - :type done_queue: multiprocessing.Queue + :type result_queue: multiprocessing.Queue """ name = current_process().name logger.info(_("{} monitoring at {}").format(name, current_process().pid)) - for task in iter(done_queue.get, 'STOP'): + for task in iter(result_queue.get, 'STOP'): + save_task(task) if task['success']: logger.info(_("Processed [{}]").format(task['name'])) else: logger.error(_("Failed [{}] - {}").format(task['name'], task['result'])) - save_task(task) logger.info(_("{} stopped monitoring results").format(name)) -def worker(task_queue, done_queue, timer): +def worker(task_queue, result_queue, timer): """ Takes a task from the task queue, tries to execute it and puts the result back in the result queue :type task_queue: multiprocessing.Queue - :type done_queue: multiprocessing.Queue + :type result_queue: multiprocessing.Queue :type timer: multiprocessing.Value """ name = current_process().name @@ -352,7 +365,7 @@ def worker(task_queue, done_queue, timer): task['result'] = result[0] task['success'] = result[1] task['stopped'] = timezone.now() - done_queue.put(task) + result_queue.put(task) timer.value = -1 # Idle # Recycle if task_count == Conf.RECYCLE: diff --git a/django_q/monitor.py b/django_q/monitor.py index d848dc7..4540b5a 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -9,8 +9,8 @@ from django.utils import timezone from django.utils.translation import ugettext as _ # local -from .conf import Conf, redis_client, logger -from .tasks import SignedPackage +from django_q.conf import Conf, redis_client, logger +from django_q.tasks import SignedPackage def monitor(run_once=False): @@ -113,13 +113,13 @@ class Stat(Status): """ def __init__(self, sentinel): - super(Stat, self).__init__(sentinel.parent_pid) + super(Stat, self).__init__(sentinel.parent_pid or sentinel.pid) self.r = sentinel.r self.tob = sentinel.tob self.reincarnations = sentinel.reincarnations self.sentinel = sentinel.pid self.status = sentinel.status() - self.done_q_size = sentinel.done_queue.qsize() + self.done_q_size = sentinel.result_queue.qsize() if sentinel.monitor: self.monitor = sentinel.monitor.pid self.task_q_size = sentinel.task_queue.qsize() From 33b7fe50ed9274cbe2c44821fdc330395bba85d6 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Tue, 7 Jul 2015 16:45:05 +0200 Subject: [PATCH 2/6] fix reincarnation bug without timeout set --- django_q/cluster.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index f037449..d981b9f 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -193,7 +193,7 @@ class Sentinel(object): else: self.pool.remove(process) self.spawn_worker() - if int(process.timer.value) >= self.timeout: + if self.timeout and int(process.timer.value) >= self.timeout: # only need to terminate on timeout, otherwise we risk destabilizing the queues process.terminate() logger.warn(_("reincarnated worker {} after timeout").format(process.name)) @@ -205,6 +205,7 @@ class Sentinel(object): self.reincarnations += 1 def spawn_cluster(self): + self.pool = [] Stat(self).save() for i in range(self.pool_size): self.spawn_worker() @@ -282,7 +283,6 @@ class Sentinel(object): count += 1 # Final status Stat(self).save() - self.pool = [] def pusher(task_queue, e, list_key=Conf.Q_LIST, r=redis_client): From b801c480f5c179445d5b3fe6ad1fe97edd02a5e2 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Tue, 7 Jul 2015 17:47:49 +0200 Subject: [PATCH 3/6] Handle Redis errors better --- django_q/cluster.py | 8 +++++++- django_q/monitor.py | 5 ++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index d981b9f..35efd03 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -294,7 +294,13 @@ def pusher(task_queue, e, list_key=Conf.Q_LIST, r=redis_client): """ logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid)) while True: - task = r.blpop(list_key, 1) + try: + task = r.blpop(list_key, 1) + except Exception as e: + logger.error(e) + # redis probably crashed. Let the sentinel handle it. + sleep(10) + break if task: task = task[1] task_queue.put(task) diff --git a/django_q/monitor.py b/django_q/monitor.py index 4540b5a..a602341 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -147,7 +147,10 @@ class Stat(Status): return '{}:{}'.format(Conf.Q_STAT, cluster_id) def save(self): - self.r.set(self.key, SignedPackage.dumps(self, True), 3) + try: + self.r.set(self.key, SignedPackage.dumps(self, True), 3) + except Exception as e: + logger.error(e) def empty_queues(self): return self.done_q_size + self.task_q_size == 0 From b05fe76fbca6fa2fff1e25a35043992b35539b85 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Tue, 7 Jul 2015 17:54:24 +0200 Subject: [PATCH 4/6] log database errors more quiet --- django_q/cluster.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 35efd03..3eed195 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -402,7 +402,7 @@ def save_task(task): result=task['result'], success=task['success']) except Exception as e: - logger.exception(e) + logger.error(e) def scheduler(list_key=Conf.Q_LIST): From 4ee47fc38f0f02c9e4feb1ccbb1f6cbde4112ea6 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Tue, 7 Jul 2015 20:20:28 +0200 Subject: [PATCH 5/6] upping the coverage --- django_q/conf.py | 6 ++---- django_q/monitor.py | 4 +--- django_q/tests/test_cluster.py | 3 +++ django_q/tests/test_scheduler.py | 1 + 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/django_q/conf.py b/django_q/conf.py index 396f937..b195637 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -46,10 +46,8 @@ class Conf(object): LABEL = conf.get('label', 'Django Q') # Use the secret key for package signing - try: - SECRET_KEY = settings.SECRET_KEY - except AttributeError: - SECRET_KEY = 'omgicantbelieveudonthaveasecretkey' + # Django itself should raise an error if it's not configured + SECRET_KEY = settings.SECRET_KEY # The redis list key Q_LIST = 'django_q:{}:q'.format(PREFIX) diff --git a/django_q/monitor.py b/django_q/monitor.py index a602341..e946f61 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -52,8 +52,6 @@ def monitor(run_once=False): status = term.red(str(Conf.STOPPED)) elif stat.status == Conf.IDLE: status = str(Conf.IDLE) - else: - status = term.yellow(str(stat.status)) # color q's tasks = stat.task_q_size if tasks > 0: @@ -98,7 +96,7 @@ class Status(object): self.reincarnations = 0 self.cluster_id = pid self.sentinel = 0 - self.status = 'Idle' + self.status = Conf.STOPPED self.done_q_size = 0 self.host = socket.gethostname() self.monitor = 0 diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 1e0f65f..5d0055a 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -39,9 +39,12 @@ def test_cluster_initial(r): r.delete(list_key) c = Cluster(list_key=list_key) assert c.sentinel is None + assert c.stat.status == Conf.STOPPED assert c.start() > 0 assert c.sentinel.is_alive() is True assert c.is_running + assert c.is_stopping is False + assert c.is_starting is False stat = c.stat assert stat.status == Conf.IDLE assert c.stop() is True diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index e10db1c..93f1239 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -59,3 +59,4 @@ def test_scheduler(r): ) assert schedule is not None assert schedule.last_run() is None + scheduler() From 79d254317c9d970a917f6f124b622eff9e1d2aee Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Tue, 7 Jul 2015 20:22:59 +0200 Subject: [PATCH 6/6] bumping to version 0.2.2 --- django_q/__init__.py | 2 +- setup.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/django_q/__init__.py b/django_q/__init__.py index a30c5c7..e622c82 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -1,6 +1,6 @@ from .tasks import async, schedule, result, fetch from .models import Task, Schedule -VERSION = (0, 2, 1) +VERSION = (0, 2, 2) default_app_config = 'django_q.apps.DjangoQConfig' diff --git a/setup.py b/setup.py index 7eef2aa..a65d299 100644 --- a/setup.py +++ b/setup.py @@ -26,7 +26,7 @@ class PyTest(Command): setup( name='django-q', - version='0.2.1.1', + version='0.2.2', author='Ilan Steemers', author_email='koed00@gmail.com', packages=['django_q'],