support concurrent renders

Author: rmorshea

src/py/reactpy/reactpy/core/_life_cycle_hook.py

@@ -5,6 +5,8 @@
 from collections.abc import Coroutine
 from typing import Any, Callable, TypeVar
 
+from anyio import Semaphore
+
 from reactpy.core._thread_local import ThreadLocal
 from reactpy.core.types import ComponentType, Context, ContextProviderType
 
@@ -93,6 +95,7 @@ class LifeCycleHook:
         "_effect_starts",
         "_effect_stops",
         "_is_rendering",
+        "_render_access",
         "_rendered_atleast_once",
         "_schedule_render_callback",
         "_schedule_render_later",
@@ -115,6 +118,7 @@ def __init__(
         self._state: tuple[Any, ...] = ()
         self._effect_starts: list[StartEffect] = []
         self._effect_stops: list[StopEffect] = []
+        self._render_access = Semaphore(1)  # ensure only one render at a time
 
     def schedule_render(self) -> None:
         if self._is_rendering:
@@ -147,6 +151,7 @@ def get_context_provider(
 
     async def affect_component_will_render(self, component: ComponentType) -> None:
         """The component is about to render"""
+        await self._render_access.acquire()
         self.component = component
         self._is_rendering = True
         self.set_current()
@@ -158,6 +163,7 @@ async def affect_component_did_render(self) -> None:
         self._is_rendering = False
         self._rendered_atleast_once = True
         self._current_state_index = 0
+        self._render_access.release()
 
     async def affect_layout_did_render(self) -> None:
         """The layout completed a render"""

src/py/reactpy/reactpy/core/hooks.py

@@ -155,16 +155,18 @@ def use_effect(
     def add_effect(function: _Effect) -> None:
         effect_func = _cast_async_effect(function)
 
-        async def start_effect() -> Callable:
+        async def start_effect() -> StopEffect:
             if stop_last_effect.current is not None:
                 await stop_last_effect.current()
 
             stop = Event()
 
-            async def run_effect() -> StopEffect:
+            async def run_effect() -> None:
                 effect_gen = effect_func()
                 # start running the effect
-                effect_task = create_task(effect_gen.asend(None))
+                effect_task = create_task(
+                    cast(Coroutine[None, None, None], effect_gen.asend(None))
+                )
                 # wait for re-render or unmount
                 await stop.wait()
                 # signal effect to stop (no-op if already complete)

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

@@ -1,7 +1,16 @@
 from __future__ import annotations
 
 import abc
-import asyncio
+from asyncio import (
+    FIRST_COMPLETED,
+    Event,
+    Queue,
+    Task,
+    create_task,
+    gather,
+    get_running_loop,
+    wait,
+)
 from collections import Counter
 from collections.abc import Iterator
 from contextlib import AsyncExitStack
@@ -41,6 +50,7 @@ class Layout:
         "root",
         "_event_handlers",
         "_rendering_queue",
+        "_render_tasks",
         "_root_life_cycle_state_id",
         "_model_states_by_life_cycle_state_id",
     )
@@ -58,6 +68,7 @@ def __init__(self, root: ComponentType) -> None:
     async def __aenter__(self) -> Layout:
         # create attributes here to avoid access before entering context manager
         self._event_handlers: EventHandlerDict = {}
+        self._render_tasks: set[Task[LayoutUpdateMessage]] = set()
 
         self._rendering_queue: _ThreadSafeQueue[_LifeCycleStateId] = _ThreadSafeQueue()
         root_model_state = _new_root_model_state(self.root, self._rendering_queue.put)
@@ -72,6 +83,7 @@ async def __aenter__(self) -> Layout:
     async def __aexit__(self, *exc: Any) -> None:
         root_csid = self._root_life_cycle_state_id
         root_model_state = self._model_states_by_life_cycle_state_id[root_csid]
+        await gather(*self._render_tasks, return_exceptions=True)
         await self._unmount_model_states([root_model_state])
 
         # delete attributes here to avoid access after exiting context manager
@@ -102,21 +114,35 @@ async def deliver(self, event: LayoutEventMessage) -> None:
     async def render(self) -> LayoutUpdateMessage:
         """Await the next available render. This will block until a component is updated"""
         while True:
-            model_state_id = await self._rendering_queue.get()
-            try:
-                model_state = self._model_states_by_life_cycle_state_id[model_state_id]
-            except KeyError:
-                logger.debug(
-                    "Did not render component with model state ID "
-                    f"{model_state_id!r} - component already unmounted"
-                )
+            render_completed = (
+                create_task(wait(self._render_tasks, return_when=FIRST_COMPLETED))
+                if self._render_tasks
+                else get_running_loop().create_future()
+            )
+            await wait(
+                (create_task(self._rendering_queue.ready()), render_completed),
+                return_when=FIRST_COMPLETED,
+            )
+            if render_completed.done():
+                done, _ = await render_completed
+                update_task: Task[LayoutUpdateMessage] = done.pop()
+                self._render_tasks.remove(update_task)
+                return update_task.result()
             else:
-                update = await self._create_layout_update(model_state)
-                if REACTPY_CHECK_VDOM_SPEC.current:
-                    root_id = self._root_life_cycle_state_id
-                    root_model = self._model_states_by_life_cycle_state_id[root_id]
-                    validate_vdom_json(root_model.model.current)
-                return update
+                model_state_id = await self._rendering_queue.get()
+                try:
+                    model_state = self._model_states_by_life_cycle_state_id[
+                        model_state_id
+                    ]
+                except KeyError:
+                    logger.debug(
+                        "Did not render component with model state ID "
+                        f"{model_state_id!r} - component already unmounted"
+                    )
+                else:
+                    self._render_tasks.add(
+                        create_task(self._create_layout_update(model_state))
+                    )
 
     async def _create_layout_update(
         self, old_state: _ModelState
@@ -127,6 +153,9 @@ async def _create_layout_update(
         async with AsyncExitStack() as exit_stack:
             await self._render_component(exit_stack, old_state, new_state, component)
 
+        if REACTPY_CHECK_VDOM_SPEC.current:
+            validate_vdom_json(new_state.model.current)
+
         return {
             "type": "layout-update",
             "path": new_state.patch_path,
@@ -540,6 +569,7 @@ class _ModelState:
     __slots__ = (
         "__weakref__",
         "_parent_ref",
+        "_render_semaphore",
         "children_by_key",
         "index",
         "key",
@@ -651,24 +681,27 @@ class _LifeCycleState(NamedTuple):
 
 
 class _ThreadSafeQueue(Generic[_Type]):
-    __slots__ = "_loop", "_queue", "_pending"
-
     def __init__(self) -> None:
-        self._loop = asyncio.get_running_loop()
-        self._queue: asyncio.Queue[_Type] = asyncio.Queue()
+        self._loop = get_running_loop()
+        self._queue: Queue[_Type] = Queue()
         self._pending: set[_Type] = set()
+        self._ready = Event()
 
     def put(self, value: _Type) -> None:
         if value not in self._pending:
             self._pending.add(value)
             self._loop.call_soon_threadsafe(self._queue.put_nowait, value)
+            self._ready.set()
+
+    async def ready(self) -> None:
+        """Return when the next value is available"""
+        await self._ready.wait()
 
     async def get(self) -> _Type:
-        while True:
-            value = await self._queue.get()
-            if value in self._pending:
-                break
+        value = await self._queue.get()
         self._pending.remove(value)
+        if not self._pending:
+            self._ready.clear()
         return value
 
 

src/py/reactpy/tests/test_client.py

@@ -42,7 +42,7 @@ def SomeComponent():
         incr = await page.wait_for_selector("#incr")
 
         for i in range(3):
-            assert (await count.get_attribute("data-count")) == str(i)
+            await poll(count.get_attribute, "data-count").until_equals(str(i))
             await incr.click()
 
     # the server is disconnected but the last view state is still shown
@@ -102,7 +102,9 @@ def ButtonWithChangingColor():
 
     for color in ["blue", "red"] * 2:
         await button.click()
-        assert (await _get_style(button))["background-color"] == color
+        await poll(_get_style, button).until(
+            lambda style, c=color: style["background-color"] == c
+        )
 
 
 async def _get_style(element):

src/py/reactpy/tests/test_core/test_serve.py

@@ -5,11 +5,13 @@
 from jsonpointer import set_pointer
 
 import reactpy
+from reactpy.core.hooks import use_effect
 from reactpy.core.layout import Layout
 from reactpy.core.serve import serve_layout
 from reactpy.core.types import LayoutUpdateMessage
 from reactpy.testing import StaticEventHandler
 from tests.tooling.common import event_message
+from tests.tooling.concurrency import WaitForEvent
 
 EVENT_NAME = "on_event"
 STATIC_EVENT_HANDLER = StaticEventHandler()
@@ -96,9 +98,10 @@ async def test_dispatch():
 
 
 async def test_dispatcher_handles_more_than_one_event_at_a_time():
-    block_and_never_set = asyncio.Event()
-    will_block = asyncio.Event()
-    second_event_did_execute = asyncio.Event()
+    did_render = WaitForEvent()
+    block_and_never_set = WaitForEvent()
+    will_block = WaitForEvent()
+    second_event_did_execute = WaitForEvent()
 
     blocked_handler = StaticEventHandler()
     non_blocked_handler = StaticEventHandler()
@@ -114,6 +117,10 @@ async def block_forever():
         async def handle_event():
             second_event_did_execute.set()
 
+        @use_effect
+        def set_did_render():
+            did_render.set()
+
         return reactpy.html.div(
             reactpy.html.button({"on_click": block_forever}),
             reactpy.html.button({"on_click": handle_event}),
@@ -129,11 +136,12 @@ async def handle_event():
             recv_queue.get,
         )
     )
-
-    await recv_queue.put(event_message(blocked_handler.target))
-    await will_block.wait()
-
-    await recv_queue.put(event_message(non_blocked_handler.target))
-    await second_event_did_execute.wait()
-
-    task.cancel()
+    try:
+        await did_render.wait()
+        await recv_queue.put(event_message(blocked_handler.target))
+        await will_block.wait()
+
+        await recv_queue.put(event_message(non_blocked_handler.target))
+        await second_event_did_execute.wait()
+    finally:
+        task.cancel()