WIP: drop stack configurability, this will come later. Cleanup for WIP

PR

config/galaxy.ini.sample

@@ -1228,6 +1228,9 @@ use_interactive = True
 # be configured through the job configuration file.
 #job_config_file = config/job_conf.xml
 
+### TODO: These options will move to a stack config section in the YAML config
+### file
+
 # Some application stacks that Galaxy run under (such as uWSGI) can start
 # arbitrary Galaxy server processes to perform functions asynchronously from
 # the Galaxy web server process(es). Such stacks may give you the option to

lib/galaxy/app.py

@@ -191,9 +191,6 @@ class UniverseApplication(object, config.ConfiguresGalaxyMixin):
         from galaxy.jobs import manager
         self.job_manager = manager.JobManager(self)
         self.application_stack.register_postfork_function(self.job_manager.start)
-        # FIXME: These are exposed directly for backward compatibility
-        #self.job_queue = self.job_manager.job_queue
-        #self.job_stop_queue = self.job_manager.job_stop_queue
         self.proxy_manager = ProxyManager(self.config)
         # Initialize the external service types
         self.external_service_types = external_service_types.ExternalServiceTypesCollection(

lib/galaxy/jobs/__init__.py

@@ -166,15 +166,6 @@ class JobConfiguration(object, ConfiguresHandlers):
         except Exception as e:
             raise config_exception(e, job_config_file)
 
-    def __config_default_handlers(self):
-        """This is a stop-gap solution until the job conf loading is rewritten
-        for YAML job conf support.
-        """
-        # FIXME: this will not support the case where uWSGI is being used w/o handler mules
-        if self.app.application_stack.handle_jobs:
-            self.handlers[self.app.config.server_name] = (self.app.config.server_name,)
-            self.default_handler_id = self.app.config.server_name
-
     def __parse_job_conf_xml(self, tree):
         """Loads the new-style job configuration from options in the job config file (by default, job_conf.xml).
 
@@ -212,13 +203,18 @@ class JobConfiguration(object, ConfiguresHandlers):
         handlers_conf = root.find('handlers')
         self._init_handlers(handlers_conf)
 
-        # Must define at least one handler to have a default.
-        if not self.handlers:
-            self.__config_default_handlers()
-            log.info("No handlers defined or empty handlers group, will use default handlers: %s", self.handlers)
-        else:
-            # Determine the default handler(s)
-            self.default_handler_id = self.__get_default(handlers, list(self.handlers.keys()))
+        # Determine the default handler(s)
+        try:
+            self.default_handler_id = self._get_default(self.app.config, handlers_conf, list(self.handlers.keys()))
+        except:
+            log.info('No default handler specified in job config, will check for job handlers managed by application stack')
+        if ((self.default_handler_id is None
+             or (len(self.handlers) == 1 and self.app.config.server_name.startswith(self.handlers.keys()[0])))
+            and self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS)):
+            # Shortcut for compatibility with existing job confs that use the default handlers block
+            # There are no defined handlers or there's only one handler and it's this server, and the stack has a handler pool
+            self.handlers = {}
+            self.default_handler_id = None
 
         # Parse destinations
         destinations = root.find('destinations')
@@ -352,7 +348,14 @@ class JobConfiguration(object, ConfiguresHandlers):
             self.runner_plugins.append(dict(id=runner, load=runner, workers=self.app.config.cluster_job_queue_workers))
 
         # Set the handlers
-        self.__config_default_handlers()
+        if self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS):
+            self.default_handler_id = None
+        else:
+            for id in self.app.config.job_handlers:
+                self.handlers[id] = (id,)
+
+            self.handlers['default_job_handlers'] = self.app.config.default_job_handlers
+            self.default_handler_id = 'default_job_handlers'
 
         # Set tool handler configs
         for id, tool_handlers in self.app.config.tool_handlers.items():

lib/galaxy/jobs/manager.py

@@ -23,27 +23,19 @@ class JobManager(object):
 
     def __init__(self, app):
         self.app = app
-        self.job_handler = NoopHandler()
-        self.job_stop_queue = NoopQueue()
-        if app.application_stack.setup_jobs_with_msg:
-            # defer setup to postfork
-            log.debug('######### registering manager init function')
-            self.app.application_stack.register_postfork_function(self.init)
-        else:
-            self.init()
-
-    def init(self):
-        log.debug("Initializing job manager interface")
+        self.job_lock = False
         if self.app.is_job_handler():
             log.debug("Starting job handler")
-            self.job_handler = handler.JobHandler(self.app)
+            self.job_handler = handler.JobHandler(app)
             self.job_stop_queue = self.job_handler.job_stop_queue
-        elif self.app.application_stack.setup_jobs_with_msg:
-            # not a handler, but notification is via the application stack
-            self.job_handler = MessageJobHandler( self.app )
+        elif app.application_stack.has_pool(app.application_stack.pools.JOB_HANDLERS):
+            log.debug("Initializing job handler messaging interface")
+            self.job_handler = MessageJobHandler(app)
+            self.job_stop_queue = NoopQueue()
+        else:
+            self.job_handler = NoopHandler()
             self.job_stop_queue = NoopQueue()
         self.job_queue = self.job_handler.job_queue
-        self.job_lock = False
 
     def start(self):
         self.job_handler.start()
@@ -64,21 +56,19 @@ class NoopHandler(object):
         pass
 
 
-class MessageJobHandler( object ):
+class MessageJobHandler(NoopHandler):
     """
     Implements the JobHandler interface but just to send setup messages on startup
 
     TODO: It should be documented that starting two Galaxy uWSGI master processes simultaneously would result in a race condition that *could* cause two handlers to pick up the same job.
 
-    The recommended config for now will be webless/main handlers if running more than one uWSGI (master) process
+    The recommended config for now will be webless handlers if running more than one uWSGI (master) process
     """
     def __init__(self, app):
+        # This runs in the web (main) process pre-fork
         self.app = app
         self.job_queue = MessageJobQueue(app)
         self.job_stop_queue = NoopQueue()
-
-    def start(self):
-        # This runs in the web (main) process pre-fork
         jobs_at_startup = self.app.model.context.query(Job).enable_eagerloads(False) \
             .filter((Job.state == Job.states.NEW) & (Job.handler == null())).all()
         if jobs_at_startup:
@@ -86,11 +76,8 @@ class MessageJobHandler( object ):
         for job in jobs_at_startup:
             self.job_queue.put(job.id, job.tool_id)
 
-    def shutdown(self, *args):
-        pass
 
-
-class MessageJobQueue(object):
+class MessageJobQueue(NoopQueue):
     """
     Implements the JobQueue / JobStopQueue interface but only sends messages to the actual job queue
     """
@@ -99,11 +86,4 @@ class MessageJobQueue(object):
 
     def put(self, job_id, tool_id):
         msg = JobHandlerMessage(task='setup', job_id=job_id)
-        # TODO: send to a specific pool
-        self.app.application_stack.send_message(self.app.application_stack.purposes.JOB_HANDLER, msg)
-
-    def put_stop(self, *args):
-        pass
-
-    def shutdown(self):
-        pass
+        self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg)

lib/galaxy/util/configdict.py

@@ -1,39 +0,0 @@
-"""Generic config parsing into dictionary.
-"""
-import errno
-import logging
-
-try:
-    import yaml
-except ImportError:
-    yaml = None
-
-
-log = logging.getLogger(__name__)
-
-
-def parse_config(config_file, root, default=None):
-    if default:
-        conf = default.copy()
-    else:
-        conf = {}
-    try:
-        load_func = __load_func(config_file)
-        with open(config_file) as fh:
-            c = load_func(fh)
-            if not c:
-                c = {}
-            conf.update(c.get(root, {}))
-    except (OSError, IOError) as exc:
-        if exc.errno == errno.ENOENT:
-            log.warning("config file '%s' does not exist, running with default config", config_file)
-        else:
-            raise
-    return conf
-
-
-def __load_func(path):
-    if path.endswith('yaml') or path.endswith('.yml'):
-        if not yaml:
-            raise RuntimeError("The 'yaml' module could not be imported, please install PyYAML to read '%s'" % path)
-        return yaml.load

lib/galaxy/util/handlers.py

@@ -101,12 +101,9 @@ class ConfiguresHandlers:
 
         :return: bool
         """
-        if self.app.application_stack.has_purpose(self.app.application_stack.purposes.JOB_HANDLER):
+        if self.app.application_stack.in_pool(self.app.application_stack.pools.JOB_HANDLERS):
             # Handlers started as uWSGI mules do not require configuration in the job conf
             return True
-        #if stack_handles_jobs is not None:
-        #    # Handlers started as uWSGI mules do not require configuration in the job conf
-        #    return stack_handles_jobs
         for collection in self.handlers.values():
             if server_name in collection:
                 return True

lib/galaxy/web/stack/__init__.py

@@ -17,63 +17,17 @@ try:
 except ImportError:
     uwsgi = None
 
-#try:
-#    from uwsgidecorators import postfork as uwsgi_postfork
-#except:
-#    uwsgi_postfork = lambda x: x    # noqa: E731
-#    if uwsgi is not None and hasattr(uwsgi, 'numproc'):
-#        print("WARNING: This is a uwsgi process but the uwsgidecorators library"
-#              " is unavailable.  This is likely due to using an external (not"
-#              " in Galaxy's virtualenv) uwsgi and you may experience errors. "
-#              "HINT:\n  {venv}/bin/pip install uwsgidecorators".format(
-#                  venv=os.environ.get('VIRTUAL_ENV', '/path/to/venv')))
-
 from six import string_types
 
-from .message import ApplicationStackMessage, JobHandlerMessage, decode
+from .message import ApplicationStackMessage, ApplicationStackMessageDispatcher, JobHandlerMessage, decode
 from .transport import ApplicationStackTransport, UWSGIFarmMessageTransport
 
 from galaxy.util.bunch import Bunch
-from galaxy.util.configdict import parse_config
 
 
 log = logging.getLogger(__name__)
 
 
-class ApplicationStackMessageDispatcher(object):
-    def __init__(self):
-        self.__funcs = {}
-
-    def __func_name(self, func, name):
-        if not name:
-            name = func.__name__
-        return name
-
-    def register_func(self, func, name=None):
-        name = self.__func_name(func, name)
-        self.__funcs[name] = func
-
-    def deregister_func(self, func=None, name=None):
-        name = self.__func_name(func, name)
-        del self.__funcs[name]
-
-    @property
-    def handler_count(self):
-        return len(self.__funcs)
-
-    def dispatch(self, msg_str):
-        msg = decode(msg_str)
-        try:
-            msg.validate()
-        except AssertionError as exc:
-            log.error('######## Invalid message received: %s, error: %s', msg_str, exc)
-            return
-        if msg.target not in self.__funcs:
-            log.error("######## Received message with target '%s' but no functions were registered with that name. Params were: %s", msg.target, msg.params)
-        else:
-            self.__funcs[msg.target](msg)
-
-
 class ApplicationStackLogFilter(logging.Filter):
     def filter(self, record):
         record.worker_id = None
@@ -91,97 +45,47 @@ class UWSGILogFilter(logging.Filter):
 class ApplicationStack(object):
     name = None
     prohibited_middleware = frozenset()
-    # TODO: this is a fairly clunky way of handling these cases
-    setup_jobs_with_msg = False # used in galaxy.jobs.manager to determine whether jobs should be sent via message
-    handle_jobs = False         # used by galaxy.jobs to determine whether this process handles jobs
-
     transport_class = ApplicationStackTransport
     log_filter_class = ApplicationStackLogFilter
+    # TODO: this belongs in the pool configuration
+    server_name_template = '{server_name}'
 
-    purposes = Bunch(
-        WEB_WORKER = 'web-worker',
-        JOB_HANDLER = 'job-handler',
+    # used both to route jobs to a pool with this name and indicate whether or
+    # not a stack is using messaging for handler assignment
+    pools = Bunch(
+        JOB_HANDLERS = 'job-handlers',
     )
 
-    default_conf = {
-        'processes': 1,
-        'threads': 4,
-        'server_name': '{server_name}',
-        'workers': [],
-        'pools': [],
-    }
-
-    web_worker_pool = {
-        'name': 'web-workers',
-        'purpose': 'web-worker',
-    }
-
     @classmethod
     def register_postfork_function(cls, f, *args, **kwargs):
         f(*args, **kwargs)
 
     def __init__(self, app=None):
         self.app = app
-        # FIXME: hardcoded path
-        self._conf = parse_config('config/stack_conf.yml', 'stack', default=self.default_conf)
-        self.server_name_template = '{server_name}'
-        self.pools = []
-        self.pool_names = []
-        self._purposes = []
-
-    def _postfork_init(self):
-        self.pools = []
-        self.pool_names = []
-        self._purposes = []
-        for pool in self._conf.get('pools', []) + [ApplicationStack.web_worker_pool]:
-            if self.in_pool(pool['name']):
-                self.pools.append(pool)
-                self.pool_names.append(pool['name'])
-                self._purposes.append(pool['purpose'])
-        for pool in self.pools:
-            if pool.get('server_name', None):
-                self.server_name_template = pool['server_name']
-                break
-        else:
-            if self._conf.get('server_name', None):
-                # default if the process is not in a pool and there is an unpooled server name set in the conf
-                self.server_name_template = self._conf['server_name']
-            if self.pools:
-                # default if the process is in a pool
-                self.server_name_template = '{server_name}.{pool_name}.{process_num}'
 
     def start(self):
-        self._postfork_init()
-
-    @property
-    def pool_name(self):
-        # TODO: in the future, allow for mapping job handlers in the job conf by something other than server_name, such
-        # as pool name and process number. but for now, the job-handlers pool should be the first one specified under a
-        # worker set's 'pools' list, if more than one pool is specified for a given worker set
-        try:
-            return self.pool_names[0]
-        except:
-            return None
-
-    def in_pool(self, pool_name):
-        return None
-
-    def has_purpose(self, purpose):
-        return purpose in self._purposes
-
-    def workers(self):
-        return []
-
-
-
-
-
+        # TODO: with a stack config the pools could be parsed here
+        pass
 
     def allowed_middleware(self, middleware):
         if hasattr(middleware, '__name__'):
             middleware = middleware.__name__
         return middleware not in self.prohibited_middleware
 
+    def workers(self):
+        return []
+
+    @property
+    def pool_name(self):
+        # TODO: ideally jobs would be mappable to handlers by pool name
+        return None
+
+    def has_pool(self, pool_name):
+        return False
+
+    def in_pool(self, pool_name):
+        return False
+
     def set_server_name(self, app, caller_tmpl_dict):
         tmpl_dict = {
             'server_name': app.config.server_name,
@@ -193,10 +97,10 @@ class ApplicationStack(object):
         }
         tmpl_dict.update(caller_tmpl_dict)
         app.config.server_name = self.server_name_template.format(**tmpl_dict)
-        log.debug('######## server_name is: %s', app.config.server_name)
+        log.debug('server_name set to: %s', app.config.server_name)
 
     def set_postfork_server_name(self, app):
-        self.set_server_name(app, tmpl_dict, {})
+        self.set_server_name(app, {})
 
     def register_message_handler(self, func, name=None):
         pass
@@ -207,13 +111,6 @@ class ApplicationStack(object):
     def send_message(self, dest, msg=None, target=None, params=None, **kwargs):
         pass
 
-    def send_purpose_message(self, purpose, **kwargs):
-        for pool in self.pools:
-            if purpose == pool['purpose']:
-                return self.send_message(pool['name'], **kwargs)
-        else:
-            raise RuntimeError('No pools defined for purpose: %s', purpose)
-
     def shutdown(self):
         pass
 
@@ -223,9 +120,6 @@ class MessageApplicationStack(ApplicationStack):
         super(MessageApplicationStack, self).__init__(app=app)
         self.dispatcher = ApplicationStackMessageDispatcher()
         self.transport = self.transport_class(app, stack=self, dispatcher=self.dispatcher)
-        #if app:
-        #    log.debug('######## registering self.start')
-        #    self.register_postfork_function(self.start)
 
     def start(self):
         super(MessageApplicationStack, self).start()
@@ -262,28 +156,11 @@ class UWSGIApplicationStack(MessageApplicationStack):
         'wrap_in_static',
         'EvalException',
     ])
