mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-15 21:47:53 +08:00
Compare commits
6 Commits
v1.10.0
...
334-fix-mu
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5ff834b794 | ||
|
|
2fe20033e3 | ||
|
|
1537872bf2 | ||
|
|
c9020ce94e | ||
|
|
5597d7ccd8 | ||
|
|
480b77dfe4 |
23
.github/workflows/test.yml
vendored
23
.github/workflows/test.yml
vendored
@@ -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"
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
~~~~~~~
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
1647
poetry.lock
generated
File diff suppressed because it is too large
Load Diff
@@ -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 }
|
||||
|
||||
Reference in New Issue
Block a user