Skip to main content

Advanced: Short-lived Parent

Drain the current workload, then complete the parent without racing new requests.

Definition graph

AdvancedShortLiveParentFlow

Valid

python · examples/python/dex_examples/patterns/parallel-subflows/advanced_short_live_parent_flow.py

InitStep fills RequestChannel, initializes CurrSubFlowNum, and starts N workers. HandleRequestStep.Execute holds the Attribute lock while incrementing the active SubFlow count. HandleSubFlowStep.Execute holds the same lock while decrementing it.

The last active worker uses ForceCompleteIfChannelsEmpty. Dex atomically checks RequestChannel and completes only when it is empty. If an RPC published a request first, Dex schedules another HandleRequestStep instead. This atomic decision is the key difference from checking Channel size in application code.

Core implementation​

DEFAULT_CONCURRENCY = 10
MAX_BUFFERED_REQUESTS = 100


class ShortLiveInitStep(Step[ParentInput]):
    def __init__(
        self,
        request_channel: Channel[str],
        curr_subflow_num: Attribute[int],
    ) -> None:
        self.request_channel = request_channel
        self.curr_subflow_num = curr_subflow_num

    def get_step_type(self) -> str:
        return "InitStep"

    def execute(self, context: Context, input: ParentInput) -> StepDecision:
        for request in input.requests:
            self.request_channel.publish(context, request)
        self.curr_subflow_num.set(context, 0)
        concurrency = input.concurrency if input.concurrency > 0 else DEFAULT_CONCURRENCY
        return go_to_many(
            *(StepMovement.of(ShortLiveHandleRequestStep, None) for _ in range(concurrency))
        )


class ShortLiveHandleRequestStep(Step[None]):
    def __init__(
        self,
        request_channel: Channel[str],
        curr_subflow_num: Attribute[int],
    ) -> None:
        self.request_channel = request_channel
        self.curr_subflow_num = curr_subflow_num

    def get_step_type(self) -> str:
        return "HandleRequestStep"

    def get_step_options(self) -> StepOptions:
        return StepOptions(execute_lock_attributes=(self.curr_subflow_num.lock(),))

    def wait_for(self, context: Context, input: None) -> Wait:
        return Wait.until(self.request_channel.for_one())

    def execute(self, context: Context, input: None) -> StepDecision:
        request = self.request_channel.results(context)[0]
        self.curr_subflow_num.set(context, (self.curr_subflow_num.get(context) or 0) + 1)
        return go_to(ShortLiveHandleSubFlowStep, request)


class ShortLiveHandleSubFlowStep(Step[str]):
    def __init__(
        self,
        example_subflow: ExampleSubFlow,
        request_channel: Channel[str],
        curr_subflow_num: Attribute[int],
    ) -> None:
        self.example_subflow = example_subflow
        self.request_channel = request_channel
        self.curr_subflow_num = curr_subflow_num

    def get_step_type(self) -> str:
        return "HandleSubFlowStep"

    def get_step_options(self) -> StepOptions:
        return StepOptions(execute_lock_attributes=(self.curr_subflow_num.lock(),))

    def wait_for(self, context: Context, request: str) -> Wait:
        return Wait.until(SubFlow.run(self.example_subflow, request))

    def execute(self, context: Context, request: str) -> StepDecision:
        current = (self.curr_subflow_num.get(context) or 0) - 1
        self.curr_subflow_num.set(context, current)
        if current == 0:
            return force_complete_if_channels_empty(
                None,
                StepMovement.of(ShortLiveHandleRequestStep, None),
                self.request_channel,
            )
        return go_to(ShortLiveHandleRequestStep, None)


class AdvancedShortLiveParentFlow(Flow[ParentInput]):
    request_channel = Channel("RequestChannel", str)
    curr_subflow_num = Attribute("CurrSubFlowNum", int)

    def __init__(self, example_subflow: ExampleSubFlow) -> None:
        self.init = ShortLiveInitStep(self.request_channel, self.curr_subflow_num)
        self.handle_request = ShortLiveHandleRequestStep(
            self.request_channel, self.curr_subflow_num
        )
        self.handle_subflow = ShortLiveHandleSubFlowStep(
            example_subflow, self.request_channel, self.curr_subflow_num
        )

    def get_steps(self) -> StepList[ParentInput]:
        return StepList.start_step(self.init).other_steps(
            self.handle_request, self.handle_subflow
        )

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(self.request_channel, self.curr_subflow_num)

    @rpc
    def send_request(self, context: Context, request: str) -> RPCResult[bool]:
        if self.request_channel.size(context) >= MAX_BUFFERED_REQUESTS:
            return RPCResult(False)
        self.request_channel.publish(context, request)
        return RPCResult(True)

Example: examples/python/dex_examples/patterns/parallel-subflows/advanced_short_live_parent_flow.py