perf: preload the app and fork production granian workers (#7079)

* 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/<pid>/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

* Defer run telemetry past the fork and drain the executor first

Address the review asking not to fork granian workers while a telemetry
thread is alive:

- run_backend_prod, run_frontend_prod, run_granian_backend_prod and
  run_uvicorn_backend_prod take an on_started callback; `reflex run
  --env prod` sends the "run-prod" event from it instead of before the
  server starts. Granian calls it from Server.startup() once the workers
  are spawned; uvicorn/gunicorn runs in a separate interpreter, so it is
  called right before that process is launched.
- Before forking, the supervisor finishes Plotly serializer setup
  (_prepare_serializers_for_fork) and drains and shuts down the telemetry
  executor, so no telemetry thread exists at the fork.
- The at-fork child hook also recreates the executor creation lock, so a
  fork taken while another thread held it cannot block the child's first
  telemetry send forever.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Bound the pre-fork telemetry drain

_shutdown_executor() waited on the telemetry executor without a limit, so
a send stalled on DNS (urlopen's timeout does not cover getaddrinfo) or a
slow event collector could block production startup forever. It now
waits up to two seconds through _flush(), then cancels what is still
queued and stops the pool without waiting for a stalled send; the forked
children reset the executor and its lock through the at-fork hook.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Keep the telemetry pool published while it shuts down

_shutdown_executor() cleared _executor before the old pool had stopped,
so a send() racing the pre-fork shutdown could start a replacement pool
whose thread was then alive at the fork. The pool now stays published
until it has stopped (a racing submit fails and is suppressed) and is
only then cleared.

The shutdown test now checks the threads of the pool it stops instead
of every "reflex-telemetry" thread in the process, and the fork-reset
tests stop the pools they orphan, so the check no longer depends on
test order.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Spawn production workers when a telemetry send outlives the drain

After the bounded drain, a stalled telemetry send (e.g. blocked in
getaddrinfo) keeps its thread alive, and that thread can hold
process-wide locks such as the resolver's across a fork. Forking then
trades a startup hang for a worker hang.

_shutdown_executor() now reports whether the pool's thread actually
stopped, _preload_for_fork() passes that on, and
run_granian_backend_prod() starts its workers with "spawn" instead of
"fork" when it did not, so the supervisor never forks with a live
telemetry thread and startup waits at most the drain timeout.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Set the start method before preloading and deflake the stalled drain test

run_granian_backend_prod() imported the app before selecting the start
method, so multiprocessing objects the app creates at import time used
the interpreter default context (forkserver on Python 3.14 Linux) rather
than the requested one. The requested method is now set first, and only
switched to "spawn" when the telemetry drain leaves a live thread.

The stalled-drain test now waits until the worker is running the stalled
task before shutting down, and joins the worker instead of calling
result() on a future that shutdown may have cancelled.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Keep telemetry from starting a thread between the drain and the fork

Once _shutdown_executor() had stopped the pool, a send() racing the
pending fork could create a fresh pool and thread that granian then
forked. The shutdown now pauses pool creation (checked under the
executor lock; such sends are dropped with a debug log) until the next
fork, which lifts it in the parent through an after_in_parent hook and
in the child through the existing after_in_child reset. A caller that
keeps forking can pass resume_after_fork=False to stay paused.

When the workers are forked, the supervisor stops the pool again after
on_started has sent "run-prod", so granian respawns do not fork while a
send is in flight.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Silence the multithreaded-fork warning in the telemetry fork test

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Send the deferred run event from a detached process

The supervisor that forks granian workers also forks again to respawn
them, so it must never hold a telemetry thread: a "run-prod" send that
stalled (DNS, the lscpu collector) after on_started kept one alive across
those respawns, and the earlier bounded re-shutdown could not help.

Once the supervisor drains and stops its pool before the first fork, it
now stays paused for good (in-process sends are dropped with a debug
log; forked children reset it). on_started hands "run-prod" to
telemetry._send_detached(), which runs the event collection and delivery
in a fresh interpreter: Popen with close_fds (so the server sockets stay
in the supervisor) runs a snippet that forks once and exits, so the
supervisor reaps its direct child at once and the sender is reparented
to init. No process starts when telemetry is disabled. The after-fork
parent hook and the post-callback shutdown are gone.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Send the detached run event in-process where nothing forks

Windows has no os.fork, so the detaching snippet cannot run there; the
supervisor never forks on that platform either, so send in-process.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Use the detached sender only in a process paused to fork

_send_detached() started a fresh interpreter and waited for it even when
nothing forks the process: the uvicorn/gunicorn path calls on_started
right before launching the server, so the launch waited on that child.
Only a process that paused telemetry before forking its workers needs
the detached sender; every other process (uvicorn, granian with spawn or
forkserver, a failed pre-fork drain that fell back to spawn) sends
in-process as before.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Reap the detached sender reliably and keep its tests offline

- A direct child that outlives the wait is now killed and reaped instead
  of being left behind when TimeoutExpired was suppressed.
- As PID 1 (a container without an init) or a Linux child subreaper, the
  double-forked sender would be reparented to the supervisor, which never
  reaps it. There the single child sends itself and is waited for
  (bounded, then killed and reaped), which only delays the supervisor's
  management loop, never the workers.
- The snippet is split into its detaching prologue and its send body, so
  the tests swap in a self-contained body instead of slicing the real
  snippet, which could have posted real telemetry if it were reworded.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Keep the run event's sender attached under a PID 1 parent

`reflex run --json` re-runs itself below an output supervisor; as a
container's PID 1 (e.g. `CMD exec reflex run --json`) that supervisor
inherits the double-forked sender but only waits for its own child, so
the sender stayed a zombie for the server's lifetime. Orphans always go
to PID 1 (absent a subreaper), so whenever PID 1 is our own parent the
sender may linger: _reaps_orphans() now also returns True for
os.getppid() == 1, which makes the server send attached and reap the
child itself. This covers reflex's output supervisor and any other
non-reaping PID 1 parent; a false positive (e.g. a service under
systemd) only costs a bounded wait in the supervisor.

The ordinary-process test no longer depends on the runner's PIDs or
subreaper flag: it fakes getpid/getppid and prctl.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Only stay attached under reflex's own output supervisor as PID 1

aebe6ec sent attached under any PID 1 parent, which includes a working
init such as systemd, where the detached sender is reaped fine and the
attached send only made the supervisor wait. The PID 1 parent case now
requires the REFLEX_OUTPUT_SUPERVISED marker, i.e. the parent is the
`reflex run --json` output supervisor, which only waits for its own
child. PID 1 itself and child subreapers are unchanged.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Skip the attached-sender test where os.fork is missing

On Windows _send_detached sends in-process and never calls Popen, so the
test's popen.call_args lookup raised. It now carries the same os.fork
guard as the other detached-sender tests; running the module with
os.fork deleted shows every test passing or skipping.

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Shorten the prod fork preload news fragment to two sentences

Claude-Session: https://claude.ai/code/session_01VDzz2un9mtjmCY24zSGmca

* Send the run-prod telemetry event synchronously instead of via on_started

Address masenf's review: the supervisor sends the run-prod event on its own
thread before granian starts, so no telemetry thread exists at the fork and
nothing needs to run after the workers start. This removes the on_started
callback threaded through run_backend_prod, run_frontend_prod,
run_uvicorn_backend_prod and run_granian_backend_prod, the NotifyingGranian
subclass, telemetry._send_detached with its detached sender script,
_reaps_orphans and the _paused flag.

The pre-fork drain (_shutdown_executor) and the at-fork executor reset stay:
the compile event is still queued on the thread pool, and a forked child must
not inherit a pool whose worker thread it does not own.

Claude-Session: https://claude.ai/code/session_01GbqrTagh776xkQvsopdytF

* Keep the alias migration on the synchronous run-prod send and floor reflex-base

A backend-only production run skips the compile event, so its run-prod event
is the only one that can carry the one-time $create_alias migration. Send it
through _process_event, and deliver the alias on the calling thread with
_send so a process about to fork still starts no telemetry thread. From the
worker thread this changes nothing but the order: the alias now precedes the
event it rides on instead of being queued behind it.

REFLEX_BACKEND_START_METHOD is new in reflex-base, so reflex's floor on it
moves to 0.10.1.dev0, the next version after the reflex-base-v0.10.0 tag.

Claude-Session: https://claude.ai/code/session_01GbqrTagh776xkQvsopdytF

Author: FarhanAliRaza

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 they share its memory instead of each importing the app again, which cuts memory use and startup time. Set `REFLEX_BACKEND_START_METHOD=spawn` for apps that are not fork-safe.

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`).

packages/reflex-base/src/reflex_base/environment.py

@@ -815,6 +815,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)
 

pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
   "starlette >=1.3.1",
   "typing_extensions >=4.13.0",
   "wrapt >=1.17.0,<2.6",
-  "reflex-base >= 0.10.0a2.dev0",
+  "reflex-base >= 0.10.1.dev0",
   "reflex-components-code >= 0.10.0",
   "reflex-components-core >= 0.10.0",
   "reflex-components-dataeditor >= 0.10.0",

reflex/reflex.py

@@ -565,8 +565,9 @@ def _run_prod(running_mode: constants.RunningMode, port: int, host: str):
 
     _skip_compile()
 
-    # Post a telemetry event.
-    telemetry.send("run-prod")
+    # Post a telemetry event. Sent synchronously: on Linux the backend forks its
+    # workers from this process, which must hold no telemetry thread by then.
+    telemetry._process_event("run-prod", config.telemetry_enabled)
 
     # Display custom message when there is a keyboard interrupt.
     atexit.register(processes.atexit_handler)

reflex/utils/exec.py

@@ -1028,6 +1028,55 @@ 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) -> bool:
+    """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.
+
+    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
+
+    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
+
+
 def run_granian_backend_prod(
     host: str, port: int, loglevel: LogLevel, app_target: str | None = None
 ):
@@ -1039,12 +1088,22 @@ 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:
+        # Set before the preload so the app's import-time multiprocessing
+        # objects use the requested context.
+        multiprocessing.set_start_method(start_method, force=True)
+        if start_method == "fork" and not _preload_for_fork(app_target):
+            logger.debug("A telemetry send is still running; spawning workers.")
+            multiprocessing.set_start_method("spawn", force=True)
+
     granian_app = Granian(
         target=app_target or get_app_instance_from_file(),
         factory=True,

reflex/utils/telemetry.py

@@ -501,6 +501,46 @@ def _get_telemetry_executor() -> ThreadPoolExecutor:
     return _executor
 
 
+def _shutdown_executor(timeout: float = 2) -> bool:
+    """Deliver queued telemetry and stop the worker thread.
+
+    Called by a supervisor right before it forks workers, so no telemetry
+    thread is alive at the fork. A send that stalls past the timeout (e.g. on
+    a DNS lookup) is abandoned so it cannot hold up startup, and the caller
+    must then not fork.
+
+    Args:
+        timeout: Maximum number of seconds to wait for queued telemetry.
+
+    Returns:
+        Whether the worker thread has stopped, i.e. whether forking is safe.
+    """
+    global _executor
+    if (executor := _executor) is None:
+        return True
+    drained = _flush(timeout)
+    if not drained:
+        logger.debug(f"Telemetry did not drain within {timeout}s before forking.")
+    # Stay published while stopping: a racing send() then fails to submit (and
+    # is suppressed) instead of starting a second pool before the fork.
+    executor.shutdown(wait=drained, cancel_futures=True)
+    with _executor_lock:
+        if _executor is executor:
+            _executor = None
+    return not any(thread.is_alive() for thread in executor._threads)
+
+
+def _reset_executor_after_fork() -> None:
+    """Drop the inherited executor and lock; the child owns neither's thread."""
+    global _executor, _executor_lock
+    _executor_lock = threading.Lock()
+    _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.
 
@@ -634,8 +674,7 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None:
             # Don't latch: a later enabled send in this process should retry.
             return
 
-        # Latch before the alias send below (which re-enters send()) so it cannot
-        # recurse and so the attempt happens at most once per process.
+        # Latch first so the attempt happens at most once per process.
         _legacy_alias_attempted = True
 
         # Resolve the installation id first: a brand-new install is created and
@@ -647,8 +686,10 @@ def _maybe_alias_legacy_distinct_id(telemetry_enabled: bool | None) -> None:
 
         # distinct_id is the UUID form (set by get_event_defaults); send the
         # legacy integer as ``alias`` so PostHog coerces it to the same float as
-        # the historic events and merges the two persons.
-        send("$create_alias", telemetry_enabled, properties={"alias": installation_id})
+        # the historic events and merges the two persons. Delivered on the
+        # calling thread: this runs from the worker, or from a caller that must
+        # not start one.
+        _send("$create_alias", telemetry_enabled, properties={"alias": installation_id})
 
         # Record the new semantics regardless of outcome; we must not retry every
         # run even if the lossy legacy id failed to match.
@@ -668,7 +709,9 @@ def send(
     pool, so neither the data gathering (blocking syscalls and subprocess calls)
     nor the synchronous HTTP request blocks the caller — in particular the
     asyncio event loop serving a running app. Delivery is best-effort: any
-    failure is suppressed and never surfaces to the caller.
+    failure is suppressed and never surfaces to the caller. A caller that must
+    not start a thread (a process about to fork) uses ``_process_event``
+    directly instead.
 
     Args:
         event: The event name.

tests/units/test_reflex.py

@@ -173,6 +173,45 @@ def test_compile_app_worker_flushes_telemetry_on_failure(mocker):
     flush.assert_called_once_with()
 
 
+@pytest.mark.parametrize(
+    ("running_mode", "launcher"),
+    [
+        (constants.RunningMode.BACKEND_ONLY, "run_backend_prod"),
+        (constants.RunningMode.FRONTEND_ONLY, "run_frontend_prod"),
+    ],
+)
+def test_run_prod_sends_telemetry_synchronously(
+    mocker, running_mode: constants.RunningMode, launcher: str
+):
+    """The run-prod event is delivered on the calling thread, before the server.
+
+    The server forks its workers from this process, so the event must not be
+    queued on the telemetry thread pool. Going through ``_process_event`` keeps
+    the one-time alias migration, which a backend-only run has no other event
+    to ride on.
+    """
+    from reflex.utils import build, exec, processes, telemetry
+
+    config = mocker.patch.object(reflex, "get_config").return_value
+    mocker.patch.object(reflex, "_compile_app")
+    mocker.patch.object(reflex, "_skip_compile")
+    mocker.patch.object(build, "setup_frontend_prod")
+    mocker.patch.object(processes, "atexit_handler")
+    mocker.patch("atexit.register")
+    mocker.patch.object(exec, "notify_app_running")
+    mocker.patch.object(exec, "notify_frontend")
+    queued = mocker.patch.object(telemetry, "send")
+    calls = mocker.Mock()
+    calls.attach_mock(mocker.patch.object(telemetry, "_process_event"), "send")
+    calls.attach_mock(mocker.patch.object(exec, launcher), "serve")
+
+    reflex._run_prod(running_mode, 8000, "127.0.0.1")
+
+    queued.assert_not_called()
+    calls.send.assert_called_once_with("run-prod", config.telemetry_enabled)
+    assert [name for name, _args, _kwargs in calls.mock_calls] == ["send", "serve"]
+
+
 def test_cloud_commands_registered():
     """The hosting CLI commands import, resolve, and dispatch only on demand."""
     probe = """