Merge pull request #1212 from jmchilton/tool_opts_6

Smaller Tool Optimization PR

config/galaxy.ini.sample

@@ -832,6 +832,15 @@ use_interactive = True
 #enable_beta_tool_command_isolation = False
 
 
+# Set the following to a number of threads greater than 1 to spawn
+# a Python task queue for dealing with large tool submissions (either
+# through the tool form or as part of an individual workflow step across
+# large collection). The size of a "large" tool request is controlled by 
+# the second parameter below and defaults to 10. This affects workflow 
+# scheduling and web processes, not job handlers.
+#tool_submission_burst_threads = 1
+#tool_submission_burst_at = 10
+
 # Enable beta workflow modules that should not yet be considered part of Galaxy's
 # stable API.
 #enable_beta_workflow_modules = False

lib/galaxy/config.py

@@ -215,6 +215,8 @@ class Configuration( object ):
         self.use_tasked_jobs = string_as_bool( kwargs.get( 'use_tasked_jobs', False ) )
         self.local_task_queue_workers = int(kwargs.get("local_task_queue_workers", 2))
         self.commands_in_new_shell = string_as_bool( kwargs.get( 'enable_beta_tool_command_isolation', "False" ) )
+        self.tool_submission_burst_threads = int( kwargs.get( 'tool_submission_burst_threads', '1' ) )
+        self.tool_submission_burst_at = int( kwargs.get( 'tool_submission_burst_at', '10' ) )
         # The transfer manager and deferred job queue
         self.enable_beta_job_managers = string_as_bool( kwargs.get( 'enable_beta_job_managers', 'False' ) )
         # These workflow modules should not be considered part of Galaxy's

lib/galaxy/jobs/__init__.py

@@ -1029,26 +1029,33 @@ class JobWrapper( object ):
         self.sa_session.add( job )
         self.sa_session.flush()
 
-    def change_state( self, state, info=False ):
-        job = self.get_job()
-        self.sa_session.refresh( job )
+    def change_state( self, state, info=False, flush=True, job=None ):
+        job_supplied = job is not None
+        if not job_supplied:
+            job = self.get_job()
+            self.sa_session.refresh( job )
+        # Else:
+        # If this is a new job (e.g. initially queued) - we are in the same
+        # thread and no other threads are working on the job yet - so don't refresh.
+
         if job.state in model.Job.terminal_states:
             log.warning( "(%s) Ignoring state change from '%s' to '%s' for job "
                          "that is already terminal", job.id, job.state, state )
             return
         for dataset_assoc in job.output_datasets + job.output_library_datasets:
             dataset = dataset_assoc.dataset
-            self.sa_session.refresh( dataset )
-            dataset.state = state
+            if not job_supplied:
+                self.sa_session.refresh( dataset )
+            dataset.raw_set_dataset_state( state )
             if info:
                 dataset.info = info
             self.sa_session.add( dataset )
-            self.sa_session.flush()
         if info:
             job.info = info
         job.set_state( state )
         self.sa_session.add( job )
-        self.sa_session.flush()
+        if flush:
+            self.sa_session.flush()
 
     def get_state( self ):
         job = self.get_job()
@@ -1059,22 +1066,23 @@ class JobWrapper( object ):
         log.warning('set_runner() is deprecated, use set_job_destination()')
         self.set_job_destination(self.job_destination, external_id)
 
-    def set_job_destination( self, job_destination, external_id=None ):
+    def set_job_destination( self, job_destination, external_id=None, flush=True, job=None ):
         """
         Persist job destination params in the database for recovery.
 
         self.job_destination is not used because a runner may choose to rewrite
         parts of the destination (e.g. the params).
         """
-        job = self.get_job()
-        self.sa_session.refresh(job)
+        if job is None:
+            job = self.get_job()
         log.debug('(%s) Persisting job destination (destination id: %s)' % (job.id, job_destination.id))
         job.destination_id = job_destination.id
         job.destination_params = job_destination.params
         job.job_runner_name = job_destination.runner
         job.job_runner_external_id = external_id
         self.sa_session.add(job)
