Merge pull request #84 from Koed00/dev

Adds cached result backend

README.rst

@@ -15,7 +15,7 @@ Features
 -  Asynchronous tasks
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
--  Failure and success database
+-  Failure and success database or cache
 -  Result hooks and groups
 -  Django Admin integration
 -  PaaS compatible with multiple instances

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_cached(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,43 @@ def save_task(task):
         logger.error(e)
 
 
+def save_cached(task, broker):
+    task_key = '{}:{}'.format(broker.list_key, task['id'])
+    timeout = task['cached']
+    if timeout is True:
+        timeout = None
+    try:
+        group = task.get('group', False)
+        iter_count = task.get('iter_count', None)
+        # if it's a group append to the group list
+        if group:
+            task_key = '{}:{}:{}'.format(broker.list_key, group, task['id'])
+            group_key = '{}:{}:keys'.format(broker.list_key, group)
+            group_list = broker.cache.get(group_key) or []
+            # if it's an inter group, check if we are ready
+            if iter_count and len(group_list) == iter_count-1:
+                group_args = '{}:{}:args'.format(broker.list_key, group)
+                # collate the results into a Task result
+                results = [signing.SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list]
+                results.append(task['result'])
+                task['result'] = results
+                task['id'] = group
+                task['args'] = signing.SignedPackage.loads(broker.cache.get(group_args))
+                save_task(task)
+                broker.cache.delete_many(group_list)
+                broker.cache.delete_many([group_key, group_args])
+                return
+            # save the group list
+            group_list.append(task_key)
+            broker.cache.set(group_key, group_list)
+        # save the task
+        broker.cache.set(task_key,
+                         signing.SignedPackage.dumps(task),
+                         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

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)

django_q/tasks.py

@@ -1,4 +1,4 @@
-"""Provides task functionalities."""
+"""Provides task functionality."""
 from multiprocessing import Queue, Value
 
 # django
@@ -15,7 +15,7 @@ from django_q.brokers import get_broker
 
 
 def async(func, *args, **kwargs):
-    """Send a task to the cluster."""
+    """Queue a task for the cluster."""
     # get options from q_options dict or direct from kwargs
     options = kwargs.pop('q_options', kwargs)
     hook = options.pop('hook', None)
@@ -23,6 +23,8 @@ 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)
+    iter_count = options.pop('iter_count', None)
     # get an id
     tag = uuid()
     # build the task package
@@ -38,6 +40,10 @@ def async(func, *args, **kwargs):
         task['group'] = group
     if save is not None:
         task['save'] = save
+    if cached:
+        task['cached'] = cached
+    if iter_count:
+        task['iter_count'] = iter_count
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync or Conf.SYNC:
@@ -83,7 +89,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.
 
@@ -91,9 +97,12 @@ def result(task_id, wait=0):
     :param task_id: the task name or uuid
     :type wait: int
     :param wait: number of milliseconds to wait for a result
+    :param bool cached: run this against the cache backend
     :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,18 +113,76 @@ 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):
