Fix async_task timeout parameter handling when cluster timeout is set…

… to None (the default)

The cluster timeout configuration default value None is documented to mean
tasks never timeout out. Also the documentation states that the timeout
can be overridden for individual tasks.

With the old implementation the timeout parameter given to async_task was
not honored if the cluster timeout was set to None.

Fixes: #335

django_q/cluster.py

@@ -172,7 +172,7 @@ def reincarnate(self, process):
         else:
             self.pool.remove(process)
             self.spawn_worker()
-            if self.timeout and int(process.timer.value) == 0:
+            if 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))
@@ -210,11 +210,11 @@ def guard(self):
             # Check Workers
             for p in self.pool:
                 # Are you alive?
-                if not p.is_alive() or (self.timeout and p.timer.value == 0):
+                if not p.is_alive() or p.timer.value == 0:
                     self.reincarnate(p)
                     continue
                 # Decrement timer if work is being done
-                if self.timeout and p.timer.value > 0:
+                if p.timer.value > 0:
                     p.timer.value -= cycle
             # Check Monitor
             if not self.monitor.is_alive():
@@ -347,6 +347,8 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
     name = current_process().name
     logger.info(_('{} ready for work at {}').format(name, current_process().pid))
     task_count = 0
+    if timeout is None:
+        timeout = -1
     # Start reading the task queue
     for task in iter(task_queue.get, 'STOP'):
         result = None
@@ -368,7 +370,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
         # We're still going
         if not result:
             db.close_old_connections()
-            timer_value = task['kwargs'].pop('timeout', timeout or 0)
+            timer_value = task['kwargs'].pop('timeout', timeout)
             # signal execution
             pre_execute.send(sender="django_q", func=f, task=task)
             # execute the payload

django_q/tests/test_cluster.py

@@ -247,6 +247,7 @@ def test_enqueue(broker, admin_user):
 @pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', (
     (1, {}),
     (10, {'timeout': 1}),
+    (None, {'timeout': 1}),
 ))
 def test_timeout(broker, cluster_config_timeout, async_task_kwargs):
     # set up the Sentinel
@@ -269,6 +270,7 @@ def test_timeout(broker, cluster_config_timeout, async_task_kwargs):
     (5, {}),
     (10, {'timeout': 5}),
     (1, {'timeout': 5}),
+    (None, {'timeout': 5}),
 ))
 def test_timeout_task_finishes(broker, cluster_config_timeout, async_task_kwargs):
     # set up the Sentinel