Support workflow scheduling via stack messaging.

config/workflow_schedulers_conf.xml.sample

@@ -11,4 +11,15 @@
        require any external dependencies. -->
   <core id="core" />
 
+  <!-- Handlers (Galaxy server processes that perform the scheduling work) can
+       be defined here in the same format as in job_conf.xml. By default, the
+       handlers defined in job_conf.xml will be used (or `main` if there is no
+       job_conf.xml).  -->
+  <!--
+  <handlers default="handlers">
+    <handler id="handler0" tags="handlers"/>
+    <handler id="handler1" tags="handlers"/>
+  </handlers>
+  -->
+
 </workflow_schedulers>

lib/galaxy/jobs/handler.py

@@ -87,7 +87,8 @@ class JobHandlerQueue(object):
         # Start the queue
         self.monitor_thread.start()
         # The stack code is initialized in the application
-        self.app.application_stack.register_message_handler(self.handle_msg, name=JobHandlerMessage.target)
+        JobHandlerMessage().bind_default_handler(self, '_handle_message')
+        self.app.application_stack.register_message_handler(self._handle_message, name=JobHandlerMessage.target)
         log.info("job handler queue started")
 
     def job_wrapper(self, job, use_persisted_destination=False):
@@ -644,17 +645,19 @@ class JobHandlerQueue(object):
         else:
             log.warning("(%s) Handler '%s' received setup message but handler '%s' is already assigned, ignoring", job.id, self.app.config.server_name, job.handler)
 
-    def handle_msg(self, msg):
-        try:
-            getattr(self, '_handle_%s_msg' % msg.task)(**msg.params)
-        except:
-            log.exception( "Exception in mule message handling" )
-
     def put(self, job_id, tool_id):
         """Add a job to the queue (by job identifier)"""
         if not self.track_jobs_in_database:
             self.queue.put((job_id, tool_id))
             self.sleeper.wake()
+        else:
+            # Workflow invocations farmed out to workers will submit jobs through here. If a handler is unassigned, we
+            # will submit for one, or else claim it ourself. TODO: This should be moved to a higher level as it's now
+            # implemented here and in MessageJobQueue
+            job = self.sa_session.query(model.Job).get(job_id)
+            if job.handler is None and self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS):
+                msg = JobHandlerMessage(task='setup', job_id=job_id)
+                self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg)
 
     def shutdown(self):
         """Attempts to gracefully shut down the worker thread"""

lib/galaxy/web/stack/message.py

@@ -4,6 +4,7 @@ from __future__ import absolute_import
 
 import json
 import logging
+from types import MethodType
 
 
 log = logging.getLogger(__name__)
@@ -45,15 +46,52 @@ class ApplicationStackMessageDispatcher(object):
 
 class ApplicationStackMessage(dict):
     target = None
+    default_handler = None
+    _validate_kwargs = ('target',)
 
-    def __init__(self, target=None, params=None, **kwargs):
+    def __init__(self, target=None, **kwargs):
         self['target'] = target or self.__class__.target
-        self['params'] = params or {}
-        for k, v in kwargs.items():
-            self['params'][k] = v
+        self._merge_class_tuples()
+
+    def _merge_class_tuples(self):
+        """Locates any class-level tuples beginning with a single (but not double) underscore in the MRO and creates a
+        property on the instance with the same name (without the leading underscore) that will return the union of
+        those tuples.
+        """
+        names = set()
+        for cls in reversed(self.__class__.mro()):
+            names.update([x for x in dir(cls) if x.startswith('_') and not x.startswith('__') and type(getattr(cls, x)) == tuple])
+        for name in names:
+            setattr(self.__class__, name.lstrip('_'), property(lambda self, name=name: self._get_list_from_mro(name)))
+
+    def _get_list_from_mro(self, name):
+        """Locate all class-level tuples with the given `name` in the MRO and return their union.
+        """
+        r = set()
+        for cls in reversed(self.__class__.mro()):
+            r.update(getattr(cls, name, []))
+        return r
+
+    def _validate_items(self, obj, items, name):
+        for item in items:
+            assert item in obj, "Missing '%s' message %" % (item, name)
 
     def validate(self):
-        assert self['target'] is not None, "Missing 'target' parameter"
+        self._validate_items(self, self.validate_kwargs, 'argument')
+
+    def encode(self):
+        self['__classname__'] = self.__class__.__name__
+        return json.dumps(self)
+
+    def bind_default_handler(self, obj, name):
+        """Bind the default handler method to `obj` as attribute `name`.
+
+        This could also be implemented as a mixin class.
+        """
+        assert self.default_handler is not None, '%s has no default handler method, cannot bind' % self.__class__.__name__
+        setattr(obj, name, MethodType(self.default_handler, obj, obj.__class__))
+        log.debug("Bound default message handler '%s.%s' to %s", self.__class__.__name__, self.default_handler.__name__,
+                  getattr(obj, name))
 
     @property
     def target(self):
@@ -63,37 +101,62 @@ class ApplicationStackMessage(dict):
     def set_target(self, target):
         self['target'] = target
 
+
+class ParamMessage(ApplicationStackMessage):
+    _validate_kwargs = ('params',)
+    _validate_params = ()
+    _exclude_params = ()
+
+    def __init__(self, target=None, params=None, **kwargs):
+        super(ParamMessage, self).__init__(target=target)
+        self['params'] = params or {}
+        for k, v in kwargs.items():
+            self['params'][k] = v
+
+    def validate(self):
+        super(ParamMessage, self).validate()
+        self._validate_items(self['params'], self.validate_params, 'parameters')
+
     @property
     def params(self):
-        return self['params']
+        d = self['params'].copy()
+        for key in self.exclude_params:
+            d.pop(key, None)
+        return d
 
     @params.setter
     def set_params(self, params):
         self['params'] = params
 
