Refactoring to make testing more stable

django_q/__init__.py

@@ -1,10 +1,7 @@
-"""
-A multiprocessing task queue application for Django
-Author: Ilan Steemers (koed00@gmail.com
-Github: https://github.com/Koed00/django-q
-"""
+from django_q.models import Task
+from django_q.core import async, Cluster
 
-from .main import Cluster, async, SignedPackage, Stat
+default_app_config = 'django_q.apps.DjangoQConfig'
 
-
-default_app_config = 'django_q.apps.SessionAdminConfig'
+def result(name):
+    return Task.get_result(name)

django_q/admin.py

@@ -1,6 +1,6 @@
 from django.contrib import admin
 
-from django_q import async
+from django_q.core import async
 
 from .models import Success, Failure
 

django_q/apps.py

@@ -1,6 +1,6 @@
 from django.apps import AppConfig
 
 
-class SessionAdminConfig(AppConfig):
+class DjangoQConfig(AppConfig):
     name = 'django_q'
     verbose_name = "Django Q"

django_q/conf.py

@@ -3,6 +3,14 @@ from multiprocessing import cpu_count
 
 VERSION = '0.1.0'
 
+"""
+Prefixes the Redis keys. Defaults to django_q
+"""
+try:
+    PREFIX = settings.Q_PREFIX
+except AttributeError:
+    PREFIX = 'django_q'
+
 """
 Sets the logging level for the app
 """
@@ -41,3 +49,8 @@ try:
     COMPRESSED = settings.Q_COMPRESSED
 except AttributeError:
     COMPRESSED = False
+
+try:
+    USE_TZ = settings.USE_TZ
+except AttributeError:
+    USE_TZ = False

django_q/core.py

@@ -1,3 +1,18 @@
+# Future
+from __future__ import unicode_literals
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+from builtins import dict
+from builtins import range
+from datetime import datetime
+from django.utils.timezone import make_aware
+
+from future import standard_library
+
+standard_library.install_aliases()
+
+# Standard
 import importlib
 import logging
 import os
@@ -6,6 +21,7 @@ from multiprocessing import Queue, Event, Process, current_process
 import socket
 import sys
 from time import sleep
+import gc
 
 # External
 import jsonpickle
@@ -15,22 +31,29 @@ import redis
 # Django
 from django.core import signing
 
-from django.utils import timezone
-
 # Local
-from .conf import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, VERSION
+from .conf import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, PREFIX, USE_TZ
 from .humanhash import uuid
 from .models import Task, Success
 
 SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n)
 
-prefix = 'django_q'
-q_list = '{}:q'.format(prefix)
 logger = logging.getLogger('django-q')
 coloredlogs.install(level=getattr(logging, LOG_LEVEL))
 
+Q_LIST = '{}:q'.format(PREFIX)
+STARTING = 'Starting'
+RUNNING = 'Running'
+STOPPED = 'Stopped'
+STOPPING = 'Stopping'
+
 r = redis.StrictRedis()
 
+# JsonPickle seems to have a problem with django's timezone.now
+def time_zone(value):
+    if USE_TZ:
+        return make_aware(value)
+    return value
 
 class Cluster(object):
     def __init__(self):
@@ -41,6 +64,8 @@ class Cluster(object):
             return
         self.sentinel = None
         self.stop_event = None
+        self.start_event = None
+        self.stopped_event = None
         self.pid = current_process().pid
         signal.signal(signal.SIGTERM, self.sig_handler)
         signal.signal(signal.SIGINT, self.sig_handler)
@@ -54,22 +79,56 @@ class Cluster(object):
             sys.stdin.close = dummy_close
         # Start Sentinel
         self.stop_event = Event()
-        self.sentinel = Process(target=Sentinel, args=(self.stop_event,))
+        self.start_event = Event()
+        self.sentinel = Process(target=Sentinel, args=(self.stop_event, self.start_event))
         self.sentinel.start()
-        logger.info('Starting Q cluster version {} at {}'.format(VERSION, self.pid))
+        logger.info('Q Cluster-{} starting.'.format(self.pid))
+        return self.pid
 
     def stop(self):
