From 90b5b0d6ab8be8abd352bd10eaaa98580dc7f5a9 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 18 Jun 2015 14:10:48 +0200 Subject: [PATCH] moved static methods to module functions medic is now sentinel and runs in it's own process sentinel shutdown procedure more stable --- django_q/__init__.py | 2 +- django_q/apps.py | 2 + django_q/main.py | 227 +++++++++++++++++++++++ django_q/management/commands/qcluster.py | 2 + django_q/q.py | 224 ---------------------- django_q/tests/tasks.py | 2 +- django_q/tests/test_qworker.py | 12 +- pytest.ini | 3 +- requirements.txt | 1 + 9 files changed, 244 insertions(+), 231 deletions(-) create mode 100644 django_q/main.py delete mode 100644 django_q/q.py diff --git a/django_q/__init__.py b/django_q/__init__.py index d695d7b..501ddf1 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -4,6 +4,6 @@ Author: Ilan Steemers (koed00@gmail.com Github: https://github.com/Koed00/django-q """ -from django_q.q import Cluster, async +from .main import Cluster, async default_app_config = 'django_q.apps.SessionAdminConfig' diff --git a/django_q/apps.py b/django_q/apps.py index f029166..e0e58a4 100644 --- a/django_q/apps.py +++ b/django_q/apps.py @@ -9,6 +9,8 @@ class SessionAdminConfig(AppConfig): verbose_name = "Django Q" +VERSION = '0.1.0' + """ Sets the logging level for the app """ diff --git a/django_q/main.py b/django_q/main.py new file mode 100644 index 0000000..14fb4b9 --- /dev/null +++ b/django_q/main.py @@ -0,0 +1,227 @@ +import importlib +import logging +import signal +from multiprocessing import Queue, Event, Process, current_process +import sys +from time import sleep + +import jsonpickle +import coloredlogs +from django.core.signing import BadSignature + +from django.utils import timezone +import redis + +from django.core import signing + +from .apps import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, VERSION +from .humanhash import uuid +from .models import Task, Success + +SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n) + +prefix = 'django_q' +q_list = '{}:q'.format(prefix) +logger = logging.getLogger('django-q') +coloredlogs.install(level=getattr(logging, LOG_LEVEL)) + +r = redis.StrictRedis() + + +class Cluster(object): + def __init__(self): + try: + r.ping() + except (): + logger.error('Can not connect to Redis server') + return + self.pool_size = WORKERS + self.pool = [] + self.task_queue = Queue() + self.done_queue = Queue() + self.event_stop = Event() + self.monitor_pid = None + self.pusher_pid = None + self.sentinel_pid = None + + def spawn_process(self, target, *args): + # This is just for PyCharm to not crash. Ignore it. + if not hasattr(sys.stdin, 'close'): + def dummy_close(): + pass + + sys.stdin.close = dummy_close + p = Process(target=target, args=args) + self.pool.append(p) + p.start() + return p.pid + + def spawn_pusher(self): + return self.spawn_process(pusher, self.task_queue, self.event_stop) + + def spawn_worker(self): + self.spawn_process(worker, self.task_queue, self.done_queue) + + def spawn_monitor(self): + return self.spawn_process(monitor, self.done_queue) + + def spawn_sentinel(self): + return self.spawn_process(self.sentinel, self.event_stop) + + def reincarnate(self, pid): + if pid == self.monitor_pid: + self.spawn_monitor() + logger.warn("reincarnated monitor after death of {}".format(pid)) + elif pid == self.pusher_pid: + self.spawn_pusher() + logger.warn("reincarnated pusher after death of {}".format(pid)) + else: + self.spawn_worker() + logger.warn("reincarnated work worker after death of {}".format(pid)) + + def sentinel(self, e): + for i in range(self.pool_size): + self.spawn_worker() + self.monitor_pid = self.spawn_monitor() + self.pusher_pid = self.spawn_pusher() + logger.info('{} checking worker health at {}'.format(current_process().name, current_process().pid)) + while not e.is_set(): + for p in list(self.pool): + if not p.is_alive(): + # Be humane + p.terminate() + self.pool.remove(p) + # Replace it with a fresh one + self.reincarnate(p.pid) + sleep(1) + self.stop() + + def start(self): + signal.signal(signal.SIGTERM, self.sig_handler) + signal.signal(signal.SIGINT, self.sig_handler) + self.sentinel_pid = self.spawn_sentinel() + logger.info('Starting Q cluster version {} at {}'.format(VERSION, current_process().pid)) + + def stop(self): + # Send the STOP signal to the pool + name = current_process().name + logger.info('{} stopping pool processes'.format(name)) + # Stopping pusher + self.event_stop.set() + # Stopping monitor + self.done_queue.put('STOP') + # Stopping workers + for _ in self.pool: + self.task_queue.put('STOP') + + def sig_handler(self, signum, frame): + logger.debug('{} got signal {}'.format(current_process().name, SIGNAL_NAMES.get(signum, 'UNKNOWN'))) + self.event_stop.set() + + +def pusher(task_queue, e): + logger.info('{} pushing tasks at {}'.format(current_process().name, current_process().pid)) + while not e.is_set(): + task = r.blpop(q_list, 1) + if task: + task = task[1] + task_queue.put(task) + logger.debug('queueing {}'.format(task)) + logger.info("{} stopped pushing".format(current_process().name)) + + +def monitor(done_queue): + name = current_process().name + logger.info("{} monitoring at {}".format(name, current_process().pid)) + for task in iter(done_queue.get, 'STOP'): + name = task[0] + func = task[1] + result = task[6] + success = task[7] + if success: + logger.info("Finished [{}:{}]".format(func, name)) + else: + logger.error("Failed [{}:{}] - {}".format(func, name, result)) + save_task(task) + logger.info("{} stopped monitoring".format(name)) + + +def worker(task_queue, done_queue): + name = current_process().name + logger.info('{} ready for work at {}'.format(name, current_process().pid)) + for pack in iter(task_queue.get, 'STOP'): + # unpickle the task + try: + task = signing.loads(pack, key=SECRET_KEY, salt='django_q.q', serializer=JSONPickleSerializer) + except TypeError as e: + logger.error(e) + continue + except BadSignature as e: + task[0] = task[0].rsplit(":", 1)[0] + task.append(timezone.now()) + task.append(e) + task.append(False) + done_queue.put(task) + continue + func = task[1] + module, func = func.rsplit('.', 1) + args = task[2] + kwargs = task[3] + logger.info('{} processing [{}:{}]'.format(name, func, task[0])) + task.append(timezone.now()) + try: + m = importlib.import_module(module) + f = getattr(m, func) + result = f(*args, **kwargs) + task.append(result) + task.append(True) + done_queue.put(task) + except Exception as e: + task.append(e) + task.append(False) + done_queue.put(task) + logger.info('{} stopped working'.format(name)) + + +def save_task(task): + if task[7] and 0 < SAVE_LIMIT < Success.objects.count(): + Success.objects.first().delete() + Task.objects.create(name=task[0], + func=task[1], + args=task[2], + kwargs=task[3], + started=task[4], + stopped=task[5], + result=task[6], + success=task[7]) + + +def async(func, *args, **kwargs): + """ + Schedules a module function + [name, func, args, kwargs, started, finished, result, success] + """ + name = uuid()[0] + pack = signing.dumps([name, func, args, kwargs, timezone.now()], + key=SECRET_KEY, + salt='django_q.q', + compress=COMPRESSED, + serializer=JSONPickleSerializer) + r.rpush(q_list, pack) + logger.debug('Pushed {}'.format(pack)) + return name + + +class JSONPickleSerializer(object): + """ + Simple wrapper around jsonpickle to be used in signing.dumps and + signing.loads. + """ + + @staticmethod + def dumps(obj): + return jsonpickle.dumps(obj).encode('latin-1') + + @staticmethod + def loads(data): + return jsonpickle.loads(data.decode('latin-1')) diff --git a/django_q/management/commands/qcluster.py b/django_q/management/commands/qcluster.py index 1016a52..671e0c7 100644 --- a/django_q/management/commands/qcluster.py +++ b/django_q/management/commands/qcluster.py @@ -7,3 +7,5 @@ class Command(BaseCommand): def handle(self, *args, **options): q = Cluster() + q.start() + diff --git a/django_q/q.py b/django_q/q.py deleted file mode 100644 index 1a24a8c..0000000 --- a/django_q/q.py +++ /dev/null @@ -1,224 +0,0 @@ -import importlib -import logging -import signal -from multiprocessing import Queue, Event, Process, current_process -import sys -from time import sleep - -import jsonpickle -import coloredlogs -from django.core.signing import BadSignature - -from django.utils import timezone -import redis - -from django.core import signing - -from django_q.apps import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED -from django_q.humanhash import uuid -from django_q.models import Task, Success - -prefix = 'django_q' -q_list = '{}:q'.format(prefix) -logger = logging.getLogger('django-q') -coloredlogs.install(level=getattr(logging, LOG_LEVEL)) - -r = redis.StrictRedis() - - -class Cluster(object): - def __init__(self): - signal.signal(signal.SIGTERM, self.sig_handler) - signal.signal(signal.SIGINT, self.sig_handler) - try: - r.ping() - except (): - logger.error('Can not connect to Redis server') - return - self.running = True - self.pool_size = WORKERS - self.pool = [] - self.task_queue = Queue() - self.done_queue = Queue() - # Spawn workers - for i in range(self.pool_size): - self.spawn_worker() - # Spawn monitor - self.monitor_pid = None - self.spawn_monitor() - # Spawn pusher - self.pusher_pid = None - self.pusher_stop = Event() - self.spawn_pusher() - # Monitor process health - while self.running: - self.medic() - sleep(1) - - def spawn_process(self, target, *args): - # This is just for PyCharm to not crash. Ignore it. - if not hasattr(sys.stdin, 'close'): - def dummy_close(): - pass - - sys.stdin.close = dummy_close - p = Process(target=target, args=args) - self.pool.append(p) - p.start() - return p.pid - - def spawn_pusher(self): - self.pusher_pid = self.spawn_process(self.pusher, self.task_queue, self.pusher_stop) - - def spawn_worker(self): - self.spawn_process(self.worker, self.task_queue, self.done_queue) - - def spawn_monitor(self): - self.monitor_pid = self.spawn_process(self.monitor, self.done_queue) - - def reincarnate(self, pid): - if pid == self.monitor_pid: - self.spawn_monitor() - logger.warn("reincarnated monitor after death of {}".format(pid)) - elif pid == self.pusher_pid: - self.spawn_pusher() - logger.warn("reincarnated pusher after death of {}".format(pid)) - else: - self.spawn_worker() - logger.warn("reincarnated work worker after death of {}".format(pid)) - - @staticmethod - def pusher(task_queue, e): - logger.info('{} pushing tasks at {}'.format(current_process().name, current_process().pid)) - while not e.is_set(): - task = r.blpop(q_list, 1) - if task: - task = task[1] - task_queue.put(task) - logger.debug('queueing {}'.format(task)) - - @staticmethod - def monitor(done_queue): - name = current_process().name - logger.info("{} monitoring at {}".format(name, current_process().pid)) - for task in iter(done_queue.get, 'STOP'): - name = task[0] - func = task[1] - result = task[6] - success = task[7] - if success: - logger.info("Finished [{}:{}]".format(func, name)) - else: - logger.error("Failed [{}:{}] - {}".format(func, name, result)) - Cluster.save_task(task) - logger.info("{} stopped".format(name)) - - @staticmethod - def worker(task_queue, done_queue): - name = current_process().name - logger.info('{} ready for work at {}'.format(name, current_process().pid)) - for pack in iter(task_queue.get, 'STOP'): - # unpickle the task - try: - task = signing.loads(pack, key=SECRET_KEY, salt='django_q.q', serializer=JSONPickleSerializer) - except TypeError as e: - logger.error(e) - continue - except BadSignature as e: - task[0] = task[0].rsplit(":", 1)[0] - task.append(timezone.now()) - task.append(e) - task.append(False) - done_queue.put(task) - continue - func = task[1] - module, func = func.rsplit('.', 1) - args = task[2] - kwargs = task[3] - logger.info('{} processing [{}:{}]'.format(name, func, task[0])) - task.append(timezone.now()) - try: - m = importlib.import_module(module) - f = getattr(m, func) - result = f(*args, **kwargs) - task.append(result) - task.append(True) - done_queue.put(task) - except Exception as e: - task.append(e) - task.append(False) - done_queue.put(task) - logger.info('{} Stopped'.format(name)) - - def medic(self): - # Check if all the workers are alive - for p in list(self.pool): - if not p.is_alive(): - # Be humane - p.terminate() - self.pool.remove(p) - # Replace it with a fresh one - self.reincarnate(p.pid) - - @staticmethod - def save_task(task): - if task[7] and 0 < SAVE_LIMIT < Success.objects.count(): - Success.objects.first().delete() - Task.objects.create(name=task[0], - func=task[1], - args=task[2], - kwargs=task[3], - started=task[4], - stopped=task[5], - result=task[6], - success=task[7]) - - def stop(self): - # Send the STOP signal to the pool - self.running = False - logger.info('Stopping') - # Wait for all the workers to finish the queue - for p in self.pool: - if p.pid == self.monitor_pid: - self.done_queue.put('STOP') - elif p.pid == self.pusher_pid: - self.pusher_stop.set() - else: - self.task_queue.put('STOP') - p.join() - - logger.info('Goodbye. Have a wonderful time.') - - def sig_handler(self, signum, frame): - self.stop() - - -def async(func, *args, **kwargs): - """ - Schedules a module function - [name, func, args, kwargs, started, finished, result, success] - """ - name = uuid()[0] - pack = signing.dumps([name, func, args, kwargs, timezone.now()], - key=SECRET_KEY, - salt='django_q.q', - compress=COMPRESSED, - serializer=JSONPickleSerializer) - r.rpush(q_list, pack) - logger.debug('Pushed {}'.format(pack)) - return name - - -class JSONPickleSerializer(object): - """ - Simple wrapper around jsonpickle to be used in signing.dumps and - signing.loads. - """ - - @staticmethod - def dumps(obj): - return jsonpickle.dumps(obj).encode('latin-1') - - @staticmethod - def loads(data): - return jsonpickle.loads(data.decode('latin-1')) diff --git a/django_q/tests/tasks.py b/django_q/tests/tasks.py index c5463a4..1c6d0fe 100644 --- a/django_q/tests/tasks.py +++ b/django_q/tests/tasks.py @@ -1,7 +1,7 @@ import time # simple countdown -def count(n): +def countdown(n): start_time = time.time() while n > 0: n -= 1 diff --git a/django_q/tests/test_qworker.py b/django_q/tests/test_qworker.py index bab5a9a..14c4222 100644 --- a/django_q/tests/test_qworker.py +++ b/django_q/tests/test_qworker.py @@ -1,11 +1,15 @@ +import sys, os +myPath = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, myPath + '/../') + import pytest from django_q import Cluster -@pytest.fixture -def qworker(): +@pytest.fixture(scope='session') +def cluster(): return Cluster() -def test_worker(qworker): - assert len(qworker.stable) == qworker.stable_size +def test_worker(cluster): + assert len(cluster.stable) == cluster.stable_size diff --git a/pytest.ini b/pytest.ini index 676f86b..748d0b0 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,2 +1,3 @@ [pytest] -DJANGO_SETTINGS_MODULE=.settings \ No newline at end of file +DJANGO_SETTINGS_MODULE=django_q.tests.settings +#django_find_project = false \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 10a3140..ea8b26b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ +-e . coloredlogs==1.0.1 django-picklefield==0.3.1 Django==1.8.2