-    setup_jobs_with_msg = True
-
     transport_class = UWSGIFarmMessageTransport
     log_filter_class = UWSGILogFilter
-    # FIXME: this is copied into UWSGIFarmMessageTransport
-    shutdown_msg = '__SHUTDOWN__'
-    postfork_functions = []
+    server_name_template = '{server_name}.{id}'
 
-    default_conf = {
-        'processes': 1,
-        'threads': 4,
-        'server_name': '{server_name}.{process_num}',
-        'workers': [{
-            'load': 'lib/galaxy/main.py',
-            'processes': 1,
-            'pools': ['job-handlers']
-        }],
-        'pools': [{
-            'name': 'job-handlers',
-            'purpose': 'job-handler',
-        }]
-    }
+    postfork_functions = []
 
     @classmethod
     def register_postfork_function(cls, f, *args, **kwargs):
@@ -291,50 +168,42 @@ class UWSGIApplicationStack(MessageApplicationStack):
             cls.postfork_functions.append((f, args, kwargs))
         else:
             # mules are forked from the master and run the master's postfork functions immediately before the forked
-            # process is replaced. that is prevented in the _do_uwsgi_postfork function, and because mules are
-            # standalone non-forking processes, they should run postfork functions immediately
+            # process is replaced. that is prevented in the _do_uwsgi_postfork function, and because programmed mules
+            # are standalone non-forking processes, they should run postfork functions immediately
             f(*args, **kwargs)
 
     def __init__(self, app=None):
