Make handler assignment configurable in preparation for new assignment

methods.

lib/galaxy/app.py

@@ -212,6 +212,11 @@ class UniverseApplication(config.ConfiguresGalaxyMixin):
         # Must be initialized after job_config.
         self.workflow_scheduling_manager = scheduling_manager.WorkflowSchedulingManager(self)
 
+        # Must be initialized after any component that might make use of stack messaging is configured. Alternatively if
+        # it becomes more commonly needed we could create a prefork function registration method like we do with
+        # postfork functions.
+        self.application_stack.init_late_prefork()
+
         self.containers = {}
         if self.config.enable_beta_containers_interface:
             self.containers = build_container_interfaces(

lib/galaxy/config.py

@@ -154,7 +154,7 @@ def find_root(kwargs):
 
 
 class Configuration(object):
-    deprecated_options = ('database_file', )
+    deprecated_options = ('database_file', 'track_jobs_in_database')
 
     def __init__(self, **kwargs):
         self.config_dict = kwargs

lib/galaxy/exceptions/__init__.py

@@ -185,6 +185,15 @@ class InternalServerError(MessageException):
     err_code = error_codes.INTERNAL_SERVER_ERROR
 
 
+class ToolExecutionError(MessageException):
+    status_code = 500
+    err_code = error_codes.TOOL_EXECUTION_ERROR
+
+    def __init__(self, err_msg, type="error", job=None):
+        super(ToolExecutionError, self).__init__(err_msg, type)
+        self.job = job
+
+
 class NotImplemented(MessageException):
     status_code = 501
     err_code = error_codes.NOT_IMPLEMENTED
@@ -221,3 +230,9 @@ class ContainerRunError(Exception):
         super(ContainerRunError, self).__init__(msg, **kwargs)
         self.image = image
         self.command = command
+
+
+class HandlerAssignmentError(Exception):
+    def __init__(self, msg=None, obj=None, **kwargs):
+        super(HandlerAssignmentError, self).__init__(msg, **kwargs)
+        self.obj = obj

lib/galaxy/exceptions/error_codes.json

@@ -144,6 +144,11 @@
     "code": 500003,
     "message": "Error in a configuration file."
    },
+   {
+    "name": "TOOL_EXECUTION_ERROR",
+    "code": 500004,
+    "message": "Tool execution failed due to an internal server error."
+   },
    {
     "name": "NOT_IMPLEMENTED",
     "code": 501001,

lib/galaxy/jobs/__init__.py

@@ -115,6 +115,8 @@ class JobConfiguration(ConfiguresHandlers):
 
     These features are configured in the job configuration, by default, ``job_conf.xml``
     """
+    DEFAULT_BASE_HANDLER_POOLS = ('job-handlers',)
+
     DEFAULT_NWORKERS = 4
 
     JOB_RESOURCE_CONDITIONAL_XML = """<conditional name="__job_resource">
@@ -135,6 +137,8 @@ class JobConfiguration(ConfiguresHandlers):
         self.handlers = {}
         self.handler_runner_plugins = {}
         self.default_handler_id = None
+        self.handler_assignment_methods = None
+        self.handler_assignment_methods_configured = False
         self.destinations = {}
         self.destination_tags = {}
         self.default_destination_id = None
@@ -150,7 +154,6 @@ class JobConfiguration(ConfiguresHandlers):
                             output_size=None,
                             destination_user_concurrent_jobs={},
                             destination_total_concurrent_jobs={})
-        self._is_handler = None
 
         default_resubmits = []
         default_resubmit_condition = self.app.config.default_job_resubmission_condition
@@ -212,23 +215,12 @@ class JobConfiguration(ConfiguresHandlers):
 
         # Parse handlers
         handlers_conf = root.find('handlers')
+        self._init_handler_assignment_methods(handlers_conf)
         self._init_handlers(handlers_conf)
-
-        # Determine the default handler(s)
-        try:
-            self.default_handler_id = self._get_default(self.app.config, handlers_conf, list(self.handlers.keys()))
-        except Exception:
-            pass
-        # For tets, this may not exist
-        try:
-            base_server_name = self.app.config.base_server_name
-        except AttributeError:
-            base_server_name = self.app.config.get('base_server_name', None)
-        if (self.default_handler_id is None
-                or (len(self.handlers) == 1 and base_server_name == next(iter(self.handlers.keys())))):
-            # 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
-            self.__set_default_job_handler()
+        self._set_default_handler_assignment_methods()
+        log.info("Job handler assignment methods set to: %s", ', '.join(self.handler_assignment_methods))
+        for tag, handlers in [(t, h) for t, h in self.handlers.items() if isinstance(h, list)]:
+            log.info("Tag [%s] handlers: %s", tag, ', '.join(handlers))
 
         # Parse destinations
         destinations = root.find('destinations')
@@ -269,7 +261,8 @@ class JobConfiguration(ConfiguresHandlers):
                     self.destinations[tag].append(job_destination)
 
         # Determine the default destination
-        self.default_destination_id = self._get_default(self.app.config, destinations, list(self.destinations.keys()))
+        self.default_destination_id = self._get_default(
+            self.app.config, destinations, list(self.destinations.keys()), auto=True)
 
         # Parse resources...
         resources = root.find('resources')
@@ -346,32 +339,14 @@ class JobConfiguration(ConfiguresHandlers):
         if self.app.config.use_tasked_jobs:
             self.runner_plugins.append(dict(id='tasks', load='tasks', workers=DEFAULT_LOCAL_WORKERS))
         # Set the handlers
-        self.__set_default_job_handler()
+        self._init_handler_assignment_methods()
+        self._init_handlers()
+        self._set_default_handler_assignment_methods()
         # Set the destination
         self.default_destination_id = 'local'
         self.destinations['local'] = [JobDestination(id='local', runner='local')]
         log.debug('Done loading job configuration')
 
-    def __set_default_job_handler(self):
-        # Called when self.default_handler_id is None
-        if self.app.application_stack.has_pool(self.app.application_stack.pools.JOB_HANDLERS):
-            if (self.app.application_stack.in_pool(self.app.application_stack.pools.JOB_HANDLERS)):
-                log.info("Found job handler pool managed by application stack, this server (%s) is a member of pool: "
-                         "%s", self.app.config.server_name, self.app.application_stack.pools.JOB_HANDLERS)
-                self._is_handler = True
-            else:
-                log.info("Found job handler pool managed by application stack, this server (%s) will submit jobs to "
-                         "pool: %s", self.app.config.server_name, self.app.application_stack.pools.JOB_HANDLERS)
-        else:
-            log.info('Did not find job handler pool managed by application stack, this server (%s) will handle jobs '
-                     'submitted to it', self.app.config.server_name)
-            self._is_handler = True
-            self.app.application_stack.register_postfork_function(self.make_self_default_handler)
-
-    def make_self_default_handler(self):
-        self.default_handler_id = self.app.config.server_name
-        self.handlers[self.app.config.server_name] = [self.app.config.server_name]
-
     def get_tool_resource_xml(self, tool_id, tool_type):
         """ Given a tool id, return XML elements describing parameters to
         insert into job resources.
@@ -484,7 +459,7 @@ class JobConfiguration(ConfiguresHandlers):
 
         :returns: JobToolConfiguration -- a representation of a <tool> element that uses the default handler and destination
         """
-        return JobToolConfiguration(id='default', handler=self.default_handler_id, destination=self.default_destination_id)
+        return JobToolConfiguration(id='_default_', handler=self.default_handler_id, destination=self.default_destination_id)
 
     # Called upon instantiation of a Tool object
     def get_job_tool_configurations(self, ids):

lib/galaxy/jobs/handler.py

