Adds a timeout override per task

* added a `timeout` keyword to tasks.
* timeouts now count down instead of up
* `timeout` in task overrides global timeout setting
* added tests for timeout override
* added a test for recycling

django_q/cluster.py

@@ -176,7 +176,7 @@ class Sentinel(object):
         return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key, self.r)
 
     def spawn_worker(self):
-        self.spawn_process(worker, self.task_queue, self.result_queue, Value('b', -1))
+        self.spawn_process(worker, self.task_queue, self.result_queue, Value('b', -1), self.timeout)
 
     def spawn_monitor(self):
         return self.spawn_process(monitor, self.result_queue)
@@ -195,7 +195,7 @@ class Sentinel(object):
         else:
             self.pool.remove(process)
             self.spawn_worker()
-            if self.timeout and int(process.timer.value) >= self.timeout:
+            if self.timeout and int(process.timer.value) == 0:
                 # only need to terminate on timeout, otherwise we risk destabilizing the queues
                 process.terminate()
                 logger.warn(_("reincarnated worker {} after timeout").format(process.name))
@@ -231,12 +231,12 @@ class Sentinel(object):
             # Check Workers
             for p in self.pool:
                 # Are you alive?
-                if not p.is_alive() or (self.timeout and int(p.timer.value) >= self.timeout):
+                if not p.is_alive() or (self.timeout and int(p.timer.value) == 0):
                     self.reincarnate(p)
                     continue
-                # Increment timer if work is being done
-                if p.timer.value >= 0:
-                    p.timer.value += 1
+                # Decrement timer if work is being done
+                if p.timer.value > 0:
+                    p.timer.value -= 1
             # Check Monitor
             if not self.monitor.is_alive():
                 self.reincarnate(self.monitor)
@@ -335,7 +335,7 @@ def monitor(result_queue):
     logger.info(_("{} stopped monitoring results").format(name))
 
 
-def worker(task_queue, result_queue, timer):
+def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
     """
     Takes a task from the task queue, tries to execute it and puts the result back in the result queue
     :type task_queue: multiprocessing.Queue
@@ -359,7 +359,7 @@ def worker(task_queue, result_queue, timer):
         # Get the function from the task
         logger.info(_('{} processing [{}]').format(name, task['name']))
         f = task['func']
-        # if it's not an instance try to get it from the string
+        # if it's not an instance try to get it from the stringIncrement
         if not callable(task['func']):
             try:
                 module, func = f.rsplit('.', 1)
@@ -370,7 +370,7 @@ def worker(task_queue, result_queue, timer):
         # We're still going
         if not result:
             # execute the payload
-            timer.value = 0  # Busy
+            timer.value = task['kwargs'].pop('timeout', timeout or 0)  # Busy
             try:
                 res = f(*task['args'], **task['kwargs'])
                 result = (res, True)
@@ -384,7 +384,7 @@ def worker(task_queue, result_queue, timer):
         timer.value = -1  # Idle
         # Recycle
         if task_count == Conf.RECYCLE:
-            timer.value = -2
+            timer.value = -2  # Recycled
             break
     logger.info(_('{} stopped doing work').format(name))
 

django_q/tests/test_cluster.py

@@ -1,8 +1,8 @@
 import sys
-import os
 from multiprocessing import Queue, Event, Value
 import threading
 
+import os
 import pytest
 
 myPath = os.path.dirname(os.path.abspath(__file__))
@@ -32,11 +32,13 @@ def r():
 def test_redis_connection(r):
     assert r.ping() is True
 
+
 @pytest.mark.django_db
 def test_sync(r):
     task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, redis=r, sync=True)
     assert result(task) == 1506
 
+
 @pytest.mark.django_db
 def test_cluster_initial(r):
     list_key = 'initial_test:q'
@@ -203,6 +205,60 @@ def test_timeout(r):
     assert start_event.is_set()
     assert s.status() == Conf.STOPPED
     assert s.reincarnations == 1
+    r.delete(list_key)
+
+
+@pytest.mark.django_db
+def test_timeout(r):
+    # set up the Sentinel
+    list_key = 'timeout_test:q'
+    async('django_q.tests.tasks.count_forever', list_key=list_key)
+    start_event = Event()
+    stop_event = Event()
+    # Set a timer to stop the Sentinel
+    threading.Timer(3, stop_event.set).start()
+    s = Sentinel(stop_event, start_event, list_key=list_key, timeout=1)
+    assert start_event.is_set()
+    assert s.status() == Conf.STOPPED
+    assert s.reincarnations == 1
+    r.delete(list_key)
+
+
+@pytest.mark.django_db
+def test_timeout_override(r):
+    # set up the Sentinel
+    list_key = 'timeout_override_test:q'
+    async('django_q.tests.tasks.count_forever', list_key=list_key, timeout=1)
+    start_event = Event()
+    stop_event = Event()
+    # Set a timer to stop the Sentinel
+    threading.Timer(3, stop_event.set).start()
+    s = Sentinel(stop_event, start_event, list_key=list_key, timeout=10)
+    assert start_event.is_set()
+    assert s.status() == Conf.STOPPED
+    assert s.reincarnations == 1
+    r.delete(list_key)
+
+
+@pytest.mark.django_db
+def test_recycle(r):
+    # set up the Sentinel
+    list_key = 'test_recycle_test:q'
+    async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key)
+    async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key)
+    async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key)
+    start_event = Event()
+    stop_event = Event()
+    # override settings
+    Conf.RECYCLE = 2
+    Conf.WORKERS = 1
+    # Set a timer to stop the Sentinel
+    threading.Timer(3, stop_event.set).start()
+    s = Sentinel(stop_event, start_event, list_key=list_key)
+    assert start_event.is_set()
+    assert s.status() == Conf.STOPPED
+    assert s.reincarnations == 1
+    r.delete(list_key)
 
 
 @pytest.mark.django_db

docs/tasks.rst

@@ -78,7 +78,7 @@ When you are making individual calls to :func:`async` a lot though, it can help
 Reference
 ---------
 
-.. py:function:: async(func, *args, hook=None, sync=False, redis=None, **kwargs)
+.. py:function:: async(func, *args, hook=None, timeout=None, sync=False, redis=None, **kwargs)
 
     Puts a task in the cluster queue
 
@@ -87,6 +87,7 @@ Reference
    :type func: object
    :param hook: Optional function to call after execution
    :type hook: object
+   :param int timeout: timeout in seconds. Overrides the cluster setting.
    :param bool sync: If set to True, async will simulate a task execution
    :param redis: Optional redis connection
    :param kwargs: Keyword arguments for the task function