-        super(UWSGIApplicationStack, self).__init__(app=app)
         self._farms_dict = None
         self._mules_list = None
-        self._is_mule = None
+        super(UWSGIApplicationStack, self).__init__(app=app)
 
     def __register_signal_handlers(self):
         for name in ('TERM', 'INT', 'HUP'):
-            log.debug('######## registered signal handler for SIG%s', name)
             sig = getattr(signal, 'SIG%s' % name)
             signal.signal(sig, self._handle_signal)
 
     def _handle_signal(self, signum, frame):
-        if signum == signal.SIGTERM:
-            log.info('######## Mule %s received SIGTERM, shutting down gracefully', uwsgi.mule_id())
-            self.shutdown()
-        elif signum == signal.SIGINT:
-            log.info('######## Mule %s received SIGINT, shutting down immediately', uwsgi.mule_id())
-            self.shutdown()
-            # This terminates the application loop in the handler entrypoint
-            self.app.exit = True
+        # uWSGI always sends SIGINT even if the master received SIGTERM
+        if signum in (signal.SIGTERM, signal.SIGINT):
+            log.info('Received SIGTERM/SIGINT, shutting down gracefully')
         elif signum == signal.SIGHUP:
-            log.debug('######## Mule %s received SIGHUP, restarting', uwsgi.mule_id())
-            self.shutdown()
-            # uWSGI master will restart us
-            self.app.exit = True
+            log.debug('Received SIGHUP, restarting')
+        self.shutdown()
+        # this terminates the application loop in the mule script, in the case of HUP, uWSGI will restart the mule
+        self.app.exit = True
 
