AttributeError when start_event is None, and guard process faster stop (#305)

* AttributeError when start_event is None, and guard process faster stop

This commit has two related fixes that come into play when frequently
shutting down and restarting the cluster.

If a SIGINT or SIGTERM signal is received while the main process is
waiting for the sentinel process to set start_event, the signal
handler sets start_event to None, and the loop that polls start_event
will raise an unhandled AttributeError attempting to check
start_event.is_set().

The second issue is that the guard process sleeps for the guard cycle
interval in each iteration before checking the stop event, preventing
the guard cycle from terminating until the sleep completes. The
cluster can be made more responsive to a shutdown request by using
the event's wait(cycle) method rather than time.sleep(cycle) so the
process wakes immediately when the event is set.

The commit also fixes the case where the guard process sleeps for an
extra cycle when the cycle counter happens to have been reset to zero
on the loop iterations when the scheduler is called.

This fix helps in our environment where we are frequently stopping
and starting the cluster and where we have increased the guard
cycle setting to several seconds.

* Add sentinel premature death check while waiting for sentinel to start.

Suggested by github copilot [here](https://github.com/django-q2/django-q2/pull/305/changes#r3143964420)

* Add test case test_cluster_early_stop

Add a test case to validate stopping the cluster before the
sentinel has set the cluster's start_event.

The test deterministically triggers the AttributeError when
start_event is None (without the PR's fix in cluster.py).

As requested by copilot: https://github.com/django-q2/django-q2/pull/305#discussion_r3143964399

* Add test test_cluster_stop_responsive

Ensure that stopping the cluster is responsive and does not wait for a full
GUARD_CYCLE to stop the cluster.

As requested by copilot: https://github.com/django-q2/django-q2/pull/305/changes#r3143964414

* Address copilot review comments.

1. Do not raise RuntimeError if sentinel early exit was caused by
   SIGINT/SIGTERM stopping the cluster while waiting for the
   sentinel to start.
   Addresses https://github.com/django-q2/django-q2/pull/305#discussion_r3298618046

2. Ensure that the monkeypatched broker instance remains pickleable
   so the test_cluster_early_stop test will also work on platforms
   that use the spawn Process start method.
   Addresses https://github.com/django-q2/django-q2/pull/305#discussion_r3298618086

3. Do not allow the test_cluster_early_stop test to hang if the
   sentinel process or main test process terminates unexpectedly
   without setting the sentinel_event or test_event.
   Addresses https://github.com/django-q2/django-q2/pull/305#discussion_r3298618102
This commit is contained in:
Andy Galasso
2026-05-25 18:07:53 -04:00
committed by GitHub
parent a699578c34
commit 480b77dfe4
2 changed files with 138 additions and 3 deletions

View File

@@ -73,7 +73,30 @@ class Cluster:
) )
self.sentinel.start() self.sentinel.start()
logger.info(_("Q Cluster %(name)s starting.") % {"name": self.name}) 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) sleep(0.1)
return self.pid return self.pid
@@ -313,7 +336,7 @@ class Sentinel:
counter = 0 counter = 0
cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds
# Guard loop. Runs at least once # Guard loop. Runs at least once
while not self.stop_event.is_set() or not counter: while True:
# Check Workers # Check Workers
for p in self.pool: for p in self.pool:
with p.timer.get_lock(): with p.timer.get_lock():
@@ -337,7 +360,8 @@ class Sentinel:
scheduler(broker=self.broker) scheduler(broker=self.broker)
# Save current status # Save current status
Stat(self).save() Stat(self).save()
sleep(cycle) if self.stop_event.wait(cycle):
break
self.stop() self.stop()
def stop(self): def stop(self):

View File

@@ -1,4 +1,5 @@
import os import os
import signal
import sys import sys
import threading import threading
import uuid as uuidlib import uuid as uuidlib
@@ -98,6 +99,116 @@ def test_cluster_initial(broker):
broker.delete_queue() 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 @pytest.mark.django_db
def test_sentinel(): def test_sentinel():
start_event = Event() start_event = Event()