pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[FIX] bus: Resume all /longpolling/poll threads on ctrl-c

Browse Source 
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 <Julien00859@users.noreply.github.com>
This commit is contained in:
Julien Castiaux
2021-01-26 16:35:01 +00:00
parent 149cc6c2ed
commit 2d042dd2bd
2 changed files with 37 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/bus/models/bus.py
+15 -1
View File
@@ -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)
odoo/service/server.py
+22 -7
View File
@@ -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)