[Feature Request] Improve wait_condition semantics

[dandavison]

Ideally, the following would not happen:

@workflow.defn
class WaitForSharedMutableStateWorkflow:
    def __init__(self) -> None:
        self.cond = False

    @workflow.run
    async def run(self):
        self.cond = True
        await asyncio.gather(self.coro("1"), self.coro("2"))

    async def coro(self, id: str):
        await workflow.wait_condition(lambda: self.cond)
        print(f"after wait coro {id} sees {self.cond}")  # ❗❗ <== coro2 sees False❗
        self.cond = False
        await asyncio.sleep(1)
        self.cond = True

Completely eliminating the possibility would be ideal, but best-effort improvements would also be welcome.

[GSmithApps]

Hey @dandavison,

I definitely agree on this.

As an interim code-around, it worked on my end if I replace

        await asyncio.gather(self.coro("1"), self.coro("2"))

with

        for task in [self.coro("1"), self.coro("2")]:
            asyncio.create_task(task)
            await asyncio.sleep(0.0000000001)

On the downside, it creates actions because it creates a temporal timer for each coro

I'm curious to hear what you think!

[cretz]

No, because the wait condition is resolved after a second which can be on another task. And even if it did work, it's not a general solution to the problem because the concern is that wait conditions all wake up if the predicate is satisfied (as opposed to other SDKs that only wake up one). Whether that happens before or after a sleep is immaterial. The snippet is a trivial example showing the issue, but that's not real world code. In real world code, you just have to make sure you code for the fact that the line of code after a wait condition that may be waited on multiple times concurrently may not match the condition.