python/ray/workflow/step_executor.py
def _wrap_run(
    func: Callable, runtime_options: "WorkflowStepRuntimeOptions", *args, **kwargs
) -> Tuple[Any, Any]:
    """Wrap the function and execute it.

    It returns two parts, persisted_output (p-out) and volatile_output (v-out).
    P-out is the part of result to persist in a storage and pass to the
    next step. V-out is the part of result to return to the user but does not
    require persistence.

    This table describes their relationships

    +-----------------------------+-------+--------+----------------------+
    | Step Type                   | p-out | v-out  | catch exception into |
    +-----------------------------+-------+--------+----------------------+
    | Function Step               | Y     | N      | p-out                |
    +-----------------------------+-------+--------+----------------------+
    | Virtual Actor Step          | Y     | Y      | v-out                |
    +-----------------------------+-------+--------+----------------------+
    | Readonly Virtual Actor Step | N     | Y      | v-out                |
    +-----------------------------+-------+--------+----------------------+

    Args:
        func: The function body.
        runtime_options: Step execution params.

    Returns:
        State and output.
    """
    exception = None
    result = None
    done = False
    # max_retries are for application level failure.
    # For ray failure, we should use max_retries.
    i = 0
    while not done:
        if i == 0:
            logger.info(f"{get_step_status_info(WorkflowStatus.RUNNING)}")
        else:
            total_retries = (
                runtime_options.max_retries
                if runtime_options.max_retries != -1
                else "inf"
            )
            logger.info(
                f"{get_step_status_info(WorkflowStatus.RUNNING)}"
                f"\tretries: [{i}/{total_retries}]"
            )
        try:
            result = func(*args, **kwargs)
            exception = None
            done = True
        except BaseException as e:
            if i == runtime_options.max_retries:
                retry_msg = "Maximum retry reached, stop retry."
                exception = e
                done = True
            else:
                retry_msg = "The step will be retried."
                i += 1
            logger.error(
                f"{workflow_context.get_name()} failed with error message"
                f" {e}. {retry_msg}"
            )
    step_type = runtime_options.step_type
    if runtime_options.catch_exceptions:
        if step_type == StepType.FUNCTION:
            if isinstance(result, Workflow):
                # When it returns a nested workflow, catch_exception
                # should be passed recursively.
                assert exception is None
                result.data.step_options.catch_exceptions = True
                persisted_output, volatile_output = result, None
            else:
                persisted_output, volatile_output = (result, exception), None
        elif step_type == StepType.ACTOR_METHOD:
            # virtual actors do not persist exception
            persisted_output, volatile_output = result[0], (result[1], exception)
        elif runtime_options.step_type == StepType.READONLY_ACTOR_METHOD:
            persisted_output, volatile_output = None, (result, exception)
        else:
            raise ValueError(f"Unknown StepType '{step_type}'")
    else:
        if exception is not None:
            if step_type != StepType.READONLY_ACTOR_METHOD:
                status = WorkflowStatus.FAILED
                _record_step_status(workflow_context.get_current_step_id(), status)
                logger.info(get_step_status_info(status))
            raise exception
        if step_type == StepType.FUNCTION:
            persisted_output, volatile_output = result, None
        elif step_type == StepType.ACTOR_METHOD:
            persisted_output, volatile_output = result
        elif step_type == StepType.READONLY_ACTOR_METHOD:
            persisted_output, volatile_output = None, result
        else:
            raise ValueError(f"Unknown StepType '{step_type}'")

    return persisted_output, volatile_output
Wrap the function and execute it.

    It returns two parts, persisted_output (p-out) and volatile_output (v-out).
    P-out is the part of result to persist in a storage and pass to the
    next step. V-out is the part of result to return to the user but does not
    require persistence.

    This table describes their relationships

    +-----------------------------+-------+--------+----------------------+
    | Step Type                   | p-out | v-out  | catch exception into |
    +-----------------------------+-------+--------+----------------------+
    | Function Step               | Y     | N      | p-out                |
    +-----------------------------+-------+--------+----------------------+
    | Virtual Actor Step          | Y     | Y      | v-out                |
    +-----------------------------+-------+--------+----------------------+
    | Readonly Virtual Actor Step | N     | Y      | v-out                |
    +-----------------------------+-------+--------+----------------------+

    Args:
        func: The function body.
        runtime_options: Step execution params.

    Returns:
        State and output.
    
Wrap the function and execute it.
It returns two parts, persisted_output (p-out) and volatile_output (v-out).
P-out is the part of result to persist in a storage and pass to the
next step. V-out is the part of result to return to the user but does not
require persistence.

This table describes their relationships


Wrap the function and execute it . It returns two parts persisted_output ( p - out ) and volatile_output ( v - out ) . P - out is the part of result to persist in a storage and pass to the next step . V - out is the part of result to return to the user but does not require persistence . This table describes their relationships
def _wrap_run(
    func: Callable, runtime_options: "WorkflowStepRuntimeOptions", *args, **kwargs
) -> Tuple[Any, Any]:
    exception = None
    result = None
    done = False
    i = 0
    while not done:
        if i == 0:
            logger.info(f"{get_step_status_info(WorkflowStatus.RUNNING)}")
        else:
            total_retries = (
                runtime_options.max_retries
                if runtime_options.max_retries != -1
                else "inf"
            )
            logger.info(
                f"{get_step_status_info(WorkflowStatus.RUNNING)}"
                f"\tretries: [{i}/{total_retries}]"
            )
        try:
            result = func(*args, **kwargs)
            exception = None
            done = True
        except BaseException as e:
            if i == runtime_options.max_retries:
                retry_msg = "Maximum retry reached, stop retry."
                exception = e
                done = True
            else:
                retry_msg = "The step will be retried."
                i += 1
            logger.error(
                f"{workflow_context.get_name()} failed with error message"
                f" {e}. {retry_msg}"
            )
    step_type = runtime_options.step_type
    if runtime_options.catch_exceptions:
        if step_type == StepType.FUNCTION:
            if isinstance(result, Workflow):
                assert exception is None
                result.data.step_options.catch_exceptions = True
                persisted_output, volatile_output = result, None
            else:
                persisted_output, volatile_output = (result, exception), None
        elif step_type == StepType.ACTOR_METHOD:
            persisted_output, volatile_output = result[0], (result[1], exception)
        elif runtime_options.step_type == StepType.READONLY_ACTOR_METHOD:
            persisted_output, volatile_output = None, (result, exception)
        else:
            raise ValueError(f"Unknown StepType '{step_type}'")
    else:
        if exception is not None:
            if step_type != StepType.READONLY_ACTOR_METHOD:
                status = WorkflowStatus.FAILED
                _record_step_status(workflow_context.get_current_step_id(), status)
                logger.info(get_step_status_info(status))
            raise exception
        if step_type == StepType.FUNCTION:
            persisted_output, volatile_output = result, None
        elif step_type == StepType.ACTOR_METHOD:
            persisted_output, volatile_output = result
        elif step_type == StepType.READONLY_ACTOR_METHOD:
            persisted_output, volatile_output = None, result
        else:
            raise ValueError(f"Unknown StepType '{step_type}'")
    return persisted_output, volatile_output
def _wrap_run ( func : Callable , runtime_options : "WorkflowStepRuntimeOptions" , * args , ** kwargs ) -> Tuple [ Any , Any ] : exception = None result = None done = False i = 0 while not done : if i == 0 : logger . info ( f"{get_step_status_info(WorkflowStatus.RUNNING)}" ) else : total_retries = ( runtime_options . max_retries if runtime_options . max_retries != - 1 else "inf" ) logger . info ( f"{get_step_status_info(WorkflowStatus.RUNNING)}" f"\tretries: [{i}/{total_retries}]" ) try : result = func ( * args , ** kwargs ) exception = None done = True except BaseException as e : if i == runtime_options . max_retries : retry_msg = "Maximum retry reached, stop retry." exception = e done = True else : retry_msg = "The step will be retried." i += 1 logger . error ( f"{workflow_context.get_name()} failed with error message" f" {e}. {retry_msg}" ) step_type = runtime_options . step_type if runtime_options . catch_exceptions : if step_type == StepType . FUNCTION : if isinstance ( result , Workflow ) : assert exception is None result . data . step_options . catch_exceptions = True persisted_output , volatile_output = result , None else : persisted_output , volatile_output = ( result , exception ) , None elif step_type == StepType . ACTOR_METHOD : persisted_output , volatile_output = result [ 0 ] , ( result [ 1 ] , exception ) elif runtime_options . step_type == StepType . READONLY_ACTOR_METHOD : persisted_output , volatile_output = None , ( result , exception ) else : raise ValueError ( f"Unknown StepType '{step_type}'" ) else : if exception is not None : if step_type != StepType . READONLY_ACTOR_METHOD : status = WorkflowStatus . FAILED _record_step_status ( workflow_context . get_current_step_id ( ) , status ) logger . info ( get_step_status_info ( status ) ) raise exception if step_type == StepType . FUNCTION : persisted_output , volatile_output = result , None elif step_type == StepType . ACTOR_METHOD : persisted_output , volatile_output = result elif step_type == StepType . READONLY_ACTOR_METHOD : persisted_output , volatile_output = None , result else : raise ValueError ( f"Unknown StepType '{step_type}'" ) return persisted_output , volatile_output
Wrap the function and execute it.
Wrap the function and execute it .
"""Wrap the function and execute it.

    It returns two parts, persisted_output (p-out) and volatile_output (v-out).
    P-out is the part of result to persist in a storage and pass to the
    next step. V-out is the part of result to return to the user but does not
    require persistence.

    This table describes their relationships

    +-----------------------------+-------+--------+----------------------+
    | Step Type                   | p-out | v-out  | catch exception into |
    +-----------------------------+-------+--------+----------------------+
    | Function Step               | Y     | N      | p-out                |
    +-----------------------------+-------+--------+----------------------+
    | Virtual Actor Step          | Y     | Y      | v-out                |
    +-----------------------------+-------+--------+----------------------+
    | Readonly Virtual Actor Step | N     | Y      | v-out                |
    +-----------------------------+-------+--------+----------------------+

    Args:
        func: The function body.
        runtime_options: Step execution params.

    Returns:
        State and output.
    """ # max_retries are for application level failure. # For ray failure, we should use max_retries. # When it returns a nested workflow, catch_exception # should be passed recursively. # virtual actors do not persist exception
State and output.
State and output .
func
Callable
The function body.
The function body .
runtime_options
"WorkflowStepRuntimeOptions"
Step execution params.
Step execution params .