diff --git a/django_q/cluster.py b/django_q/cluster.py index a250cf0..0f057ec 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -319,10 +319,17 @@ def monitor(result_queue, broker=None): name = current_process().name 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'): + # check db connection timeout + if (timezone.now() - connection_timer).total_seconds() >= Conf.DB_TIMEOUT: + db.close_old_connections() + connection_timer = timezone.now() + # acknowledge ack_id = task.pop('ack_id', False) if ack_id: broker.acknowledge(ack_id) + # save the result save_task(task) if task['success']: logger.info(_("Processed [{}]").format(task['name'])) @@ -341,6 +348,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT): name = current_process().name logger.info(_('{} ready for work at {}').format(name, current_process().pid)) db.close_old_connections() + connection_timer = timezone.now() task_count = 0 # Start reading the task queue for task in iter(task_queue.get, 'STOP'): @@ -360,6 +368,10 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT): result = (e, False) # We're still going if not result: + # check db connection timeout + if (timezone.now() - connection_timer).total_seconds() >= Conf.DB_TIMEOUT: + db.close_old_connections() + connection_timer = timezone.now() # execute the payload timer.value = task['kwargs'].pop('timeout', timeout or 0) # Busy try: diff --git a/django_q/conf.py b/django_q/conf.py index 9ca2146..f45d853 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -64,6 +64,9 @@ class Conf(object): # Failures are always saved 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 SCHEDULER = conf.get('scheduler', True) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 37dc79a..75eec79 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -113,6 +113,7 @@ def test_cluster(broker): @pytest.mark.django_db def test_async(broker, admin_user): broker.list_key = 'cluster_test:q' + Conf.DB_TIMEOUT = 0 broker.delete_queue() a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result', broker=broker) @@ -237,6 +238,7 @@ def test_async(broker, admin_user): assert fetch(k, 100) is None assert result(k, 100) is None broker.delete_queue() + Conf.DB_TIMEOUT = 60 @pytest.mark.django_db