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/news/7113.performance.md b/news/7113.performance.md new file mode 100644 index 00000000000..54d3bc5c3f2 --- /dev/null +++ b/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. 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/reflex.py b/reflex/reflex.py index aa7b7c14b1d..55f7a1b6105 100644 --- a/reflex/reflex.py +++ b/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() diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index c93b11949c3..1cf02845f0c 100644 --- a/reflex/utils/exec.py +++ b/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, 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_reflex.py b/tests/units/test_reflex.py index bd539ab727b..314f860772d 100644 --- a/tests/units/test_reflex.py +++ b/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"] 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..5c0ac154ffb 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 @@ -80,6 +83,83 @@ def serve(self): assert seen["value"] == "True" +@pytest.mark.parametrize( + ("strict", "default_method", "expected"), + [ + (True, "forkserver", "start:spawn"), + (False, "forkserver", "start:fork"), + (False, "fork", "start:fork"), + (False, "spawn", None), + ], +) +def test_set_dev_start_method( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, + strict: bool, + default_method: str, + expected: str | None, +): + """Strict hot reload spawns; otherwise the platform rule decides.""" + monkeypatch.setenv(environment.REFLEX_STRICT_HOT_RELOAD.name, str(strict)) + monkeypatch.delenv(environment.REFLEX_BACKEND_START_METHOD.name, raising=False) + calls: list[str] = [] + mocker.patch.object( + multiprocessing, "get_start_method", return_value=default_method + ) + mocker.patch.object( + multiprocessing, + "set_start_method", + side_effect=lambda method, force=False: calls.append(f"start:{method}"), + ) + + exec_utils.set_dev_start_method() + + assert calls == ([expected] if expected else []) + + +@pytest.mark.parametrize(("start_method", "frozen"), [("fork", True), ("spawn", False)]) +def test_run_granian_backend_freezes_only_for_fork( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, + start_method: str, + frozen: bool, +): + """Forked reload workers share a frozen heap; the app is never preloaded.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, start_method) + monkeypatch.setenv(environment.REFLEX_STRICT_HOT_RELOAD.name, "False") + granian_server = pytest.importorskip("granian.server") + calls: list[str] = [] + + class FakeGranian: + def __init__(self, *_args, **_kwargs): + pass + + def on_reload(self, _callback): + 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_reload_paths", return_value=[]) + mocker.patch.object(exec_utils, "reset_dev_backend_reload_marker") + mocker.patch.object(multiprocessing, "set_start_method") + mocker.patch.object(multiprocessing, "get_start_method", return_value=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")) + + exec_utils.run_granian_backend( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == (["freeze", "serve"] if frozen else ["serve"]) + + def test_with_development_condition_sets_node_and_bun_options(): """Both runtime option vars gain the development condition flag.""" env = exec_utils._with_development_condition({}) @@ -116,3 +196,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