+        if not self.sentinel.is_alive():
+            return False
+        logger.info('Q Cluster-{} stopping.'.format(self.pid))
         self.stop_event.set()
         self.sentinel.join()
-        logger.info('Q cluster has stopped.')
+        logger.info('Q Cluster-{} has stopped.'.format(self.pid))
+        self.start_event = None
+        self.stop_event = None
+        return True
 
     def sig_handler(self, signum, frame):
         logger.debug('{} got signal {}'.format(current_process().name, SIGNAL_NAMES.get(signum, 'UNKNOWN')))
         self.stop()
 
+    @property
+    def stat(self):
+        if self.sentinel:
+            return Stat.get(self.pid)
+        return Status(self.pid)
+
+    @property
+    def is_starting(self):
+        return self.stop_event and self.start_event and not self.start_event.is_set()
+
+    @property
+    def is_running(self):
+        return self.stop_event and self.start_event and self.start_event.is_set()
+
+    @property
+    def is_stopping(self):
+        return self.stop_event and self.start_event and self.start_event.is_set() and self.stop_event.is_set()
+
+    @property
+    def has_stopped(self):
+        return self.start_event is None and self.stop_event is None and self.sentinel
+
+    @property
+    def is_idle(self):
+        return self.sentinel is None
+
 
 class Sentinel(object):
-    def __init__(self, event):
+    def __init__(self, stop_event, start_event):
         signal.signal(signal.SIGINT, signal.SIG_IGN)
         signal.signal(signal.SIGTERM, signal.SIG_DFL)
         self.pid = current_process().pid
@@ -77,8 +136,9 @@ class Sentinel(object):
         self.name = current_process().name
         self.status = None
         self.reincarnations = 0
-        self.tob = timezone.now()
-        self.stop_event = event
+        self.tob = datetime.utcnow()
+        self.stop_event = stop_event
+        self.start_event = start_event
         self.pool_size = WORKERS
         self.pool = []
         self.task_queue = Queue()
@@ -124,7 +184,7 @@ class Sentinel(object):
         self.reincarnations += 1
 
     def spawn_cluster(self):
-        self.set_status('Starting')
+        self.set_status(STARTING)
         for i in range(self.pool_size):
             self.spawn_worker()
         self.monitor_pid = self.spawn_monitor()
@@ -132,21 +192,23 @@ class Sentinel(object):
 
     def guard(self):
         logger.info('{} guarding cluster at {}'.format(current_process().name, self.pid))
-        self.set_status('Running')
-        while not self.stop_event.is_set():
+        self.start_event.set()
+        self.set_status(RUNNING)
+        logger.info('Q Cluster-{} running.'.format(self.parent_pid))
+        while True:
             for p in list(self.pool):
                 if not p.is_alive():
-                    # Be humane
                     p.terminate()
                     self.pool.remove(p)
-                    # Replace it with a fresh one
                     self.reincarnate(p.pid)
-            Stat(self).publish()
+            Stat(self).save()
+            if self.stop_event.is_set():
+                break
             sleep(2)
         self.stop()
 
     def stop(self):
-        self.set_status('Stopping')
+        self.set_status(STOPPING)
         name = current_process().name
         logger.info('{} stopping pool processes'.format(name))
         # Stopping pusher
@@ -163,20 +225,22 @@ class Sentinel(object):
         # Finally stop the monitor
         self.done_queue.put('STOP')
         self.pool = []
-        self.set_status('Stopped')
+        self.set_status(STOPPED)
 
     def set_status(self, message=None):
-        Stat(self, message).publish()
+        Stat(self, message).save()
 
 
 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)
+    while True:
+        task = r.blpop(Q_LIST, 1)
         if task:
             task = task[1]
             task_queue.put(task)
             logger.debug('queueing {}'.format(task))
+        if e.is_set():
+            break
     logger.info("{} stopped pushing tasks".format(current_process().name))
 
 
@@ -185,7 +249,7 @@ def monitor(done_queue):
     logger.info("{} monitoring results at {}".format(name, current_process().pid))
     for task in iter(done_queue.get, 'STOP'):
         if task['success']:
-            logger.info("Finished [{}]".format(task['name']))
+            logger.info("Processed [{}]".format(task['name']))
         else:
             logger.error("Failed [{}] - {}".format(task['name'], task['result']))
         save_task(task)