-    def in_pool(self, pool_name):
-        if uwsgi.worker_id() > 0 and pool_name == ApplicationStack.web_worker_pool['name']:
-            return True
-        else:
-            return self._in_farm(pool_name)
-            #return self.transport._farm_name == pool_name
-
-    def workers(self):
-        return uwsgi.workers()
-
-    # FIXME: these are copied into UWSGIFarmMessageTransport
     @property
-    def _farms(self):
+    def _configured_mules(self):
+        if self._mules_list is None:
+            self._mules_list = _uwsgi_configured_mules()
+        return self._mules_list
+
+    @property
+    def _is_mule(self):
+        return uwsgi.mule_id() > 0
+
+    @property
+    def _configured_farms(self):
         if self._farms_dict is None:
             self._farms_dict = {}
             farms = uwsgi.opt.get('farm', [])
@@ -345,57 +214,54 @@ class UWSGIApplicationStack(MessageApplicationStack):
         return self._farms_dict
 
     @property
-    def _mules(self):
-        if self._mules_list is None:
-            self._mules_list = []
-            mules = uwsgi.opt.get('mule', [])
-            self._mules_list = [mules] if isinstance(mules, string_types) or mules is True else mules
-        return self._mules_list
+    def _farms(self):
+        farms = []
+        for farm, mules in self._configured_farms.items():
+            if uwsgi.mule_id() in mules:
+                farms.append(farm)
+        return farms
 