-        self.sa_session.flush()
+        if flush:
+            self.sa_session.flush()
 
     def finish( self, stdout, stderr, tool_exit_code=None, remote_working_directory=None ):
         """
@@ -1759,7 +1767,7 @@ class TaskWrapper(JobWrapper):
         self.status = 'error'
         # How do we want to handle task failure?  Fail the job and let it clean up?
 
-    def change_state( self, state, info=False ):
+    def change_state( self, state, info=False, flush=True, job=None ):
         task = self.get_task()
         self.sa_session.refresh( task )
         if info:

lib/galaxy/jobs/runners/__init__.py

@@ -106,10 +106,12 @@ class BaseJobRunner( object ):
         """Add a job to the queue (by job identifier), indicate that the job is ready to run.
         """
         put_timer = ExecutionTimer()
+        job = job_wrapper.get_job()
         # Change to queued state before handing to worker thread so the runner won't pick it up again
-        job_wrapper.change_state( model.Job.states.QUEUED )
+        job_wrapper.change_state( model.Job.states.QUEUED, flush=False, job=job )
         # Persist the destination so that the job will be included in counts if using concurrency limits
-        job_wrapper.set_job_destination( job_wrapper.job_destination, None )
+        job_wrapper.set_job_destination( job_wrapper.job_destination, None, flush=False, job=job )
+        self.sa_session.flush()
         self.mark_as_queued(job_wrapper)
         log.debug("Job [%s] queued %s" % (job_wrapper.job_id, put_timer))
 

lib/galaxy/managers/collections.py

@@ -77,6 +77,8 @@ class DatasetCollectionManager( object ):
                 for input_name, input_collection in implicit_collection_info[ "implicit_inputs" ]:
                     dataset_collection_instance.add_implicit_input_collection( input_name, input_collection )
                 for output_dataset in implicit_collection_info.get( "outputs" ):
+                    if output_dataset not in trans.sa_session:
+                        output_dataset = trans.sa_session.query( type( output_dataset ) ).get( output_dataset.id )
                     if isinstance( output_dataset, model.HistoryDatasetAssociation ):
                         output_dataset.hidden_beneath_collection_instance = dataset_collection_instance
                     elif isinstance( output_dataset, model.HistoryDatasetCollectionAssociation ):
@@ -265,7 +267,12 @@ class DatasetCollectionManager( object ):
         # Previously created collection already found in request, just pass
         # through as is.
         if "__object__" in element_identifier:
-            return element_identifier[ "__object__" ]
+            the_object = element_identifier[ "__object__" ]
+            if the_object is not None and the_object.id:
+                context = self.model.context
+                if the_object not in context:
+                    the_object = context.query( type(the_object) ).get(the_object.id)
+            return the_object
 
         # dateset_identifier is dict {src=hda|ldda|hdca|new_collection, id=<encoded_id>}
         try:

lib/galaxy/model/__init__.py

@@ -556,8 +556,11 @@ class Job( object, JobLike, Dictifiable ):
     def add_parameter( self, name, value ):
         self.parameters.append( JobParameter( name, value ) )
 
-    def add_input_dataset( self, name, dataset ):
-        self.input_datasets.append( JobToInputDatasetAssociation( name, dataset ) )
+    def add_input_dataset( self, name, dataset=None, dataset_id=None ):
+        assoc = JobToInputDatasetAssociation( name, dataset )
+        if dataset is None and dataset_id is not None:
+            assoc.dataset_id = dataset_id
+        self.input_datasets.append( assoc )
 
     def add_output_dataset( self, name, dataset ):
         self.output_datasets.append( JobToOutputDatasetAssociation( name, dataset ) )
@@ -1100,16 +1103,16 @@ class History( object, Dictifiable, UsesAnnotations, HasName ):
     def empty( self ):
         return self.hid_counter == 1
 
-    def _next_hid( self ):
+    def _next_hid( self, n=1 ):
         # this is overriden in mapping.py db_next_hid() method
         if len( self.datasets ) == 0:
-            return 1
+            return n
         else:
             last_hid = 0
             for dataset in self.datasets:
                 if dataset.hid > last_hid:
                     last_hid = dataset.hid
-            return last_hid + 1
+            return last_hid + n
 
     def add_galaxy_session( self, galaxy_session, association=None ):
         if association is None:
@@ -1144,6 +1147,41 @@ class History( object, Dictifiable, UsesAnnotations, HasName ):
         self.datasets.append( dataset )
         return dataset
 
+    def add_datasets( self, sa_session, datasets, parent_id=None, genome_build=None, set_hid=True, quota=True, flush=False ):
+        """ Optimized version of add_dataset above that minimizes database
+        interactions when adding many datasets to history at once.
+        """
+        all_hdas = all( map( lambda d: isinstance( d, HistoryDatasetAssociation ), datasets ) )
+        optimize = len( datasets) > 1 and parent_id is None and all_hdas and set_hid and not quota
+        if optimize:
+            self.__add_datasets_optimized( datasets, genome_build=genome_build )
+            sa_session.add_all( datasets )
+            if flush:
+                sa_session.flush()
+        else:
+            for dataset in datasets:
+                self.add_dataset( dataset, parent_id=parent_id, genome_build=genome_build, set_hid=set_hid, quota=quota )
+                sa_session.add( dataset )
+                if flush:
+                    sa_session.flush()
+
+    def __add_datasets_optimized( self, datasets, genome_build=None ):
+        """ Optimized version of add_dataset above that minimizes database
+        interactions when adding many datasets to history at once under
+        certain cirucumstances.
+        """
+        n = len( datasets )
+
+        base_hid = self._next_hid( n=n )
+
+        for i, dataset in enumerate( datasets ):
+            dataset.hid = base_hid + i
+            dataset.history = self
+            if genome_build not in [None, '?']:
+                self.genome_build = genome_build
+        self.datasets.extend( datasets )
+        return datasets
+
     def add_dataset_collection( self, history_dataset_collection, set_hid=True ):
         if set_hid:
             history_dataset_collection.hid = self._next_hid()
@@ -1484,10 +1522,13 @@ class DefaultQuotaAssociation( Quota, Dictifiable ):
 
 
 class DatasetPermissions( object ):
-    def __init__( self, action, dataset, role ):
+    def __init__( self, action, dataset, role=None, role_id=None ):
         self.action = action
         self.dataset = dataset
-        self.role = role
+        if role is not None:
+            self.role = role
+        else:
+            self.role_id = role_id
 
 
 class LibraryPermissions( object ):
@@ -1789,10 +1830,17 @@ class DatasetInstance( object ):
             return self._state
         return self.dataset.state
 
+    def raw_set_dataset_state( self, state ):
+        if state != self.dataset.state:
+            self.dataset.state = state
+            return True
+        else:
+            return False
+
     def set_dataset_state( self, state ):
-        self.dataset.state = state
-        object_session( self ).add( self.dataset )
-        object_session( self ).flush()  # flush here, because hda.flush() won't flush the Dataset object
+        if self.raw_set_dataset_state( state ):
+            object_session( self ).add( self.dataset )
+            object_session( self ).flush()  # flush here, because hda.flush() won't flush the Dataset object
     state = property( get_dataset_state, set_dataset_state )
 
     def get_file_name( self ):

lib/galaxy/model/mapping.py

@@ -2449,7 +2449,7 @@ class_mapper( model.HistoryDatasetCollectionAssociation ).add_property(
 
 
 # Helper methods.
-def db_next_hid( self ):
+def db_next_hid( self, n=1 ):
     """
     db_next_hid( self )
 
