Skip to main content

Await Parallel Steps

Wait for every dynamically started Step before finishing the Flow.

Definition graph

AwaitParallelStepsFlow

Valid

python · examples/python/dex_examples/patterns/parallel/await_parallel_steps_flow.py

InitStep starts DoWorkStep executions and AwaitStep together. Each worker sleeps for a random short duration, then publishes one message to CompleteCh. AwaitStep uses WaitFor for the required number of messages, then completes only after every branch has finished. Channel messages are durable, so completions published before AwaitStep begins waiting still count.

Core implementation​

The coordinator starts alongside the workers. It waits for exactly one completion message per worker.

class DoWorkStep(Step[int]):
    def __init__(self, complete_ch: Channel[None]) -> None:
        self.complete_ch = complete_ch

    async def execute(  # type: ignore[override]
        self, context: AsyncContext, input: int
    ) -> StepDecision:
        await asyncio.sleep(random.uniform(0.05, 0.5))
        self.complete_ch.publish(context, None)
        return dead_end()


class AwaitStep(Step[int]):
    def __init__(self, complete_ch: Channel[None]) -> None:
        self.complete_ch = complete_ch

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.until(self.complete_ch.for_n(input))

    def execute(self, context: Context, input: int) -> StepDecision:
        return graceful_complete(input)


class InitStep(Step[int]):
    def execute(self, context: Context, input: int) -> StepDecision:
        movements: list[StepMovement[Any]] = [StepMovement.of(AwaitStep, input)]
        movements.extend(
            StepMovement.of(DoWorkStep, index) for index in range(input)
        )
        return go_to_many(*movements)

Example: examples/python/dex_examples/patterns/parallel/await_parallel_steps_flow.py