Merge branch 'master' into 334-fix-multiprocessor-forkserver-start-method

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):

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()