@@ -204,7 +268,7 @@ def worker(task_queue, done_queue):
             continue
         except signing.BadSignature as e:
             task['name'] = task['name'].rsplit(":", 1)[0]
-            task['stopped'] = timezone.now()
+            task['stopped'] = datetime.utcnow()
             task['result'] = e
             task['success'] = False
             done_queue.put(task)
@@ -215,12 +279,13 @@ def worker(task_queue, done_queue):
             m = importlib.import_module(module)
             f = getattr(m, func)
             task['result'] = f(*task['args'], **task['kwargs'])
-            task['stopped'] = timezone.now()
+            task['stopped'] = datetime.utcnow()
             task['success'] = True
             done_queue.put(task)
+            gc.collect()
         except Exception as e:
             task['result'] = e
-            task['stopped'] = timezone.now()
+            task['stopped'] = datetime.utcnow()
             task['success'] = False
             done_queue.put(task)
     logger.info('{} stopped doing work'.format(name))
@@ -234,19 +299,24 @@ def save_task(task):
                         hook=task['hook'],
                         args=task['args'],
                         kwargs=task['kwargs'],
-                        started=task['started'],
-                        stopped=task['stopped'],
+                        started=time_zone(task['started']),
+                        stopped=time_zone(task['stopped']),
                         result=task['result'],
                         success=task['success'])
 
 
-def async(func, *args, hook=None, **kwargs):
+def async(func, *args, **kwargs):
     """
-    Schedules a task
+    Schedules a task with optional hook
     """
