diff --git a/django_q/admin.py b/django_q/admin.py index 8c38f3f..5f90e4b 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -1,3 +1,15 @@ from django.contrib import admin # Register your models here. +from .models import Task + + +class TaskAdmin(admin.ModelAdmin): + list_display = ( + u'name', + 'func', + 'started', + 'time_taken', + 'success' + ) +admin.site.register(Task, TaskAdmin) diff --git a/django_q/humanhash.py b/django_q/humanhash.py new file mode 100644 index 0000000..feac001 --- /dev/null +++ b/django_q/humanhash.py @@ -0,0 +1,143 @@ +""" +humanhash: Human-readable representations of digests. + +The simplest ways to use this module are the :func:`humanize` and :func:`uuid` +functions. For tighter control over the output, see :class:`HumanHasher`. +""" +from argparse import ArgumentError + +import operator +import uuid as uuidlib +from functools import reduce + + +DEFAULT_WORDLIST = ( + 'ack', 'alabama', 'alanine', 'alaska', 'alpha', 'angel', 'apart', 'april', + 'arizona', 'arkansas', 'artist', 'asparagus', 'aspen', 'august', 'autumn', + 'avocado', 'bacon', 'bakerloo', 'batman', 'beer', 'berlin', 'beryllium', + 'black', 'blossom', 'blue', 'bluebird', 'bravo', 'bulldog', 'burger', + 'butter', 'california', 'carbon', 'cardinal', 'carolina', 'carpet', 'cat', + 'ceiling', 'charlie', 'chicken', 'coffee', 'cola', 'cold', 'colorado', + 'comet', 'connecticut', 'crazy', 'cup', 'dakota', 'december', 'delaware', + 'delta', 'diet', 'don', 'double', 'early', 'earth', 'east', 'echo', + 'edward', 'eight', 'eighteen', 'eleven', 'emma', 'enemy', 'equal', + 'failed', 'fanta', 'fifteen', 'fillet', 'finch', 'fish', 'five', 'fix', + 'floor', 'florida', 'football', 'four', 'fourteen', 'foxtrot', 'freddie', + 'friend', 'fruit', 'gee', 'georgia', 'glucose', 'golf', 'green', 'grey', + 'hamper', 'happy', 'harry', 'hawaii', 'helium', 'high', 'hot', 'hotel', + 'hydrogen', 'idaho', 'illinois', 'india', 'indigo', 'ink', 'iowa', + 'island', 'item', 'jersey', 'jig', 'johnny', 'juliet', 'july', 'jupiter', + 'kansas', 'kentucky', 'kilo', 'king', 'kitten', 'lactose', 'lake', 'lamp', + 'lemon', 'leopard', 'lima', 'lion', 'lithium', 'london', 'louisiana', + 'low', 'magazine', 'magnesium', 'maine', 'mango', 'march', 'mars', + 'maryland', 'massachusetts', 'may', 'mexico', 'michigan', 'mike', + 'minnesota', 'mirror', 'mississippi', 'missouri', 'mobile', 'mockingbird', + 'monkey', 'montana', 'moon', 'mountain', 'muppet', 'music', 'nebraska', + 'neptune', 'network', 'nevada', 'nine', 'nineteen', 'nitrogen', 'north', + 'november', 'nuts', 'october', 'ohio', 'oklahoma', 'one', 'orange', + 'oranges', 'oregon', 'oscar', 'oven', 'oxygen', 'papa', 'paris', 'pasta', + 'pennsylvania', 'pip', 'pizza', 'pluto', 'potato', 'princess', 'purple', + 'quebec', 'queen', 'quiet', 'red', 'river', 'robert', 'robin', 'romeo', + 'rugby', 'sad', 'salami', 'saturn', 'september', 'seven', 'seventeen', + 'shade', 'sierra', 'single', 'sink', 'six', 'sixteen', 'skylark', 'snake', + 'social', 'sodium', 'solar', 'south', 'spaghetti', 'speaker', 'spring', + 'stairway', 'steak', 'stream', 'summer', 'sweet', 'table', 'tango', 'ten', + 'tennessee', 'tennis', 'texas', 'thirteen', 'three', 'timing', 'triple', + 'twelve', 'twenty', 'two', 'uncle', 'undress', 'uniform', 'uranus', 'utah', + 'vegan', 'venus', 'vermont', 'victor', 'video', 'violet', 'virginia', + 'washington', 'west', 'whiskey', 'white', 'william', 'winner', 'winter', + 'wisconsin', 'wolfram', 'wyoming', 'xray', 'yankee', 'yellow', 'zebra', + 'zulu') + + +class HumanHasher(object): + + """ + Transforms hex digests to human-readable strings. + + The format of these strings will look something like: + `victor-bacon-zulu-lima`. The output is obtained by compressing the input + digest to a fixed number of bytes, then mapping those bytes to one of 256 + words. A default wordlist is provided, but you can override this if you + prefer. + + As long as you use the same wordlist, the output will be consistent (i.e. + the same digest will always render the same representation). + """ + + def __init__(self, wordlist=DEFAULT_WORDLIST): + if len(wordlist) != 256: + raise ArgumentError("Wordlist must have exactly 256 items") + self.wordlist = wordlist + + def humanize(self, hexdigest, words=4, separator='-'): + + """ + Humanize a given hexadecimal digest. + + Change the number of words output by specifying `words`. Change the + word separator with `separator`. + + >>> digest = '60ad8d0d871b6095808297' + >>> HumanHasher().humanize(digest) + 'sodium-magnesium-nineteen-hydrogen' + """ + + # Gets a list of byte values between 0-255. + bytes = [int(x, 16) for x in list(map(''.join, list(zip(hexdigest[::2], hexdigest[1::2]))))] + # Compress an arbitrary number of bytes to `words`. + compressed = self.compress(bytes, words) + # Map the compressed byte values through the word list. + return separator.join(self.wordlist[byte] for byte in compressed) + + @staticmethod + def compress(bytes, target): + + """ + Compress a list of byte values to a fixed target length. + + >>> bytes = [96, 173, 141, 13, 135, 27, 96, 149, 128, 130, 151] + >>> HumanHasher.compress(bytes, 4) + [205, 128, 156, 96] + + Attempting to compress a smaller number of bytes to a larger number is + an error: + + >>> HumanHasher.compress(bytes, 15) # doctest: +ELLIPSIS + Traceback (most recent call last): + ... + ValueError: Fewer input bytes than requested output + """ + + length = len(bytes) + if target > length: + raise ValueError("Fewer input bytes than requested output") + + # Split `bytes` into `target` segments. + seg_size = length // target + segments = [bytes[i * seg_size:(i + 1) * seg_size] + for i in range(target)] + # Catch any left-over bytes in the last segment. + segments[-1].extend(bytes[target * seg_size:]) + + # Use a simple XOR checksum-like function for compression. + checksum = lambda bytes: reduce(operator.xor, bytes, 0) + checksums = list(map(checksum, segments)) + return checksums + + def uuid(self, **params): + + """ + Generate a UUID with a human-readable representation. + + Returns `(human_repr, full_digest)`. Accepts the same keyword arguments + as :meth:`humanize` (they'll be passed straight through). + """ + + digest = str(uuidlib.uuid4()).replace('-', '') + return self.humanize(digest, **params), digest + + +DEFAULT_HASHER = HumanHasher() +uuid = DEFAULT_HASHER.uuid +humanize = DEFAULT_HASHER.humanize diff --git a/django_q/models.py b/django_q/models.py index 13d3a25..da80f1b 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -3,20 +3,13 @@ import socket from django.utils.translation import ugettext_lazy as _ +class Task(models.Model): + name = models.CharField(max_length=100) + func = models.CharField(max_length=256) + task = models.TextField(null=True) + started = models.DateTimeField() + stopped = models.DateTimeField() + success = models.BooleanField(default=True) -class Worker(models.Model): - WORKER = 'W' - PUBLISHER = 'P' - QUEUE = 'Q' - TYPE = ( - (WORKER, _('Worker')), - (PUBLISHER, _('Publisher')), - (QUEUE, _('Queue')), - ) - worker_type = models.CharField(max_length=1, choices=TYPE, default=TYPE[0][0], verbose_name=_('Worker Type')) - ip_address = models.GenericIPAddressField(default='127.0.0.1') - port = models.PositiveSmallIntegerField() - - @staticmethod - def get_queue(): - return Worker.objects.filter(worker_type=Worker.QUEUE) + def time_taken(self): + return (self.stopped - self.started).total_seconds() diff --git a/django_q/q.py b/django_q/q.py index 9fb844d..093cdbc 100644 --- a/django_q/q.py +++ b/django_q/q.py @@ -1,56 +1,69 @@ import importlib -from time import sleep -from multiprocessing import Queue, Process, current_process, cpu_count +from random import randint +from multiprocessing import Queue, Process, Event, current_process, cpu_count import sys import signal -from uuid import uuid4 -import ujson as json +import logging +from time import sleep +import jsonpickle as json + +import coloredlogs import redis from django.conf import settings +from django.utils import timezone + +from .models import Task + +from .humanhash import uuid r = redis.StrictRedis(decode_responses=True) secret = settings.SECRET_KEY prefix = 'django_q' q_list = '{}:q'.format(prefix) +logger = logging.getLogger('django-q') +coloredlogs.install(level=logging.INFO) + def test(): - for i in range(4): - defer(u'testq.tasks.multiply', 5, i) + for i in range(20): + defer(u'testq.tasks.multiply', 5, i * randint(2, 100)) def defer(func, *args, **kwargs): - pack = json.dumps([uuid4().urn, func, args, kwargs]) + # [name, func, args, kwargs, started, finished, result] + pack = json.dumps([uuid()[0], func, args, kwargs, timezone.now()]) r.rpush(q_list, pack) + logger.debug('Pushed {}'.format(pack)) class Worker(object): def __init__(self): + signal.signal(signal.SIGTERM, self.sig_handler) + signal.signal(signal.SIGINT, self.sig_handler) self.running = True - self.stable_size = cpu_count() - 2 + self.stable_size = cpu_count() self.stable = [] self.task_queue = Queue() self.done_queue = Queue() self.fail_queue = Queue() - self.failure_monitor_pid = None - self.success_monitor_pid = None # Spawn work horses for i in range(self.stable_size): self.spawn_horse() # Spawn monitors + self.success_monitor_pid = None self.spawn_success_monitor() + self.failure_monitor_pid = None self.spawn_failure_monitor() - # Attach signal handler - signal.signal(signal.SIGTERM, self.sig_handler) - signal.signal(signal.SIGINT, self.sig_handler) - # Keep popping Redis + # Spawn pusher + self.pusher_pid = None + self.pusher_stop = Event() + self.spawn_pusher() + # Monitor process health while self.running: self.stable_boy() - sleep(0.2) - task = r.lpop(q_list) - if task: - self.task_queue.put(task) + sleep(1) def spawn_process(self, target, *args): # This is just for PyCharm to not crash. Ignore it. @@ -64,6 +77,9 @@ class Worker(object): 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_horse(self): self.spawn_process(self.horse, self.task_queue, self.done_queue, self.fail_queue) @@ -76,51 +92,71 @@ class Worker(object): def reincarnate(self, pid): if pid == self.success_monitor_pid: self.spawn_success_monitor() + logger.warn("reincarnated success monitor after death of {}".format(pid)) elif pid == self.failure_monitor_pid: self.spawn_failure_monitor() + logger.warn("reincarnated failure monitor after death of {}".format(pid)) else: self.spawn_horse() + logger.warn("reincarnated work horse 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_queue.put(task[1]) + logger.debug('queueing {}'.format(task[1])) @staticmethod def success_monitor(done_queue): name = current_process().name - print("{} monitoring successes at {}".format(name, current_process().pid)) - for result in iter(done_queue.get, 'STOP'): - task = result[0] - res = result[1] - print("Success [{}:{} - {}]".format(task[1], task[0], res)) - print("{} stopped".format(name)) + logger.info("{} monitoring successes at {}".format(name, current_process().pid)) + for task in iter(done_queue.get, 'STOP'): + logger.info("Finished [{}:{}]".format(task[1], task[0])) + Task.objects.create(name=task[0], func=task[1], task=json.dumps(task), + started=task[4], + stopped=task[5]) + logger.info("{} stopped".format(name)) @staticmethod def failure_monitor(fail_queue): name = current_process().name - print("{} monitoring failures at {}".format(name, current_process().pid)) - for result in iter(fail_queue.get, 'STOP'): - task = result[0] - e = result[1] - print("Failure [{}:{} - {}]".format(task[1], task[0], e)) - print("{} stopped".format(name)) + logger.info("{} monitoring failures at {}".format(name, current_process().pid)) + for task in iter(fail_queue.get, 'STOP'): + logger.error("Failure [{}:{} - {}]".format(task[1], task[0], task[6])) + Task.objects.create(name=task[0], func=task[1], task=json.dumps(task), + started=task[4], + stopped=task[5], success=False) + logger.info("{} stopped".format(name)) @staticmethod def horse(task_queue, done_queue, fail_queue): name = current_process().name - print('{} ready for work at {}'.format(name, current_process().pid)) + logger.info('{} ready for work at {}'.format(name, current_process().pid)) for pack in iter(task_queue.get, 'STOP'): - task = json.loads(pack) - uid = task[0] + try: + task = json.loads(pack) + except TypeError as e: + logger.error(e) + continue func = task[1] module, func = func.rsplit('.', 1) args = task[2] kwargs = task[3] - print(name, 'Processing [{}:{}]'.format(func, uid)) + 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) - done_queue.put((task, result)) + task.append(result) + done_queue.put(task) except TypeError as e: - fail_queue.put((task, e)) - print(name, 'Stopped') + task.append(e) + fail_queue.put(task) + logger.info('{} Stopped'.format(name)) def stable_boy(self): # Check if all the horses are alive @@ -135,18 +171,20 @@ class Worker(object): def stop(self): # Send the STOP signal to the stable self.running = False - print('Stopping') + logger.info('Stopping') # Wait for all the workers to finish the queue for p in self.stable: if p.pid == self.failure_monitor_pid: self.fail_queue.put('STOP') elif p.pid == self.success_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() - print('Goodbye. Have a wonderful time.') + logger.info('Goodbye. Have a wonderful time.') def sig_handler(self, signum, frame): self.stop() diff --git a/requirements.txt b/requirements.txt index 3b65774..0fed12b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,5 @@ -ujson -simplejson +jsonpickle +coloredlogs Django==1.8.2 hiredis==0.2.0 -pyzmq==14.6.0 redis==2.10.3