opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1959 from jmchilton/optimize_dataset_collection_discovery

Browse Source 
Optimization for discovering datasets for dynamic output collection.
This commit is contained in:
Martin Cech
2016-04-05 10:27:34 -04:00
parent 497cd024da ac14ad0897
commit 1543f0d925
3 changed files with 74 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/model/__init__.py
+5 -1
View File
@@ -1186,9 +1186,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()
lib/galaxy/security/__init__.py
+2 -2
View File
@@ -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 ""
lib/galaxy/tools/parameters/output_collect.py
+67 -31
View File
@@ -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"
@@ -104,14 +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
@@ -132,14 +131,47 @@ 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,
)
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,
@@ -153,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)
@@ -180,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
@@ -252,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)
@@ -483,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)