vllm-project-vllm-42585

build_rust.sh
cmake
codecov.yml
csrc
docker
docs
examples
mkdocs.yaml
pyproject.toml
requirements
rust
rust-toolchain.toml
scripts
setup.py
tests
tools
use_existing_torch.py
vllm
vllm.egg-info
vllm/v1/engine/coordinator.py:18:from vllm.v1.utils import get_engine_client_zmq_addr, shutdown
vllm/v1/engine/coordinator.py:96:                get_engine_client_zmq_addr(local_only=True, host=host)
vllm/v1/engine/utils.py:31:from vllm.v1.utils import get_engine_client_zmq_addr, shutdown
vllm/v1/engine/utils.py:983:            get_engine_client_zmq_addr(client_local_only, host)
vllm/v1/engine/utils.py:987:            get_engine_client_zmq_addr(client_local_only, host)
vllm/v1/engine/utils.py:1098:    handshake_address = get_engine_client_zmq_addr(
vllm/v1/utils.py:147:def get_engine_client_zmq_addr(local_only: bool, host: str, port: int = 0) -> str:

    def __init__(
        self,
        *size: int | torch.SymInt,
        dtype: torch.dtype,
        device: torch.device,
        pin_memory: bool,
        with_numpy: bool = True,
    ) -> None:
        self.cpu = torch.zeros(*size, dtype=dtype, device="cpu", pin_memory=pin_memory)
        self.gpu = torch.zeros_like(self.cpu, device=device)
        self.np: np.ndarray
        # To keep type hints simple (avoiding generics and subclasses), we
        # only conditionally create the numpy array attribute. This can cause
        # AttributeError if `self.np` is accessed when `with_numpy=False`.
        if with_numpy:
            if dtype == torch
…[3239 chars omitted]…
ate_address"] = stats_update_address
            if tensor_queue is not None:
                client_config["tensor_queue"] = tensor_queue

            proc = spawn_context.Process(
                target=target_server_fn or run_api_server_worker_proc,
                name=f"ApiServer_{i}",
                args=(listen_address, sock, args, client_config),
vllm/entrypoints/cli/serve.py
vllm/v1/utils.py

        """Shutdown API server processes with configurable timeout"""
        if self._finalizer.detach() is not None:
            shutdown(self.processes, timeout=timeout)


class RustFrontendProcessManager:
    """Manages a single Rust frontend subprocess.

    Launches the Rust vllm-rs binary in 'frontend' mode, passing the
    listening socket fd and ZMQ transport addresses. Provides the same
    int
…[2243 chars omitted]…
     recv, send = connection.Pipe(duplex=False)
        self._sentinel_conn = recv
        self._sentinel_send = send

        def monitor_subprocess() -> None:
            try:
                proc.wait()
            finally:
                with contextlib.suppress(Exception):
                    send.close()
vllm/entrypoints/cli/serve.py:333:            api_server_manager = APIServerProcessManager(

    if num_api_servers > 1 and envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING:
        raise ValueError(
            "VLLM_ALLOW_RUNTIME_LORA_UPDATING cannot be used with api_server_count > 1"
        )

    executor_class = Executor.get_class(vllm_config)
    log_stats = not engine_args.disable_log_stats

    parallel_config = vllm_config.parallel_config
    dp_rank = parallel_config.data_parallel_rank
    assert parallel_config.local_engines_only or dp_rank == 0

    
…[2015 chars omitted]…
loat | None) -> float | None:
            return (
                deadline if deadline is None else max(deadline - time.monotonic(), 0.0)
            )

        api_server_manager.shutdown(timeout=timeout)
        if local_engine_manager:
            local_engine_manager.shutdown(timeout=to_timeout(shutdown_by))
        if coordinator:
            coordinator.shutdown(timeout=to_timeout(shutdown_by))

    local_engine_count = parallel_config.data_parallel_size_local
    local_start_index = parallel_config.data_parallel_rank_local
    dp_size = parallel_config.data_parallel_size
    host = parallel_config.data_parallel_master_ip
    local_engines_only = parallel_config.local_engines_only

    # In offline mode there is an LLM instance per DP rank and
    # one core engine per LLM, see
    # examples/features/data_parallel/data_parallel_offline.py.
    offline_mode = local_start_index is not None

  
…[5224 chars omitted]…
R, bind=True
    ) as handshake_socket:
        # Start local engines.
        if local_engine_count:
            local_engine_manager = CoreEngineProcManager(
                vllm_config=vllm_config,
                executor_class=executor_class,
                log_stats=log_stats,
                handshake_address=handshake_address,
                client_handshake_address=client_handshake_address,

…[222 chars omitted]…
EngineZmqAddresses, None, None]:
vllm/v1/engine/core.py:1060:    ) -> EngineZmqAddresses:
vllm/v1/engine/core.py:2020:        addresses: EngineZmqAddresses,
vllm/v1/engine/core.py:2144:        addresses: EngineZmqAddresses,
vllm/v1/engine/core.py:2167:        addresses: EngineZmqAddresses,
vllm/v1/engine/utils.py:58:class EngineZmqAddresses:
vllm/v1/engine/utils.py:80:    addresses: EngineZmqAddresses

            # Background Threads and Queues for IO. These enable us to
            # overlap ZMQ socket IO with GPU since they release the GIL,
            # and to overlap some serialization/deserialization with the
            # model forward pass.
            # Threads handle Socket <-> Queues and core_busy_loop uses Queue.
            ready_event = threading.Event()
            input_thread = threading.Thread(
                target=self.process_input_sockets,
                args=(
          
…[3826 chars omitted]…
zmq.Context,
        handshake_address: str,
        identity: bytes,
        local_client: bool,
        headless: bool,
        vllm_config: VllmConfig,
        parallel_config_to_update: ParallelConfig | None = None,
    ) -> Generator[EngineZmqAddresses, None, None]:
        with make_zmq_socket(
            ctx,
            handshake_address,
            zmq.DEALER,
            identity=identity,

        input_addresses: list[str],
        coord_input_address: str | None,
        identity: bytes,
        ready_event: threading.Event,
    ):
        """Input socket IO thread."""

        # Msgpack serialization decoding with optional tensor IPC receiver.
        add_request_decoder = MsgpackDecoder(
            EngineCoreRequest, oob_tensor_provider=self.tensor_ipc_receiver
        )
        generic_decoder = MsgpackDecoder(oob_tensor_provider=self.tensor_ipc_recei
…[1103 chars omitted]…
m_gpu_blocks=self.vllm_config.cache_config.num_gpu_blocks or 0,
                dp_stats_address=self.frontend_stats_publish_address,
                dtype=str(self.vllm_config.model_config.dtype).removeprefix("torch."),
920:                target=self.process_input_sockets,
932:                target=self.process_output_sockets,
1400:    def process_input_sockets(
1495:    def process_output_sockets(

    from vllm.entrypoints.openai.api_server import run_server_worker

    client_config = client_config or {}
    server_index = client_config.get("client_index", 0)

    # Set process title and add process-specific prefix to stdout and stderr.
    set_process_title("APIServer", str(server_index))
    decorate_logs()

    uvloop.run(
        run_server_worker(listen_address, sock, args, client_config, **uvicorn_kwargs)
    )


def wait_for_completion_or_failure(
    api_server_manager: "APIServerProcessManager | RustFrontendProcessManager",
    engine_manag
…[3155 chars omitted]…
      engine_args._api_process_rank = client_config.get("client_index", 0)
103:        client_config=client_config,
113:    client_config: dict[str, Any] | None = None,
130:    # Don't mutate the input client_config
131:    client_config = dict(client_config) if client_config else {}
132:    client_count = client_config.pop("client_count", 1)
133:    client_index = client_config.pop("client_index", 0)

) -> AsyncIterator[EngineClient]:
    if os.getenv("VLLM_WORKER_MULTIPROC_METHOD") == "forkserver":
        # The executor is expected to be mp.
        # Pre-import heavy modules in the forkserver process
        logger.debug("Setup forkserver with pre-imports")
        multiprocessing.set_start_method("forkserver")
        multiprocessing.set_forkserver_preload(["vllm.v1.engine.async_llm"])
        forkserver.ensure_running()
        logger.debug("Forkserver setup complete!")

    # Con
…[3236 chars omitted]…
ster_models_api_router,
    )

    register_models_api_router(app)

    from vllm.entrypoints.sagemaker.api_router import (
        attach_router as register_sagemaker_api_router,
    )

    register_sagemaker_api_router(app, supported_tasks, model_config)

    if "generate" in supported_tasks:
        from vllm.entrypoints.openai.generate.api_router import (
            register_generate_api_routers,

encoder_cache_manager.py
kv_cache_coordinator.py
kv_cache_manager.py
kv_cache_metrics.py
kv_cache_utils.py
sched
single_type_kv_cache_manager.py

        # ZMQ setup.
        sync_ctx = zmq.Context(io_threads=2)
        self.ctx = zmq.asyncio.Context(sync_ctx) if asyncio_mode else sync_ctx

        # This will ensure resources created so far are closed
        # when the client is garbage collected, even if an
        # exception is raised mid-construction.
        self.resources = BackgroundResources(ctx=sync_ctx)
        self._finalizer = weakref.finalize(self, self.resources)
        success = False
        try:
            # State used for data parallel.
            self.engines_running = False
            paralle
…[2989 chars omitted]…
 None:
                    tensor_ipc_sender = TensorIpcSender(tensor_queue)

            self.encoder = MsgpackEncoder(oob_tensor_consumer=tensor_ipc_sender)
            self.decoder = MsgpackDecoder(EngineCoreOutputs)

            dp_size = parallel_config.data_parallel_size
            dp_rank = parallel_config.data_parallel_index
            dp_local_size = parallel_config.data_parallel_size_local

    An edge case to handle, is when we run data parallel,
    we need to avoid ports that are potentially used by
    the data parallel master process.
    Right now we reserve 10 ports for the data parallel master
    process. Currently it uses 2 ports.
    """
    if "VLLM_DP_MASTER_PORT" in os.environ:
        dp_master_port = envs.VLLM_DP_MASTER_PORT
        reserved_port_range = range(dp_master_port, dp_master_port + 10)
        while True:
            candidate_port = _get_open_port()
            if candidate_port not in reserved_port_range:
                return candidate_port
    return
…[3517 chars omitted]…
ZMQ socket"""

    ctx = zmq.Context()  # type: ignore[attr-defined]
    try:
        yield make_zmq_socket(
            ctx,
            path,
            socket_type,
            bind=bind,
            identity=identity,
            router_handover=router_handover,
        )
    except KeyboardInterrupt:
        logger.debug("Got Keyboard Interrupt.")

    finally:
        ctx.destroy(linger=linger)

  local_handshake_address, zmq.ROUTER, bind=True
vllm/v1/engine/utils.py:1112:    ) as handshake_socket:
vllm/v1/engine/utils.py:1119:                handshake_address=handshake_address,
vllm/v1/engine/utils.py:1120:                client_handshake_address=client_handshake_address,
vllm/v1/engine/utils.py:1134:            handshake_socket,
vllm/v1/engine/utils.py:1136:            engines_to_handshake,

        )


def wait_for_engine_startup(
    handshake_socket: zmq.Socket,
    addresses: EngineZmqAddresses,
    core_engines: list[CoreEngine],
    parallel_config: ParallelConfig,
    coordinated_dp: bool,
    cache_config: CacheConfig,
    proc_manager: CoreEngineProcManager | None,
    coord_process: Process | None,
):
    # Wait for engine core process(es) to send ready messages.
    local_count = parallel_config.data_parallel_size_loc
…[2773 chars omitted]…
r(
                    f"Remote engine {eng_index} must not use "
                    f"--headless in external or hybrid dp lb "
                    f"mode"
                )
            else:
                raise RuntimeError(
                    f"Remote engine {eng_index} must use "
                    f"--headless unless in external or hybrid "
                    f"dp lb mode"
                )

                        )
                    }
                    if coordinated_dp
                    else {},
                )
            )
            handshake_socket.send_multipart((eng_identity, init_message), copy=False)
            conn_pending[0 if local else 1] -= 1
     
…[1234 chars omitted]…
.READY
        else:
            raise RuntimeError(
                f"Unexpected {status} message for "
                f"{'local' if local else 'remote'} engine "
                f"{eng_index} in {engine.state} state."
            )

        logger.debug(
            "%s from %s core engine process %s.",
            status,
            "local" if local else "remote",
            eng_index,
        )

            engine_manager.monitor_engine_liveness()
            _self = self_ref()
            if not _self or not _self._finalizer.alive or _self.resources.engine_dead:
                return
            _self.resources.engine_dead = True
            _self.shutdown()
            # Note: For MPClient, we don't have a failure callback mechanism
            # like MultiprocExecutor, but we set engine_dead flag which will
  
…[3604 chars omitted]…
      poller.register(shutdown_socket, zmq.POLLIN)
                poller.register(out_socket, zmq.POLLIN)
                while True:
                    socks = poller.poll()
                    if not socks:
                        continue
                    if len(socks) == 2 or socks[0][0] == shutdown_socket:
                        # shutdown signal, exit thread.
                        break

…[293 chars omitted]…
:            common_kwargs["client_handshake_address"] = client_handshake_address
vllm/v1/engine/utils.py:1105:        client_handshake_address = local_handshake_address
vllm/v1/engine/utils.py:1108:        client_handshake_address = None
vllm/v1/engine/utils.py:1120:                client_handshake_address=client_handshake_address,
vllm/v1/engine/utils.py:1234:                EngineHandshakeMetadata(

        client_handshake_address: str | None,
    ):
        """
        For Ray, we don't need to actually perform handshake.
        All addresses information is known before the actor creation.
        Therefore, we simply yield these addresses.
        """
        yield self.addresses

    def wait_for_init(self):
        """
        Wait until the engine core is initialized.

        This is just an empty method. When ray.get() on this method
        (or any other method of the actor) returns, i
…[1703 chars omitted]…
el_size_local = 1
        vllm_config.parallel_config.data_parallel_rank = 0

        EngineCoreActorMixin.__init__(
            self, vllm_config, addresses, dp_rank, local_dp_rank
        )
        EngineCoreProc.__init__(
            self,
            vllm_config,
            local_client,
            "",
            executor_class,
            log_stats,
            engine_index=dp_rank,
        )

vllm/distributed/kv_transfer/kv_connector/v1/ssm_conv_transfer_utils.py
vllm/entrypoints/cli/serve.py
vllm/envs.py
vllm/v1/engine/core_client.py
vllm/v1/executor/abstract.py
vllm/v1/executor/multiproc_executor.py
vllm/v1/outputs.py
vllm/v1/worker/gpu_worker.py
vllm/vllm_flash_attn/cute/block_sparse_utils.py

    shutdown_requested = False

    # Catch SIGTERM and SIGINT to allow graceful shutdown.
    def signal_handler(signum, frame):
        nonlocal shutdown_requested
        logger.debug("Received %d signal.", signum)
        if not shutdown_requested:
            shutdown_requested = True
            raise SystemExit

    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

  
…[2128 chars omitted]…
 1"
        )

    if num_api_servers > 1:
        setup_multiprocess_prometheus()

    shutdown_requested = False

    # Catch SIGTERM and SIGINT to allow graceful shutdown.
    def signal_handler(signum, frame):
        nonlocal shutdown_requested
        logger.debug("Received %d signal.", signum)
        if not shutdown_requested:
            shutdown_requested = True
            raise SystemExit

        if aborted_reqs:
            # Map client_index to list of request_ids that belong to that client.
            by_client = defaultdict[int, 
…[2455 chars omitted]…
rank < dp_size

        self.dp_rank = dp_rank
        self.dp_size = dp_size
        dp_group, dp_store = parallel_config.stateless_init_dp_group(return_store=True)
        self.dp_group, self.dp_store = dp_group, dp_store

    def shutdown(self):
        super().shutdown()
        if dp_group := getattr(self, "dp_group", None):
            stateless_destroy_torch_distributed_process_group(dp_group)

942:            # Don't complete handshake until DP coordinator ready message is
951:    def _perform_handshakes(
953:        handshake_address: str,
957:        client_handshake_address: str | None,
960:        Perform star
…[3598 chars omitted]…
e stats to coordinator for "internal"
            # and "hybrid" LB modes.
            self.publish_dp_lb_stats = internal_dp_balancing

            self.addresses = addresses
            self.process_input_queue_block = True
            if envs.VLLM_ELASTIC_EP_SCALE_UP_LAUNCH:
                self._eep_send_engine_core_notification(
                    EEPNotificationType.NEW_CORE_ENGINES_INIT_READY,

                    "headless": headless,
                }
            )
        )

        # Receive initialization message.
        logger.debug("Waiting for init message from front-end.")
        if not handshake_socket.poll(timeout=HANDSHAKE_TIMEOUT_MINS * 60_000):
            raise RuntimeError(
                "Did not receive response from front-end "
                f"process within {HANDSHAKE_TIMEOUT_MINS} "
                f"minutes"
            )
        init_bytes = handshak
…[2499 chars omitted]…
dent, so treat like DP=1.
                # Note that parallel_config.data_parallel_index will still reflect
                # the original DP rank.
                parallel_config.data_parallel_size = 1
                parallel_config.data_parallel_size_local = 1
                parallel_config.data_parallel_rank = 0
                engine_core = EngineCoreProc(*args, engine_index=dp_rank, **kwargs)

@dataclass
class ElasticScalingCache:
    existing_core_engines: list[EngineIdentity]
    num_new_core_engines: int
    pending_notifications: dict[EEPNotificationType, set[int]]


class MPClient(EngineCoreClient):
    """
    MPClient: base client for multi-proc EngineCore.
        EngineCore runs in a background process busy loop, gett
…[1653 chars omitted]…
                input_address = client_addresses["input_address"]
                output_address = client_addresses["output_address"]
                self.stats_update_address = client_addresses.get("stats_update_address")
                # Tensor queues passed via client_addresses for multi-API-server case
                tensor_queue = client_addresses.get("tensor_queue")  # type: ignore[assignment]

    ) -> None:
        """
        Create an AsyncLLM.

        Args:
            vllm_config: global configuration.
            executor_class: an Executor impl, e.g. MultiprocExecutor.
            log_stats: Whether to log stats.
            usage_context: Usage context of the LLM.
            mm_registry: Multi-modal registry.
            log_requests: Whether t
…[4359 chars omitted]…
engine_args.enable_log_requests,
            log_stats=not engine_args.disable_log_stats,
            start_engine_loop=start_engine_loop,
            usage_context=usage_context,
            stat_loggers=stat_loggers,
        )

    def __del__(self):
        self.shutdown()

    def shutdown(self, timeout: float | None = None) -> None:
        """Shutdown, cleaning up the background proc and IPC."""

    Otherwise, the provided host and port will be used to construct a TCP
    address (port == 0 means assign an available port)."""

    return (
        get_open_zmq_ipc_path()
        if local_only
        else (get_tcp_uri(host, port or get_open_port()))
    )


class APIServerProcessManager:
    """Manages a group of API server processes.

    Handles creation, monitori
