Better Async Effects

[rmorshea]

_{By submitting this pull request you agree that all contributions to this project are made under the MIT license.}

Issues

Closes: #956

Solution

Async effects now accept a "stop" Event that is set when an effect needs to be re-run or a component is unmounting. The next effect will only run when the last effect has exited.

@use_effect
async def my_effect(stop):
    ... # do work
    await stop.wait()
    ... # do cleanup

Implementing this same behavior using sync effects is quite challenging:

import threading, asyncio
from reactpy import use_effect, use_ref


def use_async_effect(func):
    thread = use_ref(None)

    @use_effect
    def my_effect():
        if thread.current is not None:
            # wait for last effect to complete
            thread.current.join()

        loop = stop = None
        started = threading.Event()

        def thread_target():
            async def wrapper():
                nonlocal loop, stop
                loop = asyncio.get_running_loop()
                stop = asyncio.Event()
                started.set()
                await func(stop)

            asyncio.run(wrapper())

        # start effect
        thread.current = threading.Thread(target=thread_target, daemon=True)
        thread.current.start()
        started.wait()

        # register stop callback
        return lambda: loop.call_soon_threadsafe(stop.set)

To achieve this without requiring an implementation similar to the above, we've asyncified the internals of Layout. This has the side-effect of allowing other things to happen while the Layout is rendering (e.g. receive events, or allowing the server to respond to other requests). This potentially comes at the cost of rendering speed. For example, if a user is spamming the server with events renders may be interrupted in order to respond to them.

Checklist

• Tests have been included for all bug fixes or added functionality.
• The changelog.rst has been updated with any significant changes.

[Archmonger]

This interface should be relatively similar to the use_messenger hook we designed.

Notes

• We should timeout the effect if it takes too long to teardown
• A 2 second timeout is a good default for most situations.
• Needs a teardown: Coroutine parameter in the hook

import asyncio
from reactpy import component, use_effect

@component
def example():
    async def teardown():
       ...

    @use_effect(timeout=2, teardown=teardown)
    async def example_effect(cancel: asyncio.Event):
        while True:
            await something("my-message-type")
            ...

            if cancel.is_set():
                break

[rmorshea]

Needs a teardown: Coroutine parameter in the hook

I think this is a slight simplification - teardown should just happen after the stop event is triggered.

@use_effect
async def effect(stop):
    # do effect
    await stop.wait()
    # cleanup

We should timeout the effect if it takes too long to teardown

I thought about this for a bit and realized that a simple timeout parameter is ambiguous - does it apply to the effect creation or cleanup. I'd rather not complicate it with more parameters. Adding a timeout is also quite easy:

await asyncio.wait_for(task(), timeout...)

[Archmonger]

teardown should just happen after the stop event is triggered.

This can't always occur though. For example, in the case of the user closing their browser window. We need a timeout to cover edge cases like this.

a simple timeout parameter is ambiguous

We can call it exit_timeout, cancel_timeout, or stop_timeout

Adding a timeout is also quite easy:
await asyncio.wait_for(task(), timeout...)

What you proposed isn't equivalent to a stop_timeout.

[rmorshea]

Both comments seem to address the fact that we might want to enforce a timeout when a connection is closed. That seems reasonable, but in that case, it seems better to introduce that timeout here when a Layout is exiting:

reactpy/src/py/reactpy/reactpy/core/layout.py

Line 75 in c311345

await self._unmount_model_states([root_model_state])

Though, even without this, the user could enforce a "stop timeout" themselves if they wanted:

await wait_for(create_effect(), timeout=...)  # creation timeout
await stop.wait()
await wait_for(cleanup_effect(), timeout=...)  # cleanup timeout

[rmorshea]

So thinking through our interface some more I'm realizing that:

@use_effect
async def my_effect(stop):
    task = asyncio.create_task(do_something())
    await stop.wait()
    task.cancel()
    await finalize_it()

Is not correct since simply cancelling a task does not mean that it will have exited by the time finalize_effect starts running. The correct implementation would actually be:

@use_effect
async def my_effect(stop):
    task = asyncio.create_task(do_something())
    await stop.wait()
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    await finalize_it()

This is far too complicated for users to understand.

As such, I've done some thinking and realized that in 3.11 asyncio got some interesting new features that would allow the above to be implemented with this instead:

@use_effect
async def my_effect(effect):
    async with effect:
        await do_something()
    await finalize_it()

The behavior is that awaitables within the body will be cancelled if they have not completed by the time the effect needs to stop. The body will also pause until the effect needs to be stopped, thus allowing you to perform finalization after exiting the body. If you want to implement a "fire and forget" effect, then you simply ignore the body and do not enter it.

The implementation of the effect context manager would look something like:

class AsyncEffect:

    _task: asyncio.Task | None = None

    def __init__(self) -> None:
        self._stop = asyncio.Event()
        self._cancel_count = 0

    def stop(self) -> None:
        if self._task is not None:
            self._cancel_task()
        self._stop.set()

    async def __aenter__(self) -> None:
        self._task = asyncio.current_task()
        self._cancel_count = self._task.cancelling()
        if self._stop.is_set():
            self._cancel_task()
        return None

    async def __aexit__(self, exc_type: type[BaseException], *exc: Any) -> Any:
        if exc_type is not asyncio.CancelledError:
            # propagate non-cancellation exceptions
            return None

        if self._task.cancelling() > self._cancel_count:
            # Task has been cancelled by something else - propagate it
            return None

        await self._stop.wait()
        return True

    def _cancel_task(self) -> None:
        assert self._task is not None
        self._task.cancel()
        self._cancel_count += 1

[Archmonger]

Is there any way for us to use Python 3.11 asyncio features in older versions?

I'm pretty sure the answer is no, and if so what do we want to do in the interim while we wait for 3.11 to become our minimum version?

[rmorshea]

I'll have to play around with whether this can be achieved with anyio. It may be possible.

[Archmonger]

If we're going to introduce async specific use_effect parameters, we might want to separate async effects into their own hook.

Such as use_async_effect.

[rmorshea]

@Archmonger I'm encountering some issues with lxml during installation. I fixed a similar issue by installing setuptools but that doesn't seem to be helping.

Limiting the default python version to 3.11 seems to have fixed it.

[rmorshea]

I'm going to close this and break up the changes into two parts:

1. Concurrent renders
2. Async effects.