From 8b121ae31ca94c1c2a2c59b7f77da9c8097d2fa0 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Sat, 19 Mar 2016 22:17:28 +0000 Subject: [PATCH 1/3] Improve logging message in output_collect. Conflicts: lib/galaxy/tools/parameters/output_collect.py --- lib/galaxy/tools/parameters/output_collect.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/lib/galaxy/tools/parameters/output_collect.py b/lib/galaxy/tools/parameters/output_collect.py index 13119284c8a..95f2afd84b1 100644 --- a/lib/galaxy/tools/parameters/output_collect.py +++ b/lib/galaxy/tools/parameters/output_collect.py @@ -9,12 +9,12 @@ import json from galaxy import jobs from galaxy import util from galaxy.util import odict +from galaxy.util import ExecutionTimer from galaxy.tools.parser.output_collection_def import ( DEFAULT_DATASET_COLLECTOR_DESCRIPTION, INPUT_DBKEY_TOKEN, ) - DATASET_ID_TOKEN = "DATASET_ID" @@ -105,6 +105,7 @@ class JobContext( object ): filenames = self.find_files( collection, dataset_collectors ) for filename, extra_file_collector in filenames.iteritems(): + create_dataset_timer = ExecutionTimer() fields_match = extra_file_collector.match( collection, os.path.basename( filename ) ) if not fields_match: raise Exception( "Problem parsing metadata fields for file %s" % filename ) @@ -132,11 +133,12 @@ class JobContext( object ): metadata_source_name=output_collection_def.metadata_source, ) log.debug( - "(%s) Created dynamic collection dataset for path [%s] with element identifier [%s] for output [%s].", + "(%s) Created dynamic collection dataset for path [%s] with element identifier [%s] for output [%s] %s", self.job.id, filename, designation, output_collection_def.name, + create_dataset_timer, ) current_builder.add_dataset( element_identifiers[-1], dataset ) From 7c31a792fcdda60128848a53e9f567a99edb5c4a Mon Sep 17 00:00:00 2001 From: John Chilton Date: Sun, 20 Mar 2016 23:18:25 +0000 Subject: [PATCH 2/3] Allow add_datasets to use optimized even if quota is used... --- lib/galaxy/model/__init__.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 22929c29e76..dcd7d033589 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -1181,9 +1181,13 @@ class History( object, Dictifiable, UsesAnnotations, HasName ): interactions when adding many datasets to history at once. """ all_hdas = all( imap( is_hda, datasets ) ) - optimize = len( datasets) > 1 and parent_id is None and all_hdas and set_hid and not quota + optimize = len( datasets) > 1 and parent_id is None and all_hdas and set_hid if optimize: self.__add_datasets_optimized( datasets, genome_build=genome_build ) + if quota and self.user: + disk_usage = sum([d.get_total_size() for d in datasets]) + self.user.adjust_total_disk_usage(disk_usage) + sa_session.add_all( datasets ) if flush: sa_session.flush() From ac14ad0897d4abb7875ad16a49dda7b6a43e3c84 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Sat, 19 Mar 2016 20:52:56 +0000 Subject: [PATCH 3/3] Optimization for discovering datasets for dynamic output collection. Batch up job creation and adding datasets to the history in order to eliminate a bunch of database flushes. Using the create_input_collection tool from PR #1958 to create an output collection with 100 elements - the optimization reduces the job finish time from 45 seconds to 15 seconds. Before: ``` galaxy.jobs DEBUG 2016-03-21 01:51:01,377 job 24 ended (finish() executed in (15474.472 ms)) ``` After: ``` galaxy.jobs DEBUG 2016-03-21 01:50:06,659 job 23 ended (finish() executed in (46430.736 ms)) ``` --- lib/galaxy/security/__init__.py | 4 +- lib/galaxy/tools/parameters/output_collect.py | 92 +++++++++++++------ 2 files changed, 65 insertions(+), 31 deletions(-) diff --git a/lib/galaxy/security/__init__.py b/lib/galaxy/security/__init__.py index 44ced674516..3f2b6605c44 100644 --- a/lib/galaxy/security/__init__.py +++ b/lib/galaxy/security/__init__.py @@ -871,7 +871,7 @@ class GalaxyRBACAgent( RBACAgent ): permissions[ action ] = [ dhp.role ] return permissions - def set_all_dataset_permissions( self, dataset, permissions={}, new=False ): + def set_all_dataset_permissions( self, dataset, permissions={}, new=False, flush=True ): """ Set new full permissions on a dataset, eliminating all current permissions. Permission looks like: { Action : [ Role, Role ] } @@ -902,7 +902,7 @@ class GalaxyRBACAgent( RBACAgent ): dp = self.model.DatasetPermissions( action, dataset, role_id=role.id ) self.sa_session.add( dp ) flush_needed = True - if flush_needed: + if flush_needed and flush: self.sa_session.flush() return "" diff --git a/lib/galaxy/tools/parameters/output_collect.py b/lib/galaxy/tools/parameters/output_collect.py index 95f2afd84b1..1fb52296c77 100644 --- a/lib/galaxy/tools/parameters/output_collect.py +++ b/lib/galaxy/tools/parameters/output_collect.py @@ -104,15 +104,13 @@ class JobContext( object ): dataset_collectors = map(dataset_collector, output_collection_def.dataset_collector_descriptions) filenames = self.find_files( collection, dataset_collectors ) + element_datasets = [] for filename, extra_file_collector in filenames.iteritems(): create_dataset_timer = ExecutionTimer() fields_match = extra_file_collector.match( collection, os.path.basename( filename ) ) if not fields_match: raise Exception( "Problem parsing metadata fields for file %s" % filename ) element_identifiers = fields_match.element_identifiers - current_builder = root_collection_builder - for element_identifier in element_identifiers[:-1]: - current_builder = current_builder.get_level(element_identifier) designation = fields_match.designation visible = fields_match.visible ext = fields_match.ext @@ -140,8 +138,40 @@ class JobContext( object ): output_collection_def.name, create_dataset_timer, ) + element_datasets.append((element_identifiers, dataset)) + + app = self.app + sa_session = self.sa_session + job = self.job + + if job: + add_datasets_timer = ExecutionTimer() + job.history.add_datasets(sa_session, [d for (ei, d) in element_datasets]) + log.debug( + "(%s) Add dynamic collection datsets to history for output [%s] %s", + self.job.id, + output_collection_def.name, + add_datasets_timer, + ) + + for (element_identifiers, dataset) in element_datasets: + current_builder = root_collection_builder + for element_identifier in element_identifiers[:-1]: + current_builder = current_builder.get_level(element_identifier) current_builder.add_dataset( element_identifiers[-1], dataset ) + # Associate new dataset with job + if job: + element_identifier_str = ":".join(element_identifiers) + # Below was changed from '__new_primary_file_%s|%s__' % ( name, designation ) + assoc = app.model.JobToOutputDatasetAssociation( '__new_primary_file_%s|%s__' % ( name, element_identifier_str ), dataset ) + assoc.job = self.job + sa_session.add( assoc ) + + dataset.raw_set_dataset_state('ok') + + sa_session.flush() + def create_dataset( self, ext, @@ -155,20 +185,13 @@ class JobContext( object ): app = self.app sa_session = self.sa_session + primary_data = _new_hda(app, sa_session, ext, designation, visible, dbkey, self.permissions) + # Copy metadata from one of the inputs if requested. metadata_source = None if metadata_source_name: metadata_source = self.inp_data[ metadata_source_name ] - # Create new primary dataset - primary_data = app.model.HistoryDatasetAssociation( extension=ext, - designation=designation, - visible=visible, - dbkey=dbkey, - create_dataset=True, - sa_session=sa_session ) - app.security_agent.set_all_dataset_permissions( primary_data.dataset, self.permissions ) - sa_session.add( primary_data ) sa_session.flush() # Move data from temp location to dataset location app.object_store.update_from_file(primary_data.dataset, file_name=filename, create=True) @@ -182,16 +205,6 @@ class JobContext( object ): else: primary_data.init_meta() - # Associate new dataset with job - if self.job: - self.job.history.add_dataset( primary_data ) - - assoc = app.model.JobToOutputDatasetAssociation( '__new_primary_file_%s|%s__' % ( name, designation ), primary_data ) - assoc.job = self.job - sa_session.add( assoc ) - sa_session.flush() - - primary_data.state = 'ok' return primary_data @@ -254,14 +267,8 @@ def collect_primary_datasets( tool, output, job_working_directory, input_ext, in if dbkey == INPUT_DBKEY_TOKEN: dbkey = input_dbkey # Create new primary dataset - primary_data = app.model.HistoryDatasetAssociation( extension=ext, - designation=designation, - visible=visible, - dbkey=dbkey, - create_dataset=True, - sa_session=sa_session ) + primary_data = _new_hda(app, sa_session, ext, designation, visible, dbkey) app.security_agent.copy_dataset_permissions( outdata.dataset, primary_data.dataset ) - sa_session.add( primary_data ) sa_session.flush() # Move data from temp location to dataset location app.object_store.update_from_file(primary_data.dataset, file_name=filename, create=True) @@ -485,5 +492,32 @@ class CollectedDatasetMatch( object ): except IndexError: return self.collector.default_visible +UNSET = object() + + +def _new_hda( + app, + sa_session, + ext, + designation, + visible, + dbkey, + permissions=UNSET, +): + """Return a new unflushed HDA with dataset and permissions setup. + """ + # Create new primary dataset + primary_data = app.model.HistoryDatasetAssociation( extension=ext, + designation=designation, + visible=visible, + dbkey=dbkey, + create_dataset=True, + flush=False, + sa_session=sa_session ) + if permissions is not UNSET: + app.security_agent.set_all_dataset_permissions( primary_data.dataset, permissions, new=True, flush=False ) + sa_session.add( primary_data ) + return primary_data + DEFAULT_DATASET_COLLECTOR = DatasetCollector(DEFAULT_DATASET_COLLECTOR_DESCRIPTION)