Improve shutdown cleanliness under uWSGI.

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):

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):

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):

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')
 

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()
 
 

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(