From 2e81ba2858086b7a1ceba16bb859e0b8bc4f4e72 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 17 Jun 2015 19:14:20 +0200 Subject: [PATCH] Added some config settings Updated README --- README.md | 36 ++++++++++++++++++++++++++++-- README.rst | 58 +++++++++++++++++++++++++++++++++++++++++++++++- django_q/apps.py | 16 +++++++++++++ django_q/q.py | 20 ++++++++++++----- 4 files changed, 121 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 99d3679..21bd895 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,36 @@ # Django Q ##A multiprocessing task queue application for Django -Currently in the pre-alpha stage and Python 3 only for now. -![Django Q schema](http://i.imgur.com/jYRb1mJ.png) \ No newline at end of file +### Status +Currently in the pre-alpha stage and Python 3 only (for now). + +### Architecture +![Django Q schema](http://i.imgur.com/jYRb1mJ.png) +### Signed Tasks +Tasks are first pickled to Json and then signed using Django's own signing module before being sent to a Redis list. This ensures that task packages on the Redis server can only be excuted and read by clusters and django servers who share the same secret key. + +Optionally, packages can be compressed before transport by setting `Q_COMPRESSED = True ` + +### Pusher +The pusher process continously checks the Redis list for new task packages and pushes them on the Task Queue. + +### Worker +A worker process checks the package signing, unpacks the task, executes it and saves the return value. Irrespective of the failure or success of any of these steps, the package is then pushed onto the Result Queue. + +By default Django Q spawns a worker for each detected CPU on the host system. +This can be overridden by setting `Q_WORKERS = n`. With *n* being the numbered of desired worker processes. + +### Monitor +The result monitor checks the Result Queue for processed packages and saves both failed and succesful packages to the Django database. + +By default only the last 100 succesful packages are kept in the database. +This can be increased or decreased at will by settings `Q_SAVE_LIMIT = n`. With *n* being the desired number of records. +Set `Q_SAVE_LIMIT = 0` to save all results to the database. +Failed packages are always saved. + +### Medic + +The medic loop checks the health of all workers, including the pusher and the monitor. +In case one of them dies, the medic will reincarnate a new process to take over the duties of the deceased. + +### Todo +I'll add to this README while I'm developing the various parts. \ No newline at end of file diff --git a/README.rst b/README.rst index 6e45ec7..e5e3106 100644 --- a/README.rst +++ b/README.rst @@ -4,6 +4,62 @@ Django Q A multiprocessing task queue application for Django --------------------------------------------------- -Currently in the pre-alpha stage. |Django Q schema| +Status +~~~~~~ + +Currently in the pre-alpha stage and Python 3 only (for now). + +Architecture +~~~~~~~~~~~~ + +|Django Q schema| ### Signed Tasks Tasks are first pickled to Json and +then signed using Django's own signing module before being sent to a +Redis list. This ensures that task packages on the Redis server can only +be excuted and read by clusters and django servers who share the same +secret key. + +Optionally, packages can be compressed before transport by setting +``Q_COMPRESSED = True`` + +Pusher +~~~~~~ + +The pusher process continously checks the Redis list for new task +packages and pushes them on the Task Queue. + +Worker +~~~~~~ + +A worker process checks the package signing, unpacks the task, executes +it and saves the return value. Irrespective of the failure or success of +any of these steps, the package is then pushed onto the Result Queue. + +By default Django Q spawns a worker for each detected CPU on the host +system. This can be overridden by setting ``Q_WORKERS = n``. With *n* +being the numbered of desired worker processes. + +Monitor +~~~~~~~ + +The result monitor checks the Result Queue for processed packages and +saves both failed and succesful packages to the Django database. + +By default only the last 100 succesful packages are kept in the +database. This can be increased or decreased at will by settings +``Q_SAVE_LIMIT = n``. With *n* being the desired number of records. Set +``Q_SAVE_LIMIT = 0`` to save all results to the database. Failed +packages are always saved. + +Medic +~~~~~ + +The medic loop checks the health of all workers, including the pusher +and the monitor. In case one of them dies, the medic will reincarnate a +new process to take over the duties of the deceased. + +Todo +~~~~ + +I'll add to this README while I'm developing the various parts. .. |Django Q schema| image:: http://i.imgur.com/jYRb1mJ.png diff --git a/django_q/apps.py b/django_q/apps.py index 8ee09ea..f029166 100644 --- a/django_q/apps.py +++ b/django_q/apps.py @@ -1,3 +1,5 @@ +from multiprocessing import cpu_count + from django.apps import AppConfig from django.conf import settings @@ -6,6 +8,7 @@ class SessionAdminConfig(AppConfig): name = 'django_q' verbose_name = "Django Q" + """ Sets the logging level for the app """ @@ -31,3 +34,16 @@ try: SAVE_LIMIT = settings.Q_SAVE_LIMIT except AttributeError: SAVE_LIMIT = 100 + +try: + WORKERS = settings.Q_WORKERS +except AttributeError: + WORKERS = cpu_count() + +""" +Turns compression on/off for task packages +""" +try: + COMPRESSED = settings.Q_COMPRESSED +except AttributeError: + COMPRESSED = False diff --git a/django_q/q.py b/django_q/q.py index 0e2c2f2..1a24a8c 100644 --- a/django_q/q.py +++ b/django_q/q.py @@ -1,19 +1,20 @@ import importlib import logging import signal -from multiprocessing import cpu_count, Queue, Event, Process, current_process +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 +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 @@ -35,7 +36,7 @@ class Cluster(object): logger.error('Can not connect to Redis server') return self.running = True - self.pool_size = cpu_count() + self.pool_size = WORKERS self.pool = [] self.task_queue = Queue() self.done_queue = Queue() @@ -198,19 +199,26 @@ def async(func, *args, **kwargs): [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=True, + 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. """ - def dumps(self, obj): + + @staticmethod + def dumps(obj): return jsonpickle.dumps(obj).encode('latin-1') - def loads(self, data): + @staticmethod + def loads(data): return jsonpickle.loads(data.decode('latin-1'))