python/ray/workflow/step_executor.py
def _execute_workflow(workflow: "Workflow") -> "WorkflowExecutionResult":
    """Internal function of workflow execution."""
    if workflow.executed:
        return workflow.result

    # Stage 1: prepare inputs
    workflow_data = workflow.data
    inputs = workflow_data.inputs
    # Here A is the outer workflow step, B & C are the inner steps.
    # C is the output step for A, because C produces the output for A.
    #
    # @workflow.step
    # def A():
    #     b = B.step()
    #     return C.step(b)
    #
    # If the outer workflow step skips checkpointing, it would
    # update the checkpoint context of all inner steps except
    # the output step, marking them "detached" from the DAG.
    # Output step is not detached from the DAG because once
    # completed, it replaces the output of the outer step.
    step_context = workflow_context.get_workflow_step_context()
    checkpoint_context = step_context.checkpoint_context.copy()
    # "detached" could be defined recursively:
    # detached := already detached or the outer step skips checkpointing
    checkpoint_context.detached_from_dag = (
        checkpoint_context.detached_from_dag
        or not step_context.checkpoint_context.checkpoint
    )
    # Apply checkpoint context to input steps. Since input steps
    # further apply them to their inputs, this would eventually
    # apply to all steps except the output step. This avoids
    # detaching the output step.
    workflow_outputs = []
    with workflow_context.fork_workflow_step_context(
        outer_most_step_id=None,
        last_step_of_workflow=False,
        checkpoint_context=checkpoint_context,
    ):
        for w in inputs.workflows:
            static_ref = w.ref
            if static_ref is None:
                extra_options = w.data.step_options.ray_options
                # The input workflow is not a reference to an executed
                # workflow.
                static_ref = execute_workflow(w).persisted_output
                static_ref._resolve_like_object_ref_in_args = extra_options.get(
                    "_resolve_like_object_ref_in_args", False
                )
            workflow_outputs.append(static_ref)

    baked_inputs = _BakedWorkflowInputs(
        args=inputs.args,
        workflow_outputs=workflow_outputs,
        workflow_refs=inputs.workflow_refs,
    )

    # Stage 2: match executors
    step_options = workflow_data.step_options
    if step_options.allow_inplace:
        # TODO(suquark): For inplace execution, it is impossible
        # to get the ObjectRef of the output before execution.
        # Here we use a dummy ObjectRef, because _record_step_status does not
        # even use it (?!).
        _record_step_status(workflow.step_id, WorkflowStatus.RUNNING, [ray.put(None)])
        # Note: we need to be careful about workflow context when
        # calling the executor directly.
        # TODO(suquark): We still have recursive Python calls.
        # This would cause stack overflow if we have a really
        # deep recursive call. We should fix it later.
        if step_options.step_type == StepType.WAIT:
            executor = _workflow_wait_executor
        else:
            # Tell the executor that we are running inplace. This enables
            # tail-recursion optimization.
            executor = functools.partial(_workflow_step_executor, inplace=True)
    else:
        if step_options.step_type == StepType.WAIT:
            # This is very important to set "num_cpus=0" to
            # ensure "workflow.wait" is not blocked by other
            # tasks.
            executor = _workflow_wait_executor_remote.options(num_cpus=0).remote
        else:
            ray_options = step_options.ray_options.copy()
            # cleanup the "_resolve_like_object_ref_in_args" option, it is not for Ray.
            ray_options.pop("_resolve_like_object_ref_in_args", None)
            executor = _workflow_step_executor_remote.options(**ray_options).remote

    # Stage 3: execution
    persisted_output, volatile_output = executor(
        workflow_data.func_body,
        step_context,
        workflow.step_id,
        baked_inputs,
        workflow_data.step_options,
    )

    # Stage 4: post processing outputs
    if step_options.step_type != StepType.READONLY_ACTOR_METHOD:
        if not step_options.allow_inplace:
            # TODO: [Possible flaky bug] Here the RUNNING state may
            # be recorded earlier than SUCCESSFUL. This caused some
            # confusion during development.

            # convert into workflow static ref for step status record.
            volatile_output = WorkflowStaticRef.from_output(
                workflow.step_id, volatile_output
            )
            _record_step_status(
                workflow.step_id, WorkflowStatus.RUNNING, [volatile_output]
            )

    result = WorkflowExecutionResult(persisted_output, volatile_output)
    workflow._result = result
    workflow._executed = True
    return result
Internal function of workflow execution.
Internal function of workflow execution.
Internal function of workflow execution .
def _execute_workflow(workflow: "Workflow") -> "WorkflowExecutionResult":
    if workflow.executed:
        return workflow.result
    workflow_data = workflow.data
    inputs = workflow_data.inputs
     @workflow.step
     def A():
         b = B.step()
         return C.step(b)
     If the outer workflow step skips checkpointing, it would
     update the checkpoint context of all inner steps except
     the output step, marking them "detached" from the DAG.
     Output step is not detached from the DAG because once
     completed, it replaces the output of the outer step.
    step_context = workflow_context.get_workflow_step_context()
    checkpoint_context = step_context.checkpoint_context.copy()
     "detached" could be defined recursively:
     detached := already detached or the outer step skips checkpointing
    checkpoint_context.detached_from_dag = (
        checkpoint_context.detached_from_dag
        or not step_context.checkpoint_context.checkpoint
    )
     Apply checkpoint context to input steps. Since input steps
     further apply them to their inputs, this would eventually
     apply to all steps except the output step. This avoids
     detaching the output step.
    workflow_outputs = []
    with workflow_context.fork_workflow_step_context(
        outer_most_step_id=None,
        last_step_of_workflow=False,
        checkpoint_context=checkpoint_context,
    ):
        for w in inputs.workflows:
            static_ref = w.ref
            if static_ref is None:
                extra_options = w.data.step_options.ray_options
                 The input workflow is not a reference to an executed
                 workflow.
                static_ref = execute_workflow(w).persisted_output
                static_ref._resolve_like_object_ref_in_args = extra_options.get(
                    "_resolve_like_object_ref_in_args", False
                )
            workflow_outputs.append(static_ref)
    baked_inputs = _BakedWorkflowInputs(
        args=inputs.args,
        workflow_outputs=workflow_outputs,
        workflow_refs=inputs.workflow_refs,
    )
     Stage 2: match executors
    step_options = workflow_data.step_options
    if step_options.allow_inplace:
         TODO(suquark): For inplace execution, it is impossible
         to get the ObjectRef of the output before execution.
         Here we use a dummy ObjectRef, because _record_step_status does not
         even use it (?!).
        _record_step_status(workflow.step_id, WorkflowStatus.RUNNING, [ray.put(None)])
         Note: we need to be careful about workflow context when
         calling the executor directly.
         TODO(suquark): We still have recursive Python calls.
         This would cause stack overflow if we have a really
         deep recursive call. We should fix it later.
        if step_options.step_type == StepType.WAIT:
            executor = _workflow_wait_executor
        else:
             Tell the executor that we are running inplace. This enables
             tail-recursion optimization.
            executor = functools.partial(_workflow_step_executor, inplace=True)
    else:
        if step_options.step_type == StepType.WAIT:
             This is very important to set "num_cpus=0" to
             ensure "workflow.wait" is not blocked by other
             tasks.
            executor = _workflow_wait_executor_remote.options(num_cpus=0).remote
        else:
            ray_options = step_options.ray_options.copy()
             cleanup the "_resolve_like_object_ref_in_args" option, it is not for Ray.
            ray_options.pop("_resolve_like_object_ref_in_args", None)
            executor = _workflow_step_executor_remote.options(**ray_options).remote
     Stage 3: execution
    persisted_output, volatile_output = executor(
        workflow_data.func_body,
        step_context,
        workflow.step_id,
        baked_inputs,
        workflow_data.step_options,
    )
     Stage 4: post processing outputs
    if step_options.step_type != StepType.READONLY_ACTOR_METHOD:
        if not step_options.allow_inplace:
             TODO: [Possible flaky bug] Here the RUNNING state may
             be recorded earlier than SUCCESSFUL. This caused some
             confusion during development.
             convert into workflow static ref for step status record.
            volatile_output = WorkflowStaticRef.from_output(
                workflow.step_id, volatile_output
            )
            _record_step_status(
                workflow.step_id, WorkflowStatus.RUNNING, [volatile_output]
            )
    result = WorkflowExecutionResult(persisted_output, volatile_output)
    workflow._result = result
    workflow._executed = True
    return result
def _execute_workflow ( workflow : "Workflow" ) -> "WorkflowExecutionResult" : if workflow . executed : return workflow . result workflow_data = workflow . data inputs = workflow_data . inputs step_context = workflow_context . get_workflow_step_context ( ) checkpoint_context = step_context . checkpoint_context . copy ( ) checkpoint_context . detached_from_dag = ( checkpoint_context . detached_from_dag or not step_context . checkpoint_context . checkpoint ) workflow_outputs = [ ] with workflow_context . fork_workflow_step_context ( outer_most_step_id = None , last_step_of_workflow = False , checkpoint_context = checkpoint_context , ) : for w in inputs . workflows : static_ref = w . ref if static_ref is None : extra_options = w . data . step_options . ray_options static_ref = execute_workflow ( w ) . persisted_output static_ref . _resolve_like_object_ref_in_args = extra_options . get ( "_resolve_like_object_ref_in_args" , False ) workflow_outputs . append ( static_ref ) baked_inputs = _BakedWorkflowInputs ( args = inputs . args , workflow_outputs = workflow_outputs , workflow_refs = inputs . workflow_refs , ) step_options = workflow_data . step_options if step_options . allow_inplace : _record_step_status ( workflow . step_id , WorkflowStatus . RUNNING , [ ray . put ( None ) ] ) if step_options . step_type == StepType . WAIT : executor = _workflow_wait_executor else : executor = functools . partial ( _workflow_step_executor , inplace = True ) else : if step_options . step_type == StepType . WAIT : executor = _workflow_wait_executor_remote . options ( num_cpus = 0 ) . remote else : ray_options = step_options . ray_options . copy ( ) ray_options . pop ( "_resolve_like_object_ref_in_args" , None ) executor = _workflow_step_executor_remote . options ( ** ray_options ) . remote persisted_output , volatile_output = executor ( workflow_data . func_body , step_context , workflow . step_id , baked_inputs , workflow_data . step_options , ) if step_options . step_type != StepType . READONLY_ACTOR_METHOD : if not step_options . allow_inplace : volatile_output = WorkflowStaticRef . from_output ( workflow . step_id , volatile_output ) _record_step_status ( workflow . step_id , WorkflowStatus . RUNNING , [ volatile_output ] ) result = WorkflowExecutionResult ( persisted_output , volatile_output ) workflow . _result = result workflow . _executed = True return result
Internal function of workflow execution.
Internal function of workflow execution .
"""Internal function of workflow execution.""" # Stage 1: prepare inputs # Here A is the outer workflow step, B & C are the inner steps. # C is the output step for A, because C produces the output for A. # # @workflow.step # def A(): #     b = B.step() #     return C.step(b) # # If the outer workflow step skips checkpointing, it would # update the checkpoint context of all inner steps except # the output step, marking them "detached" from the DAG. # Output step is not detached from the DAG because once # completed, it replaces the output of the outer step. # "detached" could be defined recursively: # detached := already detached or the outer step skips checkpointing # Apply checkpoint context to input steps. Since input steps # further apply them to their inputs, this would eventually # apply to all steps except the output step. This avoids # detaching the output step. # The input workflow is not a reference to an executed # workflow. # Stage 2: match executors # TODO(suquark): For inplace execution, it is impossible # to get the ObjectRef of the output before execution. # Here we use a dummy ObjectRef, because _record_step_status does not # even use it (?!). # Note: we need to be careful about workflow context when # calling the executor directly. # TODO(suquark): We still have recursive Python calls. # This would cause stack overflow if we have a really # deep recursive call. We should fix it later. # Tell the executor that we are running inplace. This enables # tail-recursion optimization. # This is very important to set "num_cpus=0" to # ensure "workflow.wait" is not blocked by other # tasks. # cleanup the "_resolve_like_object_ref_in_args" option, it is not for Ray. # Stage 3: execution # Stage 4: post processing outputs # TODO: [Possible flaky bug] Here the RUNNING state may # be recorded earlier than SUCCESSFUL. This caused some # confusion during development. # convert into workflow static ref for step status record.
workflow
"Workflow"