From 9ec9ab7929cc79e1c92b539024ce97924079e89b Mon Sep 17 00:00:00 2001 From: Benedikt Bartscher Date: Fri, 22 Nov 2024 09:57:18 +0100 Subject: [PATCH 1/4] test fast yielding in background task --- tests/integration/test_background_task.py | 30 +++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/tests/integration/test_background_task.py b/tests/integration/test_background_task.py index d7fe208247b..2cf4d6a30ec 100644 --- a/tests/integration/test_background_task.py +++ b/tests/integration/test_background_task.py @@ -42,6 +42,11 @@ async def handle_event_yield_only(self): yield State.increment() # type: ignore await asyncio.sleep(0.005) + @rx.event(background=True) + async def fast_yielding(self): + for _ in range(100): + yield State.increment() + @rx.event def increment(self): self.counter += 1 @@ -375,3 +380,28 @@ def test_yield_in_async_with_self( yield_in_async_with_self_button.click() assert background_task._poll_for(lambda: counter.text == "2", timeout=5) + + +def test_fast_yielding( + background_task: AppHarness, + driver: WebDriver, + token: str, +) -> None: + """Test that fast yielding works as expected. + + Args: + background_task: harness for BackgroundTask app. + driver: WebDriver instance. + token: The token for the connected client. + """ + assert background_task.app_instance is not None + + # get a reference to all buttons + fast_yielding_button = driver.find_element(By.ID, "yield-increment") + + # get a reference to the counter + counter = driver.find_element(By.ID, "counter") + assert background_task._poll_for(lambda: counter.text == "0", timeout=5) + + fast_yielding_button.click() + assert background_task._poll_for(lambda: counter.text == "100", timeout=5) From 03eda2e90ecb3d7ae5ae7a6948b0da3059757f53 Mon Sep 17 00:00:00 2001 From: Benedikt Bartscher Date: Sat, 23 Nov 2024 11:44:58 +0100 Subject: [PATCH 2/4] accidentally pushed unfinished changes --- tests/integration/test_background_task.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/integration/test_background_task.py b/tests/integration/test_background_task.py index 2cf4d6a30ec..70e2202a68d 100644 --- a/tests/integration/test_background_task.py +++ b/tests/integration/test_background_task.py @@ -44,7 +44,7 @@ async def handle_event_yield_only(self): @rx.event(background=True) async def fast_yielding(self): - for _ in range(100): + for _ in range(1000): yield State.increment() @rx.event @@ -174,6 +174,11 @@ def index() -> rx.Component: on_click=State.yield_in_async_with_self, id="yield-in-async-with-self", ), + rx.button( + "Fast Yielding", + on_click=State.fast_yielding, + id="fast-yielding", + ), rx.button("Reset", on_click=State.reset_counter, id="reset"), ) @@ -397,11 +402,11 @@ def test_fast_yielding( assert background_task.app_instance is not None # get a reference to all buttons - fast_yielding_button = driver.find_element(By.ID, "yield-increment") + fast_yielding_button = driver.find_element(By.ID, "fast-yielding") # get a reference to the counter counter = driver.find_element(By.ID, "counter") assert background_task._poll_for(lambda: counter.text == "0", timeout=5) fast_yielding_button.click() - assert background_task._poll_for(lambda: counter.text == "100", timeout=5) + assert background_task._poll_for(lambda: counter.text == "1000", timeout=50) From 389f4c7196784bcbdf38260c05a32a79e3114a42 Mon Sep 17 00:00:00 2001 From: Benedikt Bartscher Date: Sat, 23 Nov 2024 20:37:39 +0100 Subject: [PATCH 3/4] fix: only open one connection/sub for each token per worker bonus: properly cleanup StateManager connections on disconnect --- reflex/app.py | 5 +++-- reflex/state.py | 42 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 44 insertions(+), 3 deletions(-) diff --git a/reflex/app.py b/reflex/app.py index afc40e3b880..9c89868f451 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -1477,7 +1477,7 @@ def __init__(self, namespace: str, app: App): super().__init__(namespace) self.app = app - def on_connect(self, sid, environ): + async def on_connect(self, sid, environ): """Event for when the websocket is connected. Args: @@ -1486,7 +1486,7 @@ def on_connect(self, sid, environ): """ pass - def on_disconnect(self, sid): + async def on_disconnect(self, sid): """Event for when the websocket disconnects. Args: @@ -1495,6 +1495,7 @@ def on_disconnect(self, sid): disconnect_token = self.sid_to_token.pop(sid, None) if disconnect_token: self.token_to_sid.pop(disconnect_token, None) + await self.app.state_manager.disconnect(sid) async def emit_update(self, update: StateUpdate, sid: str) -> None: """Emit an update to the client. diff --git a/reflex/state.py b/reflex/state.py index 349dc59e973..7a7d7f43eb8 100644 --- a/reflex/state.py +++ b/reflex/state.py @@ -2826,6 +2826,14 @@ async def modify_state(self, token: str) -> AsyncIterator[BaseState]: """ yield self.state() + async def disconnect(self, token: str) -> None: + """Disconnect the client with the given token. + + Args: + token: The token to disconnect. + """ + pass + class StateManagerMemory(StateManager): """A state manager that stores states in memory.""" @@ -2895,6 +2903,20 @@ async def modify_state(self, token: str) -> AsyncIterator[BaseState]: yield state await self.set_state(token, state) + @override + async def disconnect(self, token: str) -> None: + """Disconnect the client with the given token. + + Args: + token: The token to disconnect. + """ + if token in self.states: + del self.states[token] + if lock := self._states_locks.get(token): + if lock.locked(): + lock.release() + del self._states_locks[token] + def _default_token_expiration() -> int: """Get the default token expiration time. @@ -3183,6 +3205,9 @@ class StateManagerRedis(StateManager): b"evicted", } + # This lock is used to ensure we only subscribe to keyspace events once per token and worker + _pubsub_locks: Dict[bytes, asyncio.Lock] = pydantic.PrivateAttr({}) + async def _get_parent_state( self, token: str, state: BaseState | None = None ) -> BaseState | None: @@ -3458,7 +3483,9 @@ async def _wait_lock(self, lock_key: bytes, lock_id: bytes) -> None: # Some redis servers only allow out-of-band configuration, so ignore errors here. if not environment.REFLEX_IGNORE_REDIS_CONFIG_ERROR.get(): raise - async with self.redis.pubsub() as pubsub: + if lock_key not in self._pubsub_locks: + self._pubsub_locks[lock_key] = asyncio.Lock() + async with self._pubsub_locks[lock_key], self.redis.pubsub() as pubsub: await pubsub.psubscribe(lock_key_channel) while not state_is_locked: # wait for the lock to be released @@ -3475,6 +3502,19 @@ async def _wait_lock(self, lock_key: bytes, lock_id: bytes) -> None: break state_is_locked = await self._try_get_lock(lock_key, lock_id) + @override + async def disconnect(self, token: str): + """Disconnect the token from the redis client. + + Args: + token: The token to disconnect. + """ + lock_key = self._lock_key(token) + if lock := self._pubsub_locks.get(lock_key): + if lock.locked(): + lock.release() + del self._pubsub_locks[lock_key] + @contextlib.asynccontextmanager async def _lock(self, token: str): """Obtain a redis lock for a token. From 7fe33c9bf1917cb59f0384a1220025348deede6c Mon Sep 17 00:00:00 2001 From: Benedikt Bartscher Date: Mon, 25 Nov 2024 00:28:10 +0100 Subject: [PATCH 4/4] run ci