@@ -743,14 +743,6 @@ class JobHandlerQueue(Monitors):
         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/jobs/manager.py

@@ -1,11 +1,12 @@
 """
 Top-level Galaxy job manager, moves jobs to handler(s)
 """
-
 import logging
+from functools import partial
 
 from sqlalchemy.sql.expression import null
 
+from galaxy.exceptions import HandlerAssignmentError, ToolExecutionError
 from galaxy.jobs import handler, NoopQueue
 from galaxy.model import Job
 from galaxy.web.stack.message import JobHandlerMessage
@@ -16,35 +17,86 @@ log = logging.getLogger(__name__)
 class JobManager(object):
     """
     Highest level interface to job management.
-
-    TODO: Currently the app accesses "job_queue" and "job_stop_queue" directly.
-          This should be decoupled.
     """
-
     def __init__(self, app):
         self.app = app
         self.job_lock = False
         if self.app.is_job_handler:
             log.debug("Initializing job handler")
             self.job_handler = handler.JobHandler(app)
-            self.job_stop_queue = self.job_handler.job_stop_queue
-        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.__check_jobs_at_startup()
+
+    def __check_jobs_at_startup(self):
+        """
+        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 handlers if running more than one uWSGI (master) process
+        """
+        # FIXME: test
+        if self.app.job_config.use_messaging:
+            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:
+                log.info(
+                    'No handler assigned at startup for the following jobs, will dispatch via message: %s',
+                    ', '.join([str(j.id) for j in jobs_at_startup]))
+            for job in jobs_at_startup:
+                tool = self.app.toolbox.get_tool(job.tool_id, job.tool_version, exact=True)
+                self.enqueue(job, tool)
 
     def start(self):
         self.job_handler.start()
 
+    def _queue_callback(self, job, tool_id):
+        self.job_handler.job_queue.put(job.id, tool_id)
+
+    def _message_callback(self, job):
+        return JobHandlerMessage(task='setup', job_id=job.id)
+
+    def enqueue(self, job, tool=None):
+        tool_id = None
+        configured_handler = None
+        if tool:
+            tool_id = tool.id
+            configured_handler = tool.get_configured_job_handler(job.params)
+            if configured_handler is not None:
+                p = " (with job params: %s)" % str(job.params) if job.params else ""
+                log.debug("(%s) Configured job handler for tool '%s'%s is: %s", job.log_str(), tool_id, p, configured_handler)
+        queue_callback = partial(self._queue_callback, job, tool_id)
+        message_callback = partial(self._message_callback, job)
+        try:
+            return self.app.job_config.assign_handler(
+                job, configured=configured_handler, queue_callback=queue_callback, message_callback=message_callback)
+        except HandlerAssignmentError as exc:
+            raise ToolExecutionError(exc.args[0], job=exc.obj)
+
+    def stop(self, job, message=None):
+        self.job_handler.job_stop_queue.put(job.id, error_msg=message)
+
     def shutdown(self):
         self.job_handler.shutdown()
 
 
+class NoopManager(object):
+    """
+    Implements the JobManager interface but does nothing
+    """
+    def __init__(self, *args, **kwargs):
+        self.job_handler = NoopHandler()
+
+    def enqueue(self, *args, **kwargs):
+        pass
+
+    def stop(self, *args, **kwargs):
+        pass
+
+
 class NoopHandler(object):
+    """
+    Implements the JobHandler interface but does nothing
+    """
     def __init__(self, *args, **kwargs):
         self.job_queue = NoopQueue()
         self.job_stop_queue = NoopQueue()
@@ -54,36 +106,3 @@ class NoopHandler(object):
 
     def shutdown(self, *args):
         pass
-
-
-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 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()
-        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:
-            log.info('No handler assigned at startup for the following jobs, will dispatch via message: %s', ', '.join([str(j.id) for j in jobs_at_startup]))
-        for job in jobs_at_startup:
-            self.job_queue.put(job.id, job.tool_id)
-
-
-class MessageJobQueue(NoopQueue):
-    """
-    Implements the JobQueue / JobStopQueue interface but only sends messages to the actual job queue
-    """
-    def __init__(self, app):
-        self.app = app
-
-    def put(self, job_id, tool_id):
-        msg = JobHandlerMessage(task='setup', job_id=job_id)
-        self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg)

lib/galaxy/managers/datasets.py

@@ -296,7 +296,7 @@ class DatasetAssociationManager(base.ModelManager,
                 # Are *all* of the job's other output datasets deleted?
                 if job.check_if_output_datasets_deleted():
                     job.mark_deleted(self.app.config.track_jobs_in_database)
-                    self.app.job_manager.job_stop_queue.put(job.id)
+                    self.app.job_manager.stop(job)
                     return True
         return False
 

lib/galaxy/model/__init__.py

@@ -561,6 +561,7 @@ class Job(JobLike, UsesCreateAndUpdateTime, Dictifiable, RepresentById):
                    RUNNING='running',
                    OK='ok',
                    ERROR='error',
+                   FAILED='failed',
                    PAUSED='paused',
                    DELETED='deleted',
                    DELETED_NEW='deleted_new')
@@ -862,6 +863,22 @@ class Job(JobLike, UsesCreateAndUpdateTime, Dictifiable, RepresentById):
                 dataset.peek = 'Job deleted'
                 dataset.info = 'Job output deleted by user before job completed'
 
+    def mark_failed(self, info="Job execution failed", blurb=None, peek=None):
+        """
+        Mark this job as failed, and mark any output datasets as errored.
+        """
+        self.state = self.states.FAILED
+        self.info = info
+        for jtod in self.output_datasets:
+            jtod.dataset.state = jtod.dataset.states.ERROR
+            for hda in jtod.dataset.dataset.history_associations:
+                hda.state = hda.states.ERROR
+                if blurb:
+                    hda.blurb = blurb
+                if peek:
+                    hda.peek = peek
+                hda.info = info
+
     def resume(self, flush=True):
         if self.state == self.states.PAUSED:
             self.set_state(self.states.NEW)
@@ -4388,6 +4405,17 @@ class WorkflowInvocation(UsesCreateAndUpdateTime, Dictifiable, RepresentById):
                 target_invocation_step = invocation_step
         return target_invocation_step
 
