fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-07 04:08:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #393 from ossobv/sync

Browse Source 
Preserve database connection when sync=True
This commit is contained in:
Ilan Steemers
2020-01-17 15:34:46 +01:00
committed by GitHub
parent d2c9aa0e88 66588a28e7
commit 681375364d
1 changed files with 18 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/cluster.py
+18 -6
View File
@@ -16,7 +16,6 @@ import socket
import traceback import traceback
# Django # Django
from django import db from django import db
from django.conf import settings
from django.utils import timezone from django.utils import timezone
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
from multiprocessing import Event, Process, Value, current_process from multiprocessing import Event, Process, Value, current_process
@@ -163,7 +162,7 @@ class Sentinel(object):
:param process: the process to reincarnate :param process: the process to reincarnate
:type process: Process or None :type process: Process or None
""" """
db.connections.close_all() # Close any old connections close_old_django_connections()
if process == self.monitor: if process == self.monitor:
self.monitor = self.spawn_monitor() self.monitor = self.spawn_monitor()
logger.error(_("reincarnated monitor {} after sudden death").format(process.name)) logger.error(_("reincarnated monitor {} after sudden death").format(process.name))
@@ -187,7 +186,7 @@ class Sentinel(object):
def spawn_cluster(self): def spawn_cluster(self):
self.pool = [] self.pool = []
Stat(self).save() Stat(self).save()
db.connection.close() close_old_django_connections()
# spawn worker pool # spawn worker pool
for __ in range(self.pool_size): for __ in range(self.pool_size):
self.spawn_worker() self.spawn_worker()
@@ -370,7 +369,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
error_reporter.report() error_reporter.report()
# We're still going # We're still going
if not result: if not result:
db.close_old_connections() close_old_django_connections()
timer_value = task.pop('timeout', timeout) timer_value = task.pop('timeout', timeout)
# signal execution # signal execution
pre_execute.send(sender="django_q", func=f, task=task) pre_execute.send(sender="django_q", func=f, task=task)
@@ -408,7 +407,7 @@ def save_task(task, broker):
if task.get('chain', None): if task.get('chain', None):
django_q.tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'], sync=task['sync'], broker=broker) django_q.tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'], sync=task['sync'], broker=broker)
# SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
db.close_old_connections() close_old_django_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()
@@ -489,7 +488,7 @@ def scheduler(broker=None):
""" """
if not broker: if not broker:
broker = get_broker() broker = get_broker()
db.close_old_connections() close_old_django_connections()
try: try:
with db.transaction.atomic(): with db.transaction.atomic():
for s in Schedule.objects.select_for_update().exclude(repeats=0).filter(next_run__lt=timezone.now()): for s in Schedule.objects.select_for_update().exclude(repeats=0).filter(next_run__lt=timezone.now()):
@@ -558,6 +557,19 @@ def scheduler(broker=None):
logger.error(e) logger.error(e)
def close_old_django_connections():
'''
Close django connections unless running with sync=True.
'''
if Conf.SYNC:
logger.warning(
'Preserving django database connections because sync=True. Beware '
'that tasks are now injected in the calling context/transactions '
'which may result in unexpected bahaviour.')
else:
db.close_old_connections()
def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING): def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING):
""" """
Sets the cpu affinity for the supplied processes. Sets the cpu affinity for the supplied processes.