recipes/docker-images/karmic-kraken/patches-eugr/patch_vllm_mrv2_speculator_cudagraph_pool.py

#!/usr/bin/env python3
"""Isolate MRV2 speculator graphs during CUDA-graph memory profiling."""

from __future__ import annotations

import sys
from pathlib import Path


TARGET_REL = Path("vllm/v1/worker/gpu/cudagraph_utils.py")
MARKER = "spark-vllm-docker: isolate speculator profiling graphs"

DECLARATION_ANCHOR = """    all_wrappers: list[Any] = []
    original_pools: dict[int, Any] = {}
    try:
"""

DECLARATION_REPLACEMENT = """    all_wrappers: list[Any] = []
    original_pools: dict[int, Any] = {}
    speculator_managers: list[tuple[str, CudaGraphManager]] = []
    try:
"""

POOL_ANCHOR = """        manager.pool = current_platform.graph_pool_handle()
        if manager.use_breakable_cg:
"""

POOL_REPLACEMENT = f"""        manager.pool = current_platform.graph_pool_handle()
        # {MARKER}. Speculator prefill and decode use separate
        # CudaGraphManager instances, so redirect them before their lazy capture
        # runners are created. Otherwise their discarded profiling graphs use
        # the persistent global pool and leave its allocator handle stale.
        if runner.speculator is not None:
            speculator_managers = [
                (name, value)
                for name, value in vars(runner.speculator).items()
                if isinstance(value, CudaGraphManager)
            ]
            for _, speculator_manager in speculator_managers:
                speculator_manager.pool = manager.pool
        if manager.use_breakable_cg:
"""

CLEANUP_ANCHOR = """        CUDAGraphWrapper.clear_all_graphs()
        BreakableCUDAGraphWrapper.clear_all_graphs()
        for wrapper in all_wrappers:
"""

CLEANUP_REPLACEMENT = """        CUDAGraphWrapper.clear_all_graphs()
        BreakableCUDAGraphWrapper.clear_all_graphs()
        # Drop speculator FULL graphs and their references to the throwaway
        # pool before the profiling state is garbage-collected. The real KV
        # initialization constructs fresh managers immediately afterwards.
        for name, speculator_manager in speculator_managers:
            speculator_manager.graphs.clear()
            speculator_manager.pool = None
            if (
                runner.speculator is not None
                and getattr(runner.speculator, name, None) is speculator_manager
            ):
                setattr(runner.speculator, name, None)
        for wrapper in all_wrappers:
"""


class PatchError(RuntimeError):
    """The vulnerable profiler is present in an unsupported source layout."""


def is_fixed(source: str) -> bool:
    local_fix = (
        MARKER in source
        and "speculator_manager.pool = manager.pool" in source
        and "speculator_manager.graphs.clear()" in source
        and "setattr(runner.speculator, name, None)" in source
    )
    manager_collection_fix = all(
        anchor in source
        for anchor in (
            "def _profiling_cudagraph_managers(",
            "speculator = runner.speculator",
            "candidate = getattr(speculator, name, None)",
            "graph_managers = _profiling_cudagraph_managers(runner)",
            "graph_manager.pool = manager.pool",
            "graph_manager.graphs.clear()",
            "graph_manager.pool = original_manager_pools[id(graph_manager)]",
        )
    )
    return local_fix or manager_collection_fix


def is_affected_profiler(source: str) -> bool:
    return (
        "def profile_cudagraph_memory" in source
        and "manager.pool = current_platform.graph_pool_handle()" in source
        and "PIECEWISE, encoder and speculator graphs are measured in full" in source
    )


def replace_once(source: str, anchor: str, replacement: str, label: str) -> str:
    count = source.count(anchor)
    if count != 1:
        raise PatchError(
            f"expected exactly one {label} anchor, found {count}; "
            "refusing to patch an unknown MRV2 profiler layout"
        )
    return source.replace(anchor, replacement, 1)


def patch_source(source: str) -> tuple[str, str]:
    if is_fixed(source):
        return source, "Equivalent MRV2 speculator CUDA-graph pool fix is present; skipping"
    if not is_affected_profiler(source):
        return source, "Affected MRV2 CUDA-graph memory profiler is absent; skipping"

    patched = replace_once(
        source,
        DECLARATION_ANCHOR,
        DECLARATION_REPLACEMENT,
        "profiling-state declaration",
    )
    patched = replace_once(
        patched,
        POOL_ANCHOR,
        POOL_REPLACEMENT,
        "throwaway-pool assignment",
    )
    patched = replace_once(
        patched,
        CLEANUP_ANCHOR,
        CLEANUP_REPLACEMENT,
        "profiling cleanup",
    )
    if not is_fixed(patched):
        raise PatchError("MRV2 speculator CUDA-graph pool patch postcondition failed")
    return patched, "Applied MRV2 speculator CUDA-graph profiling-pool fix"


def main() -> None:
    source_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd()
    target = source_root / TARGET_REL
    if not target.exists():
        print(f"{TARGET_REL} is absent; targeted MRV2 profiling patch is not applicable")
        return

    source = target.read_text()
    try:
        patched, message = patch_source(source)
    except PatchError as exc:
        raise SystemExit(f"MRV2 speculator CUDA-graph pool patch failed: {exc}") from exc

    if patched != source:
        target.write_text(patched)
    print(message)


if __name__ == "__main__":
    main()