From 2d042dd2bdf5b6286c3784c3a564599d8c82e1ac Mon Sep 17 00:00:00 2001 From: Julien Castiaux Date: Thu, 14 Jan 2021 10:42:56 +0000 Subject: [PATCH] [FIX] bus: Resume all /longpolling/poll threads on ctrl-c Start odoo in threading mode with bus installed. Login in the browser using any internal user. Make sure the browser call the /longpolling/poll uri. While the browser is waiting for a response, stop the server. The server takes up to 50 seconds to stop. When started in threading mode, a request to /longpolling/poll is served by a casual http thread. It searches for messages enqueued in the bus and returns them. If there are no message for the user in the queue yet, it creates a `threading.Event`, attach it to the user in a shared dictionnary and `wait()` on it with a timeout of 50 seconds (hardcoded value). When the bus thread (the one responsible to listen on the database) receives new messages, it `set()` the events which resume any http thread that was waiting. Because when we stop the server, there is no way to server new requests, there are no way new messages arrive in the bus. All the threads that were waiting for a new message will just wait until the event timeouts which slow down the shutdown of the server. Now we actively `set()` all events in order to resume all those workers when we stop the server. The `ImDispatch.poll` signature has been changed too so it is possible to change (via code) the hardcoded default. The function was using the object referenced by `TIMEOUT` at the time the function was defined, using `timeout None` then `if None: timeout=TIMEOUT` ensures we lookup the variable. closes odoo/odoo#64530 Signed-off-by: Julien Castiaux --- addons/bus/models/bus.py | 16 +++++++++++++++- odoo/service/server.py | 29 ++++++++++++++++++++++------- 2 files changed, 37 insertions(+), 8 deletions(-) diff --git a/addons/bus/models/bus.py b/addons/bus/models/bus.py index b88a833601b..d92a7c17eed 100644 --- a/addons/bus/models/bus.py +++ b/addons/bus/models/bus.py @@ -8,6 +8,7 @@ import threading import time import odoo +import odoo.service.server as servermod from odoo import api, fields, models, SUPERUSER_ID from odoo.tools.misc import DEFAULT_SERVER_DATETIME_FORMAT from odoo.tools import date_utils @@ -100,7 +101,9 @@ class ImDispatch(object): self.channels = {} self.started = False - def poll(self, dbname, channels, last, options=None, timeout=TIMEOUT): + def poll(self, dbname, channels, last, options=None, timeout=None): + if timeout is None: + timeout = TIMEOUT if options is None: options = {} # Dont hang ctrl-c for a poll request, we need to bypass private @@ -170,6 +173,15 @@ class ImDispatch(object): for event in events: event.set() + def wakeup_workers(self): + """ + Wake up all http workers that are waiting for an event, useful + on server shutdown when they can't reveive anymore messages. + """ + for events in self.channels.values(): + for event in events: + event.set() + def run(self): while True: try: @@ -197,3 +209,5 @@ dispatch = None if not odoo.multi_process or odoo.evented: # We only use the event dispatcher in threaded and gevent mode dispatch = ImDispatch() + if servermod.server: + servermod.server.on_stop(dispatch.wakeup_workers) diff --git a/odoo/service/server.py b/odoo/service/server.py index f0628681be3..62f080701bf 100644 --- a/odoo/service/server.py +++ b/odoo/service/server.py @@ -305,6 +305,7 @@ class FSWatcherInotify(FSWatcherBase): class CommonServer(object): def __init__(self, app): self.app = app + self._on_stop_funcs = [] # config self.interface = config['http_interface'] or '0.0.0.0' self.port = config['http_port'] @@ -332,6 +333,19 @@ class CommonServer(object): raise sock.close() + def on_stop(self, func): + """ Register a cleanup function to be executed when the server stops """ + self._on_stop_funcs.append(func) + + def stop(self): + for func in self._on_stop_funcs: + try: + _logger.debug("on_close call %s", func) + func() + except Exception: + _logger.warning("Exception in %s", func.__name__, exc_info=True) + + class ThreadedServer(CommonServer): def __init__(self, app): super(ThreadedServer, self).__init__(app) @@ -501,6 +515,8 @@ class ThreadedServer(CommonServer): if self.httpd: self.httpd.shutdown() + super().stop() + # Manually join() all threads before calling sys.exit() to allow a second signal # to trigger _force_quit() in case some non-daemon threads won't exit cleanly. # threading.Thread.join() should not mask signals (at least in python 2.5). @@ -657,6 +673,7 @@ class GeventServer(CommonServer): def stop(self): import gevent self.httpd.stop() + super().stop() gevent.shutdown() def run(self, preload, stop): @@ -670,9 +687,8 @@ class PreforkServer(CommonServer): dispatcher to will parse the first HTTP request line. """ def __init__(self, app): + super().__init__(app) # config - self.address = config['http_enable'] and \ - (config['http_interface'] or '0.0.0.0', config['http_port']) self.population = config['workers'] self.timeout = config['limit_time_real'] self.limit_request = config['limit_request'] @@ -681,8 +697,6 @@ class PreforkServer(CommonServer): self.cron_timeout = self.timeout # working vars self.beat = 4 - self.app = app - self.pid = os.getpid() self.socket = None self.workers_http = {} self.workers_cron = {} @@ -845,13 +859,13 @@ class PreforkServer(CommonServer): signal.signal(signal.SIGQUIT, dumpstacks) signal.signal(signal.SIGUSR1, log_ormcache_stats) - if self.address: + if config['http_enable']: # listen to socket - _logger.info('HTTP service (werkzeug) running on %s:%s', *self.address) + _logger.info('HTTP service (werkzeug) running on %s:%s', self.interface, self.port) self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.setblocking(0) - self.socket.bind(self.address) + self.socket.bind((self.interface, self.port)) self.socket.listen(8 * self.population) def stop(self, graceful=True): @@ -861,6 +875,7 @@ class PreforkServer(CommonServer): self.long_polling_pid = None if graceful: _logger.info("Stopping gracefully") + super().stop() limit = time.time() + self.timeout for pid in self.workers: self.worker_kill(pid, signal.SIGINT)