Merge pull request #32 from Koed00/dev

Adds queue limit and encryption salt

README.rst

@@ -57,7 +57,7 @@ Installation
 -  Make sure you have a `Redis <http://redis.io/>`__ server running
    somewhere
 
-Read the more complete documentation at `http://django-q.readthedocs.org <http://django-q.readthedocs.org>`__
+Read the full documentation at `http://django-q.readthedocs.org <http://django-q.readthedocs.org>`__
 
 
 Configuration
@@ -67,7 +67,7 @@ All configuration settings are optional. e.g:
 
 .. code:: python
 
-    # settings.py
+    # settings.py example
     Q_CLUSTER = {
         'name': 'myproject',
         'workers': 8,
@@ -75,6 +75,8 @@ All configuration settings are optional. e.g:
         'timeout': 60,
         'compress': True,
         'save_limit': 250,
+        'queue_limit: 500,
+        'cpu_affinity': 1,
         'label': 'Django Q',
         'redis': {
             'host': '127.0.0.1',

django_q/cluster.py

@@ -132,7 +132,7 @@ class Sentinel(object):
         self.pool_size = Conf.WORKERS
         self.pool = []
         self.timeout = timeout
-        self.task_queue = Queue()
+        self.task_queue = Queue(maxsize=Conf.QUEUE_LIMIT) if Conf.QUEUE_LIMIT else Queue()
         self.result_queue = Queue()
         self.event_out = Event()
         self.monitor = Process()

django_q/conf.py

@@ -33,6 +33,9 @@ class Conf(object):
     # Failures are always saved
     SAVE_LIMIT = conf.get('save_limit', 250)
 
+    # Maximum number of tasks that each cluster can work on
+    QUEUE_LIMIT = conf.get('queue_limit', None)
+
     # Number of workers in the pool. Default is cpu count. +2 for monitor and pusher
     WORKERS = conf.get('workers', cpu_count())
 

django_q/signing.py

@@ -19,7 +19,7 @@ class SignedPackage(object):
     def dumps(obj, compressed=Conf.COMPRESSED):
         return signing.dumps(obj,
                              key=Conf.SECRET_KEY,
-                             salt='django_q.q',
+                             salt=Conf.PREFIX,
                              compress=compressed,
                              serializer=PickleSerializer)
 
@@ -27,7 +27,7 @@ class SignedPackage(object):
     def loads(obj):
         return signing.loads(obj,
                              key=Conf.SECRET_KEY,
-                             salt='django_q.q',
+                             salt=Conf.PREFIX,
                              serializer=PickleSerializer)
 
 

docs/cluster.rst

@@ -48,9 +48,9 @@ Multiple Clusters
 -----------------
 You can have multiple clusters on multiple machines, working on the same queue as long as:
 
-- They connect to the same Redis server.
+- They connect to the same Redis server or Redis cluster.
 - They use the same cluster name. See :ref:`configuration`
-- They share the same ``SECRET_KEY``
+- They share the same ``SECRET_KEY`` for Django.
 
 Using a Procfile
 ----------------
@@ -61,7 +61,7 @@ If you host on `Heroku <https://heroku.com>`__ or you are using `Honcho <https:/
 Process managers
 ----------------
 While you certainly can run a Django Q with a process manager like `Supervisor <http://supervisord.org/>`__ or `Circus <https://circus.readthedocs.org/en/latest/>`__ it is not strictly necessary.
-The cluster has an internal sentinel that checks the health of all the processes and recycles or reincarnates according to your settings.
+The cluster has an internal sentinel that checks the health of all the processes and recycles or reincarnates according to your settings or in case of unexpected crashes.
 Because of the multiprocessing daemonic nature of the cluster, it is impossible for a process manager to determine the clusters health and resource usage.
 
 An example :file:`circus.ini` ::
@@ -92,10 +92,9 @@ Architecture
 Signed Tasks
 """"""""""""
 
-
-Tasks are first pickled and then signed using Django's own :mod:`django.core.signing` module before being sent to a Redis list. This ensures that task
+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 Redis list. This ensures that task
 packages on the Redis server can only be executed and read by clusters
-and django servers who share the same secret key.
+and django servers who share the same secret key and cluster name.
 Optionally the packages can be compressed before transport
 
 Pusher
@@ -109,7 +108,7 @@ Worker
 
 A worker process pulls a package of the Task Queue and checks the signing and unpacks the task.
 Before executing the task it set a timer on the :ref:`sentinel` indicating its about to start work.
-Afterwards it the timer is reset and any results (including errors) are saved to the pacjage.
+Afterwards it 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.
 
 
@@ -161,7 +160,8 @@ Afterwards the sentinel waits for the monitor to empty the result and then the s
 - Signal that we have stopped
 
 .. warning::
-    If you force the cluster to terminate before the stop procedure has completed, you can lose tasks and their results.
+    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
 ---------
@@ -206,7 +206,7 @@ Reference
 
     .. py:attribute:: is_starting
 
-    Bool. Indicating if the cluster is busy starting up
+    Bool. Indicating that the cluster is busy starting up
 
     .. py:attribute:: is_running
 
@@ -218,7 +218,7 @@ Reference
 
     .. py:attribute:: has_stopped
 
-    Bool. Tells you if the cluster finished the stop procedure
+    Bool. Tells you if the cluster has finished the stop procedure
 
 
 

docs/install.rst

@@ -37,6 +37,7 @@ Configuration is handled via the ``Q_CLUSTER`` dictionary in your :file:`setting
         'timeout': 60,
         'compress': True,
         'save_limit': 250,
+        'queue_limit: 500,
         'cpu_affinity': 1,
         'label': 'Django Q',
         'redis': {
@@ -46,7 +47,6 @@ Configuration is handled via the ``Q_CLUSTER`` dictionary in your :file:`setting
     }
 
 
-
 name
 ~~~~
 
@@ -91,6 +91,16 @@ Limits the amount of successful tasks saved to Django.
  - Defaults to ``250``
  - Failures are always saved.
 
+.. _queue_limit:
+
+queue_limit
+~~~~~~~~~~~
+
+This does not limit the amount of tasks that can be queued overall on Redis, but rather how many tasks are kept in memory by a single cluster.
+Setting this to a reasonable number, can help balance the workload and the memory overhead of each individual cluster.
+It can also be used to manage the loss of data in case of a cluster failure.
+Defaults to ``None``, meaning no limit.
+
 label
 ~~~~~
 

docs/monitor.rst

@@ -46,6 +46,7 @@ TQ
 **Task Queue** counts the number of tasks in the queue
 
 If this keeps rising it means you are taking on more tasks than your cluster can handle.
+You can limit this by settings the :ref:`queue_limit` in your cluster configuration.
 
 RQ
 ~~
@@ -58,7 +59,7 @@ It's normal for the result queue to take slightly longer to clear than the task
 RC
 ~~
 
-**Reincarnations** shows the amount of processes that have been reincarnated after a sudden death or timeout.
+**Reincarnations** shows the amount of processes that have been reincarnated after a recycle, sudden death or timeout.
 If this number is unusually high, you are either suffering from repeated task errors or severe timeouts and you should check your logs for details.
 
 Up
@@ -124,7 +125,7 @@ Reference
 
     .. py:attribute:: pusher
 
-    The pid of the pushes process
+    The pid of the pusher process
 
     .. py:attribute:: monitor