diff --git a/config/galaxy.ini.sample b/config/galaxy.ini.sample index f6fb9e4ef55..322ea506346 100644 --- a/config/galaxy.ini.sample +++ b/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 diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index 95457d21b25..dd3b16231ba 100644 --- a/lib/galaxy/config.py +++ b/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 diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index efd7850c7e0..3d6b0dd66eb 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/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: diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py index 3468ea693f6..406e38ddccf 100644 --- a/lib/galaxy/jobs/runners/__init__.py +++ b/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)) diff --git a/lib/galaxy/managers/collections.py b/lib/galaxy/managers/collections.py index ded32d1ba84..24dbdff1c13 100644 --- a/lib/galaxy/managers/collections.py +++ b/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=} try: diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 54518699a78..d03974e83de 100644 --- a/lib/galaxy/model/__init__.py +++ b/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 ): diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py index 42610cf0d07..ca8f75f63f8 100644 --- a/lib/galaxy/model/mapping.py +++ b/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: diff --git a/lib/galaxy/security/__init__.py b/lib/galaxy/security/__init__.py index 685bf3db20f..44ced674516 100644 --- a/lib/galaxy/security/__init__.py +++ b/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: diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 06db9dc4926..3f845e31c65 100755 --- a/lib/galaxy/tools/__init__.py +++ b/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 ): diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py index dda93560abd..7f8c36073d1 100644 --- a/lib/galaxy/tools/actions/__init__.py +++ b/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) diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index d7456617196..6b1512c2617 100644 --- a/lib/galaxy/tools/execute.py +++ b/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 diff --git a/test/functional/tools/create_10.xml b/test/functional/tools/create_10.xml new file mode 100644 index 00000000000..af872252d68 --- /dev/null +++ b/test/functional/tools/create_10.xml @@ -0,0 +1,35 @@ + + create 10 + + 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; + + + + + + + + + + + + + + + + + + + + + + diff --git a/test/functional/tools/samples_tool_conf.xml b/test/functional/tools/samples_tool_conf.xml index 6d4d89275ce..1d3ba1f9ccc 100644 --- a/test/functional/tools/samples_tool_conf.xml +++ b/test/functional/tools/samples_tool_conf.xml @@ -33,6 +33,7 @@ +