refactor: remove port reservation system in favor of manual specification (#4142)

Signed-off-by: alec-flowers <[email protected]>

Author: alec-flowers

components/src/dynamo/vllm/args.py

@@ -4,6 +4,7 @@
 
 import logging
 import os
+import socket
 from typing import Any, Dict, Optional
 
 from vllm.config import KVTransferConfig
@@ -13,24 +14,12 @@
 
 from dynamo._core import get_reasoning_parser_names, get_tool_parser_names
 from dynamo.common.config_dump import add_config_dump_args, register_encoder
-from dynamo.runtime import DistributedRuntime
-
-from . import __version__
-from .ports import (
-    DEFAULT_DYNAMO_PORT_MAX,
-    DEFAULT_DYNAMO_PORT_MIN,
-    DynamoPortRange,
-    PortAllocationRequest,
-    PortMetadata,
-    allocate_and_reserve_port,
-    allocate_and_reserve_port_block,
-    get_host_ip,
-)
+
+from . import __version__, envs
 
 logger = logging.getLogger(__name__)
 
 DEFAULT_MODEL = "Qwen/Qwen3-0.6B"
-
 VALID_CONNECTORS = {"nixl", "lmcache", "kvbm", "null", "none"}
 
 # Global LMCache configuration - initialize once on module import
@@ -48,7 +37,6 @@ class Config:
     is_decode_worker: bool
     migration_limit: int = 0
     kv_port: Optional[int] = None
-    port_range: DynamoPortRange
     custom_jinja_template: Optional[str] = None
 
     # mirror vLLM
@@ -115,18 +103,6 @@ def parse_args() -> Config:
         default=0,
         help="Maximum number of times a request may be migrated to a different engine worker. The number may be overridden by the engine.",
     )