-    def _in_farm(self, farm_name):
-        return uwsgi.mule_id() in self._farms.get(farm_name, [])
+    @property
+    def _farm_name(self):
+        try:
+            return self._farms[0]
+        except IndexError:
+            return None
 
     def start(self):
-        self._is_mule = uwsgi.mule_id() > 0
+        # Does a generalized `is_worker` attribute make sense? Hard to say w/o other stack paradigms.
         if self._is_mule:
             self.__register_signal_handlers()
         super(UWSGIApplicationStack, self).start()
 
+    def has_pool(self, pool_name):
+        return pool_name in self._configured_farms
+
+    def in_pool(self, pool_name):
+        if not self._is_mule:
+            return False
+        else:
+            return pool_name in self._farms
+
+    def workers(self):
+        return uwsgi.workers()
+
     def set_postfork_server_name(self, app):
         tmpl_dict = {
             'id': 'worker%s' % uwsgi.worker_id() if uwsgi.mule_id() == 0 else 'mule%s' % uwsgi.mule_id(),
         }
-        if uwsgi.mule_id() == 0:
-            tmpl_dict['process_num'] = uwsgi.worker_id()
-        elif self.pool_name in self._farms:
-            tmpl_dict['process_num'] = self._farms[self.pool_name].index(uwsgi.mule_id()) + 1
-            log.debug('######## self._farms[self.pool_name]: %s', self._farms[self.pool_name])
         self.set_server_name(app, tmpl_dict)
 
-    #@property
-    #def pool_name(self):
-    #    # could get this from self._conf, or could get it from uwsgi.opts
-    #    return self.transport._farm_name
-
     def shutdown(self):
-        log.debug('######## STACK SHUTDOWN CALLED')
         super(UWSGIApplicationStack, self).shutdown()
-        # FIXME: blech
-        if not self._is_mule:
-            for farm in self._farms:
-                for mule in self._mules:
-                    # This will possibly generate more than we need, but that's ok
-                    self.transport.send_message(self.shutdown_msg, farm)
 
 
 class PasteApplicationStack(ApplicationStack):
     name = 'Python Paste'
-    handle_jobs = True
 
 
 class WeblessApplicationStack(ApplicationStack):
     name = 'Webless'
-    handle_jobs = True
 
 
 def application_stack_class():
@@ -425,22 +291,19 @@ def register_postfork_function(f, *args, **kwargs):
     application_stack_class().register_postfork_function(f, *args, **kwargs)
 
 
-def _mules():
+def _uwsgi_configured_mules():
     mules = uwsgi.opt.get('mule', [])
     return [mules] if isinstance(mules, string_types) or mules is True else mules
 
 
-#@uwsgi_postfork
 def _do_uwsgi_postfork():
-    import os
-    # FIXME: _mules duplicated again
-    for i, mule in enumerate(_mules()):
+    for i, mule in enumerate(_uwsgi_configured_mules()):
         if mule is not True and i + 1 == uwsgi.mule_id():
             # mules will inherit the postfork function list and call them immediately upon fork, but programmed mules
             # should not do that (they will call the postfork functions in-place as they start up after exec())
             UWSGIApplicationStack.postfork_functions = []
-    log.debug('######## postfork called, pid %s mule %s functions are: %s' % (os.getpid(), uwsgi.mule_id(), UWSGIApplicationStack.postfork_functions))
     for f, args, kwargs in [t for t in UWSGIApplicationStack.postfork_functions]:
+        log.debug('Calling postfork function: %s', f)
         f(*args, **kwargs)
 
 

lib/galaxy/web/stack/message.py

@@ -9,45 +9,52 @@ import logging
 log = logging.getLogger(__name__)
 
 
