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

Rework - configuration now in class - better stop handling under heavy load. - scheduler test - better monitoring - better status reporting

Browse Source
This commit is contained in:
Ilan Steemers
2015-06-30 10:58:24 +02:00
parent e7784f0d18
commit 482569b312
6 changed files with 201 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/core.py
+80 -53
View File
@@ -34,8 +34,7 @@ from django.core import signing
from django.utils import timezone
# Local
from .conf import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, REDIS, Q_LIST, SIGNAL_NAMES, STARTING, \
RUNNING, STOPPING, STOPPED, RECYCLE, Q_STAT
from .conf import Conf
from .humanhash import uuid
from .models import Task, Success, Schedule
@@ -43,18 +42,18 @@ logger = logging.getLogger('django-q')
# Set up standard logging handler in case there is none
if not logger.handlers:
logger.setLevel(level=getattr(logging, LOG_LEVEL))
logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL))
formatter = logging.Formatter(fmt='%(asctime)s [Q] %(levelname)s %(message)s',
datefmt='%H:%M:%S')
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
redis_client = redis.StrictRedis(**REDIS)
redis_client = redis.StrictRedis(**Conf.REDIS)
class Cluster(object):
def __init__(self, list_key=Q_LIST):
def __init__(self, list_key=Conf.Q_LIST):
try:
redis_client.ping()
except Exception as e:
@@ -99,7 +98,7 @@ class Cluster(object):
return True
def sig_handler(self, signum, frame):
logger.debug('{} got signal {}'.format(current_process().name, SIGNAL_NAMES.get(signum, 'UNKNOWN')))
logger.debug('{} got signal {}'.format(current_process().name, Conf.SIGNAL_NAMES.get(signum, 'UNKNOWN')))
self.stop()
@property
@@ -130,7 +129,7 @@ class Cluster(object):
class Sentinel(object):
def __init__(self, stop_event, start_event, list_key=Q_LIST, start=True):
def __init__(self, stop_event, start_event, list_key=Conf.Q_LIST, start=True):
# Make sure we catch signals for the pool
signal.signal(signal.SIGINT, signal.SIG_IGN)
signal.signal(signal.SIGTERM, signal.SIG_DFL)
@@ -139,18 +138,17 @@ class Sentinel(object):
self.name = current_process().name
self.list_key = list_key
self.r = redis_client
self.status = None
self.reincarnations = 0
self.tob = timezone.now()
self.stop_event = stop_event
self.start_event = start_event
self.pool_size = WORKERS
self.pool_size = Conf.WORKERS
self.pool = []
self.task_queue = Queue()
self.done_queue = Queue()
self.event_out = Event()
self.monitor_pid = None
self.pusher_pid = None
self.monitor = None
self.pusher = None
if start:
self.start()
@@ -158,7 +156,22 @@ class Sentinel(object):
self.spawn_cluster()
self.guard()
def status(self):
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:
return Conf.IDLE
return Conf.WORKING
elif self.stop_event.is_set() and self.start_event.is_set():
if self.monitor.is_alive() or self.pusher.is_alive() or len(self.pool) > 0:
return Conf.STOPPING
return Conf.STOPPED
def spawn_process(self, target, *args):
"""
:type target: function or class
"""
# This is just for PyCharm to not crash. Ignore it.
if not hasattr(sys.stdin, 'close'):
def dummy_close():
@@ -167,9 +180,10 @@ class Sentinel(object):
sys.stdin.close = dummy_close
p = Process(target=target, args=args)
p.daemon = True
self.pool.append(p)
if target == worker:
self.pool.append(p)
p.start()
return p.pid
return p
def spawn_pusher(self):
return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key, self.r)
@@ -181,11 +195,11 @@ class Sentinel(object):
return self.spawn_process(monitor, self.done_queue)
def reincarnate(self, pid):
if pid == self.monitor_pid:
self.spawn_monitor()
if pid == self.monitor.pid:
self.monitor = self.spawn_monitor()
logger.warn("reincarnated monitor after death of {}".format(pid))
elif pid == self.pusher_pid:
self.spawn_pusher()
elif pid == self.pusher.pid:
self.pusher = self.spawn_pusher()
logger.warn("reincarnated pusher after death of {}".format(pid))
else:
self.spawn_worker()
@@ -193,60 +207,72 @@ class Sentinel(object):
self.reincarnations += 1
def spawn_cluster(self):
self.set_status(STARTING)
Stat(self).save()
for i in range(self.pool_size):
self.spawn_worker()
self.monitor_pid = self.spawn_monitor()
self.pusher_pid = self.spawn_pusher()
self.monitor = self.spawn_monitor()
self.pusher = self.spawn_pusher()
def guard(self):
logger.info('{} guarding cluster at {}'.format(current_process().name, self.pid))
self.start_event.set()
self.set_status(RUNNING)
Stat(self).save()
logger.info('Q Cluster-{} running.'.format(self.parent_pid))
scheduler(list_key=self.list_key)
counter = 0
while True:
# Guard loop. Runs at least once
while not self.stop_event.is_set() or not counter:
# Check Workers
for p in list(self.pool):
if not p.is_alive():
p.terminate()
self.pool.remove(p)
self.reincarnate(p.pid)
Stat(self).save()
if self.stop_event.is_set():
break
# Check Monitor
if not self.monitor.is_alive():
self.reincarnate(self.monitor.pid)
# Check Pusher
if not self.pusher.is_alive():
self.reincarnate(self.monitor.pid)
# Call scheduler once a minute (or so)
counter += 1
if counter > 30:
if counter > 60:
counter = 0
scheduler()
sleep(2)
scheduler(list_key=self.list_key)
# Save current status
Stat(self).save()
sleep(1)
self.stop()
def stop(self):
self.set_status(STOPPING)
Stat(self).save()
name = current_process().name
logger.info('{} stopping pool processes'.format(name))
# Stopping pusher
self.event_out.set()
while self.pusher.is_alive():
sleep(0.2)
Stat(self).save()
# Putting poison pills in the queue
for _ in self.pool:
for _ in range(self.pool_size):
self.task_queue.put('STOP')
while len(self.pool) > 2:
for p in list(self.pool):
# Wait for all the workers to exit
while len(self.pool) > 0:
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:
sleep(0.2)
Stat(self).save()
self.pool = []
self.set_status(STOPPED)
def set_status(self, message=None):
Stat(self, message).save()
def pusher(task_queue, e, list_key=Q_LIST, r=redis_client):
def pusher(task_queue, e, list_key=Conf.Q_LIST, r=redis_client):
"""
Pulls tasks of the Redis List and puts them in the task queue
:type task_queue: multiprocessing.Queue
@@ -325,7 +351,7 @@ def worker(task_queue, done_queue):
task['stopped'] = timezone.now()
done_queue.put(task)
# Recycle
if task_count == RECYCLE:
if task_count == Conf.RECYCLE and task_queue.qsize() == 0:
break
logger.info('{} stopped doing work'.format(name))
@@ -335,10 +361,10 @@ def save_task(task):
Saves the task package to Django
"""
# SAVE LIMIT < 0 : Don't save success
if SAVE_LIMIT < 0 and task['success']:
if Conf.SAVE_LIMIT < 0 and task['success']:
return
# SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
if task['success'] and 0 < SAVE_LIMIT < Success.objects.count():
if task['success'] and 0 < Conf.SAVE_LIMIT < Success.objects.count():
Success.objects.first().delete()
try:
@@ -370,7 +396,7 @@ def async(func, *args, **kwargs):
list_key = kwargs['list_key']
del kwargs['list_key']
else:
list_key = Q_LIST
list_key = Conf.Q_LIST
# Check for redis connection override
if 'redis' in kwargs:
r = kwargs['redis']
@@ -390,9 +416,9 @@ class SignedPackage(object):
"""
@staticmethod
def dumps(obj, compressed=COMPRESSED):
def dumps(obj, compressed=Conf.COMPRESSED):
return signing.dumps(obj,
key=SECRET_KEY,
key=Conf.SECRET_KEY,
salt='django_q.q',
compress=compressed,
serializer=PickleSerializer)
@@ -400,7 +426,7 @@ class SignedPackage(object):
@staticmethod
def loads(obj):
return signing.loads(obj,
key=SECRET_KEY,
key=Conf.SECRET_KEY,
salt='django_q.q',
serializer=PickleSerializer)
@@ -445,19 +471,19 @@ class Stat(Status):
Status object for Cluster monitoring
"""
def __init__(self, sentinel, message=None):
def __init__(self, sentinel):
super(Stat, self).__init__(sentinel.parent_pid)
if message:
sentinel.status = message
self.r = sentinel.r
self.tob = sentinel.tob
self.reincarnations = sentinel.reincarnations
self.sentinel = sentinel.pid
self.status = sentinel.status
self.status = sentinel.status()
self.done_q_size = sentinel.done_queue.qsize()
self.monitor = sentinel.monitor_pid
if sentinel.monitor:
self.monitor = sentinel.monitor.pid
self.task_q_size = sentinel.task_queue.qsize()
self.pusher = sentinel.pusher_pid
if sentinel.pusher:
self.pusher = sentinel.pusher.pid
for w in sentinel.pool:
self.workers.append(w.pid)
@@ -477,7 +503,7 @@ class Stat(Status):
:param cluster_id: cluster ID
:return: redis key for the cluster statistic
"""
return '{}:{}'.format(Q_STAT, cluster_id)
return '{}:{}'.format(Conf.Q_STAT, cluster_id)
def save(self):
self.r.set(self.key, SignedPackage.dumps(self, True), 3)
@@ -508,7 +534,7 @@ class Stat(Status):
:return: Stat list
"""
stats = []
keys = r.keys(pattern='{}:*'.format(Q_STAT))
keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT))
if keys:
packs = r.mget(keys)
for pack in packs:
@@ -525,7 +551,7 @@ class Stat(Status):
return state
def scheduler():
def scheduler(list_key=Conf.Q_LIST):
"""
Creates a task from a schedule at the scheduled time and schedules next run
"""
@@ -566,6 +592,7 @@ def scheduler():
else:
schedule.repeats = 0
# send it to the cluster
kwargs['list_key'] = list_key
schedule.task = async(schedule.func, *args, **kwargs)
if not schedule.task:
logger.error('{} failed to create task from schedule {}').format(current_process().name, schedule.id)