More adjustments to queue kickoff to force webapps to initialize the connection prior to models; this (at least seems to, so far) fixes the concurrent mapper initialization race condition sqlalchemy issue.

lib/galaxy/app.py

@@ -5,6 +5,7 @@ import os
 from galaxy import config, jobs
 import galaxy.model
 import galaxy.security
+import galaxy.queues
 from galaxy.managers.collections import DatasetCollectionManager
 import galaxy.quota
 from galaxy.managers.tags import GalaxyTagManager
@@ -16,7 +17,6 @@ from galaxy.sample_tracking import external_service_types
 from galaxy.openid.providers import OpenIDProviders
 from galaxy.tools.data_manager.manager import DataManagers
 from galaxy.jobs import metrics as job_metrics
-from galaxy.web.base import pluginframework
 from galaxy.web.proxy import ProxyManager
 from galaxy.queue_worker import GalaxyQueueWorker
 from tool_shed.galaxy_install import update_repository_manager
@@ -25,6 +25,7 @@ import logging
 log = logging.getLogger( __name__ )
 app = None
 
+
 class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
     """Encapsulates the state of a Universe application"""
     def __init__( self, **kwargs ):
@@ -36,11 +37,9 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
         self.config.check()
         config.configure_logging( self.config )
         self.configure_fluent_log()
-
+        self._amqp_internal_connection_obj = galaxy.queues.connection_from_config(self.config)
         self._configure_tool_shed_registry()
-
         self._configure_object_store( fsmon=True )
-
         # Setup the database engine and ORM
         config_file = kwargs.get( 'global_conf', {} ).get( '__file__', None )
         if config_file:
@@ -96,12 +95,15 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
         # Load history import/export tools.
         load_history_imp_exp_tools( self.toolbox )
         # visualizations registry: associates resources with visualizations, controls how to render
-        self.visualizations_registry = VisualizationsRegistry( self,
+        self.visualizations_registry = VisualizationsRegistry(
+            self,
             directories_setting=self.config.visualization_plugins_directory,
             template_cache_dir=self.config.template_cache )
         # Load security policy.
         self.security_agent = self.model.security_agent
-        self.host_security_agent = galaxy.security.HostAgent( model=self.security_agent.model, permitted_actions=self.security_agent.permitted_actions )
+        self.host_security_agent = galaxy.security.HostAgent(
+            model=self.security_agent.model,
+            permitted_actions=self.security_agent.permitted_actions )
         # Load quota management.
         if self.config.enable_quotas:
             self.quota_agent = galaxy.quota.QuotaAgent( self.model )
@@ -136,7 +138,9 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
         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( self.config.external_service_type_config_file, self.config.external_service_type_path, self )
+        self.external_service_types = external_service_types.ExternalServiceTypesCollection(
+            self.config.external_service_type_config_file,
+            self.config.external_service_type_path, self )
 
         from galaxy.workflow import scheduling_manager
         # Must be initialized after job_config.
@@ -146,7 +150,8 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
 
     def setup_control_queue(self):
         self.control_worker = GalaxyQueueWorker(self, galaxy.queues.control_queue_from_config(self.config),
-                                                galaxy.queue_worker.control_message_to_task)
+                                                galaxy.queue_worker.control_message_to_task,
+                                                self._amqp_internal_connection_obj)
         self.control_worker.daemon = True
         self.control_worker.start()
 

lib/galaxy/queue_worker.py

@@ -32,7 +32,7 @@ class GalaxyQueueWorker(ConsumerMixin, threading.Thread):
     """
     def __init__(self, app, queue, task_mapping, connection=None):
         super(GalaxyQueueWorker, self).__init__()
-        log.info("Initalizing Galaxy Queue Worker on %s" % app.config.amqp_internal_connection)
+        log.info("Initalizing Galaxy Queue Worker on %s", app.config.amqp_internal_connection)
         if connection:
             self.connection = connection
         else:

lib/galaxy/queues.py

@@ -14,7 +14,7 @@ if sys.version_info < (2, 7, 0):
     eggs.require('importlib')
     eggs.require('ordereddict')
 eggs.require('kombu')
-from kombu import Exchange, Queue
+from kombu import Exchange, Queue, Connection
 
 ALL_CONTROL = "control.*"
 galaxy_exchange = Exchange('galaxy_core_exchange', type='topic')
@@ -40,3 +40,10 @@ def control_queue_from_config(config):
     galaxy process's config
     """
     return Queue("control.%s" % config.server_name, galaxy_exchange, routing_key='control')
+
+
+def connection_from_config(config):
+    if config.amqp_internal_connection:
+        return Connection(config.amqp_internal_connection)
+    else:
+        return None

lib/galaxy/tools/__init__.py

@@ -2857,7 +2857,7 @@ class DataManagerTool( OutputParameterJSONTool ):
             return True
         if user:
             user = user.id
-        log.debug( "User (%s) attempted to access a data manager tool (%s), but is not an admin.", user, self.id ) 
+        log.debug( "User (%s) attempted to access a data manager tool (%s), but is not an admin.", user, self.id )
         return False
 
 # Populate tool_type to ToolClass mappings