perf: fork the dev backend reload worker instead of using a forkserver

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 framework and app code are shared copy-on-write instead of re-imported per worker. A 4-worker blank app drops from about 720 MB to about 220 MB of proportional set size and starts faster. Set `REFLEX_BACKEND_START_METHOD=spawn` for apps that are not fork-safe.

news/7113.performance.md

@@ -0,0 +1 @@
+The `reflex run` backend reload worker is now forked from the supervisor on Linux instead of started through a forkserver, so framework code is shared copy-on-write and the extra forkserver processes are gone. Set `REFLEX_BACKEND_START_METHOD=spawn` or `REFLEX_STRICT_HOT_RELOAD=1` 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

@@ -708,6 +708,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)
 

reflex/reflex.py

@@ -360,6 +360,7 @@ def _compile_app(*, avoid_dirty_check: bool = True):
     if exec.should_use_granian() and avoid_dirty_check:
         import concurrent.futures
 
+        exec.set_dev_start_method()
         with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
             compile_future = executor.submit(app_task, *args, **kwargs)
             return_result = compile_future.result()

reflex/utils/exec.py

@@ -660,18 +660,22 @@ def run_granian_backend(host: str, port: int, loglevel: LogLevel):
         port: The app port
         loglevel: The log level.
     """
-    logger.debug("Using Granian for backend")
+    import multiprocessing
 
-    if environment.REFLEX_STRICT_HOT_RELOAD.get():
-        import multiprocessing
+    logger.debug("Using Granian for backend")
 
-        multiprocessing.set_start_method("spawn", force=True)
+    set_dev_start_method()
 
     from granian.constants import Interfaces
     from granian.log import LogLevels
     from granian.server import Server as Granian
     from reflex_base.environment import _load_dotenv_from_env
 
+    # The app itself is not imported here: the reload worker must load it
+    # fresh on every restart. Only the framework pages are shared.
+    if multiprocessing.get_start_method() == "fork":
+        _freeze_for_fork()
+
     reset_dev_backend_reload_marker()
     environment.REFLEX_DEV_BACKEND_RELOAD_ACTIVE.set(True)
 
@@ -785,6 +789,66 @@ def run_uvicorn_backend_prod(
     )
 
 
+def _backend_start_method() -> str | None:
+    """Resolve the multiprocessing start method for 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 set_dev_start_method() -> None:
+    """Fix the multiprocessing start method for the development backend.
+
+    Strict hot reload spawns workers; otherwise the platform rule of
+    ``_backend_start_method`` applies. Call this before the first child
+    process starts, or a forkserver started for the compile pool stays alive
+    for the whole session.
+    """
+    import multiprocessing
+
+    if environment.REFLEX_STRICT_HOT_RELOAD.get():
+        multiprocessing.set_start_method("spawn", force=True)
+    elif (start_method := _backend_start_method()) is not None:
+        multiprocessing.set_start_method(start_method, force=True)
+
+
+def _freeze_for_fork() -> None:
+    """Freeze the heap so forked workers keep the supervisor's pages shared.
+
+    Without this, worker GC passes write to the inherited objects' headers,
+    which copies the shared pages private again.
+    """
+    import gc
+
+    gc.collect()
+    gc.freeze()
+
+
+def _preload_for_fork(app_target: str | None) -> None:
+    """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.
+    """
+    from reflex.utils import prerequisites
+
+    if app_target is None:
+        prerequisites.get_app()
+    _freeze_for_fork()
+
+
 def run_granian_backend_prod(
     host: str, port: int, loglevel: LogLevel, app_target: str | None = None
 ):
@@ -796,12 +860,19 @@ 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:
+        multiprocessing.set_start_method(start_method, force=True)
+        if start_method == "fork":
+            _preload_for_fork(app_target)
+
     granian_app = Granian(
         target=app_target or get_app_instance_from_file(),
         factory=True,

reflex/utils/telemetry.py

@@ -496,6 +496,16 @@ def _get_telemetry_executor() -> ThreadPoolExecutor:
     return _executor
 
 
+def _reset_executor_after_fork() -> None:
+    """Drop the inherited executor; its worker thread does not exist in the child."""
+    global _executor
+    _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.
 

tests/units/test_reflex.py

@@ -9,6 +9,7 @@
 
 import click.testing
 import pytest
+from pytest_mock import MockerFixture
 
 from reflex import reflex
 
@@ -384,3 +385,42 @@ def test_init_records_version_check_after_frontend_setup(
     reflex._init("demo")
 
     assert events == ["frontend", "version"]
+
+
+def test_compile_app_sets_start_method_before_compile_pool(mocker: MockerFixture):
+    """The start method is fixed before the compile pool spawns any process."""
+    from unittest import mock
+
+    from reflex.utils import exec as exec_utils
+    from reflex.utils import prerequisites
+
+    calls: list[str] = []
+    mocker.patch.object(exec_utils, "should_use_granian", return_value=True)
+    mocker.patch.object(exec_utils, "should_prerender_routes", return_value=False)
+    mocker.patch.object(
+        exec_utils,
+        "set_dev_start_method",
+        side_effect=lambda: calls.append("start_method"),
+    )
+    mocker.patch.object(prerequisites, "compile_or_validate_app")
+
+    class FakeExecutor:
+        def __init__(self, *_args, **_kwargs):
+            calls.append("pool")
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_exc):
+            return False
+
+        def submit(self, *_args, **_kwargs):
+            future = mock.Mock()
+            future.result.return_value = True
+            return future
+
+    mocker.patch("concurrent.futures.ProcessPoolExecutor", FakeExecutor)
+
+    reflex._compile_app()
+
+    assert calls == ["start_method", "pool"]

tests/units/test_telemetry.py

@@ -801,3 +801,14 @@ def test_flush_returns_false_when_worker_does_not_drain_in_time():
     finally:
         release.set()
         blocker.result(timeout=5)
+
+
+def test_executor_is_recreated_after_fork():
+    """A forked child drops the inherited pool, whose thread it does not own."""
+    inherited = telemetry._get_telemetry_executor()
+
+    telemetry._reset_executor_after_fork()
+
+    fresh = telemetry._get_telemetry_executor()
+    assert fresh is not inherited
+    assert fresh.submit(lambda: 1).result(timeout=5) == 1