@@ -2465,7 +2465,7 @@ def db_next_hid( self ):
     trans = conn.begin()
     try:
         next_hid = select( [table.c.hid_counter], table.c.id == self.id, for_update=True ).scalar()
-        table.update( table.c.id == self.id ).execute( hid_counter=( next_hid + 1 ) )
+        table.update( table.c.id == self.id ).execute( hid_counter=( next_hid + n ) )
         trans.commit()
         return next_hid
     except:

lib/galaxy/security/__init__.py

@@ -898,7 +898,8 @@ class GalaxyRBACAgent( RBACAgent ):
         for action, roles in permissions.items():
             if isinstance( action, Action ):
                 action = action.action
-            for dp in [ self.model.DatasetPermissions( action, dataset, role ) for role in roles ]:
+            for role in roles:
+                dp = self.model.DatasetPermissions( action, dataset, role_id=role.id )
                 self.sa_session.add( dp )
                 flush_needed = True
         if flush_needed:

lib/galaxy/tools/__init__.py

@@ -39,6 +39,7 @@ from galaxy.tools.parser import get_tool_source
 from galaxy.tools.parser.xml import XmlPageSource
 from galaxy.tools.toolbox import AbstractToolBox
 from galaxy.util import rst_to_html, string_as_bool
+from galaxy.util import ExecutionTimer
 from galaxy.tools.parameters.meta import expand_meta_parameters
 from galaxy.util.bunch import Bunch
 from galaxy.util.expressions import ExpressionContext
