From 33282cb2e53bae22d39e51f025b8db0c135ee135 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 1 Oct 2015 14:58:28 +0200 Subject: [PATCH] cached result backend first version of a result backend using django's cache framework --- django_q/cluster.py | 28 +++++++- django_q/conf.py | 4 ++ django_q/tasks.py | 128 ++++++++++++++++++++++++++++++++-- django_q/tests/test_cached.py | 52 ++++++++++++++ 4 files changed, 204 insertions(+), 8 deletions(-) create mode 100644 django_q/tests/test_cached.py diff --git a/django_q/cluster.py b/django_q/cluster.py index 343c5ca..8694294 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -325,7 +325,11 @@ def monitor(result_queue, broker=None): if ack_id: broker.acknowledge(ack_id) # save the result - save_task(task) + if task.get('cached', False): + save_cache(task, broker) + else: + save_task(task) + # log the result if task['success']: logger.info(_("Processed [{}]").format(task['name'])) else: @@ -384,7 +388,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT): def save_task(task): """ - Saves the task package to Django + Saves the task package to Django or the cache """ # SAVE LIMIT < 0 : Don't save success if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']: @@ -409,6 +413,26 @@ def save_task(task): logger.error(e) +def save_cache(task, broker): + key = 'django_q:{}:results'.format(broker.list_key) + timeout = task['cached'] + if timeout is True: + timeout = None + try: + task_package = signing.SignedPackage.dumps(task) + group = task.get('group', False) + if group: + group_list = broker.cache.get('{}:{}'.format(key, group)) or [] + group_list.append(task_package) + broker.cache.set('{}:{}'.format(key, group), group_list, timeout) + else: + broker.cache.set('{}:{}'.format(key, task['id']), + task_package, + timeout) + except Exception as e: + logger.error(e) + + def scheduler(broker=None): """ Creates a task from a schedule at the scheduled time and schedules next run diff --git a/django_q/conf.py b/django_q/conf.py index 9ca2146..ccfd080 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -113,6 +113,10 @@ class Conf(object): # The Django cache to use CACHE = conf.get('cache', 'default') + # Use the cache as result backend. Can be 'True' or an integer representing the global cache timeout. + # i.e 'cached: 60' , will make all results go the cache and expire in 60 seconds. + CACHED = conf.get('cached', False) + # If set to False the scheduler won't execute tasks in the past. # Instead it will run once and reschedule the next run in the future. Defaults to True. CATCH_UP = conf.get('catch_up', True) diff --git a/django_q/tasks.py b/django_q/tasks.py index 62da8f1..cc5a13b 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -23,6 +23,7 @@ def async(func, *args, **kwargs): sync = options.pop('sync', False) group = options.pop('group', None) save = options.pop('save', None) + cached = options.pop('cached', Conf.CACHED) # get an id tag = uuid() # build the task package @@ -38,6 +39,8 @@ def async(func, *args, **kwargs): task['group'] = group if save is not None: task['save'] = save + if cached: + task['cached'] = cached # sign it pack = signing.SignedPackage.dumps(task) if sync or Conf.SYNC: @@ -83,7 +86,7 @@ def schedule(func, *args, **kwargs): ) -def result(task_id, wait=0): +def result(task_id, wait=0, cached=Conf.CACHED): """ Return the result of the named task. @@ -94,6 +97,8 @@ def result(task_id, wait=0): :return: the result object of this task :rtype: object """ + if cached: + return result_cached(task_id, wait) start = time.time() while True: r = Task.get_result(task_id) @@ -104,7 +109,21 @@ def result(task_id, wait=0): time.sleep(0.01) -def result_group(group_id, failures=False): +def result_cached(task_id, wait=0, broker=None): + if not broker: + broker = get_broker() + key = 'django_q:{}:results'.format(broker.list_key) + start = time.time() + while True: + r = broker.cache.get('{}:{}'.format(key, task_id)) + if r: + return signing.SignedPackage.loads(r)['result'] + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + + +def result_group(group_id, failures=False, cached=Conf.CACHED): """ Return a list of results for a task group. @@ -112,10 +131,26 @@ def result_group(group_id, failures=False): :param bool failures: set to True to include failures :return: list or results """ + if cached: + return result_group_cached(group_id, failures) return Task.get_result_group(group_id, failures) -def fetch(task_id, wait=0): +def result_group_cached(group_id, failures=False, broker=None): + if not broker: + broker = get_broker() + key = 'django_q:{}:results'.format(broker.list_key) + group_list = broker.cache.get('{}:{}'.format(key, group_id)) + if group_list: + result_list = [] + for task_package in group_list: + task = signing.SignedPackage.loads(task_package) + if task['success'] or failures: + result_list.append(task['result']) + return result_list + + +def fetch(task_id, wait=0, cached=Conf.CACHED): """ Return the processed task. @@ -126,6 +161,8 @@ def fetch(task_id, wait=0): :return: the full task object :rtype: Task """ + if cached: + return fetch_cached(task_id, wait) start = time.time() while True: t = Task.get_task(task_id) @@ -136,7 +173,32 @@ def fetch(task_id, wait=0): time.sleep(0.01) -def fetch_group(group_id, failures=True): +def fetch_cached(task_id, wait=0, broker=None): + if not broker: + broker = get_broker() + key = 'django_q:{}:results'.format(broker.list_key) + start = time.time() + while True: + r = broker.cache.get('{}:{}'.format(key, task_id)) + if r: + task = signing.SignedPackage.loads(r) + t = Task(id=task['id'], + name=task['name'], + func=task['func'], + hook=task.get('hook'), + args=task['args'], + kwargs=task['kwargs'], + started=task['started'], + stopped=task['stopped'], + result=task['result'], + success=task['success']) + return t + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + + +def fetch_group(group_id, failures=True, cached=True): """ Return a list of Tasks for a task group. @@ -144,10 +206,37 @@ def fetch_group(group_id, failures=True): :param bool failures: set to False to exclude failures :return: list of Tasks """ + if cached: + return fetch_group_cached(group_id, failures) return Task.get_task_group(group_id, failures) -def count_group(group_id, failures=False): +def fetch_group_cached(group_id, failures=True, broker=None): + if not broker: + broker = get_broker() + key = 'django_q:{}:results'.format(broker.list_key) + group_list = broker.cache.get('{}:{}'.format(key, group_id)) + if group_list: + task_list = [] + for task_package in group_list: + task = signing.SignedPackage.loads(task_package) + if task['success'] or failures: + t = Task(id=task['id'], + name=task['name'], + func=task['func'], + hook=task.get('hook'), + args=task['args'], + kwargs=task['kwargs'], + started=task['started'], + stopped=task['stopped'], + result=task['result'], + group=task.get('group'), + success=task['success']) + task_list.append(t) + return task_list + + +def count_group(group_id, failures=False, cached=Conf.CACHED): """ Count the results in a group. @@ -156,10 +245,28 @@ def count_group(group_id, failures=False): :return: the number of tasks/results in a group :rtype: int """ + if cached: + return count_group_cached(group_id, failures) return Task.get_group_count(group_id, failures) -def delete_group(group_id, tasks=False): +def count_group_cached(group_id, failures=False, broker=None): + if not broker: + broker = get_broker() + key = 'django_q:{}:results'.format(broker.list_key) + group_list = broker.cache.get('{}:{}'.format(key, group_id)) + if group_list: + if not failures: + return len(group_list) + failure_count = 0 + for task_package in group_list: + task = signing.SignedPackage.loads(task_package) + if not task['success']: + failure_count += 1 + return failure_count + + +def delete_group(group_id, tasks=False, cached=Conf.CACHE): """ Delete a group. @@ -168,9 +275,18 @@ def delete_group(group_id, tasks=False): Otherwise just the group label is removed. :return: """ + if cached: + return delete_group_cached(group_id) return Task.delete_group(group_id, tasks) +def delete_group_cached(group_id, broker=None): + if not broker: + broker = get_broker() + key = 'django_q:{}:results'.format(broker.list_key) + return broker.cache.delete('{}:{}'.format(key, group_id)) + + def queue_size(broker=None): """ Returns the current queue size. diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py new file mode 100644 index 0000000..ba44d11 --- /dev/null +++ b/django_q/tests/test_cached.py @@ -0,0 +1,52 @@ +from multiprocessing import Event + +import pytest + +from django_q.cluster import Sentinel +from django_q.conf import Conf +from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group +from django_q.brokers import get_broker + + +@pytest.fixture +def broker(): + Conf.DISQUE_NODES = None + Conf.IRON_MQ = None + Conf.SQS = None + Conf.ORM = None + Conf.MONGO = None + Conf.DJANGO_REDIS = 'default' + return get_broker() + + +@pytest.mark.django_db +def test_cached(broker): + broker.purge_queue() + broker.cache.clear() + group = 'cache_test' + # queue the tests + task_id = async('math.copysign', 1, -1, cached=True, broker=broker) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + # run a single cluster + start_event = Event() + stop_event = Event() + stop_event.set() + Sentinel(stop_event, start_event, broker=broker) + # assert results + assert result(task_id, wait=500, cached=True) == -1 + assert fetch(task_id,wait=500, cached=True).result == -1 + # make sure it's not in the db backend + assert fetch(task_id) is None + # assert group + assert count_group(group, cached=True) == 5 + assert count_group(group, cached=True, failures=True) == 0 + assert result_group(group, cached=True) == [-1, -1, -1, -1, -1] + assert len(fetch_group(group, cached=True)) == 5 + assert len(fetch_group(group, cached=True, failures=False)) == 5 + delete_group(group, cached=True) + assert count_group(group, cached=True) is None + broker.cache.clear()