+    @staticmethod
+    def poll_unhandled_workflow_ids(sa_session):
+        and_conditions = [
+            WorkflowInvocation.state == WorkflowInvocation.states.NEW,
+            WorkflowInvocation.handler.is_(None)
+        ]
+        query = sa_session.query(
+            WorkflowInvocation.id
+        ).filter(and_(*and_conditions)).order_by(WorkflowInvocation.table.c.id.asc())
+        return [wid for wid in query.all()]
+
     @staticmethod
     def poll_active_workflow_ids(
         sa_session,
@@ -4536,6 +4564,18 @@ class WorkflowInvocation(UsesCreateAndUpdateTime, Dictifiable, RepresentById):
                 return True
         return False
 
+    def set_handler(self, handler):
+        self.handler = handler
+
+    def log_str(self):
+        extra = ""
+        safe_id = getattr(self, "id", None)
+        if safe_id is not None:
+            extra += "id=%s" % safe_id
+        else:
+            extra += "unflushed"
+        return "%s[%s]" % (self.__class__.__name__, extra)
+
 
 class WorkflowInvocationToSubworkflowInvocationAssociation(Dictifiable, RepresentById):
     dict_collection_visible_keys = ['id', 'workflow_step_id', 'workflow_invocation_id', 'subworkflow_invocation_id']

lib/galaxy/tools/__init__.py

@@ -561,16 +561,18 @@ class Tool(Dictifiable):
         assert rval is not None, 'Could not get a job tool configuration for Tool %s with job_params %s, this is a bug' % (self.id, job_params)
         return rval
 
-    def get_job_handler(self, job_params=None):
-        """Get a suitable job handler for this `Tool` given the provided `job_params`.  If multiple handlers are valid for combination of `Tool` and `job_params` (e.g. the defined handler is a handler tag), one will be selected at random.
+    def get_configured_job_handler(self, job_params=None):
+        """Get the configured job handler for this `Tool` given the provided `job_params`.
+
+        Unlike the former ``get_job_handler()`` method, this does not perform "preassignment" (random selection of
+        a configured handler ID from a tag).
 
         :param job_params: Any params specific to this job (e.g. the job source)
         :type job_params: dict or None
 
-        :returns: str -- The id of a job handler for a job run of this `Tool`
+        :returns: str or None -- The configured handler for a job run of this `Tool`
         """
-        # convert tag to ID if necessary
-        return self.app.job_config.get_handler(self.__get_job_tool_configuration(job_params=job_params).handler)
+        return self.__get_job_tool_configuration(job_params=job_params).handler
 
     def get_job_destination(self, job_params=None):
         """
@@ -1516,7 +1518,16 @@ class Tool(Dictifiable):
         `self.tool_action`. In general this will create a `Job` that
         when run will build the tool's outputs, e.g. `DefaultToolAction`.
         """
-        return self.tool_action.execute(self, trans, incoming=incoming, set_output_hid=set_output_hid, history=history, **kwargs)
+        try:
+            return self.tool_action.execute(self, trans, incoming=incoming, set_output_hid=set_output_hid, history=history, **kwargs)
+        except exceptions.ToolExecutionError as exc:
+            job = exc.job
+            job_id = 'unknown'
+            if job is not None:
+                job.mark_failed(info=exc.err_msg, blurb=exc.err_code.default_error_message)
+                job_id = job.id
+            log.error("Tool execution failed for job: %s", job_id)
+            raise
 
     def params_to_strings(self, params, app, nested=False):
         return params_to_strings(self.inputs, params, app, nested)

lib/galaxy/tools/actions/__init__.py

@@ -517,7 +517,6 @@ class DefaultToolAction(object):
         job.object_store_id = object_store_populator.object_store_id
         if job_params:
             job.params = dumps(job_params)
-        job.set_handler(tool.get_job_handler(job_params))
         if completed_job:
             job.set_copied_from_job_id(completed_job.id)
         trans.sa_session.add(job)
@@ -529,11 +528,8 @@ class DefaultToolAction(object):
                                      rerun_remap_job_id=rerun_remap_job_id,
                                      current_job=job,
                                      out_data=out_data)
-        log.info("Setup for job %s complete, ready to flush %s" % (job.log_str(), job_setup_timer))
+        log.info("Setup for job %s complete, ready to be enqueued %s" % (job.log_str(), job_setup_timer))
 
-        job_flush_timer = ExecutionTimer()
-        trans.sa_session.flush()
-        log.info("Flushed transaction for job %s %s" % (job.log_str(), job_flush_timer))
         # Some tools are not really executable, but jobs are still created for them ( for record keeping ).
         # Examples include tools that redirect to other applications ( epigraph ).  These special tools must
         # include something that can be retrieved from the params ( e.g., REDIRECT_URL ) to keep the job
@@ -555,8 +551,8 @@ class DefaultToolAction(object):
             trans.sa_session.flush()
             trans.response.send_redirect(url_for(controller='tool_runner', action='redirect', redirect_url=redirect_url))
         else:
-            # Put the job in the queue if tracking in memory
-            app.job_manager.job_queue.put(job.id, job.tool_id)
+            # Dispatch to a job handler. enqueue() is responsible for flushing the job
+            app.job_manager.enqueue(job, tool=tool)
             trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
             return job, out_data
 

lib/galaxy/tools/actions/history_imp_exp.py

@@ -56,11 +56,9 @@ class ImportHistoryToolAction(ToolAction):
             job.add_parameter(name, value)
 
         job.state = start_job_state  # job inputs have been configured, restore initial job state
-        job.set_handler(tool.get_job_handler(None))
-        trans.sa_session.flush()
 
         # Queue the job for execution
-        trans.app.job_manager.job_queue.put(job.id, tool.id)
+        trans.app.job_manager.enqueue(job, tool=tool)
         trans.log_event("Added import history job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
 
         return job, odict()
@@ -136,11 +134,9 @@ class ExportHistoryToolAction(ToolAction):
             job.add_parameter(name, value)
 
         job.state = start_job_state  # job inputs have been configured, restore initial job state
-        job.set_handler(tool.get_job_handler(None))
-        trans.sa_session.flush()
 
         # Queue the job for execution
-        trans.app.job_manager.job_queue.put(job.id, tool.id)
+        trans.app.job_manager.enqueue(job, tool=tool)
         trans.log_event("Added export history job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
 
         return job, odict()

lib/galaxy/tools/actions/metadata.py

@@ -63,7 +63,6 @@ class SetMetadataToolAction(ToolAction):
         except AttributeError:
             job.tool_version = "1.0.1"
         job.state = job.states.WAITING  # we need to set job state to something other than NEW, or else when tracking jobs in db it will be picked up before we have added input / output parameters
-        job.set_handler(tool.get_job_handler(job_params))
         sa_session.add(job)
         sa_session.flush()  # ensure job.id is available
 
@@ -104,7 +103,7 @@ class SetMetadataToolAction(ToolAction):
         sa_session.flush()
 
         # Queue the job for execution
-        app.job_manager.job_queue.put(job.id, tool.id)
+        app.job_manager.enqueue(job, tool=tool)
         # FIXME: need to add event logging to app and log events there rather than trans.
         # trans.log_event( "Added set external metadata job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id )
 

lib/galaxy/tools/actions/upload_common.py

@@ -432,15 +432,13 @@ def create_job(trans, params, tool, json_file_path, outputs, folder=None, histor
 
     job.object_store_id = object_store_id
     job.set_state(job.states.NEW)
-    job.set_handler(tool.get_job_handler(None))
     if job_params:
         for name, value in job_params.items():
             job.add_parameter(name, value)
     trans.sa_session.add(job)
-    trans.sa_session.flush()
 
     # Queue the job for execution
-    trans.app.job_manager.job_queue.put(job.id, job.tool_id)
+    trans.app.job_manager.enqueue(job, tool=tool)
     trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
     output = odict()
     for i, v in enumerate(outputs):

lib/galaxy/util/handlers.py

@@ -8,12 +8,45 @@ from __future__ import absolute_import
 import logging
 import os
 import random
+from collections import namedtuple
 
+from sqlalchemy.orm import object_session
+
+from . import (
+    ExecutionTimer,
+    listify
+)
+from ..exceptions import HandlerAssignmentError
 
 log = logging.getLogger(__name__)
 
+_handler_assignment_methods = ('MEM_SELF', 'DB_SELF', 'DB_PREASSIGN', 'UWSGI_MULE_MESSAGE')
+HANDLER_ASSIGNMENT_METHODS = namedtuple('JOB_HANDLER_ASSIGNMENT_METHODS', _handler_assignment_methods)(
+    *[x.lower().replace('_', '-') for x in _handler_assignment_methods]
+)
+
+
+class HandlerAssignmentSkip(Exception):
+    """Exception for handler assignment methods to raise if the next method should be tried.
+    """
+    pass
+
 
 class ConfiguresHandlers(object):
+    DEFAULT_HANDLER_TAG = '_default_'
+    DEFAULT_BASE_HANDLER_POOLS = ()
+    UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS = ()
+
+    def add_handler(self, handler_id, tags):
+        if handler_id in self.handlers:
+            log.error("Handler '%s' already added, ignoring", handler_id)
+            return
+        self.handlers[handler_id] = (handler_id,)
+        for tag in tags:
+            if tag in self.handlers:
+                self.handlers[tag].append(handler_id)
+            else:
+                self.handlers[tag] = [handler_id]
 
     def _init_handlers(self, config_element):
         # Parse handlers
@@ -21,22 +54,70 @@ class ConfiguresHandlers(object):
             for handler in self._findall_with_required(config_element, 'handler'):
                 handler_id = handler.get('id')
                 if handler_id in self.handlers:
-                    log.error("Handler '%s' overlaps handler with the same name, ignoring" % handler_id)
+                    log.error("Handler '%s' overlaps handler with the same name, ignoring", handler_id)
                 else:
-                    log.debug("Read definition for handler '%s'" % handler_id)
-                    self.handlers[handler_id] = (handler_id,)
-                    self._parse_handler(handler_id, handler)
-                    if handler.get('tags', None) is not None:
-                        for tag in [x.strip() for x in handler.get('tags').split(',')]:
-                            if tag in self.handlers:
-                                self.handlers[tag].append(handler_id)
-                            else:
-                                self.handlers[tag] = [handler_id]
+                    log.debug("Read definition for handler '%s'", handler_id)
+                    self.add_handler(
+                        handler_id,
+                        [x.strip() for x in handler.get('tags', self.DEFAULT_HANDLER_TAG).split(',')]
+                    )
+            self.default_handler_id = self._get_default(self.app.config, config_element, list(self.handlers.keys()))
+
+    def _init_handler_assignment_methods(self, config_element=None):
+        self.__is_handler = None
+        # This is set by the stack job handler init code
+        self.pool_for_tag = {}
+        self._handler_assignment_method_methods = {
+            HANDLER_ASSIGNMENT_METHODS.MEM_SELF: self._assign_mem_self_handler,
+            HANDLER_ASSIGNMENT_METHODS.DB_SELF: self._assign_db_self_handler,
+            HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN: self._assign_db_preassign_handler,
+            HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE: self._assign_uwsgi_mule_message_handler,
+        }
+        if config_element is not None:
+            for method in listify(config_element.attrib.get('assign_with', []), do_strip=True):
+                method = method.lower()
+                assert method in HANDLER_ASSIGNMENT_METHODS, \
+                    "Invalid job handler assignment method '%s', must be one of: %s" % (
+                        method, ', '.join(HANDLER_ASSIGNMENT_METHODS))
+                try:
+                    self.handler_assignment_methods.append(method)
+                except AttributeError:
+                    self.handler_assignment_methods_configured = True
+                    self.handler_assignment_methods = [method]
+            if self.handler_assignment_methods == [HANDLER_ASSIGNMENT_METHODS.MEM_SELF]:
+                self.app.config.track_jobs_in_database = False
+
+    def _set_default_handler_assignment_methods(self):
+        if not self.handler_assignment_methods_configured:
+            if not self.app.config.track_jobs_in_database and \
+                    HANDLER_ASSIGNMENT_METHODS.MEM_SELF not in self.UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS:
+                # DEPRECATED: You should just set mem_self as the only method if you want this
+                # FIXME: does setting MEM_SELF still disable track_jobs_in_database like it used to?
+                # FIXME: test MEM_SELF via all config routes:
+                #  1. no assign config, track_jobs_in_databse = False in config
+                #  2. assign_with='mem-self', track_jobs_in_databse = False in config
+                #  2. assign_with='mem-self', track_jobs_in_databse = True in config
+                log.warning("The `track_jobs_in_database` option is deprecated, please set `%s` as the job"
+                            " handler assignment method in the job handler configuration",
+                            HANDLER_ASSIGNMENT_METHODS.MEM_SELF)
+                self.handler_assignment_methods = [HANDLER_ASSIGNMENT_METHODS.MEM_SELF]
+            elif not self.handlers:
+                # No handlers defined, default is for processes to handle the jobs they create
+                self.handler_assignment_methods = [HANDLER_ASSIGNMENT_METHODS.DB_SELF]
+            else:
+                # Handlers are defined, default is the value if the default attribute, or any untagged handler if
+                # default attribute is unset
+                self.handler_assignment_methods = [HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN]
+            # If the stack has handler pools it can override these defaults
+            self.app.application_stack.init_job_handling(self)
+            log.info("%s: No job handler assignment method is set, defaulting to '%s', set the `assign_with` attribute"
+                     " on <handlers> to override the default", self.__class__.__name__,
+                     self.handler_assignment_methods[0])
 
     def _parse_handler(self, handler_id, handler_def):
         pass
 
-    def _get_default(self, config, parent, names):
+    def _get_default(self, config, parent, names, auto=False):
         """
         Returns the default attribute set in a parent tag like <handlers> or
         <destinations>, or return the ID of the child, if there is no explicit
@@ -46,6 +127,8 @@ class ConfiguresHandlers(object):
         :type parent: ``xml.etree.ElementTree.Element``
         :param names: The list of destination or handler IDs or tags that were loaded.
         :type names: list of str
+        :param auto: Automatically set a default if there is no default in the parent tag and there is only one child.
+        :type auto: bool
 
         :returns: str -- id or tag representing the default.
         """
@@ -60,14 +143,12 @@ class ConfiguresHandlers(object):
 
         if rval is not None:
             # If the parent element has a 'default' attribute, use the id or tag in that attribute
-            if rval not in names:
+            if self.deterministic_handler_assignment and rval not in names:
                 raise Exception("<%s> default attribute '%s' does not match a defined id or tag in a child element" % (parent.tag, rval))
             log.debug("<%s> default set to child with id or tag '%s'" % (parent.tag, rval))
-        elif len(names) == 1:
+        elif auto and len(names) == 1:
             log.info("Setting <%s> default to child with id '%s'" % (parent.tag, names[0]))
             rval = names[0]
-        else:
-            raise Exception("No <%s> default specified, please specify a valid id or tag with the 'default' attribute" % parent.tag)
         return rval
 
     def _findall_with_required(self, parent, match, attribs=None):
@@ -95,21 +176,38 @@ class ConfiguresHandlers(object):
         return rval
 
     @property
-    def is_handler(self):
-        """Indicate whether the current server is a handler.
+    def use_messaging(self):
+        # The job manager uses this to determine whether the messaging job queue should be used
+        return HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE in self.handler_assignment_methods
 
-        :param server_name: The name to check
-        :type server_name: str
+    @property
+    def deterministic_handler_assignment(self):
+        return self.handler_assignment_methods and all(
+            filter(lambda x: x in (
+                HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE,
+                HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN,
+            ), self.handler_assignment_methods))
+
+    def _get_is_handler(self):
+        """Indicate whether the current server is configured as a handler.
 
         :return: bool
         """
-        if self._is_handler is not None:
-            return self._is_handler
+        if self.__is_handler is not None:
+            return self.__is_handler
+        if (HANDLER_ASSIGNMENT_METHODS.DB_SELF in self.handler_assignment_methods
+                or HANDLER_ASSIGNMENT_METHODS.MEM_SELF in self.handler_assignment_methods):
+            return True
         for collection in self.handlers.values():
             if self.app.config.server_name in collection:
                 return True
         return False
 
+    def _set_is_handler(self, value):
+        self.__is_handler = value
+
+    is_handler = property(_get_is_handler, _set_is_handler)
+
     def _get_single_item(self, collection, index=None):
         """Given a collection of handlers or destinations, return one item from the collection at random.
         """
@@ -121,19 +219,116 @@ class ConfiguresHandlers(object):
         else:
             return collection[index % len(collection)]
 
-    # This is called by Tool.get_job_handler()
-    def get_handler(self, id_or_tag, index=None):
-        """Given a handler ID or tag, return a handler matching it.
+    # If these get to be any more complex we should probably modularize them, or at least move to a separate class
+
+    def _assign_handler_direct(self, obj, configured):
+        if self.app.config.track_jobs_in_database and configured:
+            try:
+                resolved = self.get_preassigned_handler(configured)
+            except KeyError:
+                resolved = None
+            if resolved == configured:
+                obj.set_handler(configured)
+                _timed_flush_obj(obj)
+                return True
+        return False
+
+    def _assign_mem_self_handler(self, obj, method, configured, queue_callback=None, **kwargs):
+        assert queue_callback is not None, \
+            "Cannot perform '%s' handler assignment: `queue_callback` is None" % HANDLER_ASSIGNMENT_METHODS.MEM_SELF
+        if configured:
+            log.warning("(%s) Ignoring handler assignment to '%s' because configured handler assignment method"
+                        " '' overrides per-tool handler assignment", obj.log_str(),
+                        HANDLER_ASSIGNMENT_METHODS.MEM_SELF, configured)
+        _timed_flush_obj(obj)
+        queue_callback()
+        return self.app.config.server_name
+
+    def _assign_db_self_handler(self, obj, method, configured, **kwargs):
+        if configured:
+            return self.handler_assignment_method_methods[HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN](
+                obj, method, configured, **kwargs
+            )
+        obj.set_handler(self.app.config.server_name)
+        _timed_flush_obj(obj)
+        return self.app.config.server_name
+
+    def _assign_db_preassign_handler(self, obj, method, configured, index=None, **kwargs):
+        """Given a handler ID or tag, return a handler matching it, of those handlers that are statically configured in
+        the job configuration, or known via preconfigured pools.
 
-        :param id_or_tag: A handler ID or tag.
-        :type id_or_tag: str
         :param index: Generate "consistent" "random" handlers with this index if specified.
         :type index: int
 
         :returns: str -- A valid job handler ID.
         """
-        if id_or_tag is None and self.default_handler_id is None:
-            return None
-        if id_or_tag is None:
-            id_or_tag = self.default_handler_id
-        return self._get_single_item(self.handlers[id_or_tag], index=index)
+        # FIXME: test a combo of pool handlers and defined handlers w/ DB_PREASSIGN as the only method
+        handler = configured
+        if handler is None:
+            handler = self.default_handler_id or self.DEFAULT_HANDLER_TAG
+        # Get a random handler ID from the possible handlers. If the admin has configured a tool with a handler tag that
+        # does not exist, or if there are no default handlers and configured is None, this will raise KeyError to
+        # assign_handler, which should log it, try the next method (if any), and if no other methods succeed, raise
+        # HandlerAssigmentError.
+        handler_id = self._get_single_item(self.handlers[handler], index=index)
+        if handler != handler_id:
+            log.debug("(%s) Selected handler '%s' by random choice from handler tag '%s'", obj.log_str(),
+                      handler_id, handler)
+        obj.set_handler(handler_id)
+        _timed_flush_obj(obj)
+        return handler_id
+
+    def _assign_uwsgi_mule_message_handler(self, obj, method, configured, message_callback=None, **kwargs):
+        assert message_callback is not None, \
+            "Cannot perform '%s' handler assignment: `message_callback` is None" \
+            % HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE
+        tag = configured or self.DEFAULT_HANDLER_TAG
+        pool = self.pool_for_tag.get(tag)
+        if pool is None:
+            log.debug("(%s) No handler pool (uWSGI farm) for '%s' found", obj.log_str(), tag)
+            raise HandlerAssignmentSkip()
+        else:
+            _timed_flush_obj(obj)
+            message = message_callback()
+            self.app.application_stack.send_message(pool, message)
+        return pool
+
+    def assign_handler(self, obj, configured=None, **kwargs):
+        """Set a job handler, flush obj
+
+        :param obj: Object to assign to (must be a model object with ``handler`` attribute and ``log_str`` callable)
+        :param configured: A handler ID or tag/pool.
+
+        :returns: bool -- True on successful assignment, False otherwise.
+        """
+        # It's a bit awkward that the method that actually hands off job execution is in the JobConfiguration, but
+        # that's currently the best place for it. It's worth noting that this method is also part of the
+        # WorkflowSchedulingManager, which acts like a combined JobConfiguration and JobManager. Combining those two
+        # classes would probably be reasonable (and would remove the need for the queue callback).
+        if self._assign_handler_direct(obj, configured):
+            log.info("(%s) Skipped handler assignment logic due to explicit configuration to a single handler: %s",
+                     obj.log_str(), configured)
+            return True
+        for method in self.handler_assignment_methods:
+            try:
+                handler = self._handler_assignment_method_methods[method](
+                    obj, method, configured=configured, **kwargs)
+                log.info("(%s) Handler '%s' assigned using '%s' assignment method", obj.log_str(), handler, method)
+                break
+            except HandlerAssignmentSkip:
+                log.debug("(%s) Handler assignment method '%s' did not assign a handler, trying next method",
+                          obj.log_str(), method)
+            except Exception:
+                log.exception("Caught exception in handler assignment method: %s", method)
+        else:
+            # Ideally we could just expunge the object from the SA session here, but in most cases, some of its related
+            # objects have already been created, so instead we'll just have to fail it.
+            log.error("(%s) Failed to select handler", obj.log_str())
+            raise HandlerAssignmentError("Job handler assignment failed.", obj=obj)
+
+
+def _timed_flush_obj(obj):
+    obj_flush_timer = ExecutionTimer()
+    sa_session = object_session(obj)
+    sa_session.flush()
+    log.info("Flushed transaction for %s %s" % (obj.log_str(), obj_flush_timer))

lib/galaxy/web/stack/__init__.py

@@ -19,8 +19,8 @@ import yaml
 from six import string_types
 
 from galaxy.util import unicodify
-from galaxy.util.bunch import Bunch
 from galaxy.util.facts import get_facts
+from galaxy.util.handlers import HANDLER_ASSIGNMENT_METHODS
 from galaxy.util.path import has_ext
 from galaxy.util.properties import nice_config_parser
 from .message import ApplicationStackMessage, ApplicationStackMessageDispatcher
@@ -54,12 +54,6 @@ class ApplicationStack(object):
     server_name_template = '{server_name}'
     default_app_name = 'main'
 
-    # 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',
-    )
-
     @classmethod
     def log_filter(cls):
         return cls.log_filter_class()
@@ -77,6 +71,12 @@ class ApplicationStack(object):
         self.config = config or (app and app.config)
         self.running = False
 
+    def init_job_handling(self, job_config):
+        pass
+
+    def init_late_prefork(self):
+        pass
+
     def log_startup(self):
         log.info("Galaxy server instance '%s' is running" % self.app.config.server_name)
 
@@ -97,12 +97,18 @@ class ApplicationStack(object):
         # TODO: ideally jobs would be mappable to handlers by pool name
         return None
 
+    def has_base_pool(self, pool_name):
+        return False
+
     def has_pool(self, pool_name):
         return False
 
     def in_pool(self, pool_name):
         return False
 
+    def pool_members(self, pool_name):
+        return None
+
     @property
     def facts(self):
         facts = get_facts(config=self.config)
@@ -129,12 +135,16 @@ class ApplicationStack(object):
 class MessageApplicationStack(ApplicationStack):
     def __init__(self, app=None, config=None):
         super(MessageApplicationStack, self).__init__(app=app, config=config)
+        self.use_messaging = False
         self.dispatcher = ApplicationStackMessageDispatcher()
         self.transport = self.transport_class(app, stack=self, dispatcher=self.dispatcher)
 
+    def init_late_prefork(self):
+        self.transport.init_late_prefork()
+
     def start(self):
         super(MessageApplicationStack, self).start()
-        if not self.running:
+        if self.use_messaging and not self.running:
             self.transport.start()
             self.running = True
 
@@ -157,7 +167,7 @@ class MessageApplicationStack(ApplicationStack):
         self.transport.send_message(msg.encode(), dest)
 
     def shutdown(self):
-        if self.running:
+        if self.use_messaging and self.running:
             log.info('Application stack interface shutting down')
             self.transport.shutdown()
             self.running = False
@@ -282,8 +292,85 @@ class UWSGIApplicationStack(MessageApplicationStack):
     def __init__(self, app=None, config=None):
         self._farms_dict = None
         self._mules_list = None
+        self._farm_server_names_dict = None
+
+        # If there's more than one farm *and* more than one farm will be using farm messaging, additional locks need to
+        # be configured by the admin. This allows us to keep track of how many such farms are configured.
+        self._lock_farms = set()
+
         super(UWSGIApplicationStack, self).__init__(app=app, config=config)
 
+    def _set_default_job_handler_assignment_methods(self, job_config, pool):
+        # This is called once per job_config (job config or workflow scheduling manager), and only in the case that
+        # handler assignment is not explicitly configured by the admin. Disable DB_SELF if a valid farm (pool) is
+        # configured. Use mule messaging unless the job_config doesn't allow it (e.g. workflow scheduling manager), in
+        # which case, use DB_PREASSIGN.
+        #
+        # TODO MULTIPOOL: if there is no default in any base_pool (and no job_config.default_handler_id) then don't
+        # remove DB_SELF
+        conf_class_name = job_config.__class__.__name__
+        remove_methods = [HANDLER_ASSIGNMENT_METHODS.DB_SELF]
+        if (HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE not in job_config.UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS):
+            add_method = HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE
+        else:
+            add_method = HANDLER_ASSIGNMENT_METHODS.DB_PREASSIGN
+            remove_methods.append(HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE)
+        log.debug("%s: No job handler assignment methods were configured but a uWSGI farm named '%s' exists,"
+                  " automatically enabling the '%s' assignment method", conf_class_name, pool, add_method)
+        for m in remove_methods:
+            try:
+                job_config.handler_assignment_methods.remove(m)
+                log.debug("%s: Removed '%s' from handler assignment methods due to use of mules", conf_class_name, m)
+            except ValueError:
+                pass
+        if add_method not in job_config.handler_assignment_methods:
+            job_config.handler_assignment_methods.insert(0, add_method)
+        log.debug("%s: handler assignment methods updated to: %s", conf_class_name,
+                  ', '.join(job_config.handler_assignment_methods))
+
+    def _init_job_handler_assignment_methods(self, job_config, pool):
+        if not job_config.handler_assignment_methods_configured:
+            self._set_default_job_handler_assignment_methods(job_config, pool)
+        if HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE in job_config.handler_assignment_methods:
+            self.use_messaging = True
+
+    def _init_job_handler_subpools(self, job_config, pool):
+        # Set up members of "subpools" ("pool.*") as handlers (including the base pool itself, if it exists)
+        for farm in self._configured_farms:
+            if not (farm == pool or farm.startswith(pool + '.')):
+                continue
+            if job_config.use_messaging:
+                self._lock_farms.add(farm)
+            if farm == pool:
+                tag = job_config.DEFAULT_HANDLER_TAG
+            else:
+                tag = farm.replace(pool + '.', '', 1)
+            # Pools are hierarchical (so that you can have e.g. workflow schedulers use the job handlers pool if no
+            # workflow schedulers pool exists), so if a pool for a given tag has already been found higher in the
+            # hierarchy, don't add mules from a farm/pool lower in the hierarchy.
+            # FIXME: test
+            if tag not in job_config.pool_for_tag:
+                if self.in_pool(farm):
+                    job_config.is_handler = True
+                for handler in self.pool_members(farm):
+                    job_config.add_handler(handler, [tag])
+                job_config.pool_for_tag[tag] = farm
+
+    def init_job_handling(self, job_config):
+        """Automatically add uWSGI farms as pools if they are named per predefined names and there is not an explicit
+        job handler assignment configuration.
+
+        Also automatically add mule messaging as the preferred assignment method, if supported.
+        """
+        # Only do the stack handler assignment init once per job_config
+        stack_assignment_methods_configured = False
+        for pool in job_config.DEFAULT_BASE_HANDLER_POOLS:
+            if self.has_base_pool(pool):
+                if not stack_assignment_methods_configured:
+                    self._init_job_handler_assignment_methods(job_config, pool)
+                    stack_assignment_methods_configured = True
+                self._init_job_handler_subpools(job_config, pool)
+
     @property
     def _configured_mules(self):
         if self._mules_list is None:
@@ -314,20 +401,39 @@ class UWSGIApplicationStack(MessageApplicationStack):
                 farms.append(farm)
         return farms
 
-    def _mule_index_in_farm(self, farm_name):
+    def _mule_index_in_farm(self, farm_name, mule_id=None):
+        mule_id = mule_id or uwsgi.mule_id()
         try:
             mules = self._configured_farms[farm_name]
-            return mules.index(uwsgi.mule_id())
+            return mules.index(mule_id)
         except (KeyError, ValueError):
             return -1
 
     @property
     def _farm_name(self):
+        # TODO: to allow mules to be in multiple farms you'll need to start here
         try:
             return self._farms[0]
         except IndexError:
             return None
 
+    @property
+    def _farm_server_names(self):
+        if self._farm_server_names_dict is None:
+            self._farm_server_names_dict = {}
+            for farm_name, mules in self._configured_farms.items():
+                server_names = []
+                facts = self.facts
+                for mule in mules:
+                    facts.update({
+                        'pool_name': farm_name,
+                        'server_id': mule,
+                        'instance_id': self._mule_index_in_farm(farm_name, mule) + 1,
+                    })
+                    server_names.append(self.server_name_template.format(**facts))
+                self._farm_server_names_dict[farm_name] = server_names
+        return self._farm_server_names_dict
+
     @property
     def instance_id(self):
         if not self._is_mule:
@@ -363,6 +469,9 @@ class UWSGIApplicationStack(MessageApplicationStack):
             os.environ['_GALAXY_UWSGI_FARM_NAMES'] = ','.join(self._farms)
         super(UWSGIApplicationStack, self).start()
 
+    def has_base_pool(self, pool_name):
+        return self.has_pool(pool_name) or any([farm.startswith(pool_name + '.') for farm in self._configured_farms])
+
     def has_pool(self, pool_name):
         return pool_name in self._configured_farms
 
@@ -372,6 +481,9 @@ class UWSGIApplicationStack(MessageApplicationStack):
         else:
             return pool_name in self._farms
 
+    def pool_members(self, pool_name):
+        return self._farm_server_names.get(pool_name, None)
+
     def workers(self):
         return uwsgi.workers()
 

lib/galaxy/web/stack/transport.py

@@ -29,6 +29,9 @@ class ApplicationStackTransport(object):
         self.dispatcher = dispatcher
         self.dispatcher_thread = None
 
+    def init_late_prefork(self):
+        pass
+
     def _dispatch_messages(self):
         pass
 
@@ -70,13 +73,12 @@ class UWSGIFarmMessageTransport(ApplicationStackTransport):
     # Define any static lock names here, additional locks will be appended for each configured farm's message handler
     _locks = []
 
-    def __initialize_locks(self):
+    def init_late_prefork(self):
         num = int(uwsgi.opt.get('locks', 0)) + 1
-        farms = self.stack._configured_farms.keys()
-        need = len(farms)
+        need = len(self.stack._lock_farms)
         if num < need:
             raise RuntimeError('Need %i uWSGI locks but only %i exist(s): Set `locks = %i` in uWSGI configuration' % (need, num, need - 1))
-        self._locks.extend(['RECV_MSG_FARM_' + x for x in farms])
+        self._locks.extend(['RECV_MSG_FARM_' + x for x in sorted(self.stack._lock_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 even if the function existed as documented
         # if len(self.lock_map) > 1:
         #     uwsgi.set_option('locks', len(self.lock_map))
@@ -84,7 +86,6 @@ class UWSGIFarmMessageTransport(ApplicationStackTransport):
 
     def __init__(self, app, stack, dispatcher=None):
         super(UWSGIFarmMessageTransport, self).__init__(app, stack, dispatcher=dispatcher)
-        self.__initialize_locks()
 
     def __lock(self, name_or_id):
         try:

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -203,7 +203,7 @@ class JobController(BaseAPIController, UsesLibraryMixinItems):
         if not job.finished:
             job.mark_deleted(self.app.config.track_jobs_in_database)
             trans.sa_session.flush()
-            self.app.job_manager.job_stop_queue.put(job.id)
+            self.app.job_manager.stop(job)
             return True
         else:
             return False

lib/galaxy/webapps/galaxy/config_schema.yml

@@ -2437,6 +2437,7 @@ mapping:
           failing jobs are just failed outright.
 
       track_jobs_in_database:
+        # FIXME: note deprecated, superceded by job_handler_assignment_method
         type: bool
         default: true
         required: false

lib/galaxy/webapps/galaxy/controllers/admin.py

@@ -1648,7 +1648,7 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
                     job.set_state(trans.app.model.Job.states.DELETED_NEW)
                     trans.sa_session.add(job)
                 else:
-                    trans.app.job_manager.job_stop_queue.put(job_id, error_msg=error_msg)
+                    trans.app.job_manager.stop(job, message=error_msg)
                 deleted.append(str(job_id))
         if deleted:
             msg = 'Queued job'

lib/galaxy/webapps/galaxy/controllers/history.py

@@ -360,7 +360,7 @@ class HistoryController(BaseUIController, SharableMixin, UsesAnnotations, UsesIt
                         if job.history_id == history.id and not job.finished:
                             # No need to check other outputs since the job's parent history is this history
                             job.mark_deleted(trans.app.config.track_jobs_in_database)
-                            trans.app.job_manager.job_stop_queue.put(job.id)
+                            trans.app.job_manager.stop(job)
         trans.sa_session.flush()
         if n_deleted:
             part = "Deleted %d %s" % (n_deleted, iff(n_deleted != 1, "histories", "history"))

lib/galaxy/workflow/scheduling_manager.py

@@ -1,11 +1,13 @@
 import logging
 import os
+from functools import partial
 from xml.etree import ElementTree
 
 import galaxy.workflow.schedulers
 from galaxy import model
+from galaxy.exceptions import HandlerAssignmentError
 from galaxy.util import plugin_config
-from galaxy.util.handlers import ConfiguresHandlers
+from galaxy.util.handlers import ConfiguresHandlers, HANDLER_ASSIGNMENT_METHODS
 from galaxy.util.monitors import Monitors
 from galaxy.web.stack.message import WorkflowSchedulingMessage
 
@@ -18,6 +20,7 @@ EXCEPTION_MESSAGE_SHUTDOWN = "Exception raised while attempting to shutdown work
 EXCEPTION_MESSAGE_NO_SCHEDULERS = "Failed to defined workflow schedulers - no workflow schedulers defined."
 EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER = "Failed to defined workflow schedulers - no workflow scheduler found for default id '%s'."
 EXCEPTION_MESSAGE_DUPLICATE_SCHEDULERS = "Failed to defined workflow schedulers - workflow scheduling plugin id '%s' duplicated."
+EXCEPTION_MESSAGE_SERIALIZE = "Parallelization is not desired but handler assignment methods are non-deterministic. Set DB_PREASSIGN in workflow_schedulers_conf.xml."
 
 
 class WorkflowSchedulingManager(ConfiguresHandlers):
@@ -25,22 +28,24 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
     ``galaxy.manager.JobManager``. Only schedules workflows on handler
     processes.
     """
+    DEFAULT_BASE_HANDLER_POOLS = ('workflow-schedulers', 'job-handlers')
+    UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS = (
+        HANDLER_ASSIGNMENT_METHODS.UWSGI_MULE_MESSAGE,
+    )
 
     def __init__(self, app):
         self.app = app
         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 be used
-        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
 
         self.handlers = {}
-        self._is_handler = None
+        self.handler_assignment_methods_configured = False
+        self.handler_assignment_methods = None
+        self.default_handler_id = None
 
         self.__plugin_classes = self.__plugins_dict()
         self.__init_schedulers()
@@ -50,7 +55,17 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
             self.__start_schedulers()
             if self.active_workflow_schedulers:
                 self.__start_request_monitor()
-            if self.__use_stack_messages:
+
+        # When assinging handlers to workflows being queued - use job_conf
+        # if not explicit workflow scheduling handlers have be specified or
+        # else use those explicit workflow scheduling handlers (on self).
+        if self.__handlers_configured:
+            self.__handlers_config = self
+        else:
+            self.__handlers_config = app.job_config
+
+        if self._is_workflow_handler():
+            if self.__handlers_config.use_messaging:
                 WorkflowSchedulingMessage().bind_default_handler(self, '_handle_message')
                 self.app.application_stack.register_message_handler(
                     self._handle_message,
@@ -59,24 +74,13 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
             # 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
-        # else use those explicit workflow scheduling handlers (on self).
-        if self.__handlers_configured:
-            self.__has_handlers = self
-        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)
-                self.app.application_stack.send_message(self.app.application_stack.pools.JOB_HANDLERS, msg)
+        for invocation_id in model.WorkflowInvocation.poll_unhandled_workflow_ids(sa_session):
+            log.info("(%s) Handler unassigned at startup, resubmitting workflow invocation for assignment",
+                     invocation_id)
+            workflow_invocation = sa_session.query(model.WorkflowInvocation).get(invocation_id)
+            self._assign_handler(workflow_invocation)
 
     def _handle_setup_msg(self, workflow_invocation_id=None):
         sa_session = self.app.model.context
@@ -99,13 +103,27 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
             is_handler = self.app.is_job_handler
         return is_handler
 
-    def _get_handler(self, history_id):
+    def _queue_callback(self, workflow_invocation):
+        # There isn't an in-memory queue for workflow schedulers, so if MEM_SELF is used just assign with the DB
+        workflow_invocation.handler = self.app.config.server_name
+        sa_session = self.app.model.context
+        sa_session.add(workflow_invocation)
+        sa_session.flush()
+
+    def _message_callback(self, workflow_invocation):
+        return WorkflowSchedulingMessage(task='setup', workflow_invocation_id=workflow_invocation.id)
+
+    def _assign_handler(self, workflow_invocation):
         # Use random-ish integer history_id to produce a consistent index to pick
         # job handler with.
-        random_index = history_id
+        random_index = workflow_invocation.history_id
+        queue_callback = partial(self._queue_callback, workflow_invocation)
+        message_callback = partial(self._message_callback, workflow_invocation)
         if self.app.config.parallelize_workflow_scheduling_within_histories:
             random_index = None
-        return self.__has_handlers.get_handler(None, index=random_index)
+        return self.__handlers_config.assign_handler(
+            workflow_invocation, configured=None, index=random_index, queue_callback=queue_callback,
+            message_callback=message_callback)
 
     def shutdown(self):
         exception = None
@@ -127,25 +145,16 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
 
     def queue(self, workflow_invocation, request_params):
         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)
-
-        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
-
+        workflow_invocation.scheduler = request_params.get("scheduler", None) or self.default_scheduler_id
         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)
+        # Assign handler (also performs the flush)
+        try:
+            self._assign_handler(workflow_invocation)
+        except HandlerAssignmentError:
+            # FIXME: test, raise new exception?
+            raise RuntimeError("Unable to set a handler for workflow invocation '%s'" % workflow_invocation.id)
 
         return workflow_invocation
 
@@ -156,11 +165,17 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
     def __plugins_dict(self):
         return plugin_config.plugins_dict(galaxy.workflow.schedulers, 'plugin_type')
 
+    @property
+    def __stack_has_pool(self):
+        # TODO: In the future it should be possible to map workflows to handlers based on workflow params. When that
+        # happens, we'll need to defer pool checks until execution time.
+        return any(map(self.app.application_stack.has_pool, self.DEFAULT_BASE_HANDLER_POOLS))
+
     def __init_schedulers(self):
         config_file = self.app.config.workflow_schedulers_config_file
         use_default_scheduler = False
         if not config_file:
-            log.info("Not workflow schedulers plugin config file defined, using default scheduler.")
+            log.info("No workflow schedulers plugin config file defined, using default scheduler.")
             use_default_scheduler = True
         elif not os.path.exists(config_file):
             log.info("Cannot find workflow schedulers plugin config file '%s', using default scheduler." % config_file)