-    task = {'name': uuid()[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, 'started': timezone.now()}
+    if 'hook' in kwargs:
+        hook = kwargs['hook']
+        del kwargs['hook']
+    else:
+        hook = None
+    task = {'name': uuid()[0], 'func': func, 'hook': hook, 'args': args, 'kwargs': kwargs, 'started': datetime.utcnow()}
     pack = SignedPackage.dumps(task)
-    r.rpush(q_list, pack)
+    r.rpush(Q_LIST, pack)
     logger.debug('Pushed {}'.format(pack))
     return task['name']
 
@@ -287,32 +357,75 @@ class JSONPickleSerializer(object):
         return jsonpickle.loads(data.decode('latin-1'))
 
 
-class Stat(object):
-    def __init__(self, sentinel, message=None):
-        self.cluster_id = sentinel.parent_pid
+class Status(object):
+    def __init__(self, pid):
+        self.workers = []
+        self.tob = None
+        self.reincarnations = 0
+        self.cluster_id = pid
+        self.sentinel = 0
+        self.status = 'Idle'
+        self.done_q_size = 0
         self.host = socket.gethostname()
+        self.monitor = 0
+        self.task_q_size = 0
+        self.pusher = 0
+        self.timestamp = datetime.utcnow()
+
+
+class Stat(Status):
+    def __init__(self, sentinel, message=None):
+        super().__init__(sentinel.parent_pid)
         if message:
             sentinel.status = message
-        self.status = sentinel.status
+        self.tob = sentinel.tob
+        self.reincarnations = sentinel.reincarnations
         self.sentinel = sentinel.pid
+        self.status = sentinel.status
+        self.done_q_size = sentinel.done_queue.qsize()
         self.monitor = sentinel.monitor_pid
+        self.task_q_size = sentinel.task_queue.qsize()
         self.pusher = sentinel.pusher_pid
-        self.workers = []
         for w in sentinel.pool:
             self.workers.append(w.pid)
-        self.reincarnations = sentinel.reincarnations
-        self.task_q_size = sentinel.task_queue.qsize()
-        self.done_q_size = sentinel.done_queue.qsize()
-        self.tob = sentinel.tob
-        self.timestamp = timezone.now()
+
+    def uptime(self):
+        return (datetime.utcnow() - self.tob).total_seconds()
 
     @property
     def key(self):
-        return self.get_key()
+        return self.get_key(self.cluster_id)
 
     @staticmethod
-    def get_key():
-        return '{}:cluster'.format(prefix)
+    def get_key(cluster_id):
+        return '{}:cluster:{}'.format(PREFIX, cluster_id)
 
-    def publish(self):
-        r.publish(self.key, SignedPackage.dumps(self, True))
+    def save(self):
+        r.set(self.key, SignedPackage.dumps(self, True), 3)
+
+    def empty_queues(self):
+        return self.done_q_size + self.task_q_size == 0
+
+    @staticmethod
+    def get(cluster_id):
+        key = Stat.get_key(cluster_id)
+        if r.exists(key):
+            pack = r.get(key)
+            try:
+                return SignedPackage.loads(pack)
+            except signing.BadSignature:
+                return None
+        return Status(cluster_id)
+
+    @staticmethod
+    def get_all():
+        stats = []
+        keys = r.keys(pattern='{}:cluster:*'.format(PREFIX))
+        if keys:
+            packs = r.mget(keys)
+            for pack in packs:
+                try:
+                    stats.append(SignedPackage.loads(pack))
+                except signing.BadSignature:
+                    continue
+        return stats

django_q/management/commands/qcluster.py

@@ -1,9 +1,9 @@
 from django.core.management.base import BaseCommand
-from django_q import Cluster
+from django_q.core import Cluster
 
 
 class Command(BaseCommand):
-    help = "My shiny new management command."
+    help = "Starts a Django Q Cluster."
 
     def handle(self, *args, **options):
         q = Cluster()

django_q/migrations/0001_initial.py

@@ -14,7 +14,7 @@ class Migration(migrations.Migration):
         migrations.CreateModel(
             name='Task',
             fields=[
-                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
+                ('id', models.AutoField(auto_created=True, primary_key=True, verbose_name='ID', serialize=False)),
                 ('name', models.CharField(max_length=100)),
                 ('func', models.CharField(max_length=256)),
                 ('hook', models.CharField(max_length=256, null=True)),
@@ -31,8 +31,8 @@ class Migration(migrations.Migration):
             fields=[
             ],
             options={
-                'verbose_name': 'Failed task',
                 'proxy': True,
+                'verbose_name': 'Failed task',
             },
             bases=('django_q.task',),
         ),
@@ -41,8 +41,8 @@ class Migration(migrations.Migration):
             fields=[
             ],
             options={
-                'verbose_name': 'Successful task',
                 'proxy': True,
+                'verbose_name': 'Successful task',
             },
             bases=('django_q.task',),
         ),

django_q/models.py

@@ -20,7 +20,9 @@ class Task(models.Model):
 
     @staticmethod
     def get_result(name):
-        return Task.objects.get(name=name).result
+        if Task.objects.filter(name=name).exists():
+            return Task.objects.get(name=name).result
+        return None
 
     def time_taken(self):
         return (self.stopped - self.started).total_seconds()

django_q/tests/__init__.py

@@ -1 +0,0 @@
-__author__ = 'ilan'

django_q/tests/settings.py

@@ -103,3 +103,5 @@ USE_TZ = True
 
 STATIC_URL = '/static/'
 
+# Django Q specific
+#Q_PREFIX = 'test_django_q'

django_q/tests/test_q.py

@@ -0,0 +1,130 @@
+import sys
+import os
+from time import sleep
+from multiprocessing import Queue, Event
+
+import pytest
+
+myPath = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, myPath + '/../')
+
+from django_q.core import Cluster, r, async, Q_LIST, pusher, worker, monitor
+from django_q.humanhash import DEFAULT_WORDLIST
+from django_q import result
+
+
+class WordClass(object):
+    def __init__(self):
+        self.word_list = DEFAULT_WORDLIST
+
+    def get_words(self):
+        return self.word_list
+
+
+def test_admin_view(admin_client):
+    response = admin_client.get('/admin/django_q/')
+    assert response.status_code == 200
+    response = admin_client.get('/admin/django_q/failure/')
+    assert response.status_code == 200
+    response = admin_client.get('/admin/django_q/success/')
+    assert response.status_code == 200
+
+
+def test_redis_connection():
+    assert r.ping() is True
+
+
+def test_cluster_initial():
+    c = Cluster()
+    assert c.sentinel is None
+    assert c.is_idle
+    c.start()
+    while c.is_starting:
+        sleep(0.2)
+    assert c.sentinel.is_alive() is True
+    assert c.is_running
+    c.stop()
+    while c.is_stopping:
+        sleep(0.2)
+    assert c.sentinel.is_alive() is False
+    assert c.has_stopped
+
+
+@pytest.mark.django_db
+def test_cluster():
+    task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST)
+    task_count = r.llen(Q_LIST)
+    assert task_count >= 1
+    task_queue = Queue()
+    assert task_queue.qsize() == 0
+    result_queue = Queue()
+    assert result_queue.qsize() == 0
+    event = Event()
+    event.set()
+    # Test push
+    pusher(task_queue, event)
+    assert task_queue.qsize() == 1
+    assert r.llen(Q_LIST) == task_count - 1
+    # Test work
+    task_queue.put('STOP')
+    worker(task_queue, result_queue)
+    assert task_queue.qsize() == 0
+    assert result_queue.qsize() == 1
+    # Test monitor
+    result_queue.put('STOP')
+    monitor(result_queue)
+    assert result_queue.qsize() == 0
+
+    # check result
+    assert result(task) == 1506
+
+
+@pytest.mark.django_db
+def test_async():
+    a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_q.assert_result')
+    b = async('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_q.assert_result')
+    c = Cluster()
+    assert c.start() > 0
+    while not c.is_running:
+        sleep(0.5)
+    assert isinstance(a, str)
+    assert isinstance(b, str)
+    while c.stat.task_q_size > 0 and c.stat.done_q_size > 0:
+        sleep(0.5)
+    assert c.stop() is True
+    result_a = result(a)
+    assert result_a is not None
+    assert result_a.success is True
+    assert result_a.result == 1506
+    result_b = result(b)
+    assert result_b is not None
+    assert result_b.success is True
+    assert result_b.result == 1506
+
+
+
+@pytest.mark.django_db
+def assert_result(task):
+    assert task is not None
+    assert task.success is True
+    assert task.result == 1506
+
+
+@pytest.mark.django_db
+def broken_package():
+    a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, 'oneargumentoomany',
+              hook='django_q.tests.test_q.assert_bad_result')
+    b = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_q.assert_bad_result')
+    assert isinstance(a, str)
+    assert isinstance(b, str)
+    sleep(5)
+    result_a = result(a)
+    assert result_a.success is False
+    result_b = result(b)
+    assert result_b.success is False
+
+
+@pytest.mark.django_db
+def assert_bad_result(task):
+    assert task is not None
+    assert task.success is False

