diff --git a/MANIFEST.in b/MANIFEST.in
index 8b9a9b5..7a75164 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -2,5 +2,4 @@ include LICENSE
 include README.rst
 include django_q/management/*.py
 include django_q/management/commands/*.py
-exclude django_q/management/commands/qtest.py
 include django_q/migrations/*.py
\ No newline at end of file
diff --git a/django_q/__init__.py b/django_q/__init__.py
index c6305e3..9d412e1 100644
--- a/django_q/__init__.py
+++ b/django_q/__init__.py
@@ -7,6 +7,7 @@ sys.path.insert(0, myPath)
 from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group
 from .models import Task, Schedule, Success, Failure
 from .cluster import Cluster
+from .monitor import Stat
 
 VERSION = (0, 4, 0)
 
diff --git a/django_q/cluster.py b/django_q/cluster.py
index 6ad88c2..c474c43 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -394,12 +394,11 @@ def save_task(task):
     Saves the task package to Django
     """
     # SAVE LIMIT < 0 : Don't save success
-    if Conf.SAVE_LIMIT < 0 and task['success']:
+    if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
         return
     # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
     if task['success'] and 0 < Conf.SAVE_LIMIT < Success.objects.count():
         Success.objects.last().delete()
-
     try:
         Task.objects.create(id=task['id'],
                             name=task['name'],
@@ -435,8 +434,9 @@ def scheduler(list_key=Conf.Q_LIST):
             # single value won't eval to tuple, so:
             if type(args) != tuple:
                 args = (args,)
+        q_options = kwargs.get('q_options', {})
         if s.hook:
-            kwargs['hook'] = s.hook
+            q_options['hook'] = s.hook
         # set up the next run time
         if not s.schedule_type == s.ONCE:
             next_run = arrow.get(s.next_run)
@@ -455,8 +455,9 @@ def scheduler(list_key=Conf.Q_LIST):
             s.next_run = next_run.datetime
             s.repeats += -1
         # send it to the cluster
-        kwargs['list_key'] = list_key
-        kwargs['group'] = s.name or s.id
+        q_options['list_key'] = list_key
+        q_options['group'] = s.name or s.id
+        kwargs['q_options'] = q_options
         s.task = tasks.async(s.func, *args, **kwargs)
         # log it
         if not s.task:
diff --git a/django_q/tasks.py b/django_q/tasks.py
index 74dd52b..32cee05 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -20,16 +20,14 @@ def async(func, *args, **kwargs):
     """
     Sends a task to the cluster
     """
-    # optional hook
-    hook = kwargs.pop('hook', None)
-    # optional list_key
-    list_key = kwargs.pop('list_key', Conf.Q_LIST)
-    # optional redis connection
-    redis = kwargs.pop('redis', redis_client)
-    # optional sync mode
-    sync = kwargs.pop('sync', False)
-    # optional group
-    group = kwargs.pop('group', None)
+    # get options from q_options dict or direct from kwargs
+    options = kwargs.pop('q_options', kwargs)
+    hook = options.pop('hook', None)
+    list_key = options.pop('list_key', Conf.Q_LIST)
+    redis = options.pop('redis', redis_client)
+    sync = options.pop('sync', False)
+    group = options.pop('group', None)
+    save = options.pop('save', None)
     # get an id
     tag = uuid()
     # build the task package
@@ -40,6 +38,8 @@ def async(func, *args, **kwargs):
         task['hook'] = hook
     if group:
         task['group'] = group
+    if save is not None:
+        task['save'] = save
     # sign it
     pack = signing.SignedPackage.dumps(task)
     if sync:
diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py
index 3c7a858..32a5291 100644
--- a/django_q/tests/test_cluster.py
+++ b/django_q/tests/test_cluster.py
@@ -124,6 +124,9 @@ def test_async(r, admin_user):
     h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', list_key=list_key, redis=r)
     # args unpickle test
     j = async('django_q.tests.tasks.get_user_id', admin_user, list_key=list_key, group='test_j', redis=r)
+    # q_options and save opt_out test
+    k = async('django_q.tests.tasks.get_user_id', admin_user,
+              q_options={'list_key': list_key, 'group': 'test_k', 'redis': r, 'save': False, 'timeout': 90})
     # check if everything has a task id
     assert isinstance(a, str)
     assert isinstance(b, str)
@@ -134,8 +137,9 @@ def test_async(r, admin_user):
     assert isinstance(g, str)
     assert isinstance(h, str)
     assert isinstance(j, str)
+    assert isinstance(k, str)
     # run the cluster to execute the tasks
-    task_count = 9
+    task_count = 10
     assert r.llen(list_key) == task_count
     task_queue = Queue()
     stop_event = Event()
@@ -210,7 +214,8 @@ def test_async(r, admin_user):
     assert count_group('test_j', failures=True) == 0
     assert delete_group('test_j') == 1
     assert delete_group('test_j', tasks=True) is None
-
+    # task k should not have been saved
+    assert fetch(k) is None
     r.delete(list_key)
 
 
diff --git a/docs/admin.rst b/docs/admin.rst
index b4c57a4..821cd28 100644
--- a/docs/admin.rst
+++ b/docs/admin.rst
@@ -41,14 +41,14 @@ Repeats
 If you want a schedule to only run a finite amount of times, e.g. every hour for the next 24 hours, you can do that using the :attr:`Schedule.repeats` attribute.
 In this case you would set the schedule type to :attr:`Schedule.HOURLY` and the repeats to `24`. Every time the schedule runs the repeats count down until it hits zero and schedule is no longer run.
 
-When you set repeats to `-1` the schedule will continue indefinitely and the repeats will still count down. This can be used as an indicator of how many times the schedule has been executed.
+When you set repeats to ``-1`` the schedule will continue indefinitely and the repeats will still count down. This can be used as an indicator of how many times the schedule has been executed.
 
 An exception to this are schedules of type :attr:`Schedule.ONCE`. Negative repeats for this schedule type will cause it to be deleted from the database.
 This behavior is useful if you have many delayed actions which you do not necessarily need a result for. A positive number will keep the ONCE schedule, but it will not run again.
 
 .. note::
 
-    To run a `Once` schedule again, change the repeats to something other than `0`. Set a new run time before you do this or let it execute immediately.
+    To run a ``ONCE`` schedule again, change the repeats to something other than `0`. Set a new run time before you do this or let it execute immediately.
 
 
 Next run
diff --git a/docs/cluster.rst b/docs/cluster.rst
index 43d257d..8e1c43f 100644
--- a/docs/cluster.rst
+++ b/docs/cluster.rst
@@ -4,7 +4,7 @@ Cluster
 .. py:currentmodule:: django_q
 
 Django Q uses Python's multiprocessing module to manage a pool of workers that will handle your tasks.
-Start your cluster using Django's `manage.py` command::
+Start your cluster using Django's ``manage.py`` command::
 
     $ python manage.py qcluster
 
@@ -26,7 +26,7 @@ You should see the cluster starting ::
     10:57:40 [Q] INFO Q Cluster-31781 running.
 
 
-Stopping the cluster with ctrl-c or either the `SIGTERM` and `SIGKILL` signals, will initiate the :ref:`stop_procedure`::
+Stopping the cluster with ctrl-c or either the ``SIGTERM`` and ``SIGKILL`` signals, will initiate the :ref:`stop_procedure`::
 
     16:44:12 [Q] INFO Q Cluster-31781 stopping.
     16:44:12 [Q] INFO Process-1 stopping cluster processes
@@ -50,7 +50,7 @@ You can have multiple clusters on multiple machines, working on the same queue a
 
 - They connect to the same Redis server.
 - They use the same cluster name. See :ref:`configuration`
-- They share the same `SECRET_KEY`
+- They share the same ``SECRET_KEY``
 
 Using a Procfile
 ----------------
@@ -80,7 +80,7 @@ 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 :ref:`configuration`
+Control your cluster using the ``workers``, ``recycle`` and ``timeout`` settings in your :ref:`configuration`
 
 Architecture
 ------------
diff --git a/docs/conf.py b/docs/conf.py
index 211909e..6a4dcc3 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -72,7 +72,7 @@ author = 'Ilan Steemers'
 # The short X.Y version.
 version = '0.4'
 # The full version, including alpha/beta/rc tags.
-release = '0.4.0'
+release = '0.4.1'
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.
diff --git a/docs/index.rst b/docs/index.rst
index a4241e4..6cc3d28 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -16,7 +16,7 @@ Features
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
 -  Failure and success database
--  Result hooks
+-  Result hooks and groups
 -  Django Admin integration
 -  PaaS compatible with multiple instances
 -  Multi cluster monitor
diff --git a/docs/install.rst b/docs/install.rst
index 747e373..3ad40a3 100644
--- a/docs/install.rst
+++ b/docs/install.rst
@@ -6,7 +6,7 @@ Installation
     $ pip install django-q
 
 
--  Add :mod:`django_q` to `INSTALLED_APPS` in your projects :file:`settings.py`::
+-  Add :mod:`django_q` to ``INSTALLED_APPS`` in your projects :file:`settings.py`::
 
        INSTALLED_APPS = (
            # other apps
@@ -18,14 +18,14 @@ Installation
     $ python manage.py migrate
 
 -  Make sure you have a `Redis <http://redis.io/>`__ server running
-   somewhere
+   somewhere and know how to connect to it.
 
 .. _configuration:
 
 Configuration
 -------------
 
-Configuration is handled via the `Q_CLUSTER` dictionary in your :file:`settings.py`
+Configuration is handled via the ``Q_CLUSTER`` dictionary in your :file:`settings.py`
 
 .. code:: python
 
@@ -133,7 +133,7 @@ of the cache connection you want to use::
 
 
 .. tip::
-    Django Q uses your `SECRET_KEY` to encrypt task packages and prevent task crossover. So make sure you have it set up in your Django settings.
+    Django Q uses your ``SECRET_KEY`` to encrypt task packages and prevent task crossover. So make sure you have it set up in your Django settings.
 
 cpu_affinity
 ~~~~~~~~~~~~
@@ -175,7 +175,7 @@ As a rule of thumb; cpu_affinity 1 favors repetitive short running tasks, while
 
 .. note::
 
-    The `cpu_affinity` setting requires the optional :ref:`psutil <psutil>` module.
+    The ``cpu_affinity`` setting requires the optional :ref:`psutil <psutil>` module.
 
 Requirements
 ------------
@@ -185,7 +185,7 @@ Django Q is tested for Python 2.7 and 3.4
 -  `Django <https://www.djangoproject.com>`__
 
     Django Q aims to use as much of Django's standard offerings as possible
-    The code is tested against Django version `1.7.8` and `1.8.2`.
+    The code is tested against Django version `1.7.9` and `1.8.3`.
 
 -  `Django-picklefield <https://github.com/gintas/django-picklefield>`__
 
@@ -207,6 +207,7 @@ Django Q is tested for Python 2.7 and 3.4
 
     Django Q uses Redis as a centralized hub between your Django instances and your Q clusters.
 
+
 Optional
 ~~~~~~~~
 .. _psutil:
diff --git a/docs/schedules.rst b/docs/schedules.rst
index c726e0d..e5d36e6 100644
--- a/docs/schedules.rst
+++ b/docs/schedules.rst
@@ -25,6 +25,13 @@ You can manage them through the :ref:`admin_page` or directly from your code wit
                             schedule_type=Schedule.DAILY
                             )
 
+    # In case you want to use async options
+    schedule('math.sqrt',
+             9,
+             hook='hooks.print_result',
+             q_options={'timeout': 30},
+             schedule_type=Schedule.HOURLY)
+
 
 Management Commands
 -------------------
@@ -59,6 +66,7 @@ Reference
     :param str schedule_type: (O)nce, (H)ourly, (D)aily, (W)eekly, (M)onthly, (Q)uarterly, (Y)early or :attr:`Schedule.TYPE`
     :param int repeats: Number of times to repeat schedule. -1=Always, 0=Never, n =n.
     :param datetime next_run: Next or first scheduled execution datetime.
+    :param dict q_options: async options to use for this schedule
     :param kwargs: optional keyword arguments for the scheduled function.
 
 .. class:: Schedule
diff --git a/docs/tasks.rst b/docs/tasks.rst
index dd18214..c18a4aa 100644
--- a/docs/tasks.rst
+++ b/docs/tasks.rst
@@ -2,6 +2,8 @@ Tasks
 =====
 .. py:currentmodule:: django_q
 
+.. _async:
+
 Async
 -----
 
@@ -31,11 +33,53 @@ Use :func:`async` from your code to quickly offload tasks to the :class:`Cluster
     def print_result(task):
         print(task.result)
 
+:func:`async` can take the following optional keyword arguments:
+
+hook
+""""
+The function to call after the task has been executed. This function gets passed the complete :class:`Task` object as its argument.
+
+group
+"""""
+A group label. Check :ref:`groups` for group functions.
+
+save
+""""
+Overrides the result backend's save setting for this task.
+
+timeout
+"""""""
+Overrides the cluster's timeout setting for this task.
+
+sync
+""""
+Simulates a task execution synchronously. Useful for testing.
+
+redis
+"""""
+A redis connection. In case you want to control your own connections.
+
+q_options
+"""""""""
+None of the option keywords get passed on to the task function.
+As an alternative you can also put them in
+a single keyword dict named ``q_options``. This enables you to use these keywords for your function call::
+
+    # Async options in a dict
+
+    opts = {'hook': 'hooks.print_result',
+            'group': 'math',
+            'timeout': 30}
+
+    async('math.modf', 2.5, q_options=opts)
+
+Please not that this will override any other option keywords.
+
 .. _groups:
 
 Groups
 ------
-You can group together results by passing :func:`async` the optional `group` keyword:
+You can group together results by passing :func:`async` the optional ``group`` keyword:
 
 .. code-block:: python
 
@@ -67,7 +111,7 @@ Instead of :func:`result_group` you can also use :func:`fetch_group` to return a
     # only use the successes
     results = fetch_group('modf')
     if failure_count:
-        results.exclude(success=False)
+        results = results.exclude(success=False)
     results =  [task.result for task in successes]
 
     # this is the same as
@@ -82,13 +126,13 @@ Getting results by using :func:`result_group` is of course much faster than usin
 
 .. note::
 
-   Although :func:`fetch_group` returns a queryset, due to the nature of the PickleField , calling `Queryset.values` on it will return a list of encoded results.
+   Although :func:`fetch_group` returns a queryset, due to the nature of the PickleField , calling ``Queryset.values`` on it will return a list of encoded results.
    Use list comprehension or an iterator instead.
 
 Synchronous testing
 -------------------
 
-:func:`async` can be instructed to execute a task immediately by setting the optional keyword `sync=True`.
+:func:`async` can be instructed to execute a task immediately by setting the optional keyword ``sync=True``.
 The task will then be injected straight into a worker and the result saved by a monitor instance::
 
     from django_q import async, fetch
@@ -113,7 +157,7 @@ Connection pooling
 ------------------
 
 Django Q tries to pass redis connections around its parts as much as possible to save you from running out of connections.
-When you are making individual calls to :func:`async` a lot though, it can help to set up a redis connection to pass to :func:`async`:
+When you are making individual calls to :func:`async` a lot though, it can help to set up a redis connection to reuse for :func:`async`:
 
 .. code:: python
 
@@ -133,7 +177,7 @@ Reference
 ---------
 
 .. py:function:: async(func, *args, hook=None, group=None, timeout=None,\
-    sync=False, redis=None, **kwargs)
+    sync=False, redis=None, q_options=None, **kwargs)
 
     Puts a task in the cluster queue
 
@@ -144,6 +188,7 @@ Reference
    :param int timeout: Overrides global cluster :ref:`timeout`.
    :param bool sync: If set to True, async will simulate a task execution
    :param redis: Optional redis connection
+   :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
@@ -172,7 +217,7 @@ Reference
     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 bool failures: set this to ``True`` to include failed results
     :returns: a list of results
     :rtype: list
 
@@ -181,7 +226,7 @@ Reference
     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
+    :param bool failures: set this to ``False`` to exclude failed tasks
     :returns: a list of Tasks
     :rtype: list
 
@@ -190,7 +235,7 @@ Reference
     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 failures: counts the number of failures if ``True``
     :returns: the number of tasks or failures in a group
     :rtype: int
 
@@ -199,7 +244,7 @@ Reference
     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 tasks: also deletes the associated tasks if ``True``
     :returns: the numbers of tasks affected
     :rtype: int
 
@@ -218,7 +263,7 @@ Reference
         .. note::
 
             This is for convenience and can be used as a parameter for most functions that take a `task_id`.
-            Keep in mind however that it is not guaranteed to be unique if you store very large amounts of tasks in the database.
+            Keep in mind that it is not guaranteed to be unique if you store very large amounts of tasks in the database.
 
     .. py:attribute:: func
 
@@ -269,7 +314,7 @@ Reference
     .. py:classmethod:: get_result_group(group_id, failures=False)
 
     Returns a list of results from a task group.
-    Set failures to `True` to include failed results.
+    Set failures to ``True`` to include failed results.
 
     .. py:classmethod:: get_task(task_id)
 
@@ -278,22 +323,22 @@ Reference
     .. py:classmethod:: get_task_group(group_id, failures=True)
 
     Gets a queryset of tasks with this group id.
-    Set failures to `False` to exclude failed tasks.
+    Set failures to ``False`` to exclude failed tasks.
 
     .. py:classmethod::  get_group_count(group_id, failures=False)
 
     Returns a count of the number of tasks results in a group.
-    Returns the number of failures when `failures=True`
+    Returns the number of failures when ``failures=True``
 
     .. py:classmethod:: delete_group(group_id, objects=False)
 
     Deletes a group label only, by default.
-    If `objects=True` it will also delete the tasks in this group from the database.
+    If ``objects=True`` it will also delete the tasks in this group from the database.
 
 .. py:class:: Success
 
-    A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is True.
+    A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``True``.
 
 .. 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``.
\ No newline at end of file
diff --git a/requirements.in b/requirements.in
new file mode 100644
index 0000000..c5ce15e
--- /dev/null
+++ b/requirements.in
@@ -0,0 +1,9 @@
+arrow
+blessed
+django-picklefield
+Django
+future
+hiredis
+redis
+psutil
+django-redis
diff --git a/requirements.txt b/requirements.txt
index f609487..4b23450 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,12 +1,19 @@
+#
+# This file is autogenerated by pip-compile
+# Make changes in requirements.in, then run this to update:
+#
+#    pip-compile requirements.in
+#
 arrow==0.6.0
 blessed==1.9.5
 django-picklefield==0.3.1
-Django>=1.7.8
+django-redis==4.2.0
+django==1.8.3
 future==0.14.3
 hiredis==0.2.0
-redis==2.10.3
+msgpack-python==0.4.6     # via django-redis
 psutil==3.1.1
-django-redis==4.2.0
-
-
-
+python-dateutil==2.4.2    # via arrow
+redis==2.10.3
+six==1.9.0                # via django-picklefield, python-dateutil
+wcwidth==0.1.4            # via blessed