-class ApplicationStackMessageEncoder(json.JSONEncoder):
-    def encode(self, o):
-        if isinstance(o, AttributeDict):
-            return o.serialize(self)
-        return json.JSONEncoder.encode(self, o)
+class ApplicationStackMessageDispatcher(object):
+    def __init__(self):
+        self.__funcs = {}
+
+    def __func_name(self, func, name):
+        if not name:
+            name = func.__name__
+        return name
+
+    def register_func(self, func, name=None):
+        name = self.__func_name(func, name)
+        self.__funcs[name] = func
+
+    def deregister_func(self, func=None, name=None):
+        name = self.__func_name(func, name)
+        del self.__funcs[name]
+
+    @property
+    def handler_count(self):
+        return len(self.__funcs)
+
+    def dispatch(self, msg_str):
+        msg = decode(msg_str)
+        try:
+            msg.validate()
+        except AssertionError as exc:
+            log.error('Invalid message received: %s, error: %s', msg_str, exc)
+            return
+        if msg.target not in self.__funcs:
+            log.error("Received message with target '%s' but no functions were registered with that name. Params were: %s", msg.target, msg.params)
+        else:
+            self.__funcs[msg.target](msg)
 
 
 class ApplicationStackMessage(dict):
+    target = None
+
     def __init__(self, target=None, params=None, **kwargs):
-        """Any extra kwargs override values in params
-        """
-        self['target'] = target
+        self['target'] = target or self.__class__.target
         self['params'] = params or {}
         for k, v in kwargs.items():
             self['params'][k] = v
 
-    #@classmethod
-    #def from_string(cls, s)
-    #    #kwargs['object_hook'] = ApplicationStackMessage
-    #    d = json.loads(s)
-    #    return cls(json.loads(s, *args, **kwargs))
-
     def validate(self):
         assert self['target'] is not None, "Missing 'target' parameter"
 
-    def serialize(self, encoder):
-        #return json.JSONEncoder.encode(encoder, ApplicationStackMessage.clean(self))
-        self.validate()
-        return json.JSONEncoder.encode(encoder, self)
-
-    def dumps(self, *args, **kwargs):
-        #kwargs['cls'] = AttributeStackMessageEncoder
-        return json.dumps(self, *args, **kwargs)
-
-    def dump(self, fp, *args, **kwargs):
-        #raise Exception("This isn't using the Encoder, what gives?")
-        #kwargs['cls'] = AttributeStackMessageEncoder
-        return json.dump(self, fp, *args, **kwargs)
-
     @property
     def target(self):
         return self['target']
@@ -64,23 +71,15 @@ class ApplicationStackMessage(dict):
     def set_params(self, params):
         self['params'] = params
 
-    #property
-    #def class(self):
-    #    return globals()[self['cls']]
-
     def encode(self):
         self['__classname__'] = self.__class__.__name__
         return json.dumps(self)
 
 
-# when we add additional messages this should become generalized and subclassed
+# TODO: when additional messages are added we should refactor and improve validation and param/task separation (but not all msgs may use task)
 class JobHandlerMessage(ApplicationStackMessage):
     target = 'job_handler'
 
-    def __init__(self, target=None, params=None, **kwargs):
-        super(JobHandlerMessage, self).__init__(params=params, **kwargs)
-        self['target'] = self.target
-
     def validate(self):
         super(JobHandlerMessage, self).validate()
         for param in ('task', 'job_id'):

lib/galaxy/web/stack/transport.py

@@ -4,28 +4,18 @@ from __future__ import absolute_import
 
 import logging
 import threading
-try:
-    from queue import Empty, Queue
-except ImportError:
-    from Queue import Empty, Queue
 
 try:
     import uwsgi
 except ImportError:
     uwsgi = None
 
-from six import string_types
-
 
 log = logging.getLogger(__name__)
 
 
 class ApplicationStackTransport(object):
