From 22a1ac65a29893352b6967f1bae3599d002371c2 Mon Sep 17 00:00:00 2001 From: Farhan Date: Fri, 11 Sep 2026 01:46:52 +0500 Subject: [PATCH 1/3] 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 From 9824978dbc2d621d08f456ddd3d1f425a0922472 Mon Sep 17 00:00:00 2001 From: Farhan Date: Fri, 11 Sep 2026 20:15:07 +0500 Subject: [PATCH 2/3] perf: fork the dev backend reload worker instead of using a forkserver On Python 3.14 Linux the default start method is forkserver, so the `reflex run` supervisor kept a forkserver and resource tracker alive and the reload worker re-imported the whole framework privately. The dev path now follows the same platform rule as the production supervisor: force fork where the interpreter already defaults to fork or forkserver, keep spawn under REFLEX_STRICT_HOT_RELOAD, and honour REFLEX_BACKEND_START_METHOD. The start method is fixed before the compile pool starts, otherwise that pool creates the forkserver pair first and it survives for the session. The supervisor freezes its heap before granian forks, but never imports the user app, so every reload still loads it fresh in the worker. Measured on a blank app (PSS over the session tree, Python 3.14, on top of the prod fork branch): python 330 MB -> 201 MB (supervisor 152 -> 105, worker 151 -> 96, forkserver pair 25 -> 0), whole dev tree cold ~1110 -> ~990 MB and warm ~740 -> ~610 MB, backend ready 4.4 s -> 3.0 s. Claude-Session: https://claude.ai/code/session_015Gi5wTWZNBA61pVDdPLu8u --- news/+dev-fork-workers.performance.md | 1 + reflex/reflex.py | 1 + reflex/utils/exec.py | 49 +++++++++++++---- tests/units/test_reflex.py | 40 ++++++++++++++ tests/units/utils/test_exec.py | 77 +++++++++++++++++++++++++++ 5 files changed, 157 insertions(+), 11 deletions(-) create mode 100644 news/+dev-fork-workers.performance.md diff --git a/news/+dev-fork-workers.performance.md b/news/+dev-fork-workers.performance.md new file mode 100644 index 00000000000..54d3bc5c3f2 --- /dev/null +++ b/news/+dev-fork-workers.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/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 16786d8c254..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. """ + import multiprocessing + logger.debug("Using Granian for backend") - if environment.REFLEX_STRICT_HOT_RELOAD.get(): - import multiprocessing - - 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) @@ -786,7 +790,7 @@ def run_uvicorn_backend_prod( def _backend_start_method() -> str | None: - """Resolve the multiprocessing start method for production backend workers. + """Resolve the multiprocessing start method for backend workers. Returns: The start method to force, or None to keep the interpreter default. @@ -803,6 +807,34 @@ def _backend_start_method() -> str | None: 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. @@ -810,16 +842,11 @@ def _preload_for_fork(app_target: str | None) -> None: 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() + _freeze_for_fork() def run_granian_backend_prod( 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/utils/test_exec.py b/tests/units/utils/test_exec.py index ae85908c965..5c0ac154ffb 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -83,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({}) From 3d7695930d639f600fc7603797973e97b38e5f4e Mon Sep 17 00:00:00 2001 From: Farhan Date: Fri, 11 Sep 2026 20:15:55 +0500 Subject: [PATCH 3/3] chore: name the news fragment after the PR Claude-Session: https://claude.ai/code/session_015Gi5wTWZNBA61pVDdPLu8u --- news/{+dev-fork-workers.performance.md => 7113.performance.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename news/{+dev-fork-workers.performance.md => 7113.performance.md} (100%) diff --git a/news/+dev-fork-workers.performance.md b/news/7113.performance.md similarity index 100% rename from news/+dev-fork-workers.performance.md rename to news/7113.performance.md