@@ -1301,8 +1302,10 @@ class Tool( object, Dictifiable ):
         if rerun_remap_job_id and len( expanded_incomings ) > 1:
             raise exceptions.MessageException( 'Failure executing tool (cannot create multiple jobs when remapping existing job).' )
 
+        validation_timer = ExecutionTimer()
         all_errors = []
         all_params = []
+        validate_input = self.get_hook( 'validate_input' )
         for expanded_incoming in expanded_incomings:
             expanded_state = self.new_state( trans, history=history )
             # Process incoming data
@@ -1317,13 +1320,12 @@ class Tool( object, Dictifiable ):
                 # values from `incoming`.
                 errors = self.populate_state( trans, self.inputs, expanded_state.inputs, expanded_incoming, history, source=source )
                 # If the tool provides a `validate_input` hook, call it.
-                validate_input = self.get_hook( 'validate_input' )
                 if validate_input:
                     validate_input( trans, errors, expanded_state.inputs, self.inputs )
                 params = expanded_state.inputs
             all_errors.append( errors )
             all_params.append( params )
-
+        log.debug("Validated and populated state for tool request %s" % validation_timer)
         # If there were errors, we stay on the same page and display
         # error messages
         if any( all_errors ):

lib/galaxy/tools/actions/__init__.py

@@ -172,6 +172,8 @@ class DefaultToolAction( object ):
         # Set history.
         if not history:
             history = tool.get_default_history_by_trans( trans, create=True )
+        if history not in trans.sa_session:
+            history = trans.sa_session.query( trans.app.model.History ).get( history.id )
 
         out_data = odict()
         out_collections = {}
@@ -186,9 +188,7 @@ class DefaultToolAction( object ):
         input_names = []
         input_ext = 'data'
         input_dbkey = incoming.get( "dbkey", "?" )
-        inp_items = inp_data.items()
-        inp_items.reverse()
-        for name, data in inp_items:
+        for name, data in reversed(inp_data.items()):
             if not data:
                 data = NoneDataset( datatypes_registry=trans.app.datatypes_registry )
                 continue
@@ -396,15 +396,20 @@ class DefaultToolAction( object ):
                     handle_output( name, output )
                     log.info("Handled output %s" % handle_output_timer)
         # Add all the top-level (non-child) datasets to the history unless otherwise specified
+        add_to_history_timer = ExecutionTimer()
+        datasets_to_persist = []
         for name in out_data.keys():
             if name not in child_dataset_names and name not in incoming:  # don't add children; or already existing datasets, i.e. async created
                 data = out_data[ name ]
-                if set_output_history:
-                    # Set HID and add to history.
-                    # This is brand new and certainly empty so don't worry about quota.
-                    history.add_dataset( data, set_hid=set_output_hid, quota=False )
+                datasets_to_persist.append( data )
+        if set_output_history:
+            # Set HID and add to history.
+            # This is brand new and certainly empty so don't worry about quota.
+            history.add_datasets( trans.sa_session, datasets_to_persist, set_hid=set_output_hid, quota=False, flush=False )
+        else:
+            for data in datasets_to_persist:
                 trans.sa_session.add( data )
-                trans.sa_session.flush()
+        log.info("Add outputs to history %s" % add_to_history_timer)
         # Add all the children to their parents
         for parent_name, child_name in parent_to_child_pairs:
             parent_dataset = out_data[ parent_name ]
@@ -447,7 +452,10 @@ class DefaultToolAction( object ):
             if dataset:
                 if not trans.app.security_agent.can_access_dataset( current_user_roles, dataset.dataset ):
                     raise Exception("User does not have permission to use a dataset (%s) provided for input." % data.id)
-                job.add_input_dataset( name, dataset )
+                if dataset in trans.sa_session:
+                    job.add_input_dataset( name, dataset=dataset )
+                else:
+                    job.add_input_dataset( name, dataset_id=dataset.id )
             else:
                 job.add_input_dataset( name, None )
         log.info("Verified access to datasets %s" % access_timer)

