mirror of
https://github.com/django-q2/django-q2.git
synced 2026-10-03 22:08:12 +08:00
#76 checks for stale connections always
It turns out that checking stale connections on a timer takes between 1-2 times as long as just checking them always. This also has the benefit of catching timeouts that happen between timer loops.
This commit is contained in:
+2
-13
@@ -318,13 +318,7 @@ def monitor(result_queue, broker=None):
|
|||||||
broker = get_broker()
|
broker = get_broker()
|
||||||
name = current_process().name
|
name = current_process().name
|
||||||
logger.info(_("{} monitoring at {}").format(name, current_process().pid))
|
logger.info(_("{} monitoring at {}").format(name, current_process().pid))
|
||||||
db.close_old_connections()
|
|
||||||
connection_timer = timezone.now()
|
|
||||||
for task in iter(result_queue.get, 'STOP'):
|
for task in iter(result_queue.get, 'STOP'):
|
||||||
# check db connection timeout
|
|
||||||
if (timezone.now() - connection_timer).total_seconds() >= Conf.DB_TIMEOUT:
|
|
||||||
db.close_old_connections()
|
|
||||||
connection_timer = timezone.now()
|
|
||||||
# acknowledge
|
# acknowledge
|
||||||
ack_id = task.pop('ack_id', False)
|
ack_id = task.pop('ack_id', False)
|
||||||
if ack_id:
|
if ack_id:
|
||||||
@@ -347,8 +341,6 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
|
|||||||
"""
|
"""
|
||||||
name = current_process().name
|
name = current_process().name
|
||||||
logger.info(_('{} ready for work at {}').format(name, current_process().pid))
|
logger.info(_('{} ready for work at {}').format(name, current_process().pid))
|
||||||
db.close_old_connections()
|
|
||||||
connection_timer = timezone.now()
|
|
||||||
task_count = 0
|
task_count = 0
|
||||||
# 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'):
|
||||||
@@ -368,10 +360,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
|
|||||||
result = (e, False)
|
result = (e, False)
|
||||||
# We're still going
|
# We're still going
|
||||||
if not result:
|
if not result:
|
||||||
# check db connection timeout
|
db.close_old_connections()
|
||||||
if (timezone.now() - connection_timer).total_seconds() >= Conf.DB_TIMEOUT:
|
|
||||||
db.close_old_connections()
|
|
||||||
connection_timer = timezone.now()
|
|
||||||
# execute the payload
|
# execute the payload
|
||||||
timer.value = task['kwargs'].pop('timeout', timeout or 0) # Busy
|
timer.value = task['kwargs'].pop('timeout', timeout or 0) # Busy
|
||||||
try:
|
try:
|
||||||
@@ -400,6 +389,7 @@ def save_task(task):
|
|||||||
if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
|
if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
|
||||||
return
|
return
|
||||||
# SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
|
# SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
|
||||||
|
db.close_old_connections()
|
||||||
try:
|
try:
|
||||||
if task['success'] and 0 < Conf.SAVE_LIMIT <= Success.objects.count():
|
if task['success'] and 0 < Conf.SAVE_LIMIT <= Success.objects.count():
|
||||||
Success.objects.last().delete()
|
Success.objects.last().delete()
|
||||||
@@ -424,7 +414,6 @@ def scheduler(broker=None):
|
|||||||
"""
|
"""
|
||||||
if not broker:
|
if not broker:
|
||||||
broker = get_broker()
|
broker = get_broker()
|
||||||
# reset stale db connections
|
|
||||||
db.close_old_connections()
|
db.close_old_connections()
|
||||||
try:
|
try:
|
||||||
for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()):
|
for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()):
|
||||||
|
|||||||
@@ -64,9 +64,6 @@ class Conf(object):
|
|||||||
# Failures are always saved
|
# Failures are always saved
|
||||||
SAVE_LIMIT = conf.get('save_limit', 250)
|
SAVE_LIMIT = conf.get('save_limit', 250)
|
||||||
|
|
||||||
# Sets the time in seconds to check for stale database connections
|
|
||||||
DB_TIMEOUT = conf.get('db_timeout', 60)
|
|
||||||
|
|
||||||
# Disable the scheduler
|
# Disable the scheduler
|
||||||
SCHEDULER = conf.get('scheduler', True)
|
SCHEDULER = conf.get('scheduler', True)
|
||||||
|
|
||||||
|
|||||||
@@ -113,7 +113,6 @@ def test_cluster(broker):
|
|||||||
@pytest.mark.django_db
|
@pytest.mark.django_db
|
||||||
def test_async(broker, admin_user):
|
def test_async(broker, admin_user):
|
||||||
broker.list_key = 'cluster_test:q'
|
broker.list_key = 'cluster_test:q'
|
||||||
Conf.DB_TIMEOUT = 0
|
|
||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result',
|
a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result',
|
||||||
broker=broker)
|
broker=broker)
|
||||||
@@ -238,7 +237,6 @@ def test_async(broker, admin_user):
|
|||||||
assert fetch(k, 100) is None
|
assert fetch(k, 100) is None
|
||||||
assert result(k, 100) is None
|
assert result(k, 100) is None
|
||||||
broker.delete_queue()
|
broker.delete_queue()
|
||||||
Conf.DB_TIMEOUT = 60
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.django_db
|
@pytest.mark.django_db
|
||||||
|
|||||||
Reference in New Issue
Block a user