Keep disk state reads, sidecar stats and redis version lookup off the event loop

news/7065.performance.md

@@ -0,0 +1 @@
+Keep blocking filesystem and metadata work off the backend event loop: the disk state manager now reads state files in a worker thread, precompressed static file serving stats sidecar files off the loop, and the redis client is created with its library version pinned so redis-py no longer scans `sys.path` on every new connection.

reflex/istate/manager/disk.py

@@ -120,6 +120,9 @@ def token_path(self, token: StateToken) -> Path:
     async def load_state(self, token: StateToken[TOKEN_TYPE]) -> TOKEN_TYPE | None:
         """Load a state object based on the provided token.
 
+        The file read and unpickle run in a worker thread so a cache miss does
+        not block the event loop.
+
         Args:
             token: The token used to identify the state object.
 
@@ -128,13 +131,15 @@ async def load_state(self, token: StateToken[TOKEN_TYPE]) -> TOKEN_TYPE | None:
         """
         token_path = self.token_path(token)
 
-        if token_path.exists():
+        def _read() -> TOKEN_TYPE | None:
             try:
                 with token_path.open(mode="rb") as file:
                     return token.deserialize(fp=file)
             except Exception:
-                pass
-        return None
+                return None
+
+        # The open and unpickle are blocking, keep them off the event loop.
+        return await asyncio.to_thread(_read)
 
     async def populate_substates(
         self, token: BaseStateToken, state: BaseState, root_state: BaseState
@@ -220,11 +225,17 @@ async def set_state_for_substate(
         if token.get_and_reset_touched_state(substate):
             pickle_state = token.serialize(substate)
             if pickle_state:
-                if not self.states_directory.exists():
-                    self.states_directory.mkdir(parents=True, exist_ok=True)
-                await run_in_thread(
-                    lambda: self.token_path(substate_token).write_bytes(pickle_state),
-                )
+                token_path = self.token_path(substate_token)
+
+                def _write() -> None:
+                    try:
+                        token_path.write_bytes(pickle_state)
+                    except FileNotFoundError:
+                        # The states directory was removed at runtime.
+                        self.states_directory.mkdir(parents=True, exist_ok=True)
+                        token_path.write_bytes(pickle_state)
+
+                await run_in_thread(_write)
 
         if isinstance(token, BaseStateToken) and isinstance(substate, BaseState):
             for substate_substate in substate.substates.values():

reflex/utils/precompressed_staticfiles.py

@@ -11,6 +11,7 @@
 from os import PathLike
 from pathlib import Path
 
+import anyio.to_thread
 from starlette.datastructures import Headers
 from starlette.responses import FileResponse, Response
 from starlette.staticfiles import NotModifiedResponse, StaticFiles
@@ -129,7 +130,11 @@ def file_response(
         scope: Scope,
         status_code: int = 200,
     ) -> Response:
-        """Build a FileResponse, swapping in a precompressed sidecar when possible.
+        """Build the FileResponse for the uncompressed file.
+
+        With sidecar encodings configured this response is provisional:
+        ``get_response`` picks the sidecar off the event loop and finishes the
+        response (``Vary``/``Content-Encoding`` and the conditional check).
 
         Args:
             full_path: The resolved on-disk path to the uncompressed file.
@@ -138,37 +143,43 @@ def file_response(
             status_code: The response status code to use.
 
         Returns:
-            A file response that serves the best matching asset variant.
+            A file response for the uncompressed file.
         """
-        response_path: str | PathLike[str] = full_path
-        response_stat = stat_result
-        response_headers: dict[str, str] = {}
         media_type = (
             "text/javascript"
             if Path(full_path).suffix.lower() in {".js", ".mjs"}
             else guess_type(os.fspath(full_path))[0]
         )
-
-        if self._encodings:
-            response_headers["Vary"] = "Accept-Encoding"
-            sidecar = self._select_sidecar(full_path, scope)
-            if sidecar is not None:
-                content_encoding, response_path, response_stat = sidecar
-                response_headers["Content-Encoding"] = content_encoding
-
         response = FileResponse(
-            response_path,
+            full_path,
             status_code=status_code,
-            headers=response_headers or None,
             media_type=media_type,
-            stat_result=response_stat,
+            stat_result=stat_result,
         )
+        if self._encodings:
+            return response
+        return self._conditional(response, scope)
+
+    def _conditional(self, response: FileResponse, scope: Scope) -> Response:
+        """Turn ``response`` into a 304 when the client's cached copy is current.
+
+        Args:
+            response: The file response about to be served.
+            scope: The ASGI request scope.
+
+        Returns:
+            ``response`` or a ``NotModifiedResponse`` carrying its headers.
+        """
         if self.is_not_modified(response.headers, Headers(scope=scope)):
             return NotModifiedResponse(response.headers)
         return response
 
     async def get_response(self, path: str, scope: Scope) -> Response:
-        """Serve ``path``, re-routing the 404.html fallback through ``file_response``.
+        """Serve ``path``, swapping in a precompressed sidecar when possible.
+
+        The sidecar lookup stats files, so it runs in a worker thread like
+        Starlette's own path lookup. This also covers the 404.html fallback,
+        which Starlette builds with a bare FileResponse.
 
         Args:
             path: The requested relative file path.
@@ -178,16 +189,28 @@ async def get_response(self, path: str, scope: Scope) -> Response:
             The resolved static response for the request.
         """
         response = await super().get_response(path, scope)
-        # Starlette's get_response builds the 404.html fallback with bare FileResponse,
-        # bypassing file_response. Re-route it so the sidecar/Vary handling applies.
         if (
-            self._encodings
-            and self.html
-            and isinstance(response, FileResponse)
-            and response.status_code == 404
-            and response.stat_result is not None
+            not self._encodings
+            or not isinstance(response, FileResponse)
+            or response.stat_result is None
         ):
-            return self.file_response(
-                response.path, response.stat_result, scope, status_code=404
-            )
-        return response
+            return response
+        sidecar = await anyio.to_thread.run_sync(
+            self._select_sidecar, response.path, scope
+        )
+        headers = {"Vary": "Accept-Encoding"}
+        response_path: str | PathLike[str] = response.path
+        response_stat = response.stat_result
+        if sidecar is not None:
+            content_encoding, response_path, response_stat = sidecar
+            headers["Content-Encoding"] = content_encoding
+        return self._conditional(
+            FileResponse(
+                response_path,
+                status_code=response.status_code,
+                headers=headers,
+                media_type=response.media_type,
+                stat_result=response_stat,
+            ),
+            scope,
+        )

reflex/utils/prerequisites.py

@@ -16,7 +16,7 @@
 from os import getcwd
 from pathlib import Path
 from types import ModuleType
-from typing import NamedTuple
+from typing import Any, NamedTuple
 
 from packaging import version
 from reflex_base import constants
@@ -400,10 +400,33 @@ def get_redis() -> Redis | None:
         return Redis.from_url(
             redis_url,
             retry_on_error=[RedisError],
+            **_redis_driver_kwargs(),
         )
     return None
 
 
+def _redis_driver_kwargs() -> dict[str, Any]:
+    """Pin redis-py's own version so it is not resolved per connection.
+
+    redis-py >= 7 looks its version up through ``importlib.metadata`` every
+    time a connection object is created, which scans ``sys.path`` and reads
+    package metadata on the event loop. Passing the version up front skips
+    that lookup.
+
+    Returns:
+        Keyword arguments for the redis client constructor.
+    """
+    import redis
+
+    try:
+        # Imported dynamically: the module only exists in redis-py >= 7.
+        driver_info = importlib.import_module("redis.driver_info")
+    except ImportError:
+        # redis-py < 7 takes lib_version directly.
+        return {"lib_version": redis.__version__}
+    return {"driver_info": driver_info.DriverInfo(lib_version=redis.__version__)}
+
+
 def get_redis_sync() -> RedisSync | None:
     """Get the synchronous redis client.