From 22a1ac65a29893352b6967f1bae3599d002371c2 Mon Sep 17 00:00:00 2001 From: Farhan Date: Fri, 11 Sep 2026 01:46:52 +0500 Subject: [PATCH] perf: preload the app and fork production granian workers On Linux the production backend now forces the "fork" start method, imports the app in the supervisor, and freezes the GC before granian spawns workers, so every worker shares the framework and app pages copy-on-write instead of re-importing ~1,800 modules privately. A forked child inherits the telemetry ThreadPoolExecutor without its worker thread, so events submitted in a worker would never be sent; an at-fork hook drops the inherited pool so the child creates its own. REFLEX_BACKEND_START_METHOD overrides the start method for apps that are not fork-safe. Blank app, GRANIAN_WORKERS=4, median of 3 runs (PSS over the process tree from /proc//smaps_rollup after serving requests): before: 721 MB PSS, 692 MB private dirty, port up in 1.82 s after: 223 MB PSS, 70 MB private dirty, port up in 1.07 s Claude-Session: https://claude.ai/code/session_01CEb3ocfFLeHkjkAKKH8YCy --- news/+prod-fork-preload.performance.md | 1 + .../news/+prod-fork-preload.performance.md | 1 + .../src/reflex_base/environment.py | 7 ++ reflex/utils/exec.py | 44 +++++++++ reflex/utils/telemetry.py | 10 ++ tests/units/test_telemetry.py | 11 +++ tests/units/utils/test_exec.py | 97 +++++++++++++++++++ 7 files changed, 171 insertions(+) create mode 100644 news/+prod-fork-preload.performance.md create mode 100644 packages/reflex-base/news/+prod-fork-preload.performance.md diff --git a/news/+prod-fork-preload.performance.md b/news/+prod-fork-preload.performance.md new file mode 100644 index 00000000000..28aa7e7d14f --- /dev/null +++ b/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. diff --git a/packages/reflex-base/news/+prod-fork-preload.performance.md b/packages/reflex-base/news/+prod-fork-preload.performance.md new file mode 100644 index 00000000000..53d78489de8 --- /dev/null +++ b/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`). diff --git a/packages/reflex-base/src/reflex_base/environment.py b/packages/reflex-base/src/reflex_base/environment.py index 3bb80d6970b..aacb2741fc2 100644 --- a/packages/reflex-base/src/reflex_base/environment.py +++ b/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) diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index c93b11949c3..16786d8c254 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -785,6 +785,43 @@ 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) -> 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. + """ + import gc + + from reflex.utils import prerequisites + + if app_target is None: + prerequisites.get_app() + # 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() + + def run_granian_backend_prod( host: str, port: int, loglevel: LogLevel, app_target: str | None = None ): @@ -796,12 +833,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, diff --git a/reflex/utils/telemetry.py b/reflex/utils/telemetry.py index f2f8b794301..1d6cdd6db54 100644 --- a/reflex/utils/telemetry.py +++ b/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. diff --git a/tests/units/test_telemetry.py b/tests/units/test_telemetry.py index 113d01c4d05..4f408ef390a 100644 --- a/tests/units/test_telemetry.py +++ b/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 diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index 5dfc677c094..ae85908c965 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -1,5 +1,7 @@ """Tests for development backend launchers in ``reflex.utils.exec``.""" +import gc +import multiprocessing import os from pathlib import Path @@ -8,6 +10,7 @@ from reflex_base.environment import environment from reflex.utils import exec as exec_utils +from reflex.utils import prerequisites DEV_BACKEND_RELOAD_ENV_NAME = environment.REFLEX_DEV_BACKEND_RELOAD_ACTIVE.name @@ -116,3 +119,97 @@ def test_arbitrate_ssr_env_var_wins(monkeypatch: pytest.MonkeyPatch): monkeypatch.setenv(environment.REFLEX_SSR.name, "False") assert exec_utils.arbitrate_ssr(True) is False + + +def _fake_granian_prod(mocker: MockerFixture, calls: list[str]): + """Patch granian and the prod launcher's collaborators, recording call order.""" + granian_server = pytest.importorskip("granian.server") + + class FakeGranian: + def __init__(self, *_args, **_kwargs): + pass + + def serve(self): + calls.append("serve") + + mocker.patch.object(granian_server, "Server", FakeGranian) + mocker.patch.object( + exec_utils, "get_app_instance_from_file", return_value="app:app" + ) + mocker.patch.object(exec_utils, "_get_backend_workers", return_value=1) + mocker.patch.object( + multiprocessing, + "set_start_method", + side_effect=lambda method, force=False: calls.append(f"start:{method}"), + ) + mocker.patch.object( + prerequisites, "get_app", side_effect=lambda: calls.append("preload") + ) + mocker.patch.object(gc, "freeze", side_effect=lambda: calls.append("freeze")) + + +def test_run_granian_backend_prod_preloads_app_before_forking( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """With fork, the app is imported and the heap frozen before workers start.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") + calls: list[str] = [] + _fake_granian_prod(mocker, calls) + + exec_utils.run_granian_backend_prod( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == ["start:fork", "preload", "freeze", "serve"] + + +def test_run_granian_backend_prod_spawn_skips_preload( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """Spawned workers re-import the app, so the supervisor does not load it.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "spawn") + calls: list[str] = [] + _fake_granian_prod(mocker, calls) + + exec_utils.run_granian_backend_prod( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == ["start:spawn", "serve"] + + +def test_run_granian_backend_prod_custom_target_only_freezes( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """A non-reflex target lives in an already-imported module.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") + calls: list[str] = [] + _fake_granian_prod(mocker, calls) + + exec_utils.run_granian_backend_prod( + host="0.0.0.0", + port=8000, + loglevel=exec_utils.LogLevel.INFO, + app_target="reflex.utils.exec:_frontend_prod_app", + ) + + assert calls == ["start:fork", "freeze", "serve"] + + +@pytest.mark.parametrize( + ("default_method", "expected"), + [("fork", "fork"), ("forkserver", "fork"), ("spawn", None)], +) +def test_backend_start_method_follows_platform_default( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, + default_method: str, + expected: str | None, +): + """Fork is forced only where the interpreter already defaults to forking.""" + monkeypatch.delenv(environment.REFLEX_BACKEND_START_METHOD.name, raising=False) + mocker.patch.object( + multiprocessing, "get_start_method", return_value=default_method + ) + + assert exec_utils._backend_start_method() == expected