-    shutdown_msg = '__SHUTDOWN__'
-
-    def __init_dispatcher_thread(self):
-        self.dispatcher_thread = threading.Thread(name=self.__class__.__name__ + ".dispatcher_thread", target=self._dispatch_messages)
-        #self.dispatcher_thread.daemon = True
+    SHUTDOWN_MSG = '__SHUTDOWN__'
 
     def __init__(self, app, stack, dispatcher=None):
         """ Pre-fork initialization.
@@ -36,93 +26,64 @@ class ApplicationStackTransport(object):
         self.running = False
         self.dispatcher = dispatcher
         self.dispatcher_thread = None
-        self.__init_dispatcher_thread()
 
     def _dispatch_messages(self):
         pass
 
     def start_if_needed(self):
         # Don't unnecessarily start a thread that we don't need.
-        if self.can_run and not self.running and not self.dispatcher_thread.is_alive() and self.dispatcher and self.dispatcher.handler_count:
+        if self.can_run and not self.running and not self.dispatcher_thread and self.dispatcher and self.dispatcher.handler_count:
             self.running = True
+            self.dispatcher_thread = threading.Thread(name=self.__class__.__name__ + ".dispatcher_thread", target=self._dispatch_messages)
             self.dispatcher_thread.start()
-            log.debug('######## Web stack IPC message dispatcher thread started in mule %s', uwsgi.mule_id())
+            log.info('%s dispatcher started', self.__class__.__name__)
 
     def stop_if_unneeded(self):
-        if self.can_run and self.running and self.dispatcher_thread.is_alive() and self.dispatcher and not self.dispatcher.handler_count:
+        if self.can_run and self.running and self.dispatcher_thread and self.dispatcher and not self.dispatcher.handler_count:
             self.running = False
             self.dispatcher_thread.join()
-            self.__init_dispatcher_thread()
+            self.dispatcher_thread = None
+            log.info('%s dispatcher stopped', self.__class__.__name__)
 
     def start(self):
         """ Post-fork initialization.
         """
         self.can_run = True
-        log.debug('######## start called')
         self.start_if_needed()
 
     def send_message(self, msg, dest):
         pass
 
     def shutdown(self):
-        log.debug('######## TRANSPORT SHUTDOWN CALLED')
         self.running = False
-        if self.dispatcher_thread.is_alive():
-            # FIXME
-            #self.send_message(self.shutdown_msg, 'job-handlers')
+        if self.dispatcher_thread:
             self.dispatcher_thread.join()
-            log.debug('######## Joined dispatcher thread')
+            self.dispatcher_thread = None
 
 
 class UWSGIFarmMessageTransport(ApplicationStackTransport):
     """ Communication via uWSGI Mule Farm messages. Communication is unidirectional (workers -> mules).
     """
-    # FIXME: do you need a clear "this is a producer, this is a consumer flag?
-    #shutdown_msg = '__SHUTDOWN__'
     # Define any static lock names here, additional locks will be appended for each configured farm's message handler
     _locks = []
 
     def __initialize_locks(self):
         num = int(uwsgi.opt.get('locks', 0)) + 1
-        farms = self._farms.keys()
+        farms = self.stack._configured_farms.keys()
         need = len(farms)
-        #need = len(filter(lambda x: x.startswith('LOCK_'), dir(self)))
         if num < need:
             raise RuntimeError('Need %i uWSGI locks but only %i exist(s): Set `locks = %i` in uWSGI configuration' % (need, num, need - 1))
             sys.exit(1)
         self._locks.extend(map(lambda x: 'RECV_MSG_FARM_' + x, farms))
-        # This would be nice, but in my 2.0.15 uWSGI, the uwsgi module has no set_option function. And I don't know if it'd work.
+        # this would be nice, but in my 2.0.15 uWSGI, the uwsgi module has no set_option function, and I don't know if it'd work even if the function existed as documented
         #if len(self.lock_map) > 1:
         #    uwsgi.set_option('locks', len(self.lock_map))
         #    log.debug('Created %s uWSGI locks' % len(self.lock_map))
 
     def __init__(self, app, stack, dispatcher=None):
         super(UWSGIFarmMessageTransport, self).__init__(app, stack, dispatcher=dispatcher)
-        self._farms_dict = None
-        self._mules_list = None
-        self._is_mule = False
-        self._msg_queue = Queue()
         self.__initialize_locks()
 
-    @property
-    def _farms(self):
-        if self._farms_dict is None:
-            self._farms_dict = {}
-            farms = uwsgi.opt.get('farm', [])
-            farms = [farms] if isinstance(farms, string_types) else farms
-            for farm in farms:
-                name, mules = farm.split(':', 1)
-                self._farms_dict[name] = [int(m) for m in mules.split(',')]
-        return self._farms_dict
-
-    @property
-    def _mules(self):
-        if self._mules_list is None:
-            self._mules_list = []
-            mules = uwsgi.opt.get('mule', [])
-            self._mules_list = [mules] if isinstance(mules, string_types) or mules is True else mules
-        return self._mules_list
-
     def __lock(self, name_or_id):
         try:
             uwsgi.lock(name_or_id)
@@ -136,69 +97,59 @@ class UWSGIFarmMessageTransport(ApplicationStackTransport):
             uwsgi.unlock(self._locks.index(name_or_id))
 
     def _farm_recv_msg_lock_num(self):
-        # need one lock per farm... except mules can have multiple farms... ack
-        return self._locks.index('RECV_MSG_FARM_' + self._farm_name)
+        return self._locks.index('RECV_MSG_FARM_' + self.stack._farm_name)
 
     def _dispatch_messages(self):
-        # We are going to do this a lot, so cache the lock number
+        # this could be moved to the base class if locking was abstracted and a get_message method was added
+        log.info('Application stack message dispatcher thread starting up')
+        # we are going to do this a lot, so cache the lock number
         lock = self._farm_recv_msg_lock_num()
         while self.running:
             msg = None
             self.__lock(lock)
             try:
-                log.debug('######## Mule %s acquired message receive lock, waiting for new message', uwsgi.mule_id())
+                log.debug('Acquired message lock, waiting for new message')
                 msg = uwsgi.farm_get_msg()
-                log.debug('######## Mule %s received message: %s', uwsgi.mule_id(), msg)
-                if msg == self.shutdown_msg or msg is None:
-                    # all you need to do is pass here, self.running should already be set False by the signal handler calling the shutdown method defined in the superclass
+                log.debug('Received message: %s', msg)
+                if msg == self.SHUTDOWN_MSG:
                     self.running = False
-                    log.debug('######## SHUTTING DOWN %s', uwsgi.mule_id())
+                else:
+                    self.dispatcher.dispatch(msg)
             except:
-                log.exception( "Exception in mule message handling" )
+                log.exception('Exception in mule message handling')
             finally:
                 self.__unlock(lock)
-                log.debug('######## Mule %s released lock', uwsgi.mule_id())
-            if msg != self.shutdown_msg and msg is not None:
-                self.dispatcher.dispatch(msg)
-        log.info('######## Mule %s message handler shutting down', uwsgi.mule_id())
+                log.debug('Released lock')
+        log.info('Application stack message dispatcher thread exiting')
+
+    # TODO: start_if_needed would be called on a web worker by the stack's register_message_handler function if a
+    # function were registered in a web handler, that should probably be prevented.
 
     def start(self):
         """ Post-fork initialization.
