fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-09-19 02:48:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

Can spot difference between recycle and death now.

Browse Source 
Added recycle flag
Changed pids to names for readability
This commit is contained in:
Ilan Steemers
2015-07-06 10:42:38 +02:00
parent 92f1500808
commit 70ea211cfa
1 changed files with 10 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
django_q/cluster.py
View File

@@ -4,6 +4,7 @@ from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
from builtins import range
from future import standard_library
standard_library.install_aliases()
@@ -187,17 +188,19 @@ class Sentinel(object):
process.terminate()
if process == self.monitor:
self.monitor = self.spawn_monitor()
logger.error(_("reincarnated monitor {} after sudden.").format(process.pid))
logger.error(_("reincarnated monitor {} after sudden death").format(process.name))
elif process == self.pusher:
self.pusher = self.spawn_pusher()
logger.error(_("reincarnated pusher {} after sudden death.").format(process.pid))
logger.error(_("reincarnated pusher {} after sudden death").format(process.name))
else:
self.pool.remove(process)
self.spawn_worker()
if int(process.timer.value) >= self.timeout:
logger.warn(_("reincarnated worker {} after timeout.").format(process.pid))
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 sudden death.").format(process.pid))
logger.error(_("reincarnated worker {} after death").format(process.name))
self.reincarnations += 1
def spawn_cluster(self):
@@ -353,7 +356,8 @@ def worker(task_queue, done_queue, timer):
done_queue.put(task)
timer.value = -1 # Idle
# Recycle
if task_count == Conf.RECYCLE and task_queue.qsize() == 0:
if task_count == Conf.RECYCLE:
timer.value = -2
break
logger.info(_('{} stopped doing work').format(name))
@@ -430,4 +434,4 @@ def scheduler(list_key=Conf.Q_LIST):
logger.error(_('{} failed to create task from schedule {}').format(current_process().name, s.id))
else:
logger.info(_('{} created [{}] from schedule {}').format(current_process().name, s.task, s.id))
s.save()
s.save()