@@ -172,6 +187,11 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
             plugins_element = ElementTree.parse(config_file).getroot()
             self.__init_schedulers_for_element(plugins_element)
 
+        if not self.__handlers_configured and self.__stack_has_pool:
+            # Stack has a pool for us so override inherited config and use the pool
+            self.__init_handler_assignment_methods()
+            self.__handlers_configured = True
+
     def __init_default_scheduler(self):
         self.default_scheduler_id = DEFAULT_SCHEDULER_ID
         self.__init_plugin(DEFAULT_SCHEDULER_PLUGIN_TYPE)
@@ -198,12 +218,27 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
             raise Exception(EXCEPTION_MESSAGE_NO_SCHEDULERS)
         if self.default_scheduler_id not in self.workflow_schedulers:
             raise Exception(EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER % self.default_scheduler_id)
+        if self.app.config.parallelize_workflow_scheduling_within_histories \
+                and not self.deterministic_handler_assignment:
+            if self.__handlers_configured:
+                # There's an explicit configuration, the admin should fix it.
+                raise Exception(EXCEPTION_MESSAGE_SERIALIZE)
+            else:
+                log.warning(EXCEPTION_MESSAGE_SERIALIZE)
 
     def __init_handlers(self, config_element):
         assert not self.__handlers_configured