lib/galaxy/tools/execute.py

@@ -7,6 +7,8 @@ import collections
 import galaxy.tools
 from galaxy.util import ExecutionTimer
 from galaxy.tools.actions import on_text_for_names
+from threading import Thread
+from Queue import Queue
 
 import logging
 log = logging.getLogger( __name__ )
@@ -20,7 +22,9 @@ def execute( trans, tool, param_combinations, history, rerun_remap_job_id=None,
     failures, etc...).
     """
     execution_tracker = ToolExecutionTracker( tool, param_combinations, collection_info )
-    for params in execution_tracker.param_combinations:
+    app = trans.app
+
+    def execute_single_job(params):
         job_timer = ExecutionTimer()
         if workflow_invocation_uuid:
             params[ '__workflow_invocation_uuid__' ] = workflow_invocation_uuid
@@ -36,6 +40,34 @@ def execute( trans, tool, param_combinations, history, rerun_remap_job_id=None,
         else:
             execution_tracker.record_error( result )
 
+    all_jobs_timer = ExecutionTimer()
+    config = app.config
+    burst_at = getattr( config, 'tool_submission_burst_at', 10 )
+    burst_threads = getattr( config, 'tool_submission_burst_threads', 1 )
+
+    if len(execution_tracker.param_combinations) < burst_at or burst_threads < 2:
+        for params in execution_tracker.param_combinations:
+            execute_single_job(params)
+    else:
+        q = Queue()
+
+        def worker():
+            while True:
+                params = q.get()
+                execute_single_job(params)
+                q.task_done()
+
+        for i in range(burst_threads):
+            t = Thread(target=worker)
+            t.daemon = True
+            t.start()
+
+        for params in execution_tracker.param_combinations:
+            q.put(params)
+
+        q.join()
+
+    log.debug("Executed all jobs for tool request: %s" % all_jobs_timer)
     if collection_info:
         history = history or tool.get_default_history_by_trans( trans )
         execution_tracker.create_output_collections( trans, history, params )
@@ -139,6 +171,8 @@ class ToolExecutionTracker( object ):
                 # TODO: Think through this, may only want this for output
                 # collections - or we may be already recording data in some
                 # other way.
+                if job not in trans.sa_session:
+                    job = trans.sa_session.query( trans.app.model.Job ).get( job.id )
                 job.add_output_dataset_collection( output_name, collection )
             collections[ output_name ] = collection
 

test/functional/tools/create_10.xml

@@ -0,0 +1,35 @@
+<tool id="create_10" name="Create 10">
+    <description>create 10</description>
+    <command>
+        echo "1" > 1;
+        echo "2" > 2;
+        echo "3" > 3;
+        echo "4" > 4;
+        echo "5" > 5;
+        echo "6" > 6;
+        echo "7" > 7;
+        echo "8" > 8;
+        echo "9" > 9;
+        echo "10" > 10;
+    </command>
+    <inputs>
+        <param name="input1" type="data" label="Concatenate Dataset"/>
+        <param name="input2" type="data" label="Concatenate Dataset"/>
+    </inputs>
+    <outputs>
+        <data name="out_file1" format="txt" from_work_dir="1" />
+        <data name="out_file2" format="txt" from_work_dir="2" />
+        <data name="out_file3" format="txt" from_work_dir="3" />
+        <data name="out_file4" format="txt" from_work_dir="4" />
+        <data name="out_file5" format="txt" from_work_dir="5" />
+        <data name="out_file6" format="txt" from_work_dir="6" />
+        <data name="out_file7" format="txt" from_work_dir="7" />
+        <data name="out_file8" format="txt" from_work_dir="8" />
+        <data name="out_file9" format="txt" from_work_dir="9" />
+        <data name="out_file10" format="txt" from_work_dir="10" />
+    </outputs>
+    <tests>
+    </tests>
+    <help>
+    </help>
+</tool>

test/functional/tools/samples_tool_conf.xml

@@ -33,6 +33,7 @@
   <tool file="output_filter.xml" />
   <tool file="output_collection_filter.xml" />
   <tool file="output_auto_format.xml" />
+  <tool file="create_10.xml" />
   <tool file="disambiguate_repeats.xml" />
   <tool file="min_repeat.xml" />
   <tool file="parallelism.xml" />