From 1ec960bcf66769c682135901cbd4623b6719fdf8 Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Fri, 17 Nov 2017 13:52:24 -0500 Subject: [PATCH] Improve shutdown cleanliness under uWSGI. --- lib/galaxy/web/stack/__init__.py | 15 +++++++++++---- lib/galaxy/web/stack/message.py | 5 ++++- lib/galaxy/web/stack/transport.py | 8 ++------ lib/galaxy/webapps/galaxy/buildapp.py | 2 +- scripts/galaxy-main | 13 +++++++++++++ scripts/get_uwsgi_args.py | 4 +++- 6 files changed, 34 insertions(+), 13 deletions(-) diff --git a/lib/galaxy/web/stack/__init__.py b/lib/galaxy/web/stack/__init__.py index 7476afec21b..44a58b882d4 100644 --- a/lib/galaxy/web/stack/__init__.py +++ b/lib/galaxy/web/stack/__init__.py @@ -72,6 +72,7 @@ class ApplicationStack(object): def __init__(self, app=None, config=None): self.app = app self.config = config or (app and app.config) + self.running = False def start(self): # TODO: with a stack config the pools could be parsed here @@ -127,7 +128,9 @@ class MessageApplicationStack(ApplicationStack): def start(self): super(MessageApplicationStack, self).start() - self.transport.start() + if not self.running: + self.transport.start() + self.running = True def register_message_handler(self, func, name=None): self.dispatcher.register_func(func, name) @@ -148,7 +151,10 @@ class MessageApplicationStack(ApplicationStack): self.transport.send_message(msg.encode(), dest) def shutdown(self): - self.transport.shutdown() + if self.running: + log.info('Application stack interface shutting down') + self.transport.shutdown() + self.running = False class UWSGIApplicationStack(MessageApplicationStack): @@ -247,8 +253,9 @@ class UWSGIApplicationStack(MessageApplicationStack): def start(self): # Does a generalized `is_worker` attribute make sense? Hard to say w/o other stack paradigms. - if self._is_mule: - self.__register_signal_handlers() + if self._is_mule and self._farm_name: + # used by main.py to send a shutdown message on termination + os.environ['_GALAXY_UWSGI_FARM_NAME'] = self._farm_name super(UWSGIApplicationStack, self).start() def has_pool(self, pool_name): diff --git a/lib/galaxy/web/stack/message.py b/lib/galaxy/web/stack/message.py index e972572133b..83fac262791 100644 --- a/lib/galaxy/web/stack/message.py +++ b/lib/galaxy/web/stack/message.py @@ -25,7 +25,10 @@ class ApplicationStackMessageDispatcher(object): def deregister_func(self, func=None, name=None): name = self.__func_name(func, name) - del self.__funcs[name] + try: + del self.__funcs[name] + except KeyError: + pass @property def handler_count(self): diff --git a/lib/galaxy/web/stack/transport.py b/lib/galaxy/web/stack/transport.py index c24ab1ef756..14226eb728b 100644 --- a/lib/galaxy/web/stack/transport.py +++ b/lib/galaxy/web/stack/transport.py @@ -58,6 +58,7 @@ class ApplicationStackTransport(object): def shutdown(self): self.running = False if self.dispatcher_thread: + log.info('Joining application stack transport dispatcher thread') self.dispatcher_thread.join() self.dispatcher_thread = None @@ -143,12 +144,7 @@ class UWSGIFarmMessageTransport(ApplicationStackTransport): super(UWSGIFarmMessageTransport, self).start() def shutdown(self): - if not self.stack._is_mule: - for farm in self.stack._configured_farms.keys(): - for mule in self.stack._configured_mules: - # this could possibly generate more than we need, but that's ok - self.send_message(self.SHUTDOWN_MSG, farm) - else: + if self.stack._is_mule: super(UWSGIFarmMessageTransport, self).shutdown() def send_message(self, msg, dest): diff --git a/lib/galaxy/webapps/galaxy/buildapp.py b/lib/galaxy/webapps/galaxy/buildapp.py index 4a993dcc899..00fdf9254a5 100644 --- a/lib/galaxy/webapps/galaxy/buildapp.py +++ b/lib/galaxy/webapps/galaxy/buildapp.py @@ -53,7 +53,7 @@ def app_factory(global_conf, load_app_kwds={}, **kwargs): sys.exit(1) # Call app's shutdown method when the interpeter exits, this cleanly stops # the various Galaxy application daemon threads - atexit.register(app.shutdown) + app.application_stack.register_postfork_function(atexit.register, app.shutdown) # Create the universe WSGI application webapp = GalaxyWebApplication(app, session_cookie='galaxysession', name='galaxy') diff --git a/scripts/galaxy-main b/scripts/galaxy-main index 8e3fea4f83f..b2086995946 100755 --- a/scripts/galaxy-main +++ b/scripts/galaxy-main @@ -42,6 +42,11 @@ try: except ImportError: Daemonize = None +try: + import uwsgi +except: + uwsgi = None + REQUIRES_DAEMONIZE_MESSAGE = "Attempted to use Galaxy in daemon mode, but daemonize is unavailable." log = logging.getLogger(__name__) @@ -58,6 +63,8 @@ DEFAULT_VERBOSE = True DESCRIPTION = "Daemonized entry point for Galaxy." SHUTDOWN_MSG = '__SHUTDOWN__' +UWSGI_FARM_VAR = '_GALAXY_UWSGI_FARM_NAME' + exit = threading.Event() @@ -96,6 +103,12 @@ def load_galaxy_app( def handle_signal(signum, frame): log.info('Received signal %d, exiting', signum) + if uwsgi and 'mule_id' in dir(uwsgi) and uwsgi.mule_id() > 0: + farm = os.environ.get(UWSGI_FARM_VAR, None) + if farm: + uwsgi.farm_msg(farm, SHUTDOWN_MSG) + else: + uwsgi.mule_msg(SHUTDOWN_MSG, uwsgi.mule_id()) exit.set() diff --git a/scripts/get_uwsgi_args.py b/scripts/get_uwsgi_args.py index 20cc87b569b..05fc5e3a444 100644 --- a/scripts/get_uwsgi_args.py +++ b/scripts/get_uwsgi_args.py @@ -22,7 +22,7 @@ ALIASES = { 'http': ('httprouter', 'socket', 'uwsgi-socket', 'suwsgi-socket', 'ssl-socket'), } DEFAULT_ARGS = { - '_all_': ('virtualenv', 'pythonpath', 'master', 'threads', 'http', 'static-map', 'die-on-term', 'enable-threads'), + '_all_': ('virtualenv', 'pythonpath', 'master', 'threads', 'http', 'static-map', 'die-on-term', 'hook-master-start', 'enable-threads'), 'galaxy': ('py-call-osafterfork', 'mule', 'farm'), 'reports': (), 'tool_shed': (), @@ -97,6 +97,8 @@ def _get_uwsgi_args(cliargs, kwargs): '/static={here}/static'.format(here=os.getcwd())), 'die-on-term': True, 'enable-threads': True, + 'hook-master-start': ('unix_signal:2 gracefully_kill_them_all', + 'unix_signal:15 gracefully_kill_them_all'), 'py-call-osafterfork': True, 'mule': ('=lib/galaxy/main.py',) * handlerct, 'farm': '={name}:{mules}'.format(