diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 87fc5894eb5..fd6571660f7 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -508,7 +508,7 @@ class PauseModule( WorkflowModule ): return state def execute( self, trans, progress, invocation, step ): - progress.mark_step_outputs_delayed( step ) + progress.mark_step_outputs_delayed( step, why="executing pause step" ) return None def recover_mapping( self, step, step_invocations, progress ): @@ -522,7 +522,8 @@ class PauseModule( WorkflowModule ): return elif action is False: raise CancelWorkflowEvaluation() - raise DelayedWorkflowEvaluation() + delayed_why = "workflow paused at this step waiting for review" + raise DelayedWorkflowEvaluation(why=delayed_why) def do_invocation_step_action( self, step, action ): """ Update or set the workflow invocation state action - generic @@ -834,7 +835,8 @@ class ToolModule( WorkflowModule ): workflow_invocation_uuid=invocation.uuid.hex ) except ToolInputsNotReadyException: - raise DelayedWorkflowEvaluation() + delayed_why = "tool [%s] inputs are not ready, this special tool requires inputs to be ready" % tool.id + raise DelayedWorkflowEvaluation(why=delayed_why) if collection_info: step_outputs = dict( execution_tracker.implicit_collections ) @@ -1020,7 +1022,9 @@ def load_module_sections( trans ): class DelayedWorkflowEvaluation(Exception): - pass + + def __init__(self, why=None): + self.why = why class CancelWorkflowEvaluation(Exception): diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 7810933b104..e47694501c2 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -176,9 +176,9 @@ class WorkflowInvoker( object ): # https://github.com/galaxyproject/galaxy/issues/2259 if job: workflow_invocation_step.job_id = job.id - except modules.DelayedWorkflowEvaluation: + except modules.DelayedWorkflowEvaluation as de: step_delayed = delayed_steps = True - self.progress.mark_step_outputs_delayed( step ) + self.progress.mark_step_outputs_delayed( step, why=de.why ) except Exception: log.exception( "Failed to schedule %s, problem occurred on %s.", @@ -187,8 +187,8 @@ class WorkflowInvoker( object ): ) raise - step_verb = "invoked" if not step_delayed else "delayed" - log.debug("Workflow step %s of invocation %s %s %s" % (step.id, workflow_invocation.id, step_verb, step_timer)) + if not step_delayed: + log.debug("Workflow step %s of invocation %s invoked %s" % (step.id, workflow_invocation.id, step_timer)) if delayed_steps: state = model.WorkflowInvocation.states.READY @@ -218,14 +218,16 @@ class WorkflowInvoker( object ): # No steps created yet - have to delay evaluation. if not step_invocations: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "depends on step [%s] but that step has not been invoked yet" % output_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) for step_invocation in step_invocations: job = step_invocation.job if job: # At least one job in incomplete. if not job.finished: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "depends on step [%s] but one or more jobs created from that step have not finished yet" % output_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) if job.state != job.states.OK: raise modules.CancelWorkflowEvaluation() @@ -303,7 +305,8 @@ class WorkflowProgress( object ): raise Exception(message) step_outputs = self.outputs[ output_step_id ] if step_outputs is STEP_OUTPUT_DELAYED: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "dependent step [%s] delayed, so this step must be delayed" % output_step_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) output_name = connection.output_name try: replacement = step_outputs[ output_name ] @@ -324,7 +327,8 @@ class WorkflowProgress( object ): # TODO: consider distinguish between cancelled and failed? raise modules.CancelWorkflowEvaluation() - raise modules.DelayedWorkflowEvaluation() + delayed_why = "dependent collection [%s] not yet populated with datasets" % replacement.id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) return replacement def get_replacement_workflow_output( self, workflow_output ): @@ -349,7 +353,10 @@ class WorkflowProgress( object ): def set_step_outputs(self, step, outputs): self.outputs[ step.id ] = outputs - def mark_step_outputs_delayed(self, step): + def mark_step_outputs_delayed(self, step, why=None): + if why: + message = "Marking step %s outputs of invocation %s delayed (%s)" % (step.id, self.workflow_invocation.id, why) + log.debug(message) self.outputs[ step.id ] = STEP_OUTPUT_DELAYED def _subworkflow_invocation(self, step): @@ -406,8 +413,8 @@ class WorkflowProgress( object ): def _recover_mapping( self, step, step_invocations ): try: step.module.recover_mapping( step, step_invocations, self ) - except modules.DelayedWorkflowEvaluation: - self.mark_step_outputs_delayed( step ) + except modules.DelayedWorkflowEvaluation as de: + self.mark_step_outputs_delayed( step, de.why ) __all__ = ( 'invoke', 'WorkflowRunConfig' ) diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index 408b2ee6166..67d80aa8bfa 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -857,8 +857,6 @@ test_data: @skip_without_tool( "cat1" ) @skip_without_tool( "collection_paired_test" ) def test_workflow_run_zip_collections( self ): - # A more advanced output collection workflow, testing regression of - # https://github.com/galaxyproject/galaxy/issues/776 history_id = self.dataset_populator.new_history() workflow_id = self._upload_yaml_workflow(""" class: GalaxyWorkflow