From 888324cf48d076868302479421436a6172ae3bbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Janne=20R=C3=B6nkk=C3=B6?= Date: Sat, 26 Jan 2019 20:44:31 +0200 Subject: [PATCH 1/4] Reduce code duplication in timeout unit tests --- django_q/tests/test_cluster.py | 26 +++++++------------------- 1 file changed, 7 insertions(+), 19 deletions(-) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 3355052..016422e 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -244,32 +244,20 @@ def test_enqueue(broker, admin_user): @pytest.mark.django_db -def test_timeout(broker): +@pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', ( + (1, {}), + (10, {'timeout': 1}), +)) +def test_timeout(broker, cluster_config_timeout, async_task_kwargs): # set up the Sentinel broker.list_key = 'timeout_test:q' broker.purge_queue() - async_task('django_q.tests.tasks.count_forever', broker=broker) + async_task('django_q.tests.tasks.count_forever', broker=broker, **async_task_kwargs) 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, broker=broker, timeout=1) - assert start_event.is_set() - assert s.status() == Conf.STOPPED - assert s.reincarnations == 1 - broker.delete_queue() - - -@pytest.mark.django_db -def test_timeout_override(broker): - # set up the Sentinel - broker.list_key = 'timeout_override_test:q' - async_task('django_q.tests.tasks.count_forever', broker=broker, 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, broker=broker, timeout=10) + s = Sentinel(stop_event, start_event, broker=broker, timeout=cluster_config_timeout) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 From d09e2658becef566393503482b316318649d672d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Janne=20R=C3=B6nkk=C3=B6?= Date: Sun, 27 Jan 2019 15:28:28 +0200 Subject: [PATCH 2/4] Add test for task finishes before timeout --- django_q/tests/test_cluster.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 016422e..7ef208f 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -264,6 +264,28 @@ def test_timeout(broker, cluster_config_timeout, async_task_kwargs): broker.delete_queue() +@pytest.mark.django_db +@pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', ( + (5, {}), + (10, {'timeout': 5}), + (1, {'timeout': 5}), +)) +def test_timeout_task_finishes(broker, cluster_config_timeout, async_task_kwargs): + # set up the Sentinel + broker.list_key = 'timeout_test:q' + broker.purge_queue() + async_task('time.sleep', 3, broker=broker, **async_task_kwargs) + start_event = Event() + stop_event = Event() + # Set a timer to stop the Sentinel + threading.Timer(6, stop_event.set).start() + s = Sentinel(stop_event, start_event, broker=broker, timeout=cluster_config_timeout) + assert start_event.is_set() + assert s.status() == Conf.STOPPED + assert s.reincarnations == 0 + broker.delete_queue() + + @pytest.mark.django_db def test_recycle(broker, monkeypatch): # set up the Sentinel From 2650659da9706908dba81ef8a379aec31b75f286 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Janne=20R=C3=B6nkk=C3=B6?= Date: Sat, 26 Jan 2019 20:45:38 +0200 Subject: [PATCH 3/4] Allow timeout unit tests to fail properly The old implementation used count_forever test task that never finishes. If the timeout implementation does not work in the test, the test never end and you was required to kill the test runner. --- django_q/tests/test_cluster.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 7ef208f..7dc0959 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -252,7 +252,7 @@ def test_timeout(broker, cluster_config_timeout, async_task_kwargs): # set up the Sentinel broker.list_key = 'timeout_test:q' broker.purge_queue() - async_task('django_q.tests.tasks.count_forever', broker=broker, **async_task_kwargs) + async_task('time.sleep', 5, broker=broker, **async_task_kwargs) start_event = Event() stop_event = Event() # Set a timer to stop the Sentinel From 0e2df88d92715b7b92bf04bf0d05c56408ddf6a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Janne=20R=C3=B6nkk=C3=B6?= Date: Sun, 27 Jan 2019 15:47:06 +0200 Subject: [PATCH 4/4] 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 | 10 ++++++---- django_q/tests/test_cluster.py | 2 ++ 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index dc1b46c..534f887 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -172,7 +172,7 @@ class Sentinel(object): 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 @@ class Sentinel(object): # 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 diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 7dc0959..5e289d7 100644 --- a/django_q/tests/test_cluster.py +++ b/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