diff --git a/django_q/cluster.py b/django_q/cluster.py index 4ca18f3..276c7d4 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -420,7 +420,7 @@ def save_cached(task, broker): timeout = None try: group = task.get('group', False) - iter_count = task.get('iter_count', None) + iter_count = task.get('iter_count', 0) # if it's a group append to the group list if group: task_key = '{}:{}:{}'.format(broker.list_key, group, task['id']) @@ -435,7 +435,13 @@ def save_cached(task, broker): task['result'] = results task['id'] = group task['args'] = signing.SignedPackage.loads(broker.cache.get(group_args)) - save_task(task) + task.pop('iter_count', None) + task.pop('group', None) + if task.get('iter_cached', None): + task['cached'] = task.pop('iter_cached', None) + save_cached(task, broker=broker) + else: + save_task(task) broker.cache.delete_many(group_list) broker.cache.delete_many([group_key, group_args]) return diff --git a/django_q/tasks.py b/django_q/tasks.py index 17ae5b1..758a0f0 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -25,6 +25,7 @@ def async(func, *args, **kwargs): save = options.pop('save', None) cached = options.pop('cached', Conf.CACHED) iter_count = options.pop('iter_count', None) + iter_cached = options.pop('iter_cached', None) # get an id tag = uuid() # build the task package @@ -44,6 +45,8 @@ def async(func, *args, **kwargs): task['cached'] = cached if iter_count: task['iter_count'] = iter_count + if iter_cached: + task['iter_cached'] = iter_cached # sign it pack = signing.SignedPackage.dumps(task) if sync or Conf.SYNC: @@ -388,6 +391,8 @@ def async_iter(func, args_iter, **kwargs): options['broker'] = options.get('broker', get_broker()) options['group'] = iter_group options['iter_count'] = iter_count + if options.get('cached', None): + options['iter_cached'] = options['cached'] options['cached'] = True # save the original arguments broker = options['broker'] diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py index ee05e36..2885eae 100644 --- a/django_q/tests/test_cached.py +++ b/django_q/tests/test_cached.py @@ -1,8 +1,8 @@ -from multiprocessing import Event +from multiprocessing import Event, Queue, Value import pytest +from django_q.cluster import pusher, worker, monitor -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, delete_cached, \ async_iter @@ -33,6 +33,7 @@ def test_cached(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.popysign', 1, -1, cached=True, broker=broker, group=group) + iter_id = async_iter('math.floor', [i for i in range(10)], cached=True) # test wait on cache # test wait timeout assert result(task_id, wait=10, cached=True) is None @@ -41,11 +42,23 @@ def test_cached(broker): assert result_group(group, count=2, wait=10, cached=True) is None assert fetch_group(group, wait=10, cached=True) is None assert fetch_group(group, count=2, wait=10, cached=True) is None - # run a single cluster - start_event = Event() + # run a single inline cluster + task_count = 17 + assert broker.queue_size() == task_count + task_queue = Queue() stop_event = Event() stop_event.set() - Sentinel(stop_event, start_event, broker=broker) + for i in range(task_count): + pusher(task_queue, stop_event, broker=broker) + assert broker.queue_size() == 0 + assert task_queue.qsize() == task_count + task_queue.put('STOP') + result_queue = Queue() + worker(task_queue, result_queue, Value('f', -1)) + assert result_queue.qsize() == task_count + result_queue.put('STOP') + monitor(result_queue) + assert result_queue.qsize() == 0 # assert results assert result(task_id, wait=500, cached=True) == -1 assert fetch(task_id, wait=500, cached=True).result == -1 @@ -63,6 +76,9 @@ def test_cached(broker): delete_cached(task_id) assert result(task_id, cached=True) is None assert fetch(task_id, cached=True) is None + # iter cached + assert result(iter_id) is None + assert result(iter_id, cached=True) is not None broker.cache.clear() @@ -85,3 +101,4 @@ def test_iter(broker): assert result(t2) is not None assert result(t3) is not None assert result(t4)[0] == 1 + # test cached iter result diff --git a/docs/tasks.rst b/docs/tasks.rst index e9b1df6..d073112 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -320,11 +320,12 @@ Reference .. py:function:: async_iter(func, args_iter,**kwargs) - Runs iterable arguments against the cache backend and returns a single collated result + Runs iterable arguments against the cache backend and returns a single collated result. + Accepts the same options as :func:`async` except ``hook``. :param object func: The task function to execute :param args: An iterable containing arguments for the task function - :param dict kwargs: Keyword arguments for the task function. Ignores ``cached`` and ``hook``. + :param dict kwargs: Keyword arguments for the task function. Ignores ``hook``. :returns: The uuid of the task :rtype: str