diff --git a/queue_job/jobrunner/__init__.py b/queue_job/jobrunner/__init__.py index ffc6dd7b1..984ab46e2 100644 --- a/queue_job/jobrunner/__init__.py +++ b/queue_job/jobrunner/__init__.py @@ -4,7 +4,6 @@ import logging from threading import Thread -import time from configparser import ConfigParser from odoo.service import server @@ -50,7 +49,7 @@ def __init__(self): def run(self): # sleep a bit to let the workers start at ease - time.sleep(START_DELAY) + self.runner.sleep(START_DELAY) self.runner.run() def stop(self): @@ -80,7 +79,7 @@ def process_work(self): self.runner = QueueJobRunner.from_environ_or_config() self._recover = False _logger.debug("WorkerJobRunner (%s) starting up", self.pid) - time.sleep(START_DELAY) + self.runner.sleep(START_DELAY) self.runner.run() def signal_time_expired_handler(self, n, stack): diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 6ff2e2f26..443ef27b2 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -335,6 +335,7 @@ def __init__( self.channel_manager.simple_configure(channel_config_string) self.db_by_name = {} self._stop = False + self._stop_event = threading.Event() self._stop_pipe = os.pipe() def __del__(self): @@ -486,9 +487,18 @@ def wait_notification(self): continue key.fileobj.poll() + def sleep(self, seconds): + """Wait for ``seconds``, or less if :meth:`stop` is called meanwhile. + + A plain ``time.sleep`` would hold the thread, and so the server + shutdown that joins it, for up to the whole delay. + """ + self._stop_event.wait(seconds) + def stop(self): _logger.info("graceful stop requested") self._stop = True + self._stop_event.set() # wakeup the select() in wait_notification os.write(self._stop_pipe[1], b".") @@ -520,12 +530,12 @@ def run(self): ERROR_RECOVERY_DELAY, ) self.close_databases() - time.sleep(ERROR_RECOVERY_DELAY) + self.sleep(ERROR_RECOVERY_DELAY) except Exception: _logger.exception( "exception: sleeping %ds and retrying", ERROR_RECOVERY_DELAY ) self.close_databases() - time.sleep(ERROR_RECOVERY_DELAY) + self.sleep(ERROR_RECOVERY_DELAY) self.close_databases(remove_jobs=False) _logger.info("stopped") diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 131ce6322..2adbf610d 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -4,10 +4,12 @@ # pylint: disable=odoo-addons-relative-import # we are testing, we want to test as we were an external consumer of the API import os +import threading +from unittest.mock import patch from odoo.tests import BaseCase, tagged -from odoo.addons.queue_job.jobrunner import runner +from odoo.addons.queue_job.jobrunner import START_DELAY, QueueJobRunnerThread, runner from .common import load_doctests @@ -57,3 +59,35 @@ def test_runner_file_closed_write_descriptor(self): self.assertFalse(self._is_open_file_descriptor(read_fd)) self.assertFalse(self._is_open_file_descriptor(write_fd)) + + def test_runner_stop_interrupts_the_recovery_delay(self): + """A runner that lost the master election stops without sleeping it out. + + The server joins the runner thread on shutdown, so a stop waiting for + ERROR_RECOVERY_DELAY held the whole process for up to 5 seconds. + """ + a_runner = runner.QueueJobRunner.from_environ_or_config() + election_lost = threading.Event() + + def initialize_databases(): + election_lost.set() + raise runner.MasterElectionLost("held by another runner") + + with patch.object( + a_runner, "initialize_databases", side_effect=initialize_databases + ): + thread = threading.Thread(target=a_runner.run, daemon=True) + thread.start() + self.addCleanup(thread.join) + self.assertTrue(election_lost.wait(timeout=5)) + a_runner.stop() + thread.join(timeout=runner.ERROR_RECOVERY_DELAY / 2) + self.assertFalse(thread.is_alive()) + + def test_runner_thread_stop_interrupts_the_start_delay(self): + thread = QueueJobRunnerThread() + thread.start() + self.addCleanup(thread.join) + thread.stop() + thread.join(timeout=START_DELAY / 2) + self.assertFalse(thread.is_alive())