diff --git a/django_q/cluster.py b/django_q/cluster.py index 23c9049..2eb573b 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -73,7 +73,30 @@ class Cluster: ) self.sentinel.start() logger.info(_("Q Cluster %(name)s starting.") % {"name": self.name}) - while not self.start_event.is_set(): + # Wait for the sentinel to set start_event by polling the event's state. We + # cannot call wait() as that will deadlock if a signal is received while blocked + # in wait(), as the sentinel will block in start_event.set() while this process + # is blocked in sentinel.join(). It is also necessary to check for start_event + # being set to None by the signal handler. + while self.start_event and not self.start_event.is_set(): + # While waiting for the sentinel to start, also check for sentinel premature + # death + if not self.sentinel.is_alive(): + if self.sentinel.exitcode == 0: + # the cluster was stopped via SIGTERM/SIGINT and sentinel early + # termination was intentional -- exit quietly + break + logger.error( + _( + "Q Cluster %(name)s failed to start. Sentinel exited " + "with exit code %(exitcode)s." + ) + % {"name": self.name, "exitcode": self.sentinel.exitcode} + ) + raise RuntimeError( + f"Q Cluster {self.name} failed to start: sentinel exited " + f"with code {self.sentinel.exitcode}" + ) sleep(0.1) return self.pid @@ -313,7 +336,7 @@ class Sentinel: counter = 0 cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds # Guard loop. Runs at least once - while not self.stop_event.is_set() or not counter: + while True: # Check Workers for p in self.pool: with p.timer.get_lock(): @@ -337,7 +360,8 @@ class Sentinel: scheduler(broker=self.broker) # Save current status Stat(self).save() - sleep(cycle) + if self.stop_event.wait(cycle): + break self.stop() def stop(self): diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 6065a7a..f24d51e 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -1,4 +1,5 @@ import os +import signal import sys import threading import uuid as uuidlib @@ -98,6 +99,116 @@ def test_cluster_initial(broker): broker.delete_queue() +class TestEarlyClusterStop: + sentinel_event = Event() + test_event = Event() + + @staticmethod + def _fake_ping_for_test_cluster_early_stop(): + """ + Fake broker ping() method for test_cluster_early_stop() + + This is called in the sentinel process's start() method and overriding it allows + us to synchronize the sentinel process with the main process to provide a + deterministic ordering of operations for the test. + + This is implemented as a static method so the patched broker remains pickleable + for passing to the sentinel process, in case anyone ever runs the test on + platforms like MacOS that use spawn Process start method. + """ + # let the main process know the sentinel is waiting + TestEarlyClusterStop.sentinel_event.set() + # wait for the main process to let the sentinel proceed. The timeout prevents + # the test from hanging if the main process terminates unexpectedly before + # setting the event. + assert TestEarlyClusterStop.test_event.wait(10) + + @pytest.mark.django_db + def test_cluster_early_stop(self, broker, monkeypatch): + """ + Test stopping the cluster before the sentinel has set the cluster's start_event + """ + + def raise_sigterm(): + # wait for the sentinel to be blocked in fake_ping(). Periodically check + # that the sentinel is alive in case the sentinel dies unexpectedly before + # setting the sentinel_event -- prevents the test from hanging if sentinel + # dies unexpectedly. + while True: + if self.sentinel_event.wait(0.5): + break + if not c.sentinel.is_alive(): + break + # stop the cluster via SIGTERM + os.kill(os.getpid(), signal.SIGTERM) + # unblock the sentinel + self.test_event.set() + + monkeypatch.setattr( + broker, + "ping", + self._fake_ping_for_test_cluster_early_stop, + ) + + broker.list_key = "initial_test:q" + broker.delete_queue() + c = Cluster(broker=broker) + assert c.sentinel is None + assert c.stat.status == Conf.STOPPED + + # Use a timer to stop the cluster while it is still starting up. The timer function + # is guaranteed to run before c.start() returns, as start() must wait for the + # sentinel to set the cluster's start_event, and the sentinel will block on + # test_event before setting start_event. + threading.Timer(0.5, raise_sigterm).start() + + c.start() + stat = c.stat + assert stat.status == Conf.STOPPED + assert c.sentinel.is_alive() is False + assert c.has_stopped + broker.delete_queue() + + +@pytest.mark.django_db +def test_cluster_stop_responsive(broker, monkeypatch): + """ + Ensure that stopping the cluster is responsive and does not wait for a full + GUARD_CYCLE. + """ + timeout_seconds = 5 + + def timeout(signal, frame): + assert 0, f"cluster did not stop after {timeout_seconds}s" + + broker.list_key = "initial_test:q" + broker.delete_queue() + # set a long GUARD_CYCLE -- greater than timeout_seconds + monkeypatch.setattr(Conf, "GUARD_CYCLE", timeout_seconds * 2) + c = Cluster(broker=broker) + assert c.sentinel is None + assert c.stat.status == Conf.STOPPED + assert c.start() > 0 + assert c.sentinel.is_alive() is True + assert c.is_running + assert c.is_stopping is False + assert c.is_starting is False + sleep(0.5) + stat = c.stat + assert stat.status == Conf.IDLE + prev_handler = signal.signal(signal.SIGALRM, timeout) + try: + signal.alarm(timeout_seconds) + assert c.stop() is True + finally: + signal.alarm(0) # cancel the alarm + signal.signal(signal.SIGALRM, prev_handler) + assert c.sentinel.is_alive() is False + assert c.has_stopped + assert c.stop() is False + broker.delete_queue() + + @pytest.mark.django_db def test_sentinel(): start_event = Event()