pytest.ini

@@ -1,3 +1,2 @@
 [pytest]
 DJANGO_SETTINGS_MODULE=django_q.tests.settings
-#django_find_project = false

requirements.txt

@@ -1,3 +1,5 @@
+blessed
+future
 -e .
 coloredlogs==1.0.1
 django-picklefield==0.3.1

setup.py

@@ -1,16 +1,15 @@
 import os
 
-from setuptools import setup
-
 with open(os.path.join(os.path.dirname(__file__), 'README.rst')) as readme:
     README = readme.read()
 
 os.chdir(os.path.normpath(os.path.join(os.path.abspath(__file__), os.pardir)))
-from distutils.core import setup, Command
+from setuptools import setup, Command
 # you can also import from setuptools
 
 class PyTest(Command):
     user_options = []
+
     def initialize_options(self):
         pass
 
@@ -20,9 +19,11 @@ class PyTest(Command):
     def run(self):
         import subprocess
         import sys
+
         errno = subprocess.call([sys.executable, 'runtests.py'])
         raise SystemExit(errno)
 
+
 setup(
     name='django-q',
     version='0.1.0',
@@ -33,10 +34,8 @@ setup(
     license='MIT',
     description='A multiprocessing task queue for Django',
     long_description=README,
-    include_package_data=True,
     install_requires=['django>=1.7', 'redis', 'coloredlogs', 'django-picklefield', 'jsonpickle'],
-    test_suite='django_q.tests',
-    cmdclass = {'test': PyTest},
+    cmdclass={'test': PyTest},
     classifiers=[
         'Development Status :: 2 - PreAlpha',
         'Environment :: Web Environment',