From 8c0a26e0b47037b7ab3a86b02f5811e2d5a2dd0c Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Mon, 15 Jun 2015 16:02:50 +0200 Subject: [PATCH] Testing basic layout --- .gitignore | 5 + django_q/__init__.py | 0 django_q/admin.py | 3 + django_q/management/__init__.py | 0 django_q/management/commands/__init__.py | 0 django_q/management/commands/qworker.py | 9 ++ django_q/management/commands/testq.py | 10 ++ django_q/migrations/__init__.py | 0 django_q/models.py | 22 ++++ django_q/q.py | 137 +++++++++++++++++++++++ django_q/tests.py | 3 + django_q/views.py | 3 + requirements.txt | 4 + 13 files changed, 196 insertions(+) create mode 100644 django_q/__init__.py create mode 100644 django_q/admin.py create mode 100644 django_q/management/__init__.py create mode 100644 django_q/management/commands/__init__.py create mode 100644 django_q/management/commands/qworker.py create mode 100644 django_q/management/commands/testq.py create mode 100644 django_q/migrations/__init__.py create mode 100644 django_q/models.py create mode 100644 django_q/q.py create mode 100644 django_q/tests.py create mode 100644 django_q/views.py create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore index ba74660..5acf51f 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,8 @@ docs/_build/ # PyBuilder target/ + +dev-requirements.txt +manage.py +testq +db.sqlite3 \ No newline at end of file diff --git a/django_q/__init__.py b/django_q/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/django_q/admin.py b/django_q/admin.py new file mode 100644 index 0000000..8c38f3f --- /dev/null +++ b/django_q/admin.py @@ -0,0 +1,3 @@ +from django.contrib import admin + +# Register your models here. diff --git a/django_q/management/__init__.py b/django_q/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/django_q/management/commands/__init__.py b/django_q/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/django_q/management/commands/qworker.py b/django_q/management/commands/qworker.py new file mode 100644 index 0000000..debf90f --- /dev/null +++ b/django_q/management/commands/qworker.py @@ -0,0 +1,9 @@ +from django.core.management.base import BaseCommand +from django_q.q import Worker + + +class Command(BaseCommand): + help = "My shiny new management command." + + def handle(self, *args, **options): + q = Worker() diff --git a/django_q/management/commands/testq.py b/django_q/management/commands/testq.py new file mode 100644 index 0000000..0da1e8b --- /dev/null +++ b/django_q/management/commands/testq.py @@ -0,0 +1,10 @@ +from django.core.management.base import BaseCommand + +from django_q import q + + +class Command(BaseCommand): + help = "My shiny new management command." + + def handle(self, *args, **options): + q.test() diff --git a/django_q/migrations/__init__.py b/django_q/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/django_q/models.py b/django_q/models.py new file mode 100644 index 0000000..13d3a25 --- /dev/null +++ b/django_q/models.py @@ -0,0 +1,22 @@ +from django.db import models +import socket +from django.utils.translation import ugettext_lazy as _ + + + +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) diff --git a/django_q/q.py b/django_q/q.py new file mode 100644 index 0000000..8d00599 --- /dev/null +++ b/django_q/q.py @@ -0,0 +1,137 @@ +import importlib +from time import sleep +from multiprocessing import Queue, Process, current_process, cpu_count +import sys +import signal + +import redis +from django.conf import settings + +try: + import cPickle as pickle +except ImportError: + import pickle + +r = redis.StrictRedis() +secret = settings.SECRET_KEY +prefix = 'django_q' +q_list = '{}:q'.format(prefix) + + +def test(): + for i in range(20): + defer('testq.tasks.multiply', 5, i) + + +def defer(func, *args, **kwargs): + # serialize the func + task = {u'func': func, u'args': args, u'kwargs': kwargs} + pack = pickle.dumps(task) + r.rpush(q_list, pack) + + +class Worker(object): + def __init__(self): + self.running = True + self.stable_size = cpu_count() + self.stable = [] + self.task_queue = Queue() + self.done_queue = Queue() + # Spawn work horses + for i in range(self.stable_size): + self._spawn_horse() + # Spawn monitor + self._spawn_monitor() + # Attach signal handler + signal.signal(signal.SIGTERM, self.sighandler) + signal.signal(signal.SIGINT, self.sighandler) + # Keep popping Redis + while self.running: + self.work() + sleep(0.2) + + def _spawn_horse(self): + # 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=self.horse, args=(self.task_queue, self.done_queue)) + self.stable.append(p) + p.start() + + def _spawn_monitor(self): + # 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 + + self.mon = Process(target=self.monitor, args=(self.done_queue,)) + self.mon.start() + + @staticmethod + def monitor(queue_in): + print("Monitor started at {}".format(current_process().pid)) + for result in iter(queue_in.get, 'STOP'): + task = result['task'] + res = result['result'] + print("{} - {}".format(task['func'], res)) + print("Monitor stopped") + + @staticmethod + def horse(queue_in, queue_out): + name = current_process().name + print(name, 'Ready for work at {}'.format(current_process().pid)) + for pack in iter(queue_in.get, 'STOP'): + task = pickle.loads(pack) + func = task['func'] + module, func = func.rsplit('.', 1) + args = task['args'] + kwargs = task['kwargs'] + print(name, 'Starting Task {}'.format(func)) + try: + m = importlib.import_module(module) + f = getattr(m, func) + result = f(*args, **kwargs) + queue_out.put({'task': task, 'result': result}) + print(name, 'Finished JTask {}'.format(func)) + except TypeError: + print('job failed') + # TODO log failure to django + print(name, 'Stopped') + + def work(self): + self.stable_boy() + self.task_queue.put(r.brpop(q_list)) + + def stable_boy(self): + # Check if all the horses are alive + for p in list(self.stable): + if not p.is_alive(): + # Be humane + p.terminate() + self.stable.remove(p) + # Replace it with a fresh one + self._spawn_horse() + + def stop(self): + # Send the STOP signal to the stable + self.running = False + print('Stopping queue') + for i in range(self.stable_size): + self.task_queue.put('STOP') + # Optional: Delete everything in the queue and then add STOP + self.done_queue.put('STOP') + # Wait for all the workers to finish the queue + for p in self.stable: + p.join() + self.mon.join() + print('All horses stopped.') + print('Goodbye. Have a wonderful time.') + + def sighandler(self, signum, frame): + self.stop() diff --git a/django_q/tests.py b/django_q/tests.py new file mode 100644 index 0000000..7ce503c --- /dev/null +++ b/django_q/tests.py @@ -0,0 +1,3 @@ +from django.test import TestCase + +# Create your tests here. diff --git a/django_q/views.py b/django_q/views.py new file mode 100644 index 0000000..91ea44a --- /dev/null +++ b/django_q/views.py @@ -0,0 +1,3 @@ +from django.shortcuts import render + +# Create your views here. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a26ea06 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +Django==1.8.2 +hiredis==0.2.0 +pyzmq==14.6.0 +redis==2.10.3