diff --git a/django_q/__init__.py b/django_q/__init__.py index 8d6bce2..53e495a 100644 --- a/django_q/__init__.py +++ b/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) \ No newline at end of file diff --git a/django_q/admin.py b/django_q/admin.py index ad0180a..1435910 100644 --- a/django_q/admin.py +++ b/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 diff --git a/django_q/apps.py b/django_q/apps.py index 361609e..e59fd34 100644 --- a/django_q/apps.py +++ b/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" diff --git a/django_q/conf.py b/django_q/conf.py index 215f08c..a5a251d 100644 --- a/django_q/conf.py +++ b/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 diff --git a/django_q/main.py b/django_q/core.py similarity index 65% rename from django_q/main.py rename to django_q/core.py index 37226ac..a09748c 100644 --- a/django_q/main.py +++ b/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 diff --git a/django_q/management/commands/qcluster.py b/django_q/management/commands/qcluster.py index 671e0c7..93742e3 100644 --- a/django_q/management/commands/qcluster.py +++ b/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() diff --git a/django_q/migrations/0001_initial.py b/django_q/migrations/0001_initial.py index f3cf287..50692a1 100644 --- a/django_q/migrations/0001_initial.py +++ b/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',), ), diff --git a/django_q/models.py b/django_q/models.py index 622747a..e1f1b29 100644 --- a/django_q/models.py +++ b/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() diff --git a/django_q/tests/__init__.py b/django_q/tests/__init__.py deleted file mode 100644 index dc155fb..0000000 --- a/django_q/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -__author__ = 'ilan' diff --git a/django_q/tests/settings.py b/django_q/tests/settings.py index 1fa5e46..723cbd0 100644 --- a/django_q/tests/settings.py +++ b/django_q/tests/settings.py @@ -103,3 +103,5 @@ USE_TZ = True STATIC_URL = '/static/' +# Django Q specific +#Q_PREFIX = 'test_django_q' diff --git a/django_q/tests/test_q.py b/django_q/tests/test_q.py new file mode 100644 index 0000000..497fbef --- /dev/null +++ b/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 diff --git a/pytest.ini b/pytest.ini index 748d0b0..8e18a7f 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,3 +1,2 @@ [pytest] DJANGO_SETTINGS_MODULE=django_q.tests.settings -#django_find_project = false \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index ea8b26b..4a2ce40 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ +blessed +future -e . coloredlogs==1.0.1 django-picklefield==0.3.1 diff --git a/setup.py b/setup.py index d4bbafc..f0001ad 100644 --- a/setup.py +++ b/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',