add a greenlet to refresh the list of paused queues

Author: Florian Perucki

docs/command-line.md

@@ -56,6 +56,7 @@ You can pass additional configuration flags:
  - `--report_file`: Filepath of a json dump of the worker status. Disabled if none.
  - `--subqueues_refresh_interval`: Seconds between worker refreshes of the known subqueues.
  - `--subqueues_delimiter`: Delimiter between main queue and subqueue names.
+ - `--paused_queues_refresh_interval`: Seconds between worker refreshes of the paused queues list.
  - `--admin_port`: Start an admin server on this port, if provided. Incompatible with --processes. Defaults to **0**
  - `--admin_ip`: IP for the admin server to listen on. Use "0.0.0.0" to allow access from outside. Defaults to **127.0.0.1**.
  - `--local_ip`: Overwrite the local IP, to be displayed in the dashboard.
@@ -103,4 +104,3 @@ $ mrq-run tasks.mylib.myfile.MyTask '{"param1": 1, "param2": True}'
 # Shorter syntax which casts all values as strings (equivalent to '{"param1": "1", "param2": "ok"}')
 $ mrq-run tasks.mylib.myfile.MyTask param1 1 param2 ok
 ```
-

mrq/config.py

@@ -341,6 +341,13 @@ def add_parser_args(parser, config_type):
             type=float,
             help="Seconds between worker refreshes of the known subqueues")
 
+        parser.add_argument(
+            '--paused_queues_refresh_interval',
+            default=60,
+            action='store',
+            type=float,
+            help="Seconds between worker refreshes of the paused queues list")
+
         parser.add_argument(
             '--subqueues_delimiter',
             default='/',

mrq/queue.py

@@ -25,6 +25,7 @@ class Queue(object):
     # This is a mutable type so it is shared by all instances
     # of Queue in the current process
     known_queues = {}
+    paused_queues = set()
 
     def __init__(self, queue_id, add_to_known_queues=False):
 
@@ -162,7 +163,11 @@ def pause(self):
         context.connections.redis.sadd(Queue.redis_key_paused_queues(), self.id)
 
     def is_paused(self):
-        """ Returns wether the queue is paused or not """
+        """
+            Returns wether the queue is paused or not.
+            Warning: this does NOT ensure that the queue was effectively added to
+            the list of paused queues. See the 'paused_queues_refresh_interval' option.
+        """
         return context.connections.redis.sismember(Queue.redis_key_paused_queues(), self.id)
 
     def resume(self):

mrq/worker.py

@@ -207,7 +207,6 @@ def greenlet_subqueues(self):
 
             # Update the process-local list of known queues
             Queue.known_queues = Queue.redis_known_queues()
-            Queue.paused_queues = Queue.redis_paused_queues()
 
             queues = []
             try:
@@ -228,6 +227,14 @@ def greenlet_subqueues(self):
 
             time.sleep(self.config["subqueues_refresh_interval"])
 
+    def greenlet_paused_queues(self):
+
+      while True:
+
+          # Update the process-local list of known queues
+          Queue.paused_queues = Queue.redis_paused_queues()
+          time.sleep(self.config["paused_queues_refresh_interval"])
+
     def get_memory(self):
         mmaps = self.process.get_memory_maps()
         mem = {
@@ -419,6 +426,10 @@ def work_init(self):
 
         self.greenlets["subqueues"] = gevent.spawn(self.greenlet_subqueues)
 
+        # An interval of 0 disables the refresh
+        if self.config["paused_queues_refresh_interval"] > 0:
+            self.greenlets["paused_queues"] = gevent.spawn(self.greenlet_paused_queues)
+
         self.greenlets["report"] = gevent.spawn(self.greenlet_report)
 
         self.greenlets["logs"] = gevent.spawn(self.greenlet_logs)
@@ -470,8 +481,7 @@ def work_loop(self, max_jobs=None):
                     gevent.sleep(0.01)
 
                 jobs = []
-                paused_queues = Queue.redis_paused_queues()
-                available_queues = [queue for queue in self.queues if queue.id not in paused_queues]
+                available_queues = [queue for queue in self.queues if queue.id not in Queue.paused_queues]
 
                 for queue_i in xrange(len(available_queues)):
 

tests/test_pause.py

@@ -5,12 +5,15 @@
 
 def test_pause_resume(worker):
 
-    worker.start()
+    worker.start(flags="--paused_queues_refresh_interval=0.1")
 
     Queue("high").pause()
 
     assert Queue("high").is_paused()
 
+    # wait for the paused_queues list to be refreshed
+    time.sleep(2)
+
     job_id1 = send_task(
         "tests.tasks.general.MongoInsert", {"a": 41},
         queue="high")
@@ -41,3 +44,32 @@ def test_pause_resume(worker):
     assert job1["result"] == {"a": 41}
 
     assert worker.mongodb_jobs.tests_inserts.count() == 2
+
+    worker.stop()
+
+
+def test_pause_refresh_interval(worker):
+
+    """ Tests that a refresh interval of 0 disables the pause functionnality """
+
+    worker.start(flags="--paused_queues_refresh_interval=0")
+
+    Queue("high").pause()
+
+    assert Queue("high").is_paused()
+
+    # wait for the paused_queues list to be refreshed
+    time.sleep(2)
+
+    job_id1 = send_task(
+        "tests.tasks.general.MongoInsert", {"a": 41},
+        queue="high")
+
+    time.sleep(5)
+
+    job1 = Job(job_id1).fetch().data
+
+    assert job1["status"] == "success"
+    assert job1["result"] == {"a": 41}
+
+    worker.stop()