-    parser.add_argument(
-        "--dynamo-port-min",
-        type=int,
-        default=DEFAULT_DYNAMO_PORT_MIN,
-        help=f"Minimum port number for Dynamo services (default: {DEFAULT_DYNAMO_PORT_MIN}). Must be in registered ports range (1024-49151).",
-    )
-    parser.add_argument(
-        "--dynamo-port-max",
-        type=int,
-        default=DEFAULT_DYNAMO_PORT_MAX,
-        help=f"Maximum port number for Dynamo services (default: {DEFAULT_DYNAMO_PORT_MAX}). Must be in registered ports range (1024-49151).",
-    )
     parser.add_argument(
         "--connector",
         nargs="*",
@@ -249,9 +225,6 @@ def parse_args() -> Config:
     config.is_prefill_worker = args.is_prefill_worker
     config.is_decode_worker = args.is_decode_worker
     config.migration_limit = args.migration_limit
-    config.port_range = DynamoPortRange(
-        min=args.dynamo_port_min, max=args.dynamo_port_max
-    )
     config.tool_call_parser = args.dyn_tool_call_parser
     config.reasoning_parser = args.dyn_reasoning_parser
     config.custom_jinja_template = args.custom_jinja_template
@@ -315,67 +288,14 @@ def parse_args() -> Config:
     return config
 
 
-async def configure_ports(runtime: DistributedRuntime, config: Config):
-    """Configure including port allocation and vLLM overrides."""
-
-    dp_rank = config.engine_args.data_parallel_rank or 0
-    worker_id = f"vllm-{config.component}-dp{dp_rank}"
+async def configure_ports(config: Config):
+    """Configure port settings from dedicated environment overrides."""
 
-    # Allocate KV events port
     if config.engine_args.enable_prefix_caching:
-        kv_metadata = PortMetadata(worker_id=worker_id, reason="zmq_kv_event_port")
-        kv_port = await allocate_and_reserve_port(
-            runtime=runtime,
-            namespace=config.namespace,
-            metadata=kv_metadata,
-            port_range=config.port_range,
-        )
-        config.kv_port = kv_port
-        logger.info(f"Allocated ZMQ KV events port: {kv_port} (worker_id={worker_id})")
-
-        # Check if NIXL is needed based on connector list
-    needs_nixl = config.has_connector("nixl")
-
-    if needs_nixl:
-        # Allocate side channel ports
-        # https://github.com/vllm-project/vllm/blob/releases/v0.10.0/vllm/distributed/kv_transfer/kv_connector/v1/nixl_connector.py#L372
-        # NIXL calculates ports as: base_port + (dp_rank * tp_size) + tp_rank
-        # For dp_rank, we need to reserve tp_size consecutive ports
-        tp_size = config.engine_args.tensor_parallel_size or 1
-
-        # The first port for this dp_rank will be at: base_port + (dp_rank * tp_size)
-        # We need to allocate tp_size consecutive ports starting from there
-        nixl_metadata = PortMetadata(
-            worker_id=worker_id, reason="nixl_side_channel_port"
-        )
-        nixl_request = PortAllocationRequest(
-            metadata=nixl_metadata,
-            port_range=config.port_range,
-            block_size=tp_size,
-        )
-        allocated_ports = await allocate_and_reserve_port_block(
-            runtime, config.namespace, nixl_request
-        )
-        first_port_for_dp_rank = allocated_ports[0]
-
-        # Calculate the base port that NIXL expects
-        # base_port = first_port_for_dp_rank - (dp_rank * tp_size)
-        nixl_offset = dp_rank * tp_size
-        base_side_channel_port = first_port_for_dp_rank - nixl_offset
+        config.kv_port = envs.DYN_VLLM_KV_EVENT_PORT
 
-        if base_side_channel_port < 0:
-            raise ValueError(
-                f"NIXL base port calculation resulted in negative port: "
-                f"first_allocated_port={first_port_for_dp_rank}, offset={nixl_offset}, "
-                f"base_port={base_side_channel_port}. Current range: {config.port_range.min}-{config.port_range.max}. "
-                f"Consider using a higher port range."
-            )
-
-        logger.info(
-            f"Allocated NIXL side channel ports: base={base_side_channel_port}, "
-            f"allocated_ports={allocated_ports} (worker_id={worker_id}, dp_rank={dp_rank}, tp_size={tp_size})"
-        )
-        set_side_channel_host_and_port(base_side_channel_port)
+    if config.has_connector("nixl"):
+        ensure_side_channel_host()
 
 
 def create_kv_events_config(config: Config) -> Optional[KVEventsConfig]:
@@ -385,18 +305,18 @@ def create_kv_events_config(config: Config) -> Optional[KVEventsConfig]:
         return None
 
     # If user provided their own config, use that
-    if getattr(config.engine_args, "kv_events_config"):
-        logger.info("Using user-provided kv_events_config")
+    if c := getattr(config.engine_args, "kv_events_config"):
+        logger.info(f"Using user-provided kv_events_config {c}")
         return None
 
     # Create default events config for prefix caching
-    logger.info("Creating Dynamo default kv_events_config for prefix caching")
     if config.kv_port is None:
         raise ValueError(
             "config.kv_port is not set; call configure_ports(...) before overwrite_args "
             "or provide --kv-event-config to supply an explicit endpoint."
         )
     dp_rank = config.engine_args.data_parallel_rank or 0
+
     return KVEventsConfig(
         enable_kv_cache_events=True,
         publisher="zmq",
@@ -472,6 +392,10 @@ def overwrite_args(config):
         defaults["kv_transfer_config"] = kv_transfer_config
 
     kv_events_config = create_kv_events_config(config)
+    logger.info(
+        f"Using Dynamo default kv_events_config for publishing kv events over zmq: {kv_events_config}"
+    )
+
     if kv_events_config:
         defaults["kv_events_config"] = kv_events_config
 
@@ -484,11 +408,45 @@ def overwrite_args(config):
             raise ValueError(f"{key} not found in AsyncEngineArgs from vLLM.")
 
 
-def set_side_channel_host_and_port(side_channel_port: int):
-    """vLLM V1 NixlConnector creates a side channel to exchange metadata with other NIXL connectors.
-    This sets the port number for the side channel.
-    """
+def get_host_ip() -> str:
+    """Get the IP address of the host for side-channel coordination."""
+    try:
+        host_name = socket.gethostname()
+    except socket.error as exc:
+        logger.warning("Failed to get hostname: %s, falling back to 127.0.0.1", exc)
+        return "127.0.0.1"
+
+    try:
+        host_ip = socket.gethostbyname(host_name)
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as test_socket:
+            test_socket.bind((host_ip, 0))
+        return host_ip
+    except socket.gaierror as exc:
+        logger.warning(
+            "Hostname %s cannot be resolved: %s, falling back to 127.0.0.1",
+            host_name,
+            exc,
+        )
+        return "127.0.0.1"
+    except socket.error as exc:
+        logger.warning(
+            "Hostname %s is not usable for binding: %s, falling back to 127.0.0.1",
+            host_name,
+            exc,
+        )
+        return "127.0.0.1"
+
+
+def ensure_side_channel_host():
+    """Ensure the NIXL side-channel host is available without overriding user settings."""
+
+    existing_host = os.getenv("VLLM_NIXL_SIDE_CHANNEL_HOST")
+    if existing_host:
+        logger.debug(
+            "Preserving existing VLLM_NIXL_SIDE_CHANNEL_HOST=%s", existing_host
+        )
+        return
+
     host_ip = get_host_ip()
     os.environ["VLLM_NIXL_SIDE_CHANNEL_HOST"] = host_ip
-    os.environ["VLLM_NIXL_SIDE_CHANNEL_PORT"] = str(side_channel_port)
-    logger.debug(f"Set NIXL side channel to {host_ip}:{side_channel_port}")
+    logger.debug("Set VLLM_NIXL_SIDE_CHANNEL_HOST to %s", host_ip)

components/src/dynamo/vllm/envs.py

@@ -0,0 +1,82 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Environment variable configuration for Dynamo vLLM integration.
+
+This module provides a centralized location for managing environment variables
+used by Dynamo's vLLM backend, following vLLM's pattern.
+"""
+
+import os
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any
+
+# Port range constants
+REGISTERED_PORT_MIN = 1024
+REGISTERED_PORT_MAX = 49151
+
+if TYPE_CHECKING:
+    DYN_VLLM_KV_EVENT_PORT: int = 20080
+
+
+def _resolve_port(env_var: str, default_port: int) -> int:
+    """
+    Resolve port from environment variable with validation.
+
+    Args:
+        env_var: Environment variable name
+        default_port: Default port if env var not set
+
+    Returns:
+        Validated port number
+
+    Raises:
+        ValueError: If port is invalid or out of range
+    """
+    env_value = os.getenv(env_var)
+    if env_value is None:
+        port = default_port
+    else:
+        try:
+            port = int(env_value)
+        except ValueError as exc:
+            raise ValueError(
+                f"{env_var} must be an integer port number, got {env_value!r}."
+            ) from exc
+
+    if not (REGISTERED_PORT_MIN <= port <= REGISTERED_PORT_MAX):
+        raise ValueError(
+            f"{env_var} port {port} is outside of the registered port range "
+            f"({REGISTERED_PORT_MIN}-{REGISTERED_PORT_MAX})."
+        )
+
+    return port
+
+
+# Environment variables configuration
+environment_variables: dict[str, Callable[[], Any]] = {
+    # Port used for KV events publishing to the frontend
+    # Note: This env variable is ignored if explicitly using --kv-events-config ''
+    "DYN_VLLM_KV_EVENT_PORT": lambda: _resolve_port("DYN_VLLM_KV_EVENT_PORT", 20080),
+}
+
+
+def __getattr__(name: str):
+    """
+    Gets environment variables lazily.
+    """
+    if name in environment_variables:
+        return environment_variables[name]()
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__():
+    return list(environment_variables.keys())
+
+
+def is_set(name: str) -> bool:
+    """Check if an environment variable is explicitly set."""
+    if name in environment_variables:
+        return name in os.environ
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

components/src/dynamo/vllm/main.py

@@ -74,7 +74,7 @@ async def graceful_shutdown(runtime):
 async def worker(runtime: DistributedRuntime):
     config = parse_args()
 
-    await configure_ports(runtime, config)
+    await configure_ports(config)
     overwrite_args(config)
 
     # Set up signal handler for graceful shutdown