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/reflex/reflex.py b/reflex/reflex.py index facbd1fa24d..d053216244e 100644 --- a/reflex/reflex.py +++ b/reflex/reflex.py @@ -370,6 +370,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( _compile_app_worker, app_task, args, kwargs diff --git a/reflex/utils/exec.py b/reflex/utils/exec.py index cea18a0ff11..5c4a8fb9726 100644 --- a/reflex/utils/exec.py +++ b/reflex/utils/exec.py @@ -586,6 +586,9 @@ def run_backend( # Run the backend in development mode. if should_use_granian(): + # Fix the start method first so a backend-only run, which skips the + # compile pool, preloads under the same rule as the reload worker. + set_dev_start_method() # Forked workers inherit imported modules from the supervisor. Spawned # and forkserver workers do not, so preloading the app there only keeps # the full framework graph resident in the long-lived supervisor. @@ -774,12 +777,11 @@ 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 @@ -914,6 +916,14 @@ def shutdown(self, exit_code: int = 0): self._close_shared_socket() super().shutdown(exit_code) + # The app itself is not imported here: the reload worker must load it + # fresh on every restart. Only the framework pages are shared. + # The supervisor forks again on every reload; the run-dev event was sent + # before this point and nothing in it sends telemetry afterwards. + if multiprocessing.get_start_method() == "fork" and not _freeze_for_fork(): + logger.debug("A telemetry send is still running; spawning the worker.") + multiprocessing.set_start_method("spawn", force=True) + reset_dev_backend_reload_marker() environment.REFLEX_DEV_BACKEND_RELOAD_ACTIVE.set(True) @@ -1029,7 +1039,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. @@ -1046,6 +1056,48 @@ def _backend_start_method() -> str | None: return None +def set_dev_start_method() -> None: + """Fix the multiprocessing start method for the development backend. + + An explicit ``REFLEX_BACKEND_START_METHOD`` wins; otherwise strict hot + reload spawns workers and 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() + and environment.REFLEX_BACKEND_START_METHOD.get() is None + ): + start_method = "spawn" + else: + start_method = _backend_start_method() + if start_method is not None: + multiprocessing.set_start_method(start_method, force=True) + + +def _freeze_for_fork() -> bool: + """Stop the telemetry thread and freeze the heap before forking workers. + + Forking while the telemetry thread may hold a lock can deadlock the child. + Without the freeze, worker GC passes write to the inherited objects' + headers, which copies the shared pages private again. + + Returns: + Whether forking is safe; False when a telemetry thread is still alive. + """ + import gc + + from reflex.utils import telemetry + + if not telemetry._shutdown_executor(): + return False + gc.collect() + gc.freeze() + return True + + def _preload_for_fork(app_target: str | None) -> bool: """Import the app in the supervisor so forked workers share its pages. @@ -1056,25 +1108,16 @@ def _preload_for_fork(app_target: str | None) -> bool: Returns: Whether forking is safe; False when a telemetry thread is still alive. """ - import gc - from reflex_base.utils import serializers - from reflex.utils import prerequisites, telemetry + from reflex.utils import prerequisites if app_target is None: prerequisites.get_app() serializers._prepare_serializers_for_fork() - # Forking while the telemetry thread may hold a lock can deadlock the child. # Nothing in the supervisor sends telemetry after this point, so no thread # exists at any later fork either (e.g. a worker respawn). - if not telemetry._shutdown_executor(): - return False - # 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() - return True + return _freeze_for_fork() def run_granian_backend_prod( diff --git a/tests/units/test_reflex.py b/tests/units/test_reflex.py index b51580a7717..9e6e155bd0d 100644 --- a/tests/units/test_reflex.py +++ b/tests/units/test_reflex.py @@ -173,6 +173,45 @@ def test_compile_app_worker_flushes_telemetry_on_failure(mocker): flush.assert_called_once_with() +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"] + + @pytest.mark.parametrize( ("running_mode", "launcher"), [ diff --git a/tests/units/utils/test_exec.py b/tests/units/utils/test_exec.py index 2bc85ee2098..23ab8c31f42 100644 --- a/tests/units/utils/test_exec.py +++ b/tests/units/utils/test_exec.py @@ -134,6 +134,8 @@ def test_run_backend_skips_app_preload_for_spawn( run_granian = mocker.patch.object(exec_utils, "run_granian_backend") mocker.patch.object(exec_utils, "notify_backend") mocker.patch.object(multiprocessing, "get_start_method", return_value="spawn") + mocker.patch.object(multiprocessing, "set_start_method") + monkeypatch.delenv(environment.REFLEX_BACKEND_START_METHOD.name, raising=False) monkeypatch.setenv(environment.REFLEX_STRICT_HOT_RELOAD.name, "False") real_import = builtins.__import__ @@ -162,6 +164,8 @@ def test_run_backend_preloads_app_for_fork( mocker.patch.object(exec_utils, "run_granian_backend") mocker.patch.object(exec_utils, "notify_backend") mocker.patch.object(multiprocessing, "get_start_method", return_value="fork") + mocker.patch.object(multiprocessing, "set_start_method") + monkeypatch.delenv(environment.REFLEX_BACKEND_START_METHOD.name, raising=False) monkeypatch.setenv(environment.REFLEX_STRICT_HOT_RELOAD.name, "False") imported: list[str] = [] @@ -249,6 +253,141 @@ 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 []) + + +def test_set_dev_start_method_explicit_method_beats_strict_hot_reload( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """REFLEX_BACKEND_START_METHOD wins even when strict hot reload is on.""" + monkeypatch.setenv(environment.REFLEX_STRICT_HOT_RELOAD.name, "True") + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "forkserver") + set_start_method = mocker.patch.object(multiprocessing, "set_start_method") + + exec_utils.set_dev_start_method() + + set_start_method.assert_called_once_with("forkserver", force=True) + + +def _fake_granian_dev( + mocker: MockerFixture, + calls: list[str], + start_method: str, + telemetry_stopped: bool = True, +): + """Patch granian and the dev launcher's collaborators, recording call order.""" + granian_server = pytest.importorskip("granian.server") + + 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", + side_effect=lambda method, force=False: calls.append(f"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( + telemetry, + "_shutdown_executor", + side_effect=lambda: calls.append("drain") or telemetry_stopped, + ) + mocker.patch.object(gc, "freeze", side_effect=lambda: calls.append("freeze")) + + +@pytest.mark.parametrize( + ("start_method", "expected"), + [ + ("fork", ["start:fork", "drain", "freeze", "serve"]), + ("spawn", ["start:spawn", "serve"]), + ], +) +def test_run_granian_backend_freezes_only_for_fork( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, + start_method: str, + expected: list[str], +): + """Forked reload workers share a frozen heap; the app is never preloaded. + + The telemetry thread (e.g. from the run-dev event) is drained and stopped + first, so the supervisor has no telemetry thread when it forks the worker. + """ + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, start_method) + monkeypatch.setenv(environment.REFLEX_STRICT_HOT_RELOAD.name, "False") + calls: list[str] = [] + _fake_granian_dev(mocker, calls, start_method) + + exec_utils.run_granian_backend( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == expected + + +def test_run_granian_backend_spawns_when_telemetry_is_stuck( + mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch +): + """A telemetry thread that outlives the drain makes the dev fork unsafe.""" + monkeypatch.setenv(environment.REFLEX_BACKEND_START_METHOD.name, "fork") + monkeypatch.setenv(environment.REFLEX_STRICT_HOT_RELOAD.name, "False") + calls: list[str] = [] + _fake_granian_dev(mocker, calls, "fork", telemetry_stopped=False) + + exec_utils.run_granian_backend( + host="0.0.0.0", port=8000, loglevel=exec_utils.LogLevel.INFO + ) + + assert calls == ["start:fork", "drain", "start:spawn", "serve"] + + def test_run_granian_backend_binds_listen_socket_in_supervisor( tmp_path: Path, mocker: MockerFixture ):