From d8020f0c0bbffd21719f20936a404bda0e959b89 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Sat, 27 Jun 2015 12:08:33 +0200 Subject: [PATCH] Redis server now configurable with Q_REDIS. Moved constants to conf --- django_q/conf.py | 29 ++++++++++++++++++- django_q/core.py | 58 ++++++++++++++++++++++++++++++-------- django_q/tests/settings.py | 9 ++++++ 3 files changed, 83 insertions(+), 13 deletions(-) diff --git a/django_q/conf.py b/django_q/conf.py index a5a251d..5aa40b2 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -1,8 +1,17 @@ -from django.conf import settings +from signal import signal from multiprocessing import cpu_count +from django.conf import settings + VERSION = '0.1.0' +""" +Redis server connection +""" +try: + REDIS = settings.Q_REDIS +except AttributeError: + REDIS = {} """ Prefixes the Redis keys. Defaults to django_q """ @@ -54,3 +63,21 @@ try: USE_TZ = settings.USE_TZ except AttributeError: USE_TZ = False + +""" +Getting the SIGNAL names +""" +SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n) + +""" +Redis List name +""" +Q_LIST = '{}:q'.format(PREFIX) + +""" +Cluster status +""" +STARTING = 'Starting' +RUNNING = 'Running' +STOPPED = 'Stopped' +STOPPING = 'Stopping' diff --git a/django_q/core.py b/django_q/core.py index bc39973..7d62b46 100644 --- a/django_q/core.py +++ b/django_q/core.py @@ -4,7 +4,6 @@ from __future__ import print_function from __future__ import division from __future__ import absolute_import import ast -from builtins import dict from builtins import range from future import standard_library @@ -35,12 +34,11 @@ from django.core import signing from django.utils import timezone # Local -from .conf import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, PREFIX +from .conf import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, PREFIX, REDIS, Q_LIST, SIGNAL_NAMES, STARTING, \ + RUNNING, STOPPING, STOPPED from .humanhash import uuid from .models import Task, Success, Schedule -SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n) - logger = logging.getLogger('django-q') @@ -55,20 +53,14 @@ except ImportError: # Set up standard logging handler if not logger.handlers: logger.setLevel(level=getattr(logging, LOG_LEVEL)) - formatter = logging.Formatter(fmt='%(asctime)s [django_q] %(message)s', datefmt='%H:%M:%S') handler = logging.StreamHandler() handler.setFormatter(formatter) logger.addHandler(handler) -Q_LIST = '{}:q'.format(PREFIX) -STARTING = 'Starting' -RUNNING = 'Running' -STOPPED = 'Stopped' -STOPPING = 'Stopping' - -r = redis.StrictRedis() +# Redis +r = redis.StrictRedis(**REDIS) class Cluster(object): @@ -264,6 +256,12 @@ class Sentinel(object): def pusher(task_queue, e, list_key=Q_LIST): + """ + Pulls tasks of the Redis List and puts them in the task queue + :type task_queue: multiprocessing.Queue + :type e: multiprocessing.Event + :type list_key: str + """ logger.info('{} pushing tasks at {}'.format(current_process().name, current_process().pid)) while True: task = r.blpop(list_key, 1) @@ -277,6 +275,10 @@ def pusher(task_queue, e, list_key=Q_LIST): def monitor(done_queue): + """ + Gets finished tasks from the result queue and saves them to Django + :type done_queue: multiprocessing.Queue + """ name = current_process().name logger.info("{} monitoring at {}".format(name, current_process().pid)) for task in iter(done_queue.get, 'STOP'): @@ -289,6 +291,11 @@ def monitor(done_queue): def worker(task_queue, done_queue): + """ + Takes a task from the task queue, tries to execute it and puts the result back in the result queue + :type task_queue: multiprocessing.Queue + :type done_queue: multiprocessing.Queue + """ name = current_process().name logger.info('{} ready for work at {}'.format(name, current_process().pid)) task = {} @@ -408,6 +415,10 @@ class PickleSerializer(object): class Status(object): + """ + Cluster status base object + """ + def __init__(self, pid): self.workers = [] self.tob = None @@ -424,6 +435,10 @@ class Status(object): class Stat(Status): + """ + Status object for Cluster monitoring + """ + def __init__(self, sentinel, message=None): super(Stat, self).__init__(sentinel.parent_pid) if message: @@ -444,10 +459,17 @@ class Stat(Status): @property def key(self): + """ + :return: redis key for this cluster statistic + """ return self.get_key(self.cluster_id) @staticmethod def get_key(cluster_id): + """ + :param cluster_id: cluster ID + :return: redis key for the cluster statistic + """ return '{}:cluster:{}'.format(PREFIX, cluster_id) def save(self): @@ -458,6 +480,11 @@ class Stat(Status): @staticmethod def get(cluster_id): + """ + gets the current status for the cluster + :param cluster_id: id of the cluster + :return: Stat or Status + """ key = Stat.get_key(cluster_id) if r.exists(key): pack = r.get(key) @@ -469,6 +496,10 @@ class Stat(Status): @staticmethod def get_all(): + """ + Gets status for all currently running clusters with the same prefix and secret key + :return: Stat list + """ stats = [] keys = r.keys(pattern='{}:cluster:*'.format(PREFIX)) if keys: @@ -482,6 +513,9 @@ class Stat(Status): def scheduler(): + """ + Creates a task from a schedule at the scheduled time and schedules next run + """ for schedule in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()): args = () kwargs = {} diff --git a/django_q/tests/settings.py b/django_q/tests/settings.py index bf46c6c..d61a8ce 100644 --- a/django_q/tests/settings.py +++ b/django_q/tests/settings.py @@ -104,3 +104,12 @@ STATIC_URL = '/static/' # Django Q specific Q_PREFIX = 'test_django_q' + +Q_REDIS = { + #'host': 'localhost', + #'port': 6379, + # 'db': 1, + # 'socket_timeout': 3, + # 'password': '...', + # 'unix_socket_path': '' +}