Wait for Half
Start N SubFlows, wait for half to finish, then stop the unfinished children.
Each child runs in its own SubFlowStep execution. A completed child publishes one message to SubFlowCompletedCh. WaitSubFlowsStep waits for half of those messages, then publishes one AllDoneCh message for every possible unfinished child.
The important detail is that the parent does not complete immediately after reaching the threshold. Each unfinished SubFlowStep first consumes an AllDoneCh message, obtains its own SubFlowID, and stops that child. Only then does that Step execution complete gracefully. This gives every losing branch a chance to clean up its SubFlow.
Core implementation
class WaitForHalfInitStep(Step[list[str]]):
def execute(self, context: Context, requests: list[str]) -> StepDecision:
if not requests:
return graceful_complete()
return go_to_many(
StepMovement.of(WaitSubFlowsStep, len(requests)),
*(StepMovement.of(SubFlowStep, request) for request in requests),
)
class SubFlowStep(Step[str]):
def __init__(
self,
client_provider: Callable[[], AsyncClient],
example_subflow: ExampleSubFlow,
subflow_completed_ch: Channel[bool],
all_done_ch: Channel[bool],
) -> None:
self.client_provider = client_provider
self.example_subflow = example_subflow
self.subflow_completed_ch = subflow_completed_ch
self.all_done_ch = all_done_ch
def wait_for(self, context: Context, request: str) -> Wait:
return Wait.any_of(SubFlow.run(self.example_subflow, request), self.all_done_ch.for_one())
async def execute(self, context: AsyncContext, request: str) -> StepDecision:
if SubFlow.get_condition_results(context).status is not FlowStatus.RUNNING:
self.subflow_completed_ch.publish(context, True)
return graceful_complete()
await self.client_provider().stop_flow(SubFlow.get_flow_id(context))
return graceful_complete()
class WaitSubFlowsStep(Step[int]):
def __init__(
self, subflow_completed_ch: Channel[bool], all_done_ch: Channel[bool]
) -> None:
self.subflow_completed_ch = subflow_completed_ch
self.all_done_ch = all_done_ch
def wait_for(self, context: Context, total: int) -> Wait:
return Wait.until(self.subflow_completed_ch.for_n((total + 1) // 2))
def execute(self, context: Context, total: int) -> StepDecision:
for _ in range(total - (total + 1) // 2):
self.all_done_ch.publish(context, True)
return graceful_complete()
class WaitForHalfParentFlow(Flow[list[str]]):
subflow_completed_ch = Channel("SubFlowCompletedCh", bool)
all_done_ch = Channel("AllDoneCh", bool)
def __init__(
self,
client_provider: Callable[[], AsyncClient],
example_subflow: ExampleSubFlow,
) -> None:
self.init = WaitForHalfInitStep()
self.subflow = SubFlowStep(client_provider, example_subflow, self.subflow_completed_ch, self.all_done_ch)
self.wait_subflows = WaitSubFlowsStep(self.subflow_completed_ch, self.all_done_ch)
def get_steps(self) -> StepList[list[str]]:
return StepList.start_step(self.init).other_steps(self.subflow, self.wait_subflows)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of(self.subflow_completed_ch, self.all_done_ch)
Example: examples/python/dex_examples/patterns/parallel-subflows/wait_for_half_parent_flow.py