+        self.__init_handler_assignment_methods(config_element)
         self._init_handlers(config_element)
         self.__handlers_configured = True
 
+    def __init_handler_assignment_methods(self, config_element=None):
+        self._init_handler_assignment_methods(config_element)
+        self._set_default_handler_assignment_methods()
+        log.info("Workflow scheduling handler assignment method(s): %s", ', '.join(self.handler_assignment_methods))
+        for tag, handlers in [(t, h) for t, h in self.handlers.items() if isinstance(h, list)]:
+            log.info("Tag [%s] handlers: %s", tag, ', '.join(handlers))
+
     def __init_plugin(self, plugin_type, workflow_scheduler_id=None, **kwds):
         workflow_scheduler_id = workflow_scheduler_id or self.default_scheduler_id
 
@@ -241,6 +276,7 @@ class WorkflowRequestMonitor(Monitors):
     def __schedule(self, workflow_scheduler_id, workflow_scheduler):
         invocation_ids = self.__active_invocation_ids(workflow_scheduler_id)
         for invocation_id in invocation_ids:
+            log.debug("Attempting to schedule workflow invocation [%s]", invocation_id)
             self.__attempt_schedule(invocation_id, workflow_scheduler)
             if not self.monitor_running:
                 return
@@ -261,6 +297,7 @@ class WorkflowRequestMonitor(Monitors):
                     if i.active and i.id < workflow_invocation.id:
                         return False
             workflow_scheduler.schedule(workflow_invocation)
