Skip to main content

Dynamic Parallel Steps

Start a runtime-sized set of executions of the same Step type.

Definition graph

DynamicParallelStepsFlow

Valid

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

InitStep builds one DoWorkStep movement per input item. Dex schedules those movements concurrently. Each worker sleeps for a random short duration to make the concurrent completion order visible. Use this pattern when the degree of parallelism is known only at runtime.

Core implementation​

The starting Step builds the movement list at runtime. The worker's random delay represents work with variable latency.

class DoWorkStep(Step[int]):
    async def execute(  # type: ignore[override]
        self, context: AsyncContext, input: int
    ) -> StepDecision:
        await asyncio.sleep(random.uniform(0.05, 0.5))
        return graceful_complete(input)


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

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