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

Generalize tracking objects to add to history

Browse Source 
This will use the optimized route of minimizing flushes during execution
handling for all database operation tools. This also seems signifcantly
simpler than passing around the datasets_to_persist list.
This commit is contained in:
mvdbeek
2020-10-26 12:01:59 +01:00
parent b3ace06a60
commit 12d80f7159
6 changed files with 51 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/model/__init__.py
+23
View File
@@ -40,6 +40,7 @@ from sqlalchemy.orm import (
joinedload,
object_session,
Query,
reconstructor,
)
from sqlalchemy.schema import UniqueConstraint
@@ -1757,11 +1758,33 @@ class History(HasTags, Dictifiable, UsesAnnotations, HasName, RepresentById):
self.datasets = []
self.galaxy_sessions = []
self.tags = []
# Objects to eventually add to history
self._pending_additions = []
@reconstructor
def init_on_load(self):
# Restores properties that are not tracked in the database
self._pending_additions = []
def stage_addition(self, items):
history_id = self.id
for item in listify(items):
if history_id:
item.history_id = history_id
else:
item.history = self
self._pending_additions.append(item)
@property
def empty(self):
return self.hid_counter == 1
def add_pending_datasets(self, set_output_hid=True):
# These are assumed to be either copies of existing datasets or new, empty datasets,
# so we don't need to set the quota.
self.add_datasets(object_session(self), self._pending_additions, set_hid=set_output_hid, quota=False, flush=False)
self._pending_additions = []
def _next_hid(self, n=1):
# this is overriden in mapping.py db_next_hid() method
if len(self.datasets) == 0:
lib/galaxy/tools/__init__.py
+9 -16
View File
@@ -1646,9 +1646,8 @@ class Tool(Dictifiable):
)
job = rval[0]
out_data = rval[1]
if len(rval) == 4:
execution_slice.datasets_to_persist = rval[2]
execution_slice.history = rval[3]
if len(rval) > 2:
execution_slice.history = rval[2]
except (webob.exc.HTTPFound, exceptions.MessageException) as e:
# if it's a webob redirect exception, pass it up the stack
raise e
@@ -2777,14 +2776,10 @@ class DatabaseOperationTool(Tool):
check_dataset_state(state)
def _add_datasets_to_history(self, history, elements, datasets_visible=False):
datasets = []
for element_object in elements:
if getattr(element_object, "history_content_type", None) == "dataset":
element_object.visible = datasets_visible
datasets.append(element_object)
if datasets:
history.add_datasets(self.sa_session, datasets, set_hid=True)
history.stage_addition(element_object)
def produce_outputs(self, trans, out_data, output_collections, incoming, history, **kwds):
return self._outputs_dict()
@@ -3156,25 +3151,23 @@ class ApplyRulesTool(DatabaseOperationTool):
def produce_outputs(self, trans, out_data, output_collections, incoming, history, tag_handler, **kwds):
hdca = incoming["input"]
rule_set = RuleSet(incoming["rules"])
datasets_to_persist = []
copied_datasets = []
def copy_dataset(dataset, tags):
copied_dataset = dataset.copy(flush=False)
if tags is not None:
tag_handler.set_tags_from_list(trans.get_user(), copied_dataset, tags, flush=False)
copied_dataset.history_id = history.id
datasets_to_persist.append(copied_dataset)
copied_datasets.append(copied_dataset)
return copied_dataset
new_elements = self.app.dataset_collections_service.apply_rules(
hdca, rule_set, copy_dataset
)
hdca = output_collections.create_collection(
next(iter(self.outputs.values())), "output", collection_type=rule_set.collection_type, elements=new_elements, flush=False, set_hid=False,
self._add_datasets_to_history(history, copied_datasets)
output_collections.create_collection(
next(iter(self.outputs.values())), "output", collection_type=rule_set.collection_type, elements=new_elements,
)
if hdca:
datasets_to_persist.append(hdca)
return datasets_to_persist
class TagFromFileTool(DatabaseOperationTool):
@@ -3270,10 +3263,10 @@ class FilterFromFileTool(DatabaseOperationTool):
discarded_elements[element_identifier] = copied_value
self._add_datasets_to_history(history, filtered_elements.values())
self._add_datasets_to_history(history, discarded_elements.values())
output_collections.create_collection(
self.outputs["output_filtered"], "output_filtered", elements=filtered_elements
)
self._add_datasets_to_history(history, discarded_elements.values())
output_collections.create_collection(
self.outputs["output_discarded"], "output_discarded", elements=discarded_elements
)
lib/galaxy/tools/actions/__init__.py
+11 -18
View File
@@ -459,7 +459,6 @@ class DefaultToolAction:
# Flush all datasets at once.
return data
datasets_to_persist = []
for name, output in tool.outputs.items():
if not filter_output(output, incoming):
handle_output_timer = ExecutionTimer()
@@ -496,7 +495,7 @@ class DefaultToolAction:
effective_output_name = output_part_def.effective_output_name
element = handle_output(effective_output_name, output_part_def.output_def, hidden=True)
datasets_to_persist.append(element)
history.stage_addition(element)
# TODO: this shouldn't exist in the top-level of the history at all
# but for now we are still working around that by hiding the contents
# there.
@@ -513,17 +512,13 @@ class DefaultToolAction:
element_kwds = dict(elements=collections_manager.ELEMENTS_UNINITIALIZED)
else:
element_kwds = dict(element_identifiers=element_identifiers)
hdca = output_collections.create_collection(
output_collections.create_collection(
output=output,
name=name,
set_hid=True if flush_job else False,
flush=flush_job,
completed_job=completed_job,
**element_kwds
)
if hdca:
datasets_to_persist.append(hdca)
log.info(f"Handled collection output named {name} for tool {tool.id} {handle_output_timer}")
log.info("Handled collection output named {} for tool {} {}".format(name, tool.id, handle_output_timer))
else:
handle_output(name, output)
log.info(f"Handled output named {name} for tool {tool.id} {handle_output_timer}")
@@ -535,7 +530,7 @@ class DefaultToolAction:
# Add all the top-level (non-child) datasets to the history unless otherwise specified
for name, data in out_data.items():
if name not in child_dataset_names and name not in incoming: # don't add children; or already existing datasets, i.e. async created
datasets_to_persist.append(data)
history.stage_addition(data)
# Add all the children to their parents
for parent_name, child_name in parent_to_child_pairs:
@@ -560,14 +555,13 @@ class DefaultToolAction:
if app.config.track_jobs_in_database and rerun_remap_job_id is not None:
# We need a flush here and get hids in order to rewrite jobs parameter,
# but remapping jobs should only affect single jobs anyway, so this is not too costly.
history.add_pending_datasets(set_output_hid=set_output_hid)
trans.sa_session.flush()
history.add_datasets(trans.sa_session, datasets_to_persist, set_hid=set_output_hid, quota=False, flush=False)
self._remap_job_on_rerun(trans=trans,
galaxy_session=galaxy_session,
rerun_remap_job_id=rerun_remap_job_id,
current_job=job,
out_data=out_data)
datasets_to_persist = []
log.info(f"Setup for job {job.log_str()} complete, ready to be enqueued {job_setup_timer}")
# Some tools are not really executable, but jobs are still created for them ( for record keeping ).
@@ -593,8 +587,7 @@ class DefaultToolAction:
else:
if flush_job:
# 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)
history.add_pending_datasets(set_output_hid=set_output_hid)
job_flush_timer = ExecutionTimer()
trans.sa_session.flush()
log.info(f"Flushed transaction for job {job.log_str()} {job_flush_timer}")
@@ -602,7 +595,7 @@ class DefaultToolAction:
# Dispatch to a job handler. enqueue() is responsible for flushing the job
app.job_manager.enqueue(job, tool=tool)
trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
return job, out_data, datasets_to_persist, history
return job, out_data, history
def _remap_job_on_rerun(self, trans, galaxy_session, rerun_remap_job_id, current_job, out_data):
"""
@@ -825,7 +818,7 @@ class OutputCollections:
self.out_collection_instances = {}
self.tags = tags
def create_collection(self, output, name, collection_type=None, set_hid=True, flush=True, completed_job=None, **element_kwds):
def create_collection(self, output, name, collection_type=None, completed_job=None, **element_kwds):
input_collections = self.input_collections
collections_manager = self.trans.app.dataset_collections_service
collection_type = collection_type or output.structure.collection_type
@@ -904,16 +897,16 @@ class OutputCollections:
collection_type=collection_type,
trusted_identifiers=True,
tags=self.tags,
set_hid=set_hid,
flush=flush,
set_hid=False,
flush=False,
completed_job=completed_job,
output_name=name,
**element_kwds
)
# name here is name of the output element - not name
# of the hdca.
self.history.stage_addition(hdca)
self.out_collection_instances[name] = hdca
return hdca
def on_text_for_names(input_names):
lib/galaxy/tools/actions/model_operations.py
+3 -4
View File
@@ -57,7 +57,7 @@ class ModelOperationToolAction(DefaultToolAction):
# Create job.
#
job, galaxy_session = self._new_job_for_session(trans, tool, history)
datasets_to_persist = self._produce_outputs(trans, tool, out_data, output_collections, incoming=incoming, history=history, tags=preserved_tags)
self._produce_outputs(trans, tool, out_data, output_collections, incoming=incoming, history=history, tags=preserved_tags)
self._record_inputs(trans, tool, job, incoming, inp_data, inp_dataset_collections)
self._record_outputs(job, out_data, output_collections)
job.state = job.states.OK
@@ -67,11 +67,11 @@ class ModelOperationToolAction(DefaultToolAction):
# trans.app.job_manager.job_queue.put( job.id, tool.id )
# trans.log_event( "Added database job action to the job queue, id: %s" % str(job.id), tool_id=job.tool_id )
log.info("Calling produce_outputs, tool is %s" % tool)
return job, out_data, datasets_to_persist, history
return job, out_data, history
def _produce_outputs(self, trans, tool, out_data, output_collections, incoming, history, tags):
tag_handler = trans.app.tag_handler.create_tag_handler_session()
datasets_to_persist = tool.produce_outputs(trans, out_data, output_collections, incoming, history=history, tags=tags, tag_handler=tag_handler)
tool.produce_outputs(trans, out_data, output_collections, incoming, history=history, tags=tags, tag_handler=tag_handler)
mapped_over_elements = output_collections.dataset_collection_elements
if mapped_over_elements:
for name, value in out_data.items():
@@ -80,4 +80,3 @@ class ModelOperationToolAction(DefaultToolAction):
mapped_over_elements[name].hda = value
trans.sa_session.add_all(out_data.values())
return datasets_to_persist
lib/galaxy/tools/execute.py
+4 -7
View File
@@ -90,7 +90,7 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle
jobs_executed = 0
has_remaining_jobs = False
datasets_to_persist = []
execution_slice = None
for i, execution_slice in enumerate(execution_tracker.new_execution_slices()):
if max_num_jobs and jobs_executed >= max_num_jobs:
@@ -100,12 +100,10 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle
execute_single_job(execution_slice, completed_jobs[i])
history = execution_slice.history or history
jobs_executed += 1
if execution_slice.datasets_to_persist:
datasets_to_persist.extend(execution_slice.datasets_to_persist)
if datasets_to_persist:
history.add_datasets(trans.sa_session, datasets_to_persist, set_hid=True, quota=False, flush=False)
# a side effect of history.add_datasets is a commit within db_next_hid (even with flush=False).
if execution_slice:
# a side effect of adding datasets to a history is a commit within db_next_hid (even with flush=False).
history.add_pending_datasets()
else:
# Make sure collections, implicit jobs etc are flushed even if there are no precreated output datasets
trans.sa_session.flush()
@@ -130,7 +128,6 @@ class ExecutionSlice:
self.job_index = job_index
self.param_combination = param_combination
self.dataset_collection_elements = dataset_collection_elements
self.datasets_to_persist = None
self.history = None
test/unit/tools/test_actions.py
+1 -1
View File
@@ -134,7 +134,7 @@ class DefaultToolActionTestCase(unittest.TestCase, tools_support.UsesApp, tools_
if incoming is None:
incoming = dict(param1="moo")
self._init_tool(contents)
job, out_data, _, _ = self.action.execute(
job, out_data, _, = self.action.execute(
tool=self.tool,
trans=self.trans,
history=self.history,