+            log.debug("Workflow invocation [%s] scheduled", workflow_invocation.id)
         except Exception:
             # TODO: eventually fail this - or fail it right away?
             log.exception("Exception raised while attempting to schedule workflow request.")

test/unit/jobs/test_job_configuration.py

@@ -25,6 +25,7 @@ class JobConfXmlParserTestCase(unittest.TestCase):
             job_resource_params_file="/tmp/fake_absent_path",
             config_dict={},
             default_job_resubmission_condition="",
+            track_jobs_in_database=True,
             server_name="main",
         )
         self.__write_config_from(SIMPLE_JOB_CONF)
@@ -50,12 +51,10 @@ class JobConfXmlParserTestCase(unittest.TestCase):
         assert len(task_runners) == 1
         assert task_runners[0]["workers"] == 5
 
-    def test_load_simple_handler(self):
-        main_handler = self.job_config.handlers["main"]
-        assert main_handler[0] == "main"
-
-    def test_if_one_handler_implicit_default(self):
-        assert self.job_config.default_handler_id == "main"
+    def test_if_no_handlers_implict_db_self(self):
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers == {}
+        assert self.job_config.handler_assignment_methods == ['db-self']
 
     def test_explicit_handler_default(self):
         self.__with_advanced_config()

test/unit/tools/test_actions.py

