From 6dcead7310c636fb9503c5262496c8f9ade6c6e0 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sat, 17 Oct 2015 12:13:29 +0200
Subject: [PATCH 1/8] Adds task chains

---
 django_q/cluster.py |  10 +++-
 django_q/models.py  |   2 +-
 django_q/tasks.py   | 115 ++++++++++++++++++++++++++++++++++++--------
 3 files changed, 103 insertions(+), 24 deletions(-)

diff --git a/django_q/cluster.py b/django_q/cluster.py
index 276c7d4..d69a34a 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -393,6 +393,9 @@ def save_task(task):
     # SAVE LIMIT < 0 : Don't save success
     if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
         return
+    # async next in a chain
+    if task.get('chain', None):
+        tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'])
     # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
     db.close_old_connections()
     try:
@@ -419,14 +422,14 @@ def save_cached(task, broker):
     if timeout is True:
         timeout = None
     try:
-        group = task.get('group', False)
+        group = task.get('group', 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'])
             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 it's an iter 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
@@ -448,6 +451,9 @@ def save_cached(task, broker):
             # save the group list
             group_list.append(task_key)
             broker.cache.set(group_key, group_list)
+            # async next in a chain
+            if task.get('chain', None):
+                tasks.async_chain(task['chain'], group=group, cached=task['cached'])
         # save the task
         broker.cache.set(task_key,
                          signing.SignedPackage.dumps(task),
diff --git a/django_q/models.py b/django_q/models.py
index 0111727..ceba5ad 100644
--- a/django_q/models.py
+++ b/django_q/models.py
@@ -79,7 +79,7 @@ class Task(models.Model):
         return (self.stopped - self.started).total_seconds()
 
     def __unicode__(self):
-        return self.name
+        return u'{}'.format(self.name or self.id)
 
     class Meta:
         app_label = 'django_q'
diff --git a/django_q/tasks.py b/django_q/tasks.py
index 758a0f0..6c04f80 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -18,15 +18,19 @@ def async(func, *args, **kwargs):
     """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)
     broker = options.pop('broker', get_broker())
     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)
-    iter_cached = options.pop('iter_cached', None)
-    # get an id
+    # pop optionals
+    opts = {'hook': None,
+            'group': None,
+            'save': None,
+            'cached': Conf.CACHED,
+            'iter_count': None,
+            'iter_cached': None,
+            'chain': None}
+    for key in opts:
+        opts[key] = options.pop(key, opts[key])
+        # get an id
     tag = uuid()
     # build the task package
     task = {'id': tag[1], 'name': tag[0],
@@ -34,19 +38,10 @@ def async(func, *args, **kwargs):
             'args': args,
             'kwargs': kwargs,
             'started': timezone.now()}
-    # add optionals
-    if hook:
-        task['hook'] = hook
-    if group:
-        task['group'] = group
-    if save is not None:
-        task['save'] = save
-    if cached:
-        task['cached'] = cached
-    if iter_count:
-        task['iter_count'] = iter_count
-    if iter_cached:
-        task['iter_cached'] = iter_cached
+    # push optionals
+    for key in opts:
+        if opts[key] is not None:
+            task[key] = opts[key]
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync or Conf.SYNC:
@@ -371,7 +366,7 @@ def delete_cached(task_id, broker=None):
 def queue_size(broker=None):
     """
     Returns the current queue size.
-    Note that this doesn't count any tasks curren    key = 'django_q:{}:results'.format(broker.list_key)tly being processed by workers.
+    Note that this doesn't count any tasks currently being processed by workers.
 
     :param broker: optional broker
     :return: current queue size
@@ -383,6 +378,9 @@ def queue_size(broker=None):
 
 
 def async_iter(func, args_iter, **kwargs):
+    """
+    async a function with iterable arguments
+    """
     iter_count = len(args_iter)
     iter_group = uuid()[1]
     # clean up the kwargs
@@ -404,6 +402,81 @@ def async_iter(func, args_iter, **kwargs):
     return iter_group
 
 
+def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC):
+    """
+    async a chain of tasks
+    the chain must be in the format [(func,(args),{kwargs}),(func,(args),{kwargs})]
+    """
+    if not group:
+        group = uuid()[1]
+    args = ()
+    kwargs = {}
+    task = chain.pop(0)
+    if type(task) is not tuple:
+        task = (task,)
+    if len(task) > 1:
+        args = task[1]
+    if len(task) > 2:
+        kwargs = task[2]
+    kwargs['chain'] = chain
+    kwargs['group'] = group
+    kwargs['cached'] = cached
+    kwargs['sync'] = sync
+    async(task[0], *args, **kwargs)
+    return group
+
+
+class Chain(object):
+    """
+    A sequential chain of tasks
+    """
+    def __init__(self, chain=None, group=None, cached=Conf.CACHED, sync=Conf.SYNC):
+        self.chain = chain or []
+        self.group = group or ''
+        self.cached = cached
+        self.sync = sync
+
+    def append(self, func, *args, **kwargs):
+        """
+        add a task to the chain
+        takes the same parameters as async()
+        """
+        task = (func, args, kwargs)
+        self.chain.append(task)
+
+    def run(self):
+        """
+        Start queueing the chain to the worker cluster
+        :return: the chain's group id
+        """
+        self.group = async_chain(self.chain, group=self.group, cached=self.cached, sync=self.sync)
+        return self.group
+
+    def result(self, wait=0):
+        """
+        return the full list of results from the chain when it finishes. blocks until timeout.
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+        """
+        return result_group(self.group, wait=wait, count=len(self.chain), cached=self.cached)
+
+    def fetch(self, failures=True, wait=0):
+        """
+        get the task result objects from the chain when it finishes. blocks until timeout.
+        :param failures: include failed tasks
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+        """
+        return fetch_group(self.group, failures=failures, wait=wait, count=len(self.chain), cached=self.cached)
+
+    def current(self):
+        """
+        get the index of the currently executing chain element
+        :return int: current chain index
+        """
+        return count_group(self.group, cached=self.cached)
+
+
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
     task_queue = Queue()

From 2df29b908c07e3bb809b75ac8cb53a3fdafeb26f Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sat, 17 Oct 2015 18:49:45 +0200
Subject: [PATCH 2/8] Adds tests for task chains

adds several small improvements based on the problems that surfaced during writing the tests
---
 django_q/cluster.py           | 12 ++++++------
 django_q/tasks.py             | 36 +++++++++++++++++++++++++++--------
 django_q/tests/tasks.py       |  4 ++++
 django_q/tests/test_cached.py | 35 +++++++++++++++++++++++++++++++---
 4 files changed, 70 insertions(+), 17 deletions(-)

diff --git a/django_q/cluster.py b/django_q/cluster.py
index d69a34a..b5e8d0c 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -328,7 +328,7 @@ def monitor(result_queue, broker=None):
         if task.get('cached', False):
             save_cached(task, broker)
         else:
-            save_task(task)
+            save_task(task, broker)
         # log the result
         if task['success']:
             logger.info(_("Processed [{}]").format(task['name']))
@@ -386,7 +386,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
     logger.info(_('{} stopped doing work').format(name))
 
 
-def save_task(task):
+def save_task(task, broker):
     """
     Saves the task package to Django or the cache
     """
@@ -395,7 +395,7 @@ def save_task(task):
         return
     # async next in a chain
     if task.get('chain', None):
-        tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'])
+        tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'], sync=task['sync'], broker=broker)
     # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
     db.close_old_connections()
     try:
@@ -430,7 +430,7 @@ def save_cached(task, broker):
             group_key = '{}:{}:keys'.format(broker.list_key, group)
             group_list = broker.cache.get(group_key) or []
             # if it's an iter group, check if we are ready
-            if iter_count and len(group_list) == iter_count-1:
+            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]
@@ -444,7 +444,7 @@ def save_cached(task, broker):
                     task['cached'] = task.pop('iter_cached', None)
                     save_cached(task, broker=broker)
                 else:
-                    save_task(task)
+                    save_task(task, broker)
                 broker.cache.delete_many(group_list)
                 broker.cache.delete_many([group_key, group_args])
                 return
@@ -453,7 +453,7 @@ def save_cached(task, broker):
             broker.cache.set(group_key, group_list)
             # async next in a chain
             if task.get('chain', None):
-                tasks.async_chain(task['chain'], group=group, cached=task['cached'])
+                tasks.async_chain(task['chain'], group=group, cached=task['cached'], sync=task['sync'], broker=broker)
         # save the task
         broker.cache.set(task_key,
                          signing.SignedPackage.dumps(task),
diff --git a/django_q/tasks.py b/django_q/tasks.py
index 6c04f80..e572c7c 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -19,11 +19,11 @@ def async(func, *args, **kwargs):
     # get options from q_options dict or direct from kwargs
     options = kwargs.pop('q_options', kwargs)
     broker = options.pop('broker', get_broker())
-    sync = options.pop('sync', False)
     # pop optionals
     opts = {'hook': None,
             'group': None,
             'save': None,
+            'sync': None,
             'cached': Conf.CACHED,
             'iter_count': None,
             'iter_cached': None,
@@ -44,7 +44,7 @@ def async(func, *args, **kwargs):
             task[key] = opts[key]
     # sign it
     pack = signing.SignedPackage.dumps(task)
-    if sync or Conf.SYNC:
+    if task.get('sync', False) or Conf.SYNC:
         return _sync(pack)
     # push it
     broker.enqueue(pack)
@@ -402,7 +402,7 @@ def async_iter(func, args_iter, **kwargs):
     return iter_group
 
 
-def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC):
+def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None):
     """
     async a chain of tasks
     the chain must be in the format [(func,(args),{kwargs}),(func,(args),{kwargs})]
@@ -422,6 +422,7 @@ def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC):
     kwargs['group'] = group
     kwargs['cached'] = cached
     kwargs['sync'] = sync
+    kwargs['broker'] = broker or get_broker()
     async(task[0], *args, **kwargs)
     return group
 
@@ -430,26 +431,34 @@ class Chain(object):
     """
     A sequential chain of tasks
     """
+
     def __init__(self, chain=None, group=None, cached=Conf.CACHED, sync=Conf.SYNC):
         self.chain = chain or []
         self.group = group or ''
+        self.broker = get_broker()
         self.cached = cached
         self.sync = sync
+        self.started = False
 
     def append(self, func, *args, **kwargs):
         """
         add a task to the chain
         takes the same parameters as async()
         """
-        task = (func, args, kwargs)
-        self.chain.append(task)
+        self.chain.append((func, args, kwargs))
+        # remove existing results
+        if self.started:
+            delete_group(self.group)
+            self.started = False
 
     def run(self):
         """
         Start queueing the chain to the worker cluster
         :return: the chain's group id
         """
-        self.group = async_chain(self.chain, group=self.group, cached=self.cached, sync=self.sync)
+        self.group = async_chain(chain=self.chain.copy(), group=self.group, cached=self.cached, sync=self.sync,
+                                 broker=self.broker)
+        self.started = True
         return self.group
 
     def result(self, wait=0):
@@ -458,7 +467,8 @@ class Chain(object):
         :param int wait: how many milliseconds to wait for a result
         :return: an unsorted list of results
         """