+    """
+     Return the result from the cache backend
+    """
+    if not broker:
+        broker = get_broker()
+    start = time.time()
+    while True:
+        r = broker.cache.get('{}:{}'.format(broker.list_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, wait=0, count=None, cached=Conf.CACHED):
     """
     Return a list of results for a task group.
 
     :param str group_id: the group id
     :param bool failures: set to True to include failures
+    :param int count: Block until there are this many results in the group
+    :param bool cached: run this against the cache backend
     :return: list or results
     """
-    return Task.get_result_group(group_id, failures)
+    if cached:
+        return result_group_cached(group_id, failures, wait, count)
+    start = time.time()
+    if count:
+        while True:
+            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
+    while True:
+        r = Task.get_result_group(group_id, failures)
+        if r:
+            return r
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
 
 
-def fetch(task_id, wait=0):
+def result_group_cached(group_id, failures=False, wait=0, count=None, broker=None):
+    """
+    Return a list of results for a task group from the cache backend
+    """
+    if not broker:
+        broker = get_broker()
+    start = time.time()
+    if count:
+        while True:
+            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
+    while True:
+        group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
+        if group_list:
+            result_list = []
+            for task_key in group_list:
+                task = signing.SignedPackage.loads(broker.cache.get(task_key))
+                if task['success'] or failures:
+                    result_list.append(task['result'])
+            return result_list
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
+
+
+def fetch(task_id, wait=0, cached=Conf.CACHED):
     """
     Return the processed task.
 
@@ -123,9 +190,12 @@ def fetch(task_id, wait=0):
     :type task_id: str or uuid
     :param wait: the number of milliseconds to wait for a result
     :type wait: int
+    :param bool cached: run this against the cache backend
     :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,45 +206,169 @@ 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):
+    """
+    Return the processed task from the cache backend
+    """
+    if not broker:
+        broker = get_broker()
+    start = time.time()
+    while True:
+        r = broker.cache.get('{}:{}'.format(broker.list_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, wait=0, count=None, cached=Conf.CACHED):
     """
     Return a list of Tasks for a task group.
 
     :param str group_id: the group id
     :param bool failures: set to False to exclude failures
+    :param bool cached: run this against the cache backend
     :return: list of Tasks
     """
-    return Task.get_task_group(group_id, failures)
+    if cached:
+        return fetch_group_cached(group_id, failures, wait, count)
+    start = time.time()
+    if count:
+        while True:
+            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
+    while True:
+        r = Task.get_task_group(group_id, failures)
+        if r:
+            return r
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
 
 
-def count_group(group_id, failures=False):
+def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None):
+    """
+    Return a list of Tasks for a task group in the cache backend
+    """
+    if not broker:
+        broker = get_broker()
+    start = time.time()
+    if count:
+        while True:
+            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+                break
+            time.sleep(0.01)
+    while True:
+        group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
+        if group_list:
+            task_list = []
+            for task_key in group_list:
+                task = signing.SignedPackage.loads(broker.cache.get(task_key))
+                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
+        if (time.time() - start) * 1000 >= wait:
+            break
+        time.sleep(0.01)
+
+
+def count_group(group_id, failures=False, cached=Conf.CACHED):
     """
     Count the results in a group.
 
     :param str group_id: the group id
     :param bool failures: Returns failure count if True
+    :param bool cached: run this against the cache backend
     :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):
+    """
+    Count the results in a group in the cache backend
+    """
+    if not broker:
+        broker = get_broker()
+    group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
+    if group_list:
+        if not failures:
+            return len(group_list)
+        failure_count = 0
+        for task_key in group_list:
+            task = signing.SignedPackage.loads(broker.cache.get(task_key))
+            if not task['success']:
+                failure_count += 1
+        return failure_count
+
+
+def delete_group(group_id, tasks=False, cached=Conf.CACHED):
     """
     Delete a group.
 
     :param str group_id: the group id
     :param bool tasks: If set to True this will also delete the group tasks.
     Otherwise just the group label is removed.
+    :param bool cached: run this against the cache backend
     :return:
     """
+    if cached:
+        return delete_group_cached(group_id)
     return Task.delete_group(group_id, tasks)
 
 
