6 Commits

Author SHA1 Message Date
Stan
5ff834b794 More str 2026-08-01 00:55:17 +02:00
Stan
2fe20033e3 more str 2026-07-31 23:25:25 +02:00
Stan
1537872bf2 use strings instead of integers/floats 2026-07-31 23:19:15 +02:00
Stan
c9020ce94e Merge branch 'master' into 334-fix-multiprocessor-forkserver-start-method 2026-07-15 17:27:59 +02:00
Stan
5597d7ccd8 update all dependencies to match current 2026-07-15 17:17:41 +02:00
Andy Galasso
480b77dfe4 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
2026-05-26 00:07:53 +02:00
8 changed files with 1015 additions and 832 deletions

View File

@@ -27,34 +27,19 @@ jobs:
strategy:
matrix:
python-version:
- "3.9"
- "3.10"
- "3.11"
- "3.12"
- "3.13"
- "3.14"
django:
- "4.2"
- "5.0"
- "5.1"
- "5.2"
- "6.0"
exclude:
# django 5.2 does not support 3.9
- python-version: "3.9"
django: "5.2"
# django 5.1 does not support 3.9
- python-version: "3.9"
django: "5.1"
# django 5.0 does not support 3.9
- python-version: "3.9"
django: "5.0"
# django 4.2 does not support 3.13
- python-version: "3.13"
django: "4.2"
# django 6.0 does not support earlier than 3.12
- python-version: "3.9"
django: "6.0"
# django 6 does not support 3.10
- python-version: "3.10"
django: "6.0"
# django 6 does not support 3.11
- python-version: "3.11"
django: "6.0"

View File

@@ -1,5 +1,5 @@
# Sets the python version
FROM python:3.9-slim-bookworm
FROM python:3.12-slim-bookworm
# Allows the logs generated by python apps to be rendered in the terminal
ENV PYTHONUNBUFFERED 1

View File

@@ -35,13 +35,13 @@ See the `changelog <https://github.com/GDay/django-q2/blob/master/CHANGELOG.md>`
Requirements
~~~~~~~~~~~~
- `Django <https://www.djangoproject.com>`__ > = 4.2
- `Django <https://www.djangoproject.com>`__ 5.2 and 6.0
- `Django-picklefield <https://github.com/gintas/django-picklefield>`__
Tested with:
* Python 3.9 to 3.13.
* Django 4.2 to 6.0.
* Python 3.10 to 3.14.
* Django 5.2 and 6.0.
Brokers
~~~~~~~

View File

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

View File

@@ -97,6 +97,8 @@ def monitor(run_once=False, broker=None):
for stat in stats:
status = stat.status
# color status
if stat.status == Conf.STARTING:
status = term.yellow(str(Conf.STARTING))
if stat.status == Conf.WORKING:
status = term.green(str(Conf.WORKING))
elif stat.status == Conf.STOPPING:
@@ -105,6 +107,8 @@ def monitor(run_once=False, broker=None):
status = term.red(str(Conf.STOPPED))
elif stat.status == Conf.IDLE:
status = str(Conf.IDLE)
else:
status = _("Unknown")
# color q's
tasks = str(stat.task_q_size)
if stat.task_q_size > 0:
@@ -138,7 +142,7 @@ def monitor(run_once=False, broker=None):
)
print(
term.move(i, 3 * col_width)
+ term.center(workers, width=col_width - 1)
+ term.center(str(workers), width=col_width - 1)
)
print(
term.move(i, 4 * col_width)
@@ -146,11 +150,11 @@ def monitor(run_once=False, broker=None):
)
print(
term.move(i, 5 * col_width)
+ term.center(results, width=col_width - 1)
+ term.center(str(results), width=col_width - 1)
)
print(
term.move(i, 6 * col_width)
+ term.center(stat.reincarnations, width=col_width - 1)
+ term.center(str(stat.reincarnations), width=col_width - 1)
)
print(
term.move(i, 7 * col_width)
@@ -411,27 +415,27 @@ def memory(run_once=False, workers=False, broker=None):
)
print(
term.move(row, 2 * col_width)
+ term.center(memory_available_percentage, width=col_width - 1)
+ term.center(str(memory_available_percentage), width=col_width - 1)
)
print(
term.move(row, 3 * col_width)
+ term.center(memory_available, width=col_width - 1)
+ term.center(str(memory_available), width=col_width - 1)
)
print(
term.move(row, 4 * col_width)
+ term.center(
round(psutil.virtual_memory().total / 1024**2, 2),
str(round(psutil.virtual_memory().total / 1024**2, 2)),
width=col_width - 1,
)
)
print(
term.move(row, 5 * col_width)
+ term.center(get_process_mb(stat.sentinel), width=col_width - 1)
+ term.center(str(get_process_mb(stat.sentinel)), width=col_width - 1)
)
print(
term.move(row, 6 * col_width)
+ term.center(
get_process_mb(getattr(stat, "monitor", None)),
str(get_process_mb(getattr(stat, "monitor", None))),
width=col_width - 1,
)
)
@@ -444,7 +448,7 @@ def memory(run_once=False, workers=False, broker=None):
print(
term.move(row, 7 * col_width)
+ term.center(
workers_mb or "NO_PROCESSES_FOUND", width=col_width - 1
str(workers_mb) or "NO_PROCESSES_FOUND", width=col_width - 1
)
)
row += 1
@@ -476,7 +480,7 @@ def memory(run_once=False, workers=False, broker=None):
mb_used = get_process_mb(worker_pid)
print(
term.move(row, (idx + 1) * col_width)
+ term.center(mb_used, width=col_width - 1)
+ term.center(str(mb_used), width=col_width - 1)
)
row += 1
row += 1

View File

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

1647
poetry.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -52,8 +52,8 @@ include = [
[tool.poetry.dependencies]
python = ">=3.9,<4"
django = ">=4.2"
python = ">=3.10,<4"
django = ">=5.2"
django-picklefield = "^3.1"
blessed = { version = "^1.19.1", optional = true }