+
+        This is mainly done here for the future possibility that we'll be able to run mules post-fork without exec()ing. In a programmed mule it could be done at __init__ time.
         """
-        # TODO: what happens if workers > 1??
-        self._is_mule = uwsgi.mule_id() > 0
-        if self._is_mule:
+        if self.stack._is_mule:
             if not uwsgi.in_farm():
                 raise RuntimeError('Mule %s is not in a farm! Set `farm = <pool_name>:%s` in uWSGI configuration'
                                    % (uwsgi.mule_id(),
-                                      ','.join(map(str, range(1, len(filter(lambda x: x.endswith('galaxy/main.py'), self._mules)) + 1)))))
-        super(UWSGIFarmMessageTransport, self).start()
-        log.info('######## Mule transport started, worker id: %s, mule id: %s, farm names: %s, server name: %s', uwsgi.worker_id(), uwsgi.mule_id(), self.stack.farm_names, self.app.config.server_name)
-        #self._send_all_messages()
-
-    #@property
-    #def _farm_name(self):
-    #    for name, mules in self._farms.items():
-    #        if uwsgi.mule_id() in mules:
-    #            return name
-    #    return None
-
-    def _send_all_messages(self):
-        # the sender doesn't have a running thread, all we are concerned with here is whether or not we've forked yet
-        if self.can_run and not self._is_mule:
-            while True:
-                try:
-                    msg, dest = self._msg_queue.get_nowait()
-                except Empty:
-                    break
-                log.debug('######## Sending message in mule %s to farm %s: %s', uwsgi.mule_id(), dest, msg)
-                uwsgi.farm_msg(dest, msg)
-                log.debug('######## Message sent')
+                                      ','.join(map(str, range(1, len(filter(lambda x: x.endswith('galaxy/main.py'), self.stack._configured_mules)) + 1)))))
+            elif len(self.stack._farms) > 1:
+                raise RuntimeError('Mule %s is in multiple farms! This configuration is not supported due to locking issues' % uwsgi.mule_id())
+            # only mules receive messages so don't bother starting the dispatcher if we're not a mule (although
+            # currently it doesn't have any registered handlers and so wouldn't start anyway)
+            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:
+            super(UWSGIFarmMessageTransport, self).shutdown()
 
     def send_message(self, msg, dest):
-        #log.debug('######## Queing message in mule %s to farm %s: %s', uwsgi.mule_id(), dest, msg)
-        #self._msg_queue.put((msg, dest))
-        #self._send_all_messages()
-        log.debug('######## Sending message to farm %s: %s', dest, msg)
+        log.debug('Sending message to farm %s: %s', dest, msg)
         uwsgi.farm_msg(dest, msg)

scripts/get_uwsgi_args.py

@@ -12,7 +12,6 @@ from galaxy.config import (
     parse_dependency_options,
 )
 from galaxy.script import main_factory
-from galaxy.util.configdict import parse_config
 
 
 DESCRIPTION = "Script to determine uWSGI command line arguments."
@@ -20,9 +19,6 @@ COMMAND_TEMPLATE = '{virtualenv}--ini-paste {galaxy_ini} --paste-logger --die-on
 
 
 def _get_uwsgi_args(args, kwargs):
-    # FIXME: hardcoded
-    stack_conf = parse_config('config/stack_conf.yml', 'stack', {})
-    # FIXME: these belong in stack conf
     handlerct = int(kwargs.get('job_handler_count', 1))
     pool_name = kwargs.get('job_handler_pool_name', 'job-handlers')
     virtualenv = os.environ.get('VIRTUAL_ENV', None)