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