Added some config settings

Updated README
This commit is contained in:
Ilan Steemers
2015-06-17 19:14:20 +02:00
parent b28754de93
commit 2e81ba2858
4 changed files with 121 additions and 9 deletions

View File

@@ -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)
### 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.

View File

@@ -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

View File

@@ -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

View File

@@ -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'))