From 12d80f71595ccaad284f7d6ef264fe78662ba93c Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Sun, 27 Sep 2020 17:52:45 +0200 Subject: [PATCH] Generalize tracking objects to add to history 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. --- lib/galaxy/model/__init__.py | 23 ++++++++++++++++ lib/galaxy/tools/__init__.py | 25 ++++++----------- lib/galaxy/tools/actions/__init__.py | 29 ++++++++------------ lib/galaxy/tools/actions/model_operations.py | 7 ++--- lib/galaxy/tools/execute.py | 11 +++----- test/unit/tools/test_actions.py | 2 +- 6 files changed, 51 insertions(+), 46 deletions(-) diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index be537bed3df..db92f3f64e8 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -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: diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 33bc0df6993..f6517f45f80 100644 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -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 ) diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py index 6d517914c78..138c503268d 100644 --- a/lib/galaxy/tools/actions/__init__.py +++ b/lib/galaxy/tools/actions/__init__.py @@ -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): diff --git a/lib/galaxy/tools/actions/model_operations.py b/lib/galaxy/tools/actions/model_operations.py index 81f545240ca..f45dd4c6baa 100644 --- a/lib/galaxy/tools/actions/model_operations.py +++ b/lib/galaxy/tools/actions/model_operations.py @@ -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 diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index 86008f91544..c8f0a1cff3c 100644 --- a/lib/galaxy/tools/execute.py +++ b/lib/galaxy/tools/execute.py @@ -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 diff --git a/test/unit/tools/test_actions.py b/test/unit/tools/test_actions.py index d8d6e69bbf8..3f7347ee286 100644 --- a/test/unit/tools/test_actions.py +++ b/test/unit/tools/test_actions.py @@ -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,