mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-06 08:48:11 +08:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1941f1e3f0 |
@@ -1,5 +1,15 @@
|
|||||||
# Changelog
|
# Changelog
|
||||||
|
|
||||||
|
## [v1.11.0](https://github.com/django-q2/django-q2/tree/v1.11.0) (2026-08-10)
|
||||||
|
|
||||||
|
- AttributeError when start_event is None, and guard process faster stop (#305) https://github.com/django-q2/django-q2/pull/305
|
||||||
|
- Add croniter as optional extra. (#336) https://github.com/django-q2/django-q2/pull/336
|
||||||
|
- Migrate poetry to uv (#337) https://github.com/django-q2/django-q2/pull/337
|
||||||
|
- Add support for python 3.13 and 3.14 (#338) https://github.com/django-q2/django-q2/pull/338
|
||||||
|
- Add django 6.1 support (#340) https://github.com/django-q2/django-q2/pull/340
|
||||||
|
- Postpone SECRET_KEY evaluation to the qcluster command execution (#332) https://github.com/django-q2/django-q2/pull/332
|
||||||
|
|
||||||
|
|
||||||
## [v1.10.0](https://github.com/django-q2/django-q2/tree/v1.10.0) (2026-05-01)
|
## [v1.10.0](https://github.com/django-q2/django-q2/tree/v1.10.0) (2026-05-01)
|
||||||
|
|
||||||
- fix: Fix incorrect signal import (#308) https://github.com/django-q2/django-q2/pull/308
|
- fix: Fix incorrect signal import (#308) https://github.com/django-q2/django-q2/pull/308
|
||||||
|
|||||||
@@ -1,3 +1,3 @@
|
|||||||
VERSION = (1, 10, 0)
|
VERSION = (1, 11, 0)
|
||||||
|
|
||||||
__all__ = ["conf", "cluster", "models", "tasks"]
|
__all__ = ["conf", "cluster", "models", "tasks"]
|
||||||
|
|||||||
+5
-10
@@ -31,7 +31,6 @@ from django_q.conf import (
|
|||||||
psutil,
|
psutil,
|
||||||
setproctitle,
|
setproctitle,
|
||||||
)
|
)
|
||||||
from django_q.enums import TimerStatus
|
|
||||||
from django_q.humanhash import humanize
|
from django_q.humanhash import humanize
|
||||||
from django_q.monitor import monitor
|
from django_q.monitor import monitor
|
||||||
from django_q.pusher import pusher
|
from django_q.pusher import pusher
|
||||||
@@ -235,11 +234,7 @@ class Sentinel:
|
|||||||
|
|
||||||
def spawn_worker(self):
|
def spawn_worker(self):
|
||||||
self.spawn_process(
|
self.spawn_process(
|
||||||
worker,
|
worker, self.task_queue, self.result_queue, Value("f", -1), self.timeout
|
||||||
self.task_queue,
|
|
||||||
self.result_queue,
|
|
||||||
Value("f", TimerStatus.IDLE),
|
|
||||||
self.timeout,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def spawn_monitor(self) -> Process:
|
def spawn_monitor(self) -> Process:
|
||||||
@@ -276,7 +271,7 @@ class Sentinel:
|
|||||||
|
|
||||||
self.pool.remove(process)
|
self.pool.remove(process)
|
||||||
self.spawn_worker()
|
self.spawn_worker()
|
||||||
if process.timer.value == TimerStatus.TIMEOUT:
|
if process.timer.value == 0:
|
||||||
# only need to terminate on timeout, otherwise we risk destabilizing
|
# only need to terminate on timeout, otherwise we risk destabilizing
|
||||||
# the queues
|
# the queues
|
||||||
task_name = ""
|
task_name = ""
|
||||||
@@ -301,7 +296,7 @@ class Sentinel:
|
|||||||
"name": process.name
|
"name": process.name
|
||||||
}
|
}
|
||||||
logger.critical(msg)
|
logger.critical(msg)
|
||||||
elif int(process.timer.value) == TimerStatus.RECYCLED:
|
elif int(process.timer.value) == -2:
|
||||||
logger.info(_("recycled worker %(name)s") % {"name": process.name})
|
logger.info(_("recycled worker %(name)s") % {"name": process.name})
|
||||||
else:
|
else:
|
||||||
logger.critical(
|
logger.critical(
|
||||||
@@ -353,11 +348,11 @@ class Sentinel:
|
|||||||
for p in self.pool:
|
for p in self.pool:
|
||||||
with p.timer.get_lock():
|
with p.timer.get_lock():
|
||||||
# Are you alive?
|
# Are you alive?
|
||||||
if not p.is_alive() or p.timer.value == TimerStatus.TIMEOUT:
|
if not p.is_alive() or p.timer.value == 0:
|
||||||
self.reincarnate(p)
|
self.reincarnate(p)
|
||||||
continue
|
continue
|
||||||
# Decrement timer if work is being done
|
# Decrement timer if work is being done
|
||||||
if p.timer.value > TimerStatus.TIMEOUT:
|
if p.timer.value > 0:
|
||||||
p.timer.value -= cycle
|
p.timer.value -= cycle
|
||||||
# Check Monitor
|
# Check Monitor
|
||||||
if not self.monitor.is_alive():
|
if not self.monitor.is_alive():
|
||||||
|
|||||||
@@ -1,14 +0,0 @@
|
|||||||
from enum import IntEnum
|
|
||||||
|
|
||||||
|
|
||||||
class TimerStatus(IntEnum):
|
|
||||||
"""
|
|
||||||
Sentinel values for the countdown timer a worker shares with the sentinel.
|
|
||||||
|
|
||||||
Any positive value is the number of seconds left before the sentinel
|
|
||||||
considers the task timed out and reincarnates the worker.
|
|
||||||
"""
|
|
||||||
|
|
||||||
TIMEOUT = 0 # task timed out, the worker has to be terminated
|
|
||||||
IDLE = -1 # waiting for work, or working without a timeout
|
|
||||||
RECYCLED = -2 # recycle limit reached, the worker stopped on its own
|
|
||||||
@@ -6,7 +6,6 @@ import pytest
|
|||||||
|
|
||||||
from django_q.brokers import get_broker
|
from django_q.brokers import get_broker
|
||||||
from django_q.conf import Conf
|
from django_q.conf import Conf
|
||||||
from django_q.enums import TimerStatus
|
|
||||||
from django_q.monitor import monitor
|
from django_q.monitor import monitor
|
||||||
from django_q.pusher import pusher
|
from django_q.pusher import pusher
|
||||||
from django_q.queues import Queue
|
from django_q.queues import Queue
|
||||||
@@ -68,7 +67,7 @@ def test_cached(broker):
|
|||||||
assert task_queue.qsize() == task_count
|
assert task_queue.qsize() == task_count
|
||||||
task_queue.put("STOP")
|
task_queue.put("STOP")
|
||||||
result_queue = Queue()
|
result_queue = Queue()
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
assert result_queue.qsize() == task_count
|
assert result_queue.qsize() == task_count
|
||||||
result_queue.put("STOP")
|
result_queue.put("STOP")
|
||||||
monitor(result_queue)
|
monitor(result_queue)
|
||||||
|
|||||||
@@ -15,7 +15,6 @@ from django.utils import timezone
|
|||||||
from django_q.brokers import Broker, get_broker
|
from django_q.brokers import Broker, get_broker
|
||||||
from django_q.cluster import Cluster, Sentinel
|
from django_q.cluster import Cluster, Sentinel
|
||||||
from django_q.conf import Conf
|
from django_q.conf import Conf
|
||||||
from django_q.enums import TimerStatus
|
|
||||||
from django_q.humanhash import DEFAULT_WORDLIST, uuid
|
from django_q.humanhash import DEFAULT_WORDLIST, uuid
|
||||||
from django_q.models import Success, Task
|
from django_q.models import Success, Task
|
||||||
from django_q.monitor import monitor, save_task
|
from django_q.monitor import monitor, save_task
|
||||||
@@ -246,7 +245,7 @@ def test_cluster(broker):
|
|||||||
assert queue_size(broker=broker) == 0
|
assert queue_size(broker=broker) == 0
|
||||||
# Test work
|
# Test work
|
||||||
task_queue.put("STOP")
|
task_queue.put("STOP")
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
assert task_queue.qsize() == 0
|
assert task_queue.qsize() == 0
|
||||||
assert result_queue.qsize() == 1
|
assert result_queue.qsize() == 1
|
||||||
# Test monitor
|
# Test monitor
|
||||||
@@ -273,7 +272,7 @@ def test_results(broker):
|
|||||||
pusher(task_queue, stop_event, broker=broker)
|
pusher(task_queue, stop_event, broker=broker)
|
||||||
task_queue.put("STOP")
|
task_queue.put("STOP")
|
||||||
result_queue = Queue()
|
result_queue = Queue()
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
result_queue.put("STOP")
|
result_queue.put("STOP")
|
||||||
monitor(result_queue)
|
monitor(result_queue)
|
||||||
|
|
||||||
@@ -373,7 +372,7 @@ def test_enqueue(broker, admin_user):
|
|||||||
assert fetch_group("test_j", count=2, wait=10) is None
|
assert fetch_group("test_j", count=2, wait=10) is None
|
||||||
# let a worker handle them
|
# let a worker handle them
|
||||||
result_queue = Queue()
|
result_queue = Queue()
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
assert result_queue.qsize() == task_count
|
assert result_queue.qsize() == task_count
|
||||||
result_queue.put("STOP")
|
result_queue.put("STOP")
|
||||||
# store the results
|
# store the results
|
||||||
@@ -542,7 +541,7 @@ def test_recycle(broker, monkeypatch, django_assert_num_queries):
|
|||||||
pusher(task_queue, stop_event, broker=broker)
|
pusher(task_queue, stop_event, broker=broker)
|
||||||
pusher(task_queue, stop_event, broker=broker)
|
pusher(task_queue, stop_event, broker=broker)
|
||||||
# worker should exit on recycle
|
# worker should exit on recycle
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
# check if the work has been done
|
# check if the work has been done
|
||||||
assert result_queue.qsize() == 2
|
assert result_queue.qsize() == 2
|
||||||
# save_limit test
|
# save_limit test
|
||||||
@@ -574,7 +573,7 @@ def test_save_limit_per_func(broker, monkeypatch):
|
|||||||
threading.Timer(3, stop_event.set).start()
|
threading.Timer(3, stop_event.set).start()
|
||||||
for i in range(3):
|
for i in range(3):
|
||||||
pusher(task_queue, stop_event, broker=broker)
|
pusher(task_queue, stop_event, broker=broker)
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker)
|
s = Sentinel(stop_event, start_event, cluster_id=cluster_id, broker=broker)
|
||||||
assert start_event.is_set()
|
assert start_event.is_set()
|
||||||
assert s.status() == Conf.STOPPED
|
assert s.status() == Conf.STOPPED
|
||||||
@@ -619,7 +618,7 @@ def test_max_rss(broker, monkeypatch):
|
|||||||
# push the task
|
# push the task
|
||||||
pusher(task_queue, stop_event, broker=broker)
|
pusher(task_queue, stop_event, broker=broker)
|
||||||
# worker should exit on recycle
|
# worker should exit on recycle
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
# check if the work has been done
|
# check if the work has been done
|
||||||
assert result_queue.qsize() == 1
|
assert result_queue.qsize() == 1
|
||||||
# save_limit test
|
# save_limit test
|
||||||
@@ -655,7 +654,7 @@ def test_bad_secret(broker, monkeypatch):
|
|||||||
worker(
|
worker(
|
||||||
task_queue,
|
task_queue,
|
||||||
result_queue,
|
result_queue,
|
||||||
Value("f", TimerStatus.IDLE),
|
Value("f", -1),
|
||||||
)
|
)
|
||||||
assert result_queue.qsize() == 0
|
assert result_queue.qsize() == 0
|
||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
@@ -839,7 +838,7 @@ class TestSignals:
|
|||||||
event.set()
|
event.set()
|
||||||
pusher(task_queue, event, broker=broker)
|
pusher(task_queue, event, broker=broker)
|
||||||
task_queue.put("STOP")
|
task_queue.put("STOP")
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
result_queue.put("STOP")
|
result_queue.put("STOP")
|
||||||
monitor(result_queue, broker)
|
monitor(result_queue, broker)
|
||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
@@ -868,7 +867,7 @@ class TestSignals:
|
|||||||
event.set()
|
event.set()
|
||||||
pusher(task_queue, event, broker=broker)
|
pusher(task_queue, event, broker=broker)
|
||||||
task_queue.put("STOP")
|
task_queue.put("STOP")
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
result_queue.put("STOP")
|
result_queue.put("STOP")
|
||||||
monitor(result_queue, broker)
|
monitor(result_queue, broker)
|
||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
@@ -897,7 +896,7 @@ class TestSignals:
|
|||||||
event.set()
|
event.set()
|
||||||
pusher(task_queue, event, broker=broker)
|
pusher(task_queue, event, broker=broker)
|
||||||
task_queue.put("STOP")
|
task_queue.put("STOP")
|
||||||
worker(task_queue, result_queue, Value("f", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("f", -1))
|
||||||
result_queue.put("STOP")
|
result_queue.put("STOP")
|
||||||
monitor(result_queue, broker)
|
monitor(result_queue, broker)
|
||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
|
|||||||
@@ -13,7 +13,6 @@ from django.utils.timezone import is_naive
|
|||||||
|
|
||||||
from django_q.brokers import Broker, get_broker
|
from django_q.brokers import Broker, get_broker
|
||||||
from django_q.conf import Conf
|
from django_q.conf import Conf
|
||||||
from django_q.enums import TimerStatus
|
|
||||||
from django_q.monitor import monitor
|
from django_q.monitor import monitor
|
||||||
from django_q.pusher import pusher
|
from django_q.pusher import pusher
|
||||||
from django_q.queues import Queue
|
from django_q.queues import Queue
|
||||||
@@ -223,7 +222,7 @@ def test_scheduler(broker, monkeypatch):
|
|||||||
task_queue.put("STOP")
|
task_queue.put("STOP")
|
||||||
# let a worker handle them
|
# let a worker handle them
|
||||||
result_queue = Queue()
|
result_queue = Queue()
|
||||||
worker(task_queue, result_queue, Value("b", TimerStatus.IDLE))
|
worker(task_queue, result_queue, Value("b", -1))
|
||||||
assert result_queue.qsize() == 1
|
assert result_queue.qsize() == 1
|
||||||
result_queue.put("STOP")
|
result_queue.put("STOP")
|
||||||
# store the results
|
# store the results
|
||||||
|
|||||||
+6
-7
@@ -17,7 +17,6 @@ except core.exceptions.AppRegistryNotReady:
|
|||||||
django.setup()
|
django.setup()
|
||||||
|
|
||||||
from django_q.conf import Conf, error_reporter, logger, resource, setproctitle
|
from django_q.conf import Conf, error_reporter, logger, resource, setproctitle
|
||||||
from django_q.enums import TimerStatus
|
|
||||||
from django_q.exceptions import TimeoutException
|
from django_q.exceptions import TimeoutException
|
||||||
from django_q.signals import post_execute_in_worker, post_spawn, pre_execute
|
from django_q.signals import post_execute_in_worker, post_spawn, pre_execute
|
||||||
from django_q.timeout import TimeoutHandler
|
from django_q.timeout import TimeoutHandler
|
||||||
@@ -55,11 +54,11 @@ def worker(
|
|||||||
setproctitle.setproctitle(f"qcluster {proc_name} idle")
|
setproctitle.setproctitle(f"qcluster {proc_name} idle")
|
||||||
task_count = 0
|
task_count = 0
|
||||||
if timeout is None:
|
if timeout is None:
|
||||||
timeout = TimerStatus.IDLE
|
timeout = -1
|
||||||
# Start reading the task queue
|
# Start reading the task queue
|
||||||
for task in iter(task_queue.get, "STOP"):
|
for task in iter(task_queue.get, "STOP"):
|
||||||
result = None
|
result = None
|
||||||
timer.value = TimerStatus.IDLE
|
timer.value = -1 # Idle
|
||||||
task_count += 1
|
task_count += 1
|
||||||
f = task["func"]
|
f = task["func"]
|
||||||
|
|
||||||
@@ -93,7 +92,7 @@ def worker(
|
|||||||
pre_execute.send(sender="django_q", func=f, task=task)
|
pre_execute.send(sender="django_q", func=f, task=task)
|
||||||
# execute the payload
|
# execute the payload
|
||||||
timer.value = timer_value # Busy
|
timer.value = timer_value # Busy
|
||||||
if timer.value != TimerStatus.IDLE:
|
if timer.value != -1:
|
||||||
timer.value += 3 # Add buffer so that guard doesn't kill the process on timeout before it gets processed
|
timer.value += 3 # Add buffer so that guard doesn't kill the process on timeout before it gets processed
|
||||||
|
|
||||||
timeout_error = False
|
timeout_error = False
|
||||||
@@ -123,15 +122,15 @@ def worker(
|
|||||||
result_queue.put(task)
|
result_queue.put(task)
|
||||||
if timeout_error:
|
if timeout_error:
|
||||||
# force destroy process due to timeout
|
# force destroy process due to timeout
|
||||||
timer.value = TimerStatus.TIMEOUT
|
timer.value = 0
|
||||||
break
|
break
|
||||||
|
|
||||||
timer.value = TimerStatus.IDLE
|
timer.value = -1 # Idle
|
||||||
if setproctitle:
|
if setproctitle:
|
||||||
setproctitle.setproctitle(f"qcluster {proc_name} idle")
|
setproctitle.setproctitle(f"qcluster {proc_name} idle")
|
||||||
# Recycle
|
# Recycle
|
||||||
if task_count == Conf.RECYCLE or rss_check():
|
if task_count == Conf.RECYCLE or rss_check():
|
||||||
timer.value = TimerStatus.RECYCLED
|
timer.value = -2 # Recycled
|
||||||
break
|
break
|
||||||
logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name})
|
logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name})
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -73,9 +73,9 @@ author = "Ilan Steemers, Stan Triepels"
|
|||||||
# built documents.
|
# built documents.
|
||||||
#
|
#
|
||||||
# The short X.Y version.
|
# The short X.Y version.
|
||||||
version = "1.10"
|
version = "1.11"
|
||||||
# The full version, including alpha/beta/rc tags.
|
# The full version, including alpha/beta/rc tags.
|
||||||
release = "1.10.0"
|
release = "1.11.0"
|
||||||
|
|
||||||
# The language for content autogenerated by Sphinx. Refer to documentation
|
# The language for content autogenerated by Sphinx. Refer to documentation
|
||||||
# for a list of supported languages.
|
# for a list of supported languages.
|
||||||
|
|||||||
+2
-2
@@ -4,7 +4,7 @@ build-backend = "uv_build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "django-q2"
|
name = "django-q2"
|
||||||
version = "1.10.0"
|
version = "1.11.0"
|
||||||
packages = [
|
packages = [
|
||||||
{ include = "django_q" },
|
{ include = "django_q" },
|
||||||
]
|
]
|
||||||
@@ -33,11 +33,11 @@ classifiers = [
|
|||||||
"Operating System :: MacOS",
|
"Operating System :: MacOS",
|
||||||
"Programming Language :: Python",
|
"Programming Language :: Python",
|
||||||
"Programming Language :: Python :: 3",
|
"Programming Language :: Python :: 3",
|
||||||
"Programming Language :: Python :: 3.9",
|
|
||||||
"Programming Language :: Python :: 3.10",
|
"Programming Language :: Python :: 3.10",
|
||||||
"Programming Language :: Python :: 3.11",
|
"Programming Language :: Python :: 3.11",
|
||||||
"Programming Language :: Python :: 3.12",
|
"Programming Language :: Python :: 3.12",
|
||||||
"Programming Language :: Python :: 3.13",
|
"Programming Language :: Python :: 3.13",
|
||||||
|
"Programming Language :: Python :: 3.14",
|
||||||
"Topic :: Internet :: WWW/HTTP",
|
"Topic :: Internet :: WWW/HTTP",
|
||||||
"Topic :: System :: Distributed Computing",
|
"Topic :: System :: Distributed Computing",
|
||||||
"Topic :: Software Development :: Libraries :: Python Modules",
|
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||||
|
|||||||
Reference in New Issue
Block a user