perf: preload the app and fork production granian workers

news/+prod-fork-preload.performance.md

@@ -0,0 +1 @@
+Production backend workers on Linux are now forked from a supervisor that has already imported the app, so they share its memory instead of each importing the app again, which cuts memory use and startup time. Set `REFLEX_BACKEND_START_METHOD=spawn` for apps that are not fork-safe.

packages/reflex-base/news/+prod-fork-preload.performance.md

@@ -0,0 +1 @@
+Add the `REFLEX_BACKEND_START_METHOD` environment variable to choose how production backend workers are started (`fork`, `spawn`, or `forkserver`).

packages/reflex-base/src/reflex_base/environment.py

@@ -815,6 +815,13 @@ class EnvironmentVariables:
     # Whether to run Granian in a spawn process. This enables Reflex to pick up on environment variable changes between hot reloads.
     REFLEX_STRICT_HOT_RELOAD: EnvVar[bool] = env_var(False)
 
+    # The multiprocessing start method for production backend workers. Unset means "fork" wherever
+    # the interpreter itself defaults to a fork-based method (Linux), so workers share the app
+    # preloaded by the supervisor. Set to "spawn" for apps that are not fork-safe.
+    REFLEX_BACKEND_START_METHOD: EnvVar[
+        Literal["fork", "spawn", "forkserver"] | None
+    ] = env_var(None)
+
     # The path to the reflex log file. If not set, the log file will be stored in the reflex user directory.
     REFLEX_LOG_FILE: EnvVar[Path | None] = env_var(None)
 

pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
   "starlette >=1.3.1",
   "typing_extensions >=4.13.0",
   "wrapt >=1.17.0,<2.6",
-  "reflex-base >= 0.10.0a2.dev0",
+  "reflex-base >= 0.10.1.dev0",
   "reflex-components-code >= 0.10.0",
   "reflex-components-core >= 0.10.0",
   "reflex-components-dataeditor >= 0.10.0",

reflex/reflex.py

@@ -565,8 +565,9 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str):
 
     _skip_compile()
 
-    # Post a telemetry event.
-    telemetry.send("run-prod")
+    # Post a telemetry event. Sent synchronously: on Linux the backend forks its
+    # workers from this process, which must hold no telemetry thread by then.
+    telemetry._process_event("run-prod", config.telemetry_enabled)
 
     # Display custom message when there is a keyboard interrupt.
     atexit.register(processes.atexit_handler)

reflex/utils/exec.py

@@ -1028,6 +1028,55 @@ def run_uvicorn_backend_prod(
     )
 
 
+def _backend_start_method() -> str | None:
+    """Resolve the multiprocessing start method for production backend workers.
+
+    Returns:
+        The start method to force, or None to keep the interpreter default.
+    """
+    if (method := environment.REFLEX_BACKEND_START_METHOD.get()) is not None:
+        return method
+    import multiprocessing
+
+    # Python defaults to fork (<3.14) or forkserver (3.14+) on Linux and to
+    # spawn elsewhere; only where fork is already the platform norm do we rely
+    # on it so workers can share the supervisor's pages.
+    if multiprocessing.get_start_method() in ("fork", "forkserver"):
+        return "fork"
+    return None
+
+
+def _preload_for_fork(app_target: str | None) -> bool:
+    """Import the app in the supervisor so forked workers share its pages.
+
+    Args:
+        app_target: The ASGI app target; None means the reflex app, which is
+            imported here. Any other target lives in an already-loaded module.
+
+    Returns:
+        Whether forking is safe; False when a telemetry thread is still alive.
+    """
+    import gc
+
+    from reflex_base.utils import serializers
+
+    from reflex.utils import prerequisites, telemetry
+
+    if app_target is None:
+        prerequisites.get_app()
+    serializers._prepare_serializers_for_fork()
+    # Forking while the telemetry thread may hold a lock can deadlock the child.
+    # Nothing in the supervisor sends telemetry after this point, so no thread
+    # exists at any later fork either (e.g. a worker respawn).
+    if not telemetry._shutdown_executor():
+        return False
+    # Freezing keeps worker GC passes from writing to the preloaded objects'
+    # headers, which would copy-on-write the shared pages private again.
+    gc.collect()
+    gc.freeze()
+    return True
+
+
 def run_granian_backend_prod(
     host: str, port: int, loglevel: LogLevel, app_target: str | None = None
 ):