…[1076 chars omitted]…
lm.utils.network_utils import get_open_port, get_open_zmq_ipc_path, get_tcp_uri
from vllm.utils.system_utils import decorate_logs, kill_process_tree, set_process_title
from vllm.v1.core.sched.output import SchedulerOutput

if TYPE_CHECKING:
    import numpy as np

    from vllm.v1.engine.coordinator import DPCoordinator
    from vllm.v1.engine.utils import CoreEngineActorManager, CoreEngineProcManager

        Raises:
            RuntimeError: If a child process exits before reporting its
                endpoints, or if the timeout expires.
        """
        if self._gathered_addresses is not None:
            return self._gathered_addresses

        num_servers = len(self.processes)
        input_addresses: list[str | None] = [None] * num_servers
        output_addresses: list[str | None] = [None] * num_servers
        pipe_to_index = {pipe: i for i, pipe in enumerate(se
…[1997 chars omitted]…
               f"API servers reported duplicate addresses: "
                f"inputs={inputs}, outputs={outputs}"
            )

        self._gathered_addresses = (inputs, outputs)
        for pipe in self._address_pipes:
            with contextlib.suppress(Exception):
                pipe.close()
        return self._gathered_addresses

    def shutdown(self, timeout: float | None = None) -> None:

14:from multiprocessing.queues import Queue
15:from threading import Thread
16:from typing import Any, TypeAlias, TypeVar
18:import msgspec.msgpack
19:import zmq
20:import zmq.asyncio
22:from vllm.config import VllmConfig
23:from vllm.envs import VLLM_ENGINE_READY_TIMEOUT_S
24:from vllm.logger import init_logger
25:from vllm.lora.request import LoRARequest
26:from vllm.tasks import SupportedTask
27:from vllm.tracing import instrument
28:from vllm.utils.async_utils import in_loop
29:from vllm.utils.network_utils import (
34:from vllm.v1.engine import (
46:from vllm.v1.engine.coordinator import DPCoordinator
47:from vllm.v1.engine.core import EngineCore, EngineCoreProc
48:from vllm.v1.engine.exceptions import EngineDeadError
49:from vllm.v1.engine.tensor_ipc import TensorIpcSender

  File "/usr/local/lib/python3.12/multiprocessing/spawn.py", line 297, in _fixup_main_from_path
    main_content = runpy.run_path(main_path,
                   ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<frozen runpy>", line 287, in run_path
  File "<frozen runpy>", line 98, in _run_modu
…[7715 chars omitted]…
or: API server process ApiServer_0 (PID: 953) exited with code 1 before reporting its addresses.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/tmp/repro.py", line 35, in <module>
    assert expect_fail, e
           ^^^^^^^^^^^
AssertionError: API server process ApiServer_0 (PID: 953) exited with code 1 before reporting its addresses.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/tmp/repro.py", line 54, in <module>
    main()
  File "/tmp/repro.py", line 40, in main
    assert "reporting" in str(e), e
           ^^^^^^^^^^^^^^^^^^^^^
AssertionError: Timed out after 15 seconds waiting for API server processes to report their addresses. Still waiting on: ['ApiServer_1', 'ApiServer_2', 'ApiServer_0']

# Global variables to control worker behavior
WORKER_RUNTIME_SECONDS = 0.5


# Mock implementation of run_api_server_worker
def mock_run_api_server_worker(listen_address, sock, args, client_config=None):
    """Mock run_api_server_worker that runs for a specific time."""
    print(f"Mock worker started with client_config: {client_config}")
    time.sleep(WORKER_RUNTIME_SECONDS)
    print("Mock worker completed successfully")


@pytest.fixture
def api_server_args():
    """Fixture to provide arguments for APIServerProcessManager."""
    sock = socket.socket()
    return {
        "targe
…[2508 chars omitted]…
ver_manager=manager)
            except Exception as e:
                result["exception"] = e
            finally:
                manager.shutdown()

        # Start a thread to run wait_for_completion_or_failure
        wait_thread = threading.Thread(target=run_with_exception_capture, daemon=True)
        wait_thread.start()

        # Let all processes run for a short time
        time.sleep(0.2)

no changes added to commit (use "git add" and/or "git commit -a")
Dropped refs/stash@{0} (c0529829e6efa1562c66cbab4c67b5e6209dbe1f)

+            addresses.inputs = actual_inputs
+            addresses.outputs = actual_outputs
 
     # Wait for API servers.
     try:
diff --git a/vllm/v1/engine/core_client.py b/vllm/v1/engine/core_client.py
index 2f2c15d..be4a3c7 100644
---
…[8628 chars omitted]…
+
+        self._gathered_addresses = (inputs, outputs)
+        for pipe in self._address_pipes:
+            with contextlib.suppress(Exception):
+                pipe.close()
+        return self._gathered_addresses
+
     def shutdown(self, timeout: float | None = None) -> None:
         """Shutdown API server processes with configurable timeout"""
         if self._finalizer.detach() is not None: