Skip to content

executors

Chain executors.

Production workers are their own executors: their orchestration loops report progress into a ChainExecutionContext (typically via advance_for_progress) and consult ready_nodes for routing. The LocalChainExecutor here is the reference implementation for simple consumers and tests: a synchronous topological walk with a pluggable work callable.

StageWorkCallable module-attribute 

StageWorkCallable = Callable[
    [ChainStageNode], GENERATION_PROGRESS
]

Performs a stage's work and returns the generation progress it reached.

Returning the stage's completion_progress marks the stage completed; returning a failing progress fails it.

ChainExecutor

Bases: Protocol

Anything that can traverse a chain flow and produce a finished execution context.

Source code in horde_sdk/worker/chaining/executors.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class ChainExecutor(Protocol):
    """Anything that can traverse a chain flow and produce a finished execution context."""

    def execute(
        self,
        flow: ChainFlow,
        work: StageWorkCallable,
    ) -> ChainExecutionContext:
        """Traverse the flow, performing each stage's work.

        Args:
            flow (ChainFlow): The flow to traverse.
            work (StageWorkCallable): The callable that performs a stage's work.

        Returns:
            ChainExecutionContext: The finished execution context.
        """
        ...

execute 

execute(
    flow: ChainFlow, work: StageWorkCallable
) -> ChainExecutionContext

Traverse the flow, performing each stage's work.

Parameters:

Returns:

Source code in horde_sdk/worker/chaining/executors.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def execute(
    self,
    flow: ChainFlow,
    work: StageWorkCallable,
) -> ChainExecutionContext:
    """Traverse the flow, performing each stage's work.

    Args:
        flow (ChainFlow): The flow to traverse.
        work (StageWorkCallable): The callable that performs a stage's work.

    Returns:
        ChainExecutionContext: The finished execution context.
    """
    ...

LocalChainExecutor

A synchronous, single-threaded reference executor.

Source code in horde_sdk/worker/chaining/executors.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class LocalChainExecutor:
    """A synchronous, single-threaded reference executor."""

    def execute(
        self,
        flow: ChainFlow,
        work: StageWorkCallable,
    ) -> ChainExecutionContext:
        """Walk the flow in topological order, performing each stage's work.

        A stage whose work returns a progress other than its completion progress is treated as failed, which
        skips all downstream stages. Stages skipped by an earlier failure are not executed.

        Args:
            flow (ChainFlow): The flow to traverse.
            work (StageWorkCallable): The callable that performs a stage's work.

        Returns:
            ChainExecutionContext: The finished execution context.
        """
        context = ChainExecutionContext(flow)

        for handle in flow.handles:
            if context.node_state(handle) != CHAIN_NODE_STATE.PENDING:
                continue

            node = flow.get_node(handle)
            context.mark_executing(handle)
            reached_progress = work(node)

            if reached_progress == node.completion_progress:
                context.mark_completed(handle)
            else:
                context.mark_failed(
                    handle,
                    error=f"stage work reached {reached_progress}, expected {node.completion_progress}",
                )

        return context

execute 

execute(
    flow: ChainFlow, work: StageWorkCallable
) -> ChainExecutionContext

Walk the flow in topological order, performing each stage's work.

A stage whose work returns a progress other than its completion progress is treated as failed, which skips all downstream stages. Stages skipped by an earlier failure are not executed.

Parameters:

Returns:

Source code in horde_sdk/worker/chaining/executors.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def execute(
    self,
    flow: ChainFlow,
    work: StageWorkCallable,
) -> ChainExecutionContext:
    """Walk the flow in topological order, performing each stage's work.

    A stage whose work returns a progress other than its completion progress is treated as failed, which
    skips all downstream stages. Stages skipped by an earlier failure are not executed.

    Args:
        flow (ChainFlow): The flow to traverse.
        work (StageWorkCallable): The callable that performs a stage's work.

    Returns:
        ChainExecutionContext: The finished execution context.
    """
    context = ChainExecutionContext(flow)

    for handle in flow.handles:
        if context.node_state(handle) != CHAIN_NODE_STATE.PENDING:
            continue

        node = flow.get_node(handle)
        context.mark_executing(handle)
        reached_progress = work(node)

        if reached_progress == node.completion_progress:
            context.mark_completed(handle)
        else:
            context.mark_failed(
                handle,
                error=f"stage work reached {reached_progress}, expected {node.completion_progress}",
            )

    return context