fix(vllm): add env-gated CPython GC freeze policy for FPM self-benchmark

components/src/dynamo/vllm/args.py

@@ -392,6 +392,27 @@ def update_engine_config_with_dynamo(
                 f"--scheduler-cls is set to '{existing_cls}'. Either remove "
                 f"--scheduler-cls or use a subclass of InstrumentedScheduler."
             )
+        if os.environ.get("DYN_FPM_GC_POLICY", "").strip().lower() == "freeze":
+            # Class path as a literal, not an import: importing
+            # dynamo.vllm.gc_policy auto-starts the policy in the importing
+            # process, and this launcher process must stay untouched.
+            worker_extension_cls = "dynamo.vllm.gc_policy.FpmGcWorkerExtension"
+            existing_ext = getattr(engine_config, "worker_extension_cls", None)
+            if not existing_ext:
+                defaults["worker_extension_cls"] = worker_extension_cls
+                logger.info(
+                    "Benchmark mode: DYN_FPM_GC_POLICY set, injecting "
+                    "worker_extension_cls=%s",
+                    worker_extension_cls,
+                )
+            elif str(existing_ext) != worker_extension_cls:
+                raise ValueError(
+                    f"DYN_FPM_GC_POLICY requires "
+                    f"worker_extension_cls='{worker_extension_cls}' so model "
+                    f"workers apply the GC policy, but --worker-extension-cls "
+                    f"is set to '{existing_ext}'. Remove it or unset "
+                    f"DYN_FPM_GC_POLICY."
+                )
         benchmark_config: Dict[str, Any] = {
             "mode": dynamo_config.benchmark_mode,
             "warmup_iterations": dynamo_config.benchmark_warmup_iterations,

components/src/dynamo/vllm/gc_policy.py

@@ -0,0 +1,212 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""GC pause mitigation for FPM self-benchmarking.
+
+CPython gen2 (full) collections walk every tracked object and trigger on
+the 25% long-lived-growth rule, so their pauses grow with the heap: over a
+30k-point benchmark sweep we measured pauses rising from ~0.4s to ~9s at
+geometrically spaced iteration indices (ratio ~1.23), inside the vLLM
+worker processes. Under TP/DP lockstep a single worker's pause stalls the
+whole forward pass, and with the scheduler's host-side wall_time it lands
+in the measured latency as a 3-200x outlier.
+
+``gc.freeze()`` moves currently tracked objects into the permanent
+generation, excluding them from future collections: after a periodic
+freeze, gen2 only walks objects allocated since the previous tick, so the
+pause is bounded by the allocation rate instead of total heap size.
+The policy's only goal is keeping GC pauses out of measured wall_time:
+automatic gen2 collections are disabled entirely and the periodic tick
+only freezes (no traversal, no pause). Cyclic garbage is therefore NOT
+reclaimed while the policy is active - acceptable for benchmark sweeps;
+long-running processes can reclaim it by calling :func:`gc_maintain`
+from an untimed window.
+
+The policy is env-gated and off by default:
+
+  DYN_FPM_GC_POLICY=freeze        enable the periodic freeze loop
+  DYN_FPM_GC_FREEZE_INTERVAL_S=60 tick interval in seconds
+
+Coverage:
+  * worker processes  - pass ``--worker-extension-cls
+    dynamo.vllm.gc_policy.FpmGcWorkerExtension``; resolving the class
+    imports this module inside every worker, which auto-starts the
+    policy (no collective_rpc required). The RPC methods remain
+    available for explicit control.
+  * engine-core process - ``InstrumentedScheduler`` imports this module,
+    calls :func:`start_gc_policy` when benchmarking starts, and
+    :func:`stop_gc_policy` when it deactivates.
+"""
+
+from __future__ import annotations
+
+import gc
+import logging
+import math
+import os
+import threading
+
+logger = logging.getLogger(__name__)
+
+_lock = threading.Lock()
+_started = False
+_stop_event: threading.Event | None = None
+_freeze_thread: threading.Thread | None = None
+_saved_gen2_threshold: int | None = None
+
+
+def _policy() -> str:
+    return os.environ.get("DYN_FPM_GC_POLICY", "").strip().lower()
+
+
+def _interval_seconds() -> float:
+    raw = os.environ.get("DYN_FPM_GC_FREEZE_INTERVAL_S", "60")
+    try:
+        value = float(raw)
+    except ValueError:
+        logger.warning("invalid DYN_FPM_GC_FREEZE_INTERVAL_S=%r, using 60", raw)
+        return 60.0
+    if not math.isfinite(value):
+        logger.warning("invalid DYN_FPM_GC_FREEZE_INTERVAL_S=%r, using 60", raw)
+        return 60.0
+    return max(1.0, value)
+
+
+def gc_maintain() -> int:
+    """Reclaim cyclic garbage from an untimed window, then re-freeze.
+
+    Unfreeze first: the periodic ticks freeze unreachable-but-uncollected
+    cycles into the permanent generation, which ``gc.collect()`` never
+    scans. The full-heap walk this costs is why callers must be outside
+    any measured window.
+    """
+    # Serialize with the periodic tick: a freeze fired between unfreeze()
+    # and collect() would re-freeze the garbage this call is meant to
+    # reclaim, silently turning the maintenance pass into a no-op.
+    with _lock:
+        gc.unfreeze()
+        gc.collect()
+        gc.freeze()
+    return gc.get_freeze_count()
+
+
+def _freeze_loop(interval: float, stop: threading.Event) -> None:
+    while not stop.wait(interval):
+        try:
+            # freeze only: O(1) generation-list splice, no heap traversal,
+            # no pause. Never collect on the timer (a periodic collect walks
+            # the unfrozen heap) and never call gc.get_freeze_count() here
+            # (it walks the whole permanent generation: measured 0.4-4s
+            # pauses once millions of objects are frozen). Serialized with
+            # gc_maintain() through _lock.
+            with _lock:
+                gc.freeze()
+        except Exception:  # pragma: no cover - defensive
+            logger.exception("fpm gc tick failed; freeze loop exiting")
+            raise
+
+
+def start_gc_policy() -> bool:
+    """Idempotently start the env-gated freeze loop in this process."""
+    global _started, _stop_event, _freeze_thread, _saved_gen2_threshold
+    if _policy() != "freeze":
+        return False
+    with _lock:
+        if _started:
+            return True
+        # One-time setup. Automatic full (gen2) collections are the pause
+        # source: they walk every tracked object and grow with the heap
+        # (measured 0.4s -> 9s over a sweep). Disable them outright; gen0/1
+        # stay enabled and only walk objects younger than the last freeze.
+        t0, t1, _saved_gen2_threshold = gc.get_threshold()
+        gc.set_threshold(t0, t1, 1 << 30)
+        gc.freeze()
+        interval = _interval_seconds()
+        _stop_event = threading.Event()
+        _freeze_thread = threading.Thread(
+            target=_freeze_loop,
+            args=(interval, _stop_event),
+            name="fpm-gc-freeze",
+            daemon=True,
+        )
+        _freeze_thread.start()
+        _started = True
+    logger.info(
+        "FPM GC policy active: auto-gen2 disabled, gc.freeze() every %.0fs (pid=%d)",
+        interval,
+        os.getpid(),
+    )
+    return True
+
+
+def stop_gc_policy() -> None:
+    """Restore the serving GC state once benchmarking ends.
+
+    Stops the freeze loop, re-enables automatic gen2 collections, reclaims
+    the benchmark-era garbage (including cycles the periodic ticks froze),
+    and then re-establishes vLLM's own serving freeze. vLLM deliberately
+    freezes the EngineCore startup heap and each GPU worker's post-warmup
+    heap (``vllm.utils.gc_utils.freeze_gc_heap``: collect every generation,
+    then ``gc.freeze()``) so gen2 never scans model weights, KV caches, or
+    CUDA graphs during inference; a bare ``gc.unfreeze()`` here would strip
+    that baseline and leave the process paying larger gen2 pauses than one
+    that never benchmarked. Idempotent; no-op if the policy never started.
+    """
+    global _started, _stop_event, _freeze_thread, _saved_gen2_threshold
+    with _lock:
+        if not _started:
+            return
+        stop_event = _stop_event
+        thread = _freeze_thread
+        saved_gen2_threshold = _saved_gen2_threshold
+        if stop_event is not None:
+            stop_event.set()
+        if saved_gen2_threshold is not None:
+            t0, t1, _ = gc.get_threshold()
+            gc.set_threshold(t0, t1, saved_gen2_threshold)
+        _started = False
+        _stop_event = None
+        _freeze_thread = None
+        _saved_gen2_threshold = None
+    if thread is not None:
+        thread.join(timeout=5.0)
+    # Serialized like gc_maintain(): a late tick must not interleave with
+    # the unfreeze/collect/freeze sequence.
+    with _lock:
+        gc.unfreeze()
+        gc.collect()
+        # Mirror freeze_gc_heap(): a full collection has already promoted
+        # every survivor to the oldest generation, so freezing now pins the
+        # static serving heap vLLM intended to exclude from collection plus
+        # whatever benchmark-era objects are still alive at this point (a
+        # small superset of the baseline; anything in it that dies later is
+        # never reclaimed, which matches the behaviour of vLLM's own freeze
+        # for long-lived startup objects).
+        gc.freeze()
+    logger.info(
+        "FPM GC policy stopped: auto-gen2 restored, serving freeze "
+        "re-established (pid=%d)",
+        os.getpid(),
+    )
+
+
+class FpmGcWorkerExtension:
+    """vLLM worker extension exposing GC control inside worker processes."""
+
+    def fpm_gc_start(self) -> bool:
+        return start_gc_policy()
+
+    def fpm_gc_stop(self) -> None:
+        return stop_gc_policy()
+
+    def fpm_gc_maintain(self) -> int:
+        return gc_maintain()
+
+
+# Auto-start when a worker process imports this module while resolving
+# worker_extension_cls (args.py injects the class path as a literal for the
+# same reason: importing this module starts the policy in the importing
+# process). The engine-core process instead imports lazily and starts from
+# InstrumentedScheduler._bench_init, so serving without an active benchmark
+# never runs the policy. No-op unless DYN_FPM_GC_POLICY is set.
+start_gc_policy()

components/src/dynamo/vllm/instrumented_scheduler.py

@@ -2265,6 +2265,12 @@ def _bench_init(self, vllm_config: "VllmConfig") -> None:
             self._bench_cudagraph_capture_sizes,
         )
 
+        # Started last so a config validation error above never leaves the
+        # engine-core process with automatic gen2 collection disabled.
+        from dynamo.vllm import gc_policy as _fpm_gc_policy
+
+        _fpm_gc_policy.start_gc_policy()
+
     # -- Grid generation ------------------------------------------------
 
     def _bench_grid_invariants_digest(self) -> str:
@@ -3681,6 +3687,11 @@ def _bench_deactivate(self, *, resume_publisher: bool = True) -> None:
         self._last_update_time = 0.0
         if resume_publisher:
             self._publisher.resume()
+        # Benchmark over: re-enable automatic gen2 collections and reclaim
+        # the frozen heap before regular serving resumes.
+        from dynamo.vllm import gc_policy as _fpm_gc_policy
+
+        _fpm_gc_policy.stop_gc_policy()
 
     def _bench_abort(self, error: Exception) -> None:
         if self._bench_synchronizer is not None: