moved static methods to module functions

medic is now sentinel and runs in it's own process
sentinel shutdown procedure more stable

django_q/__init__.py

@@ -4,6 +4,6 @@ Author: Ilan Steemers (koed00@gmail.com
 Github: https://github.com/Koed00/django-q
 """
 
-from django_q.q import Cluster, async
+from .main import Cluster, async
 
 default_app_config = 'django_q.apps.SessionAdminConfig'

django_q/apps.py

@@ -9,6 +9,8 @@ class SessionAdminConfig(AppConfig):
     verbose_name = "Django Q"
 
 
+VERSION = '0.1.0'
+
 """
 Sets the logging level for the app
 """

django_q/main.py

@@ -0,0 +1,227 @@
+import importlib
+import logging
+import signal
+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 .apps import LOG_LEVEL, SECRET_KEY, SAVE_LIMIT, WORKERS, COMPRESSED, VERSION
+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))
+
+r = redis.StrictRedis()
+
+
+class Cluster(object):
+    def __init__(self):
+        try:
+            r.ping()
+        except ():
+            logger.error('Can not connect to Redis server')
+            return
+        self.pool_size = WORKERS
+        self.pool = []
+        self.task_queue = Queue()
+        self.done_queue = Queue()
+        self.event_stop = Event()
+        self.monitor_pid = None
+        self.pusher_pid = None
+        self.sentinel_pid = None
+
+    def spawn_process(self, target, *args):
+        # 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=target, args=args)
+        self.pool.append(p)
+        p.start()
+        return p.pid
+
+    def spawn_pusher(self):
+        return self.spawn_process(pusher, self.task_queue, self.event_stop)
+
+    def spawn_worker(self):
+        self.spawn_process(worker, self.task_queue, self.done_queue)
+
+    def spawn_monitor(self):
+        return self.spawn_process(monitor, self.done_queue)
+
+    def spawn_sentinel(self):
+        return self.spawn_process(self.sentinel, self.event_stop)
+
+    def reincarnate(self, pid):
+        if pid == self.monitor_pid:
+            self.spawn_monitor()
+            logger.warn("reincarnated monitor after death of {}".format(pid))
+        elif pid == self.pusher_pid:
+            self.spawn_pusher()
+            logger.warn("reincarnated pusher after death of {}".format(pid))
+        else:
+            self.spawn_worker()
+            logger.warn("reincarnated work worker after death of {}".format(pid))
+
+    def sentinel(self, e):
+        for i in range(self.pool_size):
+            self.spawn_worker()
+        self.monitor_pid = self.spawn_monitor()
+        self.pusher_pid = self.spawn_pusher()
+        logger.info('{} checking worker health at {}'.format(current_process().name, current_process().pid))
+        while not e.is_set():
+            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)
+            sleep(1)
+        self.stop()
+
+    def start(self):
+        signal.signal(signal.SIGTERM, self.sig_handler)
+        signal.signal(signal.SIGINT, self.sig_handler)
+        self.sentinel_pid = self.spawn_sentinel()
+        logger.info('Starting Q cluster version {} at {}'.format(VERSION, current_process().pid))
+
+    def stop(self):
+        # Send the STOP signal to the pool
+        name = current_process().name
+        logger.info('{} stopping pool processes'.format(name))
+        # Stopping pusher
+        self.event_stop.set()
+        # Stopping monitor
+        self.done_queue.put('STOP')
+        # Stopping workers
+        for _ in self.pool:
+            self.task_queue.put('STOP')
+
+    def sig_handler(self, signum, frame):
+        logger.debug('{} got signal {}'.format(current_process().name, SIGNAL_NAMES.get(signum, 'UNKNOWN')))
+        self.event_stop.set()
+
+
+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 = task[1]
+            task_queue.put(task)
+            logger.debug('queueing {}'.format(task))
+    logger.info("{} stopped pushing".format(current_process().name))
+
+
+def monitor(done_queue):
+    name = current_process().name
+    logger.info("{} monitoring at {}".format(name, current_process().pid))
+    for task in iter(done_queue.get, 'STOP'):
+        name = task[0]
+        func = task[1]
+        result = task[6]
+        success = task[7]
+        if success:
+            logger.info("Finished [{}:{}]".format(func, name))
+        else:
+            logger.error("Failed [{}:{}] - {}".format(func, name, result))
+        save_task(task)
+    logger.info("{} stopped monitoring".format(name))
+
+
+def worker(task_queue, done_queue):
+    name = current_process().name
+    logger.info('{} ready for work at {}'.format(name, current_process().pid))
+    for pack in iter(task_queue.get, 'STOP'):
+        # unpickle the task
+        try:
+            task = signing.loads(pack, key=SECRET_KEY, salt='django_q.q', serializer=JSONPickleSerializer)
+        except TypeError as e:
+            logger.error(e)
+            continue
+        except BadSignature as e:
+            task[0] = task[0].rsplit(":", 1)[0]
+            task.append(timezone.now())
+            task.append(e)
+            task.append(False)
+            done_queue.put(task)
+            continue
+        func = task[1]
+        module, func = func.rsplit('.', 1)
+        args = task[2]
+        kwargs = task[3]
+        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)
+            task.append(result)
+            task.append(True)
+            done_queue.put(task)
+        except Exception as e:
+            task.append(e)
+            task.append(False)
+            done_queue.put(task)
+    logger.info('{} stopped working'.format(name))
+
+
+def save_task(task):
+    if task[7] and 0 < SAVE_LIMIT < Success.objects.count():
+        Success.objects.first().delete()
+    Task.objects.create(name=task[0],
+                        func=task[1],
+                        args=task[2],
+                        kwargs=task[3],
+                        started=task[4],
+                        stopped=task[5],
+                        result=task[6],
+                        success=task[7])
+
+
+def async(func, *args, **kwargs):
+    """
+    Schedules a module function
+    [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=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.
+    """
+
+    @staticmethod
+    def dumps(obj):
+        return jsonpickle.dumps(obj).encode('latin-1')
+
+    @staticmethod
+    def loads(data):
+        return jsonpickle.loads(data.decode('latin-1'))

django_q/management/commands/qcluster.py

@@ -7,3 +7,5 @@ class Command(BaseCommand):
 
     def handle(self, *args, **options):
         q = Cluster()
+        q.start()
+

django_q/q.py

@@ -1,224 +0,0 @@
-import importlib
-import logging
-import signal
-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, WORKERS, COMPRESSED
-from django_q.humanhash import uuid
-from django_q.models import Task, Success
-
-prefix = 'django_q'
-q_list = '{}:q'.format(prefix)
-logger = logging.getLogger('django-q')
-coloredlogs.install(level=getattr(logging, LOG_LEVEL))
-
-r = redis.StrictRedis()
-
-
-class Cluster(object):
-    def __init__(self):
-        signal.signal(signal.SIGTERM, self.sig_handler)
-        signal.signal(signal.SIGINT, self.sig_handler)
-        try:
-            r.ping()
-        except ():
-            logger.error('Can not connect to Redis server')
-            return
-        self.running = True
-        self.pool_size = WORKERS
-        self.pool = []
-        self.task_queue = Queue()
-        self.done_queue = Queue()
-        # Spawn workers
-        for i in range(self.pool_size):
-            self.spawn_worker()
-        # Spawn monitor
-        self.monitor_pid = None
-        self.spawn_monitor()
-        # Spawn pusher
-        self.pusher_pid = None
-        self.pusher_stop = Event()
-        self.spawn_pusher()
-        # Monitor process health
-        while self.running:
-            self.medic()
-            sleep(1)
-
-    def spawn_process(self, target, *args):
-        # 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=target, args=args)
-        self.pool.append(p)
-        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_worker(self):
-        self.spawn_process(self.worker, self.task_queue, self.done_queue)
-
-    def spawn_monitor(self):
-        self.monitor_pid = self.spawn_process(self.monitor, self.done_queue)
-
-    def reincarnate(self, pid):
-        if pid == self.monitor_pid:
-            self.spawn_monitor()
-            logger.warn("reincarnated monitor after death of {}".format(pid))
-        elif pid == self.pusher_pid:
-            self.spawn_pusher()
-            logger.warn("reincarnated pusher after death of {}".format(pid))
-        else:
-            self.spawn_worker()
-            logger.warn("reincarnated work worker 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 = task[1]
-                task_queue.put(task)
-                logger.debug('queueing {}'.format(task))
-
-    @staticmethod
-    def monitor(done_queue):
-        name = current_process().name
-        logger.info("{} monitoring at {}".format(name, current_process().pid))
-        for task in iter(done_queue.get, 'STOP'):
-            name = task[0]
-            func = task[1]
-            result = task[6]
-            success = task[7]
-            if success:
-                logger.info("Finished [{}:{}]".format(func, name))
-            else:
-                logger.error("Failed [{}:{}] - {}".format(func, name, result))
-            Cluster.save_task(task)
-        logger.info("{} stopped".format(name))
-
-    @staticmethod
-    def worker(task_queue, done_queue):
-        name = current_process().name
-        logger.info('{} ready for work at {}'.format(name, current_process().pid))
-        for pack in iter(task_queue.get, 'STOP'):
-            # unpickle the task
-            try:
-                task = signing.loads(pack, key=SECRET_KEY, salt='django_q.q', serializer=JSONPickleSerializer)
-            except TypeError as e:
-                logger.error(e)
-                continue
-            except BadSignature as e:
-                task[0] = task[0].rsplit(":", 1)[0]
-                task.append(timezone.now())
-                task.append(e)
-                task.append(False)
-                done_queue.put(task)
-                continue
-            func = task[1]
-            module, func = func.rsplit('.', 1)
-            args = task[2]
-            kwargs = task[3]
-            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)
-                task.append(result)
-                task.append(True)
-                done_queue.put(task)
-            except Exception as e:
-                task.append(e)
-                task.append(False)
-                done_queue.put(task)
-        logger.info('{} Stopped'.format(name))
-
-    def medic(self):
-        # Check if all the workers are alive
-        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)
-
-    @staticmethod
-    def save_task(task):
-        if task[7] and 0 < SAVE_LIMIT < Success.objects.count():
-            Success.objects.first().delete()
-        Task.objects.create(name=task[0],
-                            func=task[1],
-                            args=task[2],
-                            kwargs=task[3],
-                            started=task[4],
-                            stopped=task[5],
-                            result=task[6],
-                            success=task[7])
-
-    def stop(self):
-        # Send the STOP signal to the pool
-        self.running = False
-        logger.info('Stopping')
-        # Wait for all the workers to finish the queue
-        for p in self.pool:
-            if p.pid == self.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()
-
-        logger.info('Goodbye. Have a wonderful time.')
-
-    def sig_handler(self, signum, frame):
-        self.stop()
-
-
-def async(func, *args, **kwargs):
-    """
-    Schedules a module function
-    [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=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.
-    """
-
-    @staticmethod
-    def dumps(obj):
-        return jsonpickle.dumps(obj).encode('latin-1')
-
-    @staticmethod
-    def loads(data):
-        return jsonpickle.loads(data.decode('latin-1'))

django_q/tests/tasks.py

@@ -1,7 +1,7 @@
 import time
 
 # simple countdown
-def count(n):
+def countdown(n):
     start_time = time.time()
     while n > 0:
         n -= 1

django_q/tests/test_qworker.py

@@ -1,11 +1,15 @@
+import sys, os
+myPath = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, myPath + '/../')
+
 import pytest
 from django_q import Cluster
 
 
-@pytest.fixture
+@pytest.fixture(scope='session')
-def qworker():
+def cluster():
     return Cluster()
 
 
-def test_worker(qworker):
+def test_worker(cluster):
-    assert len(qworker.stable) == qworker.stable_size
+    assert len(cluster.stable) == cluster.stable_size

pytest.ini

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

requirements.txt

@@ -1,3 +1,4 @@
+-e .
 coloredlogs==1.0.1
 django-picklefield==0.3.1
 Django==1.8.2