-    def encode(self):
-        self['__classname__'] = self.__class__.__name__
-        return json.dumps(self)
 
+class TaskMessage(ParamMessage):
+    _validate_params = ('task',)
+    _exclude_params = ('task',)
 
-# 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 validate(self):
-        super(JobHandlerMessage, self).validate()
-        for param in ('task', 'job_id'):
-            assert param in self['params'], "Missing required parameter '%s'" % param
+    @staticmethod
+    def default_handler(self, msg):
+        """Can be bound to an instance of any class that has message handling methods named like `_handle_{task}_method`
+        """
+        name = '_handle_{task}_msg'.format(task=msg.task)
+        assert name in dir(self), "{cls} has no method _handle_{task}_msg, cannot handle message: {msg}".format(
+            cls=self.__class__.__name__,
+            task=msg.task,
+            msg=msg)
+        getattr(self, '_handle_%s_msg' % msg.task)(**msg.params)
 
     @property
     def task(self):
         return self['params']['task']
 
-    @property
-    def params(self):
-        d = self['params'].copy()
-        del d['task']
-        return d
+
+class JobHandlerMessage(TaskMessage):
+    target = 'job_handler'
+    _validate_params = ('job_id',)
+
+
+class WorkflowSchedulingMessage(TaskMessage):
+    target = 'workflow_scheduling'
+    _validate_params = ('workflow_invocation_id',)
 
 
 def decode(msg_str):

lib/galaxy/workflow/scheduling_manager.py

@@ -8,6 +8,7 @@ import galaxy.workflow.schedulers
 from galaxy import model
 from galaxy.util import plugin_config
 from galaxy.util.handlers import ConfiguresHandlers
+from galaxy.web.stack.message import WorkflowSchedulingMessage
 
 log = logging.getLogger(__name__)
 
@@ -31,6 +32,10 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers):
         self.__handlers_configured = False
         self.workflow_schedulers = {}
         self.active_workflow_schedulers = {}
+        # TODO: this should not hardcode the job handlers pool
+        self.__handler_pool = self.app.application_stack.pools.JOB_HANDLERS
+        # TODO: and we need a better way to indicate messaging should
+        self.__use_stack_messages = app.application_stack.has_pool(self.__handler_pool)
         # Passive workflow schedulers won't need to be monitored I guess.
 
         self.request_monitor = None
@@ -45,9 +50,14 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers):
             self.__start_schedulers()
             if self.active_workflow_schedulers:
                 self.__start_request_monitor()
+            if self.__use_stack_messages:
+                WorkflowSchedulingMessage().bind_default_handler(self, '_handle_message')
+                self.app.application_stack.register_message_handler(
+                    self._handle_message,
+                    name=WorkflowSchedulingMessage.target)
         else:
-            # Process should not schedule workflows - do nothing.
-            pass
+            # Process should not schedule workflows but should check for any unassigned to handlers
+            self.__startup_recovery()
 
         # When assinging handlers to workflows being queued - use job_conf
         # if not explicit workflow scheduling handlers have be specified or
@@ -57,6 +67,29 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers):
         else:
             self.__has_handlers = app.job_config
 
+    def __startup_recovery(self):
+        sa_session = self.app.model.context
+        if self.__use_stack_messages:
+            for workflow_invocation in model.WorkflowInvocation.poll_active_workflow_ids(
+                    sa_session,
+                    handler=None,
+                ):
+                log.info("(%s) Handler unassigned at startup, queueing workflow invocation via stack messaging for pool"
+                         " [%s]", workflow_invocation.id, self.__handler_pool)
+                msg = WorkflowSchedulingMessage(task='setup', workflow_invocation_id=workflow_invocation.id)
+
+    def _handle_setup_msg(self, workflow_invocation_id=None):
+        sa_session = self.app.model.context
+        workflow_invocation = sa_session.query(model.WorkflowInvocation).get(workflow_invocation_id)
+        if workflow_invocation.handler is None:
+            workflow_invocation.handler = self.app.config.server_name
+            sa_session.add(workflow_invocation)
+            sa_session.flush()
+        else:
+            log.warning("(%s) Handler '%s' received setup message for workflow invocation but handler '%s' is"
+                        " already assigned, ignoring", workflow_invocation.id, self.app.config.server_name,
+                        workflow_invocation.handler)
+
     def _is_workflow_handler(self):
         # If we have explicitly configured handlers, check them.
         # Else just make sure we are a job handler.
@@ -90,14 +123,24 @@ class WorkflowSchedulingManager(object, ConfiguresHandlers):
         workflow_invocation.state = model.WorkflowInvocation.states.NEW
         scheduler = request_params.get("scheduler", None) or self.default_scheduler_id
         handler = self._get_handler(workflow_invocation.history.id)
-        log.info("Queueing workflow invocation for handler [%s]" % handler)
 
+        if handler is None and not self.__use_stack_messages:
+            raise RuntimeError("Unable to set a handler for workflow invocation '%s'" % workflow_invocation.id)
+
+        log.info("Queueing workflow invocation for handler [%s]", handler)
         workflow_invocation.scheduler = scheduler
         workflow_invocation.handler = handler
 
         sa_session = self.app.model.context
         sa_session.add(workflow_invocation)
         sa_session.flush()
+
+        if handler is None and self.__use_stack_messages:
+            log.info("(%s) Queueing workflow invocation via stack messaging for pool [%s]",
+                     workflow_invocation.id, self.__handler_pool)
+            msg = WorkflowSchedulingMessage(task='setup', workflow_invocation_id=workflow_invocation.id)
+            self.app.application_stack.send_message(self.__handler_pool, msg)
+
         return workflow_invocation
 
     def __start_schedulers(self):