perf: fork the dev backend reload worker instead of using a forkserver

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.

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

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(

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"),
     [

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
 ):