@@ -12,8 +12,6 @@ from galaxy.tools.actions import (
 from galaxy.tools.parser.output_objects import ToolOutput
 from .. import tools_support
 
-TEST_HANDLER_NAME = "test_handler_1"
-
 
 # I cannot think of a saner way to test if data is being wrapped than use a
 # data param in the output label - though you would probably never want to do
@@ -73,7 +71,6 @@ class DefaultToolActionTestCase(unittest.TestCase, tools_support.UsesApp, tools_
         self.app.model.context.flush()
         self.action = DefaultToolAction()
         self.app.config.len_file_path = "moocow"
-        self.app.job_config["get_handler"] = lambda h: TEST_HANDLER_NAME
         self.app.object_store = MockObjectStore()
 
     def test_output_created(self):
@@ -114,10 +111,6 @@ class DefaultToolActionTestCase(unittest.TestCase, tools_support.UsesApp, tools_
         # Again this is a stupid way to ensure data parameters are wrapped.
         self.assertEqual(output["out1"].name, "Output (%s)" % hda1.dataset.get_file_name())
 
-    def test_handler_set(self):
-        job, _ = self._simple_execute()
-        assert job.handler == TEST_HANDLER_NAME
-
     def test_inactive_user_job_create_failure(self):
         self.trans.user_is_active = False
         try:

test/unit/unittest_utils/galaxy_mock.py

@@ -11,7 +11,7 @@ from galaxy import (
     quota
 )
 from galaxy.datatypes import registry
-from galaxy.jobs import NoopQueue
+from galaxy.jobs.manager import NoopManager
 from galaxy.managers import tags
 from galaxy.model import mapping
 from galaxy.tools.deps.containers import NullContainerFinder
@@ -76,7 +76,7 @@ class MockApp(object):
         self.container_finder = NullContainerFinder()
         self._toolbox_lock = MockLock()
         self.genome_builds = GenomeBuilds(self)
-        self.job_manager = Bunch(job_queue=NoopQueue())
+        self.job_manager = NoopManager()
         self.application_stack = ApplicationStack()
 
     def init_datatypes(self):