From ac3d0d21614bcb64654588f94136928d165061d5 Mon Sep 17 00:00:00 2001 From: Ilan Date: Fri, 24 Jul 2015 00:26:40 +0200 Subject: [PATCH 1/6] use PREFIX as salt --- django_q/signing.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/django_q/signing.py b/django_q/signing.py index 19e6341..69a0ae6 100644 --- a/django_q/signing.py +++ b/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) From 805d160bbeac1ae64c91cf7c30d8e134f3785e66 Mon Sep 17 00:00:00 2001 From: Ilan Date: Fri, 24 Jul 2015 00:38:51 +0200 Subject: [PATCH 2/6] use PREFIX as salt Updated docs for this --- docs/cluster.rst | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/docs/cluster.rst b/docs/cluster.rst index 7d99db3..09798ae 100644 --- a/docs/cluster.rst +++ b/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 `__ or you are using `Honcho `__ or `Circus `__ 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,7 @@ 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 the queues. Reference --------- @@ -206,7 +205,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 +217,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 From d31e9e2b970ff4dfcdf7802031a1d1d0a5b5b5dc Mon Sep 17 00:00:00 2001 From: Ilan Date: Fri, 24 Jul 2015 01:04:58 +0200 Subject: [PATCH 3/6] adds a limit to the queue --- django_q/cluster.py | 4 ++-- django_q/conf.py | 3 +++ 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 4dad340..59e57b1 100644 --- a/django_q/cluster.py +++ b/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() @@ -314,7 +314,7 @@ def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client): sleep(10) break if task: - task_queue.put(task[1]) + task_queue.put(task[1], block=True) logger.debug(_('queueing from {}').format(list_key)) if event.is_set(): break diff --git a/django_q/conf.py b/django_q/conf.py index 407d17a..a2b3517 100644 --- a/django_q/conf.py +++ b/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()) From cf325f819947c6a999db1e19518f0a37dba8e5df Mon Sep 17 00:00:00 2001 From: Ilan Date: Fri, 24 Jul 2015 10:52:30 +0200 Subject: [PATCH 4/6] queue block is true by default. --- django_q/cluster.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 59e57b1..00e1929 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -314,7 +314,7 @@ def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client): sleep(10) break if task: - task_queue.put(task[1], block=True) + task_queue.put(task[1]) logger.debug(_('queueing from {}').format(list_key)) if event.is_set(): break From 0e7a5c4afb612c844ea762d140bc5fca28dcb958 Mon Sep 17 00:00:00 2001 From: Ilan Date: Fri, 24 Jul 2015 11:13:58 +0200 Subject: [PATCH 5/6] docs: added `queue_limit` information --- docs/install.rst | 10 ++++++++++ docs/monitor.rst | 7 ++++--- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/docs/install.rst b/docs/install.rst index 3ad40a3..1bd5e5c 100644 --- a/docs/install.rst +++ b/docs/install.rst @@ -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 ~~~~~ diff --git a/docs/monitor.rst b/docs/monitor.rst index b657f34..0b40c02 100644 --- a/docs/monitor.rst +++ b/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,9 +125,9 @@ Reference .. py:attribute:: pusher - The pid of the pushes process + The pid of the pusher process - .. py:attribute:: monitor + .. py:attribute:: monitormight The pid of the monitor process From a44fc6668255f9ba86847a1534acf447106538d1 Mon Sep 17 00:00:00 2001 From: Ilan Date: Fri, 24 Jul 2015 11:28:52 +0200 Subject: [PATCH 6/6] docs: added `queue_limit` information --- README.rst | 6 ++++-- docs/cluster.rst | 3 ++- docs/install.rst | 2 +- docs/monitor.rst | 2 +- 4 files changed, 8 insertions(+), 5 deletions(-) diff --git a/README.rst b/README.rst index cb6cca3..0b9cd71 100644 --- a/README.rst +++ b/README.rst @@ -57,7 +57,7 @@ Installation - Make sure you have a `Redis `__ server running somewhere -Read the more complete documentation at `http://django-q.readthedocs.org `__ +Read the full documentation at `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', diff --git a/docs/cluster.rst b/docs/cluster.rst index 09798ae..c752849 100644 --- a/docs/cluster.rst +++ b/docs/cluster.rst @@ -160,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 or results still being held in the queues. + 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/install.rst b/docs/install.rst index 1bd5e5c..6b2aa46 100644 --- a/docs/install.rst +++ b/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 ~~~~ diff --git a/docs/monitor.rst b/docs/monitor.rst index 0b40c02..63a0a03 100644 --- a/docs/monitor.rst +++ b/docs/monitor.rst @@ -127,7 +127,7 @@ Reference The pid of the pusher process - .. py:attribute:: monitormight + .. py:attribute:: monitor The pid of the monitor process