+def delete_group_cached(group_id, broker=None):
+    """
+    Delete a group from the cache backend
+    """
+    if not broker:
+        broker = get_broker()
+    group_key = '{}:{}:keys'.format(broker.list_key, group_id)
+    group_list = broker.cache.get(group_key)
+    broker.cache.delete_many(group_list)
+    broker.cache.delete(group_key)
+
+
+def delete_cached(task_id, broker=None):
+    """
+    Delete a task from the cache backend
+    """
+    if not broker:
+        broker = get_broker()
+    return broker.cache.delete('{}:{}'.format(broker.list_key, task_id))
+
+
 def queue_size(broker=None):
     """
     Returns the current queue size.
-    Note that this doesn't count any tasks currently being processed by workers.
+    Note that this doesn't count any tasks curren    key = 'django_q:{}:results'.format(broker.list_key)tly being processed by workers.
 
     :param broker: optional broker
     :return: current queue size
@@ -185,6 +379,26 @@ def queue_size(broker=None):
     return broker.queue_size()
 
 
+def async_iter(func, args_iter, **kwargs):
+    iter_count = len(args_iter)
+    iter_group = uuid()[1]
+    # clean up the kwargs
+    options = kwargs.get('q_options', kwargs)
+    options.pop('hook', None)
+    options['broker'] = options.get('broker', get_broker())
+    options['group'] = iter_group
+    options['iter_count'] = iter_count
+    options['cached'] = True
+    # save the original arguments
+    broker = options['broker']
+    broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), signing.SignedPackage.dumps(args_iter))
+    for args in args_iter:
+        if type(args) is not tuple:
+            args = (args,)
+        async(func, *args, **options)
+    return iter_group
+
+
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
     task_queue = Queue()

django_q/tests/test_cached.py

@@ -0,0 +1,87 @@
+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, delete_cached, \
+    async_iter
+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)
+    async('math.popysign', 1, -1, cached=True, broker=broker, group=group)
+    # test wait on cache
+    # test wait timeout
+    assert result(task_id, wait=10, cached=True) is None
+    assert fetch(task_id, wait=10, cached=True) is None
+    assert result_group(group, wait=10, cached=True) is None
+    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()
+    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) == 6
+    assert count_group(group, cached=True, failures=True) == 1
+    assert result_group(group, cached=True) == [-1, -1, -1, -1, -1]
+    assert len(result_group(group, cached=True, failures=True)) == 6
+    assert len(fetch_group(group, cached=True)) == 6
+    assert len(fetch_group(group, cached=True, failures=False)) == 5
+    delete_group(group, cached=True)
+    assert count_group(group, cached=True) is None
+    delete_cached(task_id)
+    assert result(task_id, cached=True) is None
+    assert fetch(task_id, cached=True) is None
+    broker.cache.clear()
+
+
+@pytest.mark.django_db
+def test_iter(broker):
+    broker.purge_queue()
+    broker.cache.clear()
+    it = [i for i in range(10)]
+    it2 = [(1, -1), (2, -1), (3, -4), (5, 6)]
+    it3 = (1, 2, 3, 4, 5)
+    t = async_iter('math.floor', it, sync=True)
+    t2 = async_iter('math.copysign', it2, sync=True)
+    t3 = async_iter('math.floor', it3, sync=True)
+    t4 = async_iter('math.floor', (1,), sync=True)
+    result_t = result(t)
+    assert result_t is not None
+    task_t = fetch(t)
+    assert task_t. __unicode__ is not None
+    assert task_t.result == result_t
+    assert result(t2) is not None
+    assert result(t3) is not None
+    assert result(t4)[0] == 1

django_q/tests/test_cluster.py

@@ -160,6 +160,13 @@ def test_async(broker, admin_user):
     assert broker.queue_size() == 0
     assert task_queue.qsize() == task_count
     task_queue.put('STOP')
+    # test wait timeout
+    assert result(j, wait=10) is None
+    assert fetch(j, wait=10) is None
+    assert result_group('test_j', wait=10) is None
+    assert result_group('test_j', count=2, wait=10) is None
+    assert fetch_group('test_j', wait=10) is None
+    assert fetch_group('test_j', count=2, wait=10) is None
     # let a worker handle them
     result_queue = Queue()
     worker(task_queue, result_queue, Value('f', -1))

docs/configure.rst

@@ -308,6 +308,13 @@ cache
 For some brokers, you will need to set up the Django `cache framework <https://docs.djangoproject.com/en/1.8/topics/cache/#setting-up-the-cache>`__
 to gather statistics for the monitor. You can indicate which cache to use by setting this value. Defaults to ``default``.
 
+.. _cached:
+
+cached
+~~~~~~
+Switches all task and result functions from using the database backend to the cache backend. This is the same as setting the keyword ``cached=True`` on all task functions.
+Instead of a bool this can also be set to the number of seconds you want the cache to retain results. e.g. ``cached=60``
+
 scheduler
 ~~~~~~~~~
 You can disable the scheduler by setting this option to ``False``. This will reduce a little overhead if you're not using schedules, but is most useful if you want to temporarily disable all schedules.

docs/examples.rst

@@ -253,8 +253,7 @@ Adapted from `Sebastian Raschka's blog <http://sebastianraschka.com/Articles/201
     # Group example with Parzen-window estimation
     import numpy
 
-    from django_q.tasks import async, result_group,\
+    from django_q.tasks import async, result_group, delete_group
-        count_group, delete_group
 
     # the estimation function
     def parzen_estimation(x_samples, point_x, h):
@@ -268,30 +267,47 @@ Adapted from `Sebastian Raschka's blog <http://sebastianraschka.com/Articles/201
                 k_n += 1
         return h, (k_n / len(x_samples)) / (h ** point_x.shape[1])
 
-
     # create 100 calculations and send them to the cluster
     def parzen_async():
         # clear the previous results
-        delete_group('parzen', tasks=True)
+        delete_group('parzen', cached=True)
         mu_vec = numpy.array([0, 0])
         cov_mat = numpy.array([[1, 0], [0, 1]])
-        sample = numpy.random.\
+        sample = numpy.random. \
             multivariate_normal(mu_vec, cov_mat, 10000)
         widths = numpy.linspace(1.0, 1.2, 100)
         x = numpy.array([[0], [0]])
-        # async them with a group label and a hook
+        # async them with a group label to the cache backend
         for w in widths:
             async(parzen_estimation, sample, x, w,
-             group='parzen', hook=parzen_hook)
+                  group='parzen', cached=True)
+        # return after 100 results
+        return result_group('parzen', count=100, cached=True)
 
-    # wait for 100 results to return and print it.
-    def parzen_hook(task):
-        if task.group_count() == 100:
-            print(task.group_result())
 
 
 Django Q is not optimized for distributed computing, but this example will give you an idea of what you can do with task :ref:`groups`.
 
+Alternatively the ``parzen_async()`` function can also be written with :func:`async_iter`, which automatically utilizes the cache backend and groups to return a single result from an iterable:
+
+.. code-block:: python
+
+    # create 100 calculations and send them to the cluster
+    # with async_iter
+    def parzen_async():
+        mu_vec = numpy.array([0, 0])
+        cov_mat = numpy.array([[1, 0], [0, 1]])
+        sample = numpy.random. \
+            multivariate_normal(mu_vec, cov_mat, 10000)
+        widths = numpy.linspace(1.0, 1.2, 100)
+        x = numpy.array([[0], [0]])
+        # async them with async iterable
+        args = [(sample, x, w) for w in widths]
+        result_id = async_iter(parzen_estimation, args)
+        # return the result or timeout after 10 seconds
+        return result(result_id, wait=10000)
+
+
 .. note::
 
     If you have an example you want to share, please submit a pull request on `github <https://github.com/Koed00/django-q/>`__.

docs/index.rst

@@ -15,7 +15,7 @@ Features
 -  Asynchronous tasks
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
--  Failure and success database
+-  Failure and success database or cache
 -  Result hooks and groups
 -  Django Admin integration
 -  PaaS compatible with multiple instances

docs/tasks.rst

@@ -59,6 +59,12 @@ sync
 Simulates a task execution synchronously. Useful for testing.
 Can also be forced globally via the :ref:`sync` configuration option.
 
+cached
+""""""
+Redirects the result to the cache backend instead of the database if set to ``True`` or to an integer indicating the cache timeout in seconds.
+e.g. ``cached=60``. Especially useful with large and group operations where you don't need the all results in your
+database and want to take advantage of the speed of your cache backend.
+
 broker
 """"""
 A broker instance, in case you want to control your own connections.
@@ -83,6 +89,28 @@ Please not that this will override any other option keywords.
     For tasks to be processed you will need to have a worker cluster running in the background using ``python manage.py qcluster``
     or you need to configure Django Q to run in synchronous mode for testing using the :ref:`sync` option.
 
+
+
+Async Iterable
+--------------
+If you have an iterable object with arguments for a function, you can use :func:`async_iter` to async them with a single command::
+
+    # Async Iterable example
+    from django_q.tasks import async_iter, result
+
+    # set up a list of arguments for math.floor
+    iter = [i for i in range(100)]
+
+    # async iter them
+    id=async_iter('math.floor',iter)
+
+    # wait for the collated result for 1 second
+    result_list = result(id, wait=1000)
+
+This will individually queue 100 tasks to the worker cluster, which will save their results in the cache backend for speed.
+Once all the 100 results are in the cache, they are collated into a list and saved as a single result in the database. The cache results are then cleared.
+Needs the Django cache framework.
+
 .. _groups:
 
 Groups
@@ -97,14 +125,16 @@ You can group together results by passing :func:`async` the optional ``group`` k
     for i in range(4):
         async('math.modf', i, group='modf')
 
-    # after the tasks have finished you can get the group results
+    # wait until the group has 4 results
-    result = result_group('modf')
+    result = result_group('modf', count=4)
     print(result)
 
 .. code-block:: python
 
     [(0.0, 0.0), (0.0, 1.0), (0.0, 2.0), (0.0, 3.0)]
 
+Note that the same can be achieved much faster with :func:`async_iter`
+
 Take care to not limit your results database too much and call :func:`delete_group` before each run, unless you want your results to keep adding up.
 Instead of :func:`result_group` you can also use :func:`fetch_group` to return a queryset of :class:`Task` objects.:
 
@@ -149,6 +179,54 @@ You can also access group functions from a task result instance:
         task.group_delete()
         print('Deleted group {}'.format(task.group))
 
+Cached operations
+-----------------
+You can run your tasks results against the Django cache backend instead of the database backend by either using the global :ref:`cached` setting or by supplying the ``cached`` keyword to individual functions.
+This can be useful if you are not interested in persistent results or if you run large group tasks where you only want the final result.
+By using a cache backend like Redis or Memcached you can speed up access to your task results significantly compared to a relational database.
+
+When you set ``cached=True``, results will be saved permanently in the cache and you will have to rely on your backend's cleanup strategies (like LRU) to
+manage stale results.
+You can also opt to set a manual timeout on the results, by setting ``cached=60``. Meaning the result will be evicted from the cache after 60 seconds.
+This works both globally or on individual async executions.::
+
+    # simple cached example
+    from django_q.tasks import async, result
+
+    # cache the result for 10 seconds
+    id = async('math.floor', 100, cached=10)
+
+    # wait max 50ms for the result to appear in the cache
+    result(id, wait=50, cached=True)
+
+    # o fetch the task object
+    task = fetch(id, cache=True)
+
+    # and then save it to the database
+    task.save()
+
+This also works for group actions::
+
+    # cached group example
+    from django_q.tasks import async, result_group
+    from django_q.brokers import get_broker
+
+    # set up a broker instance for better performance
+    broker = get_broker()
+
+    # async a hundred functions under a group label
+    for i in range(100):
+        async('math.frexp',
+              i,
+              group='frexp',
+              cached=True,
+              broker=broker)
+
+    # wait max 50ms for one hundred results to return
+    result_group('frexp', wait=50, count=100, cached=True)
+
+Note that exact same result can be achieved by using the more convenient :func:`async_iter` in this case, but without hook support.
+
 Synchronous testing
 -------------------
 
@@ -199,7 +277,7 @@ Reference
 ---------
 
 .. py:function:: async(func, *args, hook=None, group=None, timeout=None,\
-    save=None, sync=False, broker=None, q_options=None, **kwargs)
+    save=None, sync=False, cached=False, broker=None, q_options=None, **kwargs)
 
     Puts a task in the cluster queue
 
@@ -210,26 +288,29 @@ Reference
    :param int timeout: Overrides global cluster :ref:`timeout`.
    :param bool save: Overrides global save setting for this task.
    :param bool sync: If set to True, async will simulate a task execution
+   :param cached: Output the result to the cache backend. Bool or timeout in seconds
    :param broker: Optional broker connection from :func:`brokers.get_broker`
    :param dict q_options: Options dict, overrides option keywords
    :param dict kwargs: Keyword arguments for the task function
    :returns: The uuid of the task
    :rtype: str
 
-.. py:function:: result(task_id, wait=0)
+.. py:function:: result(task_id, wait=0, cached=False)
 
     Gets the result of a previously executed task
 
     :param str task_id: the uuid or name of the task
     :param int wait: optional milliseconds to wait for a result
+    :param bool cached: run this against the cache backend.
     :returns: The result of the executed task
 
-.. py:function:: fetch(task_id, wait=0)
+.. py:function:: fetch(task_id, wait=0, cached=False)
 
     Returns a previously executed task
 
     :param str name: the uuid or name of the task
     :param int wait: optional milliseconds to wait for a result
+    :param bool cached: run this against the cache backend.
     :returns: A task object
     :rtype: Task
 
@@ -237,6 +318,16 @@ Reference
 
     Renamed from get_task
 
+.. py:function:: async_iter(func, args_iter,**kwargs)
+
+   Runs iterable arguments against the cache backend and returns a single collated result
+
+   :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``.
+   :returns: The uuid of the task
+   :rtype: str
+
 .. py:function:: queue_size()
 
     Returns the size of the broker queue.
@@ -245,42 +336,58 @@ Reference
     :returns: The amount of task packages in the broker
     :rtype: int
 
-.. py:function:: result_group(group_id, failures=False)
+.. py:function:: result_group(group_id, failures=False, wait=0, count=None, cached=False)
 
     Returns the results of a task group
 
     :param str group_id: the group identifier
     :param bool failures: set this to ``True`` to include failed results
+    :param int wait: optional milliseconds to wait for a result or count
+    :param int count: block until there are this many results in the group
+    :param bool cached: run this against the cache backend
     :returns: a list of results
     :rtype: list
 
-.. py:function:: fetch_group(group_id, failures=True)
+.. py:function:: fetch_group(group_id, failures=True, wait=0, count=None, cached=False)
 
     Returns a list of tasks in a group
 
     :param str group_id: the group identifier
     :param bool failures: set this to ``False`` to exclude failed tasks
-    :returns: a list of Tasks
+    :param int wait: optional milliseconds to wait for a task or count
+    :param int count: block until there are this many tasks in the group
+    :param bool cached: run this against the cache backend.
+    :returns: a list of :class:`Task`
     :rtype: list
 
-.. py:function:: count_group(group_id, failures=False)
+.. py:function:: count_group(group_id, failures=False, cached=False)
 
     Counts the number of task results in a group.
 
     :param str group_id: the group identifier
     :param bool failures: counts the number of failures if ``True``
+    :param bool cached: run this against the cache backend.
     :returns: the number of tasks or failures in a group
     :rtype: int
 
-.. py:function:: delete_group(group_id, tasks=False)
+.. py:function:: delete_group(group_id, tasks=False, cached=False)
 
     Deletes a group label from the database.
 
     :param str group_id: the group identifier
     :param bool tasks: also deletes the associated tasks if ``True``
+    :param bool cached: run this against the cache backend.
     :returns: the numbers of tasks affected
     :rtype: int
 
+.. py:function:: delete_cached(task_id, broker=None)
+
+    Deletes a task from the cache backend
+
+    :param task_id: the uuid of the task
+    :param broker: an optional broker instance
+
+
 .. py:class:: Task
 
     Database model describing an executed task