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 190c2d9d67e..0619f19d6df 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -147,6 +147,7 @@ class WorkflowInvoker( object ): workflow_invocation = self.workflow_invocation remaining_steps = self.progress.remaining_steps() delayed_steps = False + more_info = "" for step in remaining_steps: step_delayed = False step_timer = ExecutionTimer() @@ -165,8 +166,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 + more_info = "(%s)" % de.why self.progress.mark_step_outputs_delayed( step ) except Exception: log.exception( @@ -177,7 +179,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)) + status = step_verb + (" %s" % more_info if more_info else "") + log.debug("Workflow step %s of invocation %s %s %s" % (step.id, workflow_invocation.id, status, step_timer)) if delayed_steps: state = model.WorkflowInvocation.states.READY @@ -207,14 +210,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() @@ -292,7 +297,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 ] @@ -313,7 +319,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 ): @@ -338,7 +345,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 delayed (%s)" % (step.id, why) + log.debug(message) self.outputs[ step.id ] = STEP_OUTPUT_DELAYED def _subworkflow_invocation(self, step): @@ -395,8 +405,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 38261d06812..2c42a528596 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -869,8 +869,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