@@ -1039,12 +1088,22 @@ def run_granian_backend_prod(
         loglevel: The log level.
         app_target: The ASGI app target to run. Defaults to the reflex app instance.
     """
+    import multiprocessing
+
     from granian.constants import Interfaces
     from granian.log import LogLevels
     from granian.server import Server as Granian
 
     logger.debug("Using Granian for backend")
 
+    if (start_method := _backend_start_method()) is not None:
+        # Set before the preload so the app's import-time multiprocessing
+        # objects use the requested context.
+        multiprocessing.set_start_method(start_method, force=True)
+        if start_method == "fork" and not _preload_for_fork(app_target):
+            logger.debug("A telemetry send is still running; spawning workers.")
+            multiprocessing.set_start_method("spawn", force=True)
+
     granian_app = Granian(
         target=app_target or get_app_instance_from_file(),
         factory=True,

reflex/utils/telemetry.py

@@ -501,6 +501,46 @@ def _get_telemetry_executor() -> ThreadPoolExecutor:
     return _executor
 
 
+def _shutdown_executor(timeout: float = 2) -> bool:
+    """Deliver queued telemetry and stop the worker thread.
+
+    Called by a supervisor right before it forks workers, so no telemetry
+    thread is alive at the fork. A send that stalls past the timeout (e.g. on
+    a DNS lookup) is abandoned so it cannot hold up startup, and the caller
+    must then not fork.
+
+    Args:
+        timeout: Maximum number of seconds to wait for queued telemetry.
+
+    Returns:
+        Whether the worker thread has stopped, i.e. whether forking is safe.
+    """
+    global _executor
+    if (executor := _executor) is None:
+        return True
+    drained = _flush(timeout)
+    if not drained:
+        logger.debug(f"Telemetry did not drain within {timeout}s before forking.")
+    # Stay published while stopping: a racing send() then fails to submit (and
+    # is suppressed) instead of starting a second pool before the fork.
+    executor.shutdown(wait=drained, cancel_futures=True)
+    with _executor_lock:
+        if _executor is executor:
+            _executor = None
+    return not any(thread.is_alive() for thread in executor._threads)
+
+
+def _reset_executor_after_fork() -> None:
+    """Drop the inherited executor and lock; the child owns neither's thread."""
+    global _executor, _executor_lock
+    _executor_lock = threading.Lock()
+    _executor = None
+
+
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(after_in_child=_reset_executor_after_fork)
+
+
 def _current_registration_context() -> RegistrationContext | None:
     """Return the caller's RegistrationContext, or None if none is attached.
 
@@ -634,8 +674,7 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None:
             # Don't latch: a later enabled send in this process should retry.
             return
 
-        # Latch before the alias send below (which re-enters send()) so it cannot
-        # recurse and so the attempt happens at most once per process.
+        # Latch first so the attempt happens at most once per process.
         _legacy_alias_attempted = True
 
         # Resolve the installation id first: a brand-new install is created and
@@ -647,8 +686,10 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None:
 
         # distinct_id is the UUID form (set by get_event_defaults); send the
         # legacy integer as ``alias`` so PostHog coerces it to the same float as
-        # the historic events and merges the two persons.
-        send("$create_alias", telemetry_enabled, properties={"alias": installation_id})
+        # the historic events and merges the two persons. Delivered on the
+        # calling thread: this runs from the worker, or from a caller that must
+        # not start one.
+        _send("$create_alias", telemetry_enabled, properties={"alias": installation_id})
 
         # Record the new semantics regardless of outcome; we must not retry every
         # run even if the lossy legacy id failed to match.
@@ -668,7 +709,9 @@ def send(
     pool, so neither the data gathering (blocking syscalls and subprocess calls)
     nor the synchronous HTTP request blocks the caller — in particular the
     asyncio event loop serving a running app. Delivery is best-effort: any
-    failure is suppressed and never surfaces to the caller.
+    failure is suppressed and never surfaces to the caller. A caller that must
+    not start a thread (a process about to fork) uses ``_process_event``
+    directly instead.
 
     Args:
         event: The event name.

tests/units/test_reflex.py

@@ -173,6 +173,45 @@ def test_compile_app_worker_flushes_telemetry_on_failure(mocker):
     flush.assert_called_once_with()
 
 
+@pytest.mark.parametrize(
+    ("running_mode", "launcher"),
+    [
+        (constants.RunningMode.BACKEND_ONLY, "run_backend_prod"),
+        (constants.RunningMode.FRONTEND_ONLY, "run_frontend_prod"),
+    ],
+)
+def test_run_prod_sends_telemetry_synchronously(
+    mocker, running_mode: constants.RunningMode, launcher: str
+):
+    """The run-prod event is delivered on the calling thread, before the server.
+
+    The server forks its workers from this process, so the event must not be
+    queued on the telemetry thread pool. Going through ``_process_event`` keeps
+    the one-time alias migration, which a backend-only run has no other event
+    to ride on.
+    """
+    from reflex.utils import build, exec, processes, telemetry
+
+    config = mocker.patch.object(reflex, "get_config").return_value
+    mocker.patch.object(reflex, "_compile_app")
+    mocker.patch.object(reflex, "_skip_compile")
+    mocker.patch.object(build, "setup_frontend_prod")
+    mocker.patch.object(processes, "atexit_handler")
+    mocker.patch("atexit.register")
+    mocker.patch.object(exec, "notify_app_running")
+    mocker.patch.object(exec, "notify_frontend")
+    queued = mocker.patch.object(telemetry, "send")
+    calls = mocker.Mock()
+    calls.attach_mock(mocker.patch.object(telemetry, "_process_event"), "send")
+    calls.attach_mock(mocker.patch.object(exec, launcher), "serve")
+
+    reflex._run_prod(running_mode, 8000, "127.0.0.1")
+
+    queued.assert_not_called()
+    calls.send.assert_called_once_with("run-prod", config.telemetry_enabled)
+    assert [name for name, _args, _kwargs in calls.mock_calls] == ["send", "serve"]
+
+
 def test_cloud_commands_registered():
     """The hosting CLI commands import, resolve, and dispatch only on demand."""
     probe = """