-        return result_group(self.group, wait=wait, count=len(self.chain), cached=self.cached)
+        if self.started:
+            return result_group(self.group, wait=wait, count=self.length(), cached=self.cached)
 
     def fetch(self, failures=True, wait=0):
         """
@@ -467,15 +477,25 @@ class Chain(object):
         :param int wait: how many milliseconds to wait for a result
         :return: an unsorted list of task objects
         """
-        return fetch_group(self.group, failures=failures, wait=wait, count=len(self.chain), cached=self.cached)
+        if self.started:
+            return fetch_group(self.group, failures=failures, wait=wait, count=self.length(), cached=self.cached)
 
     def current(self):
         """
         get the index of the currently executing chain element
         :return int: current chain index
         """
+        if not self.started:
+            return None
         return count_group(self.group, cached=self.cached)
 
+    def length(self):
+        """
+        get the length of the chain
+        :return int: length of the chain
+        """
+        return len(self.chain)
+
 
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
diff --git a/django_q/tests/tasks.py b/django_q/tests/tasks.py
index 62e5674..27ad6bb 100644
--- a/django_q/tests/tasks.py
+++ b/django_q/tests/tasks.py
@@ -38,5 +38,9 @@ def get_user_id(user):
     return user.id
 
 
+def hello():
+    return 'hello'
+
+
 def result(obj):
     print('RESULT HOOK {} : {}'.format(obj.name, obj.result))
diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index 2885eae..58bf11b 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -1,11 +1,11 @@
 from multiprocessing import Event, Queue, Value
 
 import pytest
-from django_q.cluster import pusher, worker, monitor
 
+from django_q.cluster import pusher, worker, monitor
 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
+    async_iter, Chain, async_chain
 from django_q.brokers import get_broker
 
 
@@ -96,9 +96,38 @@ def test_iter(broker):
     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
     # test cached iter result
+
+
+@pytest.mark.django_db
+def test_chain(broker):
+    broker.purge_queue()
+    broker.cache.clear()
+    task_chain = Chain(sync=True)
+    task_chain.append('math.floor', 1)
+    task_chain.append('math.copysign', 1, -1)
+    task_chain.append('math.floor', 2)
+    assert task_chain.length() == 3
+    assert task_chain.current() is None
+    task_chain.run()
+    r = task_chain.result(wait=1000)
+    assert task_chain.current() == task_chain.length()
+    assert len(r) == task_chain.length()
+    t = task_chain.fetch()
+    assert len(t) == task_chain.length()
+    task_chain.cached = True
+    task_chain.append('math.floor', 3)
+    assert task_chain.length() == 4
+    task_chain.run()
+    r = task_chain.result(wait=1000)
+    assert task_chain.current() == task_chain.length()
+    assert len(r) == task_chain.length()
+    t = task_chain.fetch()
+    assert len(t) == task_chain.length()
+    # test single
+    rid = async_chain(['django_q.tests.tasks.hello', 'django_q.tests.tasks.hello'], sync=True, cached=True)
+    assert result_group(rid, cached=True) == ['hello', 'hello']

From 083715f9bdcd893bacd14b84323af2e3c8f474cf Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sat, 17 Oct 2015 18:59:17 +0200
Subject: [PATCH 3/8] Replaces list.copy with a slice for python 2.7

---
 django_q/tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index e572c7c..b36a9ba 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -456,7 +456,7 @@ class Chain(object):
         Start queueing the chain to the worker cluster
         :return: the chain's group id
         """
-        self.group = async_chain(chain=self.chain.copy(), group=self.group, cached=self.cached, sync=self.sync,
+        self.group = async_chain(chain=self.chain[:], group=self.group, cached=self.cached, sync=self.sync,
                                  broker=self.broker)
         self.started = True
         return self.group

From eb72eb33d9edba6fe0ccda2d6313f6277b8fd67e Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 18 Oct 2015 13:53:32 +0200
Subject: [PATCH 4/8] Adds an `Iter` class

The Iter class serves as a convenience rwrapper around the `async_iter` function
---
 django_q/tasks.py             | 61 +++++++++++++++++++++++++++++++++++
 django_q/tests/test_cached.py | 18 +++++++++--
 2 files changed, 77 insertions(+), 2 deletions(-)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index b36a9ba..70d1539 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -427,6 +427,67 @@ def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=No
     return group
 
 
+class Iter(object):
+    """
+    An async task with iterable arguments
+    """
+
+    def __init__(self, func=None, args=None, kwargs=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None):
+        self.func = func
+        self.args = args or []
+        self.kwargs = kwargs or {}
+        self.id = ''
+        self.broker = broker or get_broker()
+        self.cached = cached
+        self.sync = sync
+        self.started = False
+
+    def append(self, *args):
+        """
+        add arguments to the set
+        """
+        self.args.append(args)
+        if self.started:
+            self.started = False
+
+    def run(self):
+        """
+        Start queueing the tasks to the worker cluster
+        :return: the task id
+        """
+        self.kwargs['cached'] = self.cached
+        self.kwargs['sync'] = self.sync
+        self.kwargs['broker'] = self.broker
+        self.id = async_iter(self.func, self.args, **self.kwargs)
+        self.started = True
+        return self.id
+
+    def result(self, wait=0):
+        """
+        return the full list of results.
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+        """
+        if self.started:
+            return result(self.id, wait=wait, cached=self.cached)
+
+    def fetch(self,  wait=0):
+        """
+        get the task result objects.
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+        """
+        if self.started:
+            return fetch(self.id, wait=wait, cached=self.cached)
+
+    def length(self):
+        """
+        get the length of the arguments list
+        :return int: length of the argument list
+        """
+        return len(self.args)
+
+
 class Chain(object):
     """
     A sequential chain of tasks
diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index 58bf11b..00bfe19 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -5,7 +5,7 @@ import pytest
 from django_q.cluster import pusher, worker, monitor
 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, Chain, async_chain
+    async_iter, Chain, async_chain, Iter
 from django_q.brokers import get_broker
 
 
@@ -100,7 +100,21 @@ 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
+    # test iter class
+    i = Iter('math.copysign', sync=True, cached=True)
+    i.append(1, -1)
+    i.append(2, -1)
+    i.append(3, -4)
+    i.append(5, 6)
+    assert i.started is False
+    assert i.length() == 4
+    assert i.run() is not None
+    assert len(i.result()) == 4
+    assert len(i.fetch().result) == 4
+    i.append(1, -7)
+    assert i.result() is None
+    i.run()
+    assert len(i.result()) == 5
 
 
 @pytest.mark.django_db

From 7eefb8cb64ef373f6d32b5c998f27b531f47dce9 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 18 Oct 2015 15:57:23 +0200
Subject: [PATCH 5/8] returns the size of the chain or iter on append

---
 django_q/tasks.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/django_q/tasks.py b/django_q/tasks.py
index 70d1539..d4e1d66 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -449,6 +449,7 @@ class Iter(object):
         self.args.append(args)
         if self.started:
             self.started = False
+        return self.length()
 
     def run(self):
         """
@@ -511,6 +512,7 @@ class Chain(object):
         if self.started:
             delete_group(self.group)
             self.started = False
+        return self.length()
 
     def run(self):
         """

From 37f1c4b42f0801f5bcf8d6efa237efc82d065e39 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 18 Oct 2015 15:57:54 +0200
Subject: [PATCH 6/8] docs: adds reference for `Iter` `Chain` and `async_chain`

---
 docs/tasks.rst | 122 ++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 120 insertions(+), 2 deletions(-)

diff --git a/docs/tasks.rst b/docs/tasks.rst
index eead685..1378f87 100644
--- a/docs/tasks.rst
+++ b/docs/tasks.rst
@@ -323,7 +323,7 @@ Reference
 .. py:function:: async_iter(func, args_iter,**kwargs)
 
    Runs iterable arguments against the cache backend and returns a single collated result.
-   Accepts the same options as :func:`async` except ``hook``.
+   Accepts the same options as :func:`async` except ``hook``. See also the :class:`Iter` class.
 
    :param object func: The task function to execute
    :param args: An iterable containing arguments for the task function
@@ -331,6 +331,17 @@ Reference
    :returns: The uuid of the task
    :rtype: str
 
+
+.. py:function:: async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None)
+
+    Async a chain of tasks. See also the :class:`Chain` class.
+
+    :param list chain: a list of tasks in the format [(func,(args),{kwargs}), (func,(args),{kwargs})]
+    :param str group: an optional group name.
+    :param bool cached: run this against the cache backend
+    :param bool sync: execute this inline instead of asynchronous
+
+
 .. py:function:: queue_size()
 
     Returns the size of the broker queue.
@@ -499,4 +510,111 @@ Reference
 
 .. py:class:: Failure
 
-     A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``False``.
\ No newline at end of file
+     A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``False``.
+
+
+.. py:class:: Iter(func=None, args=None, kwargs=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None)
+
+    An async task with iterable arguments. Serves as a convenient wrapper for :func:`async_iter`
+    You can pass the iterable arguments at construction or you can append individual argument tuples.
+
+        :param func: the function to execute
+        :param args: an iterable of arguments.
+        :param kwargs: the keyword arguments
+        :param bool cached: run this against the cache backend
+        :param bool sync: execute this inline instead of asynchronous
+        :param broker: optional broker instance
+
+
+    .. py:method:: append(*args)
+
+    Append arguments to the iter set. Returns the current set count.
+
+        :param args: the arguments for a single execution
+        :return: the current set count
+        :rtype: int
+
+
+    .. py:method:: run()
+
+    Start queueing the tasks to the worker cluster.
+
+        :return: the task result id
+
+
+    .. py:method:: result(wait=0)
+
+    return the full list of results.
+
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+
+
+    .. py:method:: fetch(wait=0)
+
+    get the task result objects.
+
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+
+
+    .. py:method:: length()
+
+    get the length of the arguments list
+
+        :return int: length of the argument list
+
+
+.. py:class:: Chain(chain=None, group=None, cached=Conf.CACHED, sync=Conf.SYNC)
+
+    A sequential chain of tasks. Acts as a convenient wrapper for :func:`async_chain`
+    You can pass the task chain at construction or you can append individual tasks before running them.
+
+        :param list chain: a list of task in the format [(func,(args),{kwargs}), (func,(args),{kwargs})]
+        :param str group: an optional group name.
+        :param bool cached: run this against the cache backend
+        :param bool sync: execute this inline instead of asynchronous
+
+
+    .. py:method:: append(func, *args, **kwargs)
+
+    Append a task to the chain. Takes the same arguments as :func:`async`
+
+        :return: the current number of tasks in the chain
+        :rtype: int
+
+
+    .. py:method:: run()
+
+    Start queueing the chain to the worker cluster.
+
+        :return: the chains group id
+
+
+    .. py:method:: result(wait=0)
+
+    return the full list of results from the chain when it finishes. Blocks until timeout or result.
+
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of results
+
+
+    .. py:method:: fetch(failures=True, wait=0)
+
+    get the task result objects from the chain when it finishes. Blocks until timeout or result.
+
+        :param failures: include failed tasks
+        :param int wait: how many milliseconds to wait for a result
+        :return: an unsorted list of task objects
+
+    .. py:method:: current()
+
+    get the index of the currently executing chain element
+
+        :return int: current chain index
+
+    .. py:method:: length()
+
+    get the length of the chain
+
+        :return int: length of the chain
\ No newline at end of file

From 79d9d99d2a090aee319b874e588f2db4f1c8ca26 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Sun, 18 Oct 2015 23:13:03 +0200
Subject: [PATCH 7/8] docs: moves architecture to its own main section

Also updated the architecture text a little
---
 docs/architecture.rst | 91 +++++++++++++++++++++++++++++++++++++++++++
 docs/cluster.rst      | 81 --------------------------------------
 docs/index.rst        |  3 +-
 3 files changed, 93 insertions(+), 82 deletions(-)
 create mode 100644 docs/architecture.rst

diff --git a/docs/architecture.rst b/docs/architecture.rst
new file mode 100644
index 0000000..5fab41a
--- /dev/null
+++ b/docs/architecture.rst
@@ -0,0 +1,91 @@
+Architecture
+------------
+
+.. image:: _static/cluster.png
+   :alt: Django Q schema
+
+
+Signed Tasks
+""""""""""""
+
+Tasks are first pickled and then signed using Django's own :mod:`django.core.signing` module using the ``SECRET_KEY`` and cluster name as salt, before being sent to a message broker. This ensures that task
+packages on the broker can only be executed and read by clusters
+and django servers who share the same secret key and cluster name.
+If a package fails to unpack, it will be marked failed with the broker and discarded.
+Optionally the packages can be compressed before transport.
+
+Broker
+""""""
+
+The broker collects task packages from the django instances and queues them for pick up by a cluster.
+If the broker supports message receipts, it will keep a copy of the tasks around until a cluster acknowledges the processing of the task.
+Otherwise it is put back in the queue after a timeout period. This ensure at-least-once delivery.
+Note that even if the task errors when processed by the cluster, this is considered a successful delivery.
+Most failed deliveries will be the result of a worker or the cluster crashing before the task was saved.
+
+Pusher
+""""""
+
+The pusher process continuously checks the broker for new task
+packages. It checks the signing and unpacks the task to the internal Task Queue.
+The amount of tasks in the Task Queue can be configured to control memory usage and minimize data loss in case of a failure.
+
+Worker
+""""""
+
+A worker process pulls a task of the Task Queue and it sets a shared countdown timer with :ref:`sentinel` indicating it is about to start work.
+The worker then tries to execute the task and afterwards the timer is reset and any results (including errors) are saved to the package.
+Irrespective of the failure or success of any of these steps, the package is then pushed onto the Result Queue.
+
+Monitor
+"""""""
+
+The result monitor checks the Result Queue for processed packages and
+saves both failed and successful packages to the Django database or cache backend.
+If the broker supports it, a delivery receipt is sent.
+In case the task was part of a chain, the next task is queued.
+
+.. _sentinel:
+
+Sentinel
+""""""""
+
+The sentinel spawns all process and then checks the health of all
+workers, including the pusher and the monitor. This includes checking timers on each worker for timeouts.
+In case of a sudden death or timeout, it will reincarnate the failing processes. When a stop signal is received, the sentinel will halt the
+pusher and instruct the workers and monitor to finish the remaining items. See :ref:`stop_procedure`
+
+Timeouts
+""""""""
+Before each task execution the worker sets a countdown timer on the sentinel and resets it again after execution.
+Meanwhile the sentinel checks if the timers don't reach zero, in which case it will terminate the worker and reincarnate a new one.
+
+Scheduler
+"""""""""
+Twice a minute the scheduler checks for any scheduled tasks that should be starting.
+
+- Creates a task from the schedule
+- Subtracts 1 from :attr:`django_q.Schedule.repeats`
+- Sets the next run time if there are repeats left or if it has a negative value.
+
+.. _stop_procedure:
+
+Stop procedure
+""""""""""""""
+
+When a stop signal is received, the sentinel exits the guard loop and instructs the pusher to stop pushing.
+Once this is confirmed, the sentinel pushes poison pills onto the task queue and will wait for all the workers to exit.
+This ensures that the task queue is emptied before the workers exit.
+Afterwards the sentinel waits for the monitor to empty the result queue and the stop procedure is complete.
+
+- Send stop event to pusher
+- Wait for pusher to exit
+- Put poison pills in the Task Queue
+- Wait for all the workers to clear the queue and stop
+- Put a poison pill on the Result Queue
+- Wait for monitor to process remaining results and exit
+- Signal that we have stopped
+
+.. warning::
+    If you force the cluster to terminate before the stop procedure has completed, you can lose tasks or results still being held in memory.
+    You can manage the amount of tasks in a clusters memory by setting the :ref:`queue_limit`.
diff --git a/docs/cluster.rst b/docs/cluster.rst
index 278521a..e25a062 100644
--- a/docs/cluster.rst
+++ b/docs/cluster.rst
@@ -82,87 +82,6 @@ An example :file:`circus.ini` ::
 Note that we only start one process. It is not a good idea to run multiple instances of the cluster in the same environment since this does nothing to increase performance and in all likelihood will diminish it.
 Control your cluster using the ``workers``, ``recycle`` and ``timeout`` settings in your :doc:`configure`
 
-Architecture
-------------
-
-.. image:: _static/cluster.png
-   :alt: Django Q schema
-
-
-Signed Tasks
-""""""""""""
-
-Tasks are first pickled and then signed using Django's own :mod:`django.core.signing` module using the ``SECRET_KEY`` and cluster name as salt, before being sent to a message broker. This ensures that task
-packages on the broker can only be executed and read by clusters
-and django servers who share the same secret key and cluster name.
-If a package fails to unpack, it will be marked failed with the broker and discarded.
-Optionally the packages can be compressed before transport.
-
-Pusher
-""""""
-
-The pusher process continuously checks the broker for new task
-packages. It checks the signing and unpacks the task to the Task Queue.
-
-Worker
-""""""
-
-A worker process pulls a task of the Task Queue and it sets a shared countdown timer with :ref:`sentinel` indicating it is about to start work.
-The worker then tries to execute the task and afterwards the timer is reset and any results (including errors) are saved to the package.
-Irrespective of the failure or success of any of these steps, the package is then pushed onto the Result Queue.
-
-Monitor
-"""""""
-
-The result monitor checks the Result Queue for processed packages and
-saves both failed and successful packages to the Django database.
-If the broker supports it, a delivery receipt is sent.
-
-.. _sentinel:
-
-Sentinel
-""""""""
-
-The sentinel spawns all process and then checks the health of all
-workers, including the pusher and the monitor. This includes checking timers on each worker for timeouts.
-In case of a sudden death or timeout, it will reincarnate the failing processes. When a stop signal is received, the sentinel will halt the
-pusher and instruct the workers and monitor to finish the remaining items. See :ref:`stop_procedure`
-
-Timeouts
-""""""""
-Before each task execution the worker sets a countdown timer on the sentinel and resets it again after execution.
-Meanwhile the sentinel checks if the timers don't reach zero, in which case it will terminate the worker and reincarnate a new one.
-
-Scheduler
-"""""""""
-Twice a minute the scheduler checks for any scheduled tasks that should be starting.
-
-- Creates a task from the schedule
-- Subtracts 1 from :attr:`django_q.Schedule.repeats`
-- Sets the next run time if there are repeats left or if it has a negative value.
-
-.. _stop_procedure:
-
-Stop procedure
-""""""""""""""
-
-When a stop signal is received, the sentinel exits the guard loop and instructs the pusher to stop pushing.
-Once this is confirmed, the sentinel pushes poison pills onto the task queue and will wait for all the workers to exit.
-This ensures that the task queue is emptied before the workers exit.
-Afterwards the sentinel waits for the monitor to empty the result queue and the stop procedure is complete.
-
-- Send stop event to pusher
-- Wait for pusher to exit
-- Put poison pills in the Task Queue
-- Wait for all the workers to clear the queue and stop
-- Put a poison pill on the Result Queue
-- Wait for monitor to process remaining results and exit
-- Signal that we have stopped
-
-.. warning::
-    If you force the cluster to terminate before the stop procedure has completed, you can lose tasks or results still being held in memory.
-    You can manage the amount of tasks in a clusters memory by setting the :ref:`queue_limit`.
-
 Reference
 ---------
 
diff --git a/docs/index.rst b/docs/index.rst
index 761b8a6..481c874 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -11,7 +11,7 @@ Django Q is a native Django task queue, scheduler and worker application using P
 Features
 --------
 
--  Multiprocessing worker pool
+-  Multiprocessing worker poole7u-CG
 -  Asynchronous tasks
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
@@ -39,6 +39,7 @@ Contents:
     Cluster <cluster>
     Monitor <monitor>
     Admin <admin>
+    Architecture <architecture>
     Examples <examples>
 
 * :ref:`genindex`

From 37b41b15b1d82083435708131d43777a4bf49be2 Mon Sep 17 00:00:00 2001
From: Ilan Steemers <koed00@gmail.com>
Date: Mon, 19 Oct 2015 11:32:22 +0200
Subject: [PATCH 8/8] docs: adds `Iter` and `Chain` examples

---
 docs/tasks.rst | 67 ++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 65 insertions(+), 2 deletions(-)

diff --git a/docs/tasks.rst b/docs/tasks.rst
index 1378f87..6d337ef 100644
--- a/docs/tasks.rst
+++ b/docs/tasks.rst
@@ -91,8 +91,8 @@ Please not that this will override any other option keywords.
 
 
 
-Async Iterable
---------------
+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
@@ -109,6 +109,30 @@ If you have an iterable object with arguments for a function, you can use :func:
 
 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.
+
+You can also use an :class:`Iter` instance which can sometimes be more convenient:
+
+.. code-block:: python
+
+    from django_q.tasks import Iter
+
+    i = Iter('math.copysign')
+
+    # add some arguments
+    i.append(1, -1)
+    i.append(2, -1)
+    i.append(3, -1)
+
+    # run it
+    i.run()
+
+    # get the results
+    print(i.result())
+
+.. code-block:: python
+
+    [-1.0, -2.0, -3.0]
+
 Needs the Django cache framework.
 
 .. _groups:
@@ -179,6 +203,45 @@ You can also access group functions from a task result instance:
         task.group_delete()
         print('Deleted group {}'.format(task.group))
 
+Chains
+------
+Sometimes you want to run tasks sequentially. For that you can use the :func:`async_chain` function:
+
+.. code-block:: python
+
+    # Async a chain of tasks
+    from django_q.tasks import async_chain, result_group
+
+    # the chain must be in the format
+    # [(func,(args),{kwargs}),(func,(args),{kwargs}),..]
+    group_id = async_chain([('math.copysign', (1, -1)),
+                            ('math.floor', (1,))])
+
+    # get group result
+    result_group(group_id, count=2)
+
+A slightly more convenient way is to use a :class:`Chain` instance:
+
+.. code-block:: python
+
+    # Chain async
+    from django_q.tasks import Chain
+
+    # create a chain that uses the cache backend
+    chain = Chain(cached=True)
+
+    # add some tasks
+    chain.append('math.copysign', 1, -1)
+    chain.append('math.floor', 1)
+
+    # run it
+    chain.run()
+
+    print(chain.result())
+.. code-block:: python
+
+    [-1.0, 1]
+
 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.