main.py 28.5 KB
Newer Older
1
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
Alec's avatar
Alec committed
2
3
# SPDX-License-Identifier: Apache-2.0

4
import argparse
Alec's avatar
Alec committed
5
6
7
import asyncio
import logging
import os
8
import tempfile
9
import time
10
11
12
13
from typing import TYPE_CHECKING, Any, Optional

if TYPE_CHECKING:
    from dynamo.vllm.omni.args import OmniConfig
Alec's avatar
Alec committed
14
15

import uvloop
16
from prometheus_client import REGISTRY, CollectorRegistry, multiprocess
17
from vllm.config import VllmConfig
Alec's avatar
Alec committed
18
19
20
from vllm.distributed.kv_events import ZmqEventPublisher
from vllm.usage.usage_lib import UsageContext
from vllm.v1.engine.async_llm import AsyncLLM
21
from vllm.v1.metrics.prometheus import setup_multiprocess_prometheus
Alec's avatar
Alec committed
22

23
from dynamo.common.config_dump import dump_config
24
from dynamo.common.utils.graceful_shutdown import install_signal_handlers
25
26
27
28
from dynamo.common.utils.prometheus import (
    LLMBackendMetrics,
    register_engine_metrics_callback,
)
29
from dynamo.common.utils.runtime import create_runtime
Alec's avatar
Alec committed
30
from dynamo.llm import (
31
    KvEventPublisher,
32
    ModelInput,
33
    ModelRuntimeConfig,
Alec's avatar
Alec committed
34
    ModelType,
35
36
    fetch_model,
    register_model,
Alec's avatar
Alec committed
37
)
38
from dynamo.runtime import Endpoint
Alec's avatar
Alec committed
39
from dynamo.runtime.logging import configure_dynamo_logging
40
from dynamo.vllm.worker_factory import WorkerFactory
Alec's avatar
Alec committed
41

42
from . import envs
43
from .args import Config, _uses_dynamo_connector, parse_args
44
from .constants import DisaggregationMode
45
from .handlers import get_dp_range_for_worker
46
from .publisher import DYNAMO_COMPONENT_REGISTRY, StatLoggerFactory
47
from .snapshot import prepare_snapshot_engine
Alec's avatar
Alec committed
48

jh-nv's avatar
jh-nv committed
49
50
51
52
53
54
55
56
57
58
59
60
# Optional imports for frontend decoding support
MediaDecoder: type | None = None
MediaFetcher: type | None = None
try:
    from dynamo.llm import MediaDecoder, MediaFetcher

    MEDIA_DECODER_AVAILABLE = True
except ImportError:
    MediaDecoder = None
    MediaFetcher = None
    MEDIA_DECODER_AVAILABLE = False

Alec's avatar
Alec committed
61
62
configure_dynamo_logging()
logger = logging.getLogger(__name__)
63
shutdown_endpoints: list = []
Alec's avatar
Alec committed
64
65


66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def build_headless_namespace(config: Config) -> argparse.Namespace:
    """Build an argparse Namespace from engine_args for vLLM's run_headless().

    run_headless() expects the raw CLI namespace. We reconstruct it from
    the already-parsed AsyncEngineArgs so parse_args() doesn't need to
    leak transport details.
    """
    ns = argparse.Namespace(**vars(config.engine_args))
    # run_headless() reads api_server_count; default to 0 (no API server)
    if not hasattr(ns, "api_server_count"):
        ns.api_server_count = 0
    return ns


def run_dynamo_headless(config: Config) -> None:
    """Run in headless mode for multi-node TP/PP.

    Secondary nodes spawn vLLM workers only — no engine core, no scheduler,
    no Dynamo endpoints. Bypasses DistributedRuntime entirely (no NATS/etcd).
    """
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
    # Propagate worker_cls for custom load formats so headless workers use
    # the same model loader and patches as the leader node.
    if config.engine_args.load_format == "gms":
        config.engine_args.worker_cls = (
            "gpu_memory_service.integrations.vllm.worker.GMSWorker"
        )

        if config.gms_shadow_mode:
            from gpu_memory_service.integrations.vllm.utils import (
                configure_gms_lock_mode,
                validate_cudagraph_mode,
            )

            os.environ["DYN_GMS_SHADOW_MODE"] = "1"
            configure_gms_lock_mode(config.engine_args)
            validate_cudagraph_mode(config.engine_args)

    elif config.engine_args.load_format in ("mx-source", "mx-target"):
        config.engine_args.worker_cls = "modelexpress.vllm_worker.ModelExpressWorker"

106
107
108
109
    # Keep the upstream CLI import local so tests that only exercise
    # build_headless_namespace() do not pull in vLLM's full CLI import graph.
    from vllm.entrypoints.cli.serve import run_headless

110
111
112
113
    args = build_headless_namespace(config)
    run_headless(args)


jh-nv's avatar
jh-nv committed
114
async def worker() -> None:
Alec's avatar
Alec committed
115
116
    config = parse_args()

117
118
119
120
121
122
123
124
125
126
    dump_config(config.dump_config_to, config)

    # Name the model. Use either the full path (vllm and sglang do the same),
    # or the HF name (e.g. "Qwen/Qwen3-0.6B"), depending on cmd line params.
    if not config.served_model_name:
        config.served_model_name = config.engine_args.served_model_name = config.model

    # Download the model if necessary using modelexpress.
    # We want it on disk before we start vllm to avoid downloading from HuggingFace.
    #
127
    # We don't set `config.engine_args.model` to the local path fetch_model returns
128
129
130
131
132
133
    # because vllm will send that name to its Ray pipeline-parallel workers, which
    # may not have the local path.
    # vllm will attempt to download the model again, but find it in the HF cache.
    # For non-HF models use a path instead of an HF name, and ensure all workers have
    # that path (ideally via a shared folder).
    if not os.path.exists(config.model):
134
        await fetch_model(config.model)
135

136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
    # CHECKPOINT MODE: Load engine BEFORE runtime creation
    # This allows checkpointing GPU state before runtime connections are established
    snapshot_controller = await prepare_snapshot_engine(
        config,
        setup_vllm_engine,
    )

    snapshot_engine = None
    if snapshot_controller is not None:
        snapshot_engine = snapshot_controller.engine
        (
            config.namespace,
            config.discovery_backend,
        ) = snapshot_controller.reload_restore_identity()

151
152
153
154
155
156
    # HEADLESS MODE: bypass DistributedRuntime entirely.
    # Workers run vLLM only (no NATS, etcd, or dynamo endpoints).
    if config.headless:
        run_dynamo_headless(config)
        return

157
    shutdown_event = asyncio.Event()
158
    runtime, loop = create_runtime(
159
        discovery_backend=config.discovery_backend,
160
161
162
        request_plane=config.request_plane,
        event_plane=config.event_plane,
        use_kv_events=config.use_kv_events,
163
164
    )

165
166
    # [gluo FIXME] should be after init() below? 'shutdown_endpoints' are populated
    # there
167
168
    install_signal_handlers(loop, runtime, shutdown_endpoints, shutdown_event)

169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
    # Use WorkerFactory to appropriate initialize worker based on config flags
    factory = WorkerFactory(
        setup_vllm_engine_fn=setup_vllm_engine,
        setup_kv_event_publisher_fn=setup_kv_event_publisher,
        register_vllm_model_fn=register_vllm_model,
        setup_fpm_relay_fn=setup_fpm_relay,
        setup_metrics_collection_fn=setup_metrics_collection,
    )
    await factory.create(
        runtime,
        config,
        shutdown_event,
        shutdown_endpoints,
        snapshot_engine=snapshot_engine,
    )
184
185

    logger.debug("Worker function completed, exiting...")
Alec's avatar
Alec committed
186
187


jh-nv's avatar
jh-nv committed
188
def setup_metrics_collection(
189
    config: "Config | OmniConfig", generate_endpoint: Endpoint, logger: logging.Logger
jh-nv's avatar
jh-nv committed
190
) -> None:
191
192
193
194
195
196
197
198
199
200
201
202
203
    """Set up metrics collection for vLLM and LMCache metrics.

    In multiprocess mode (PROMETHEUS_MULTIPROC_DIR set), metrics are stored:
      1. In-memory: Metric objects in global REGISTRY
      2. On-disk: Metric values in .db files (PROMETHEUS_MULTIPROC_DIR)

    MultiProcessCollector reads from .db files but adding it to REGISTRY can fail
    with "Duplicated timeseries" if PROMETHEUS_MULTIPROC_DIR was set before process
    started (K8s deployments) because metrics are already in REGISTRY.

    Solution: Try adding MultiProcessCollector to REGISTRY. If that fails, use
    separate registry for multiprocess collection and register callbacks to both
    registries to ensure all metrics (vllm, lmcache, dynamo_component) are collected.
204
205
206
207
208

    Auto-label injection:
        Hierarchy labels (dynamo_namespace, dynamo_component, dynamo_endpoint) are automatically
        injected into engine metrics to align Python metrics with Rust auto-labels.
        Additional labels can be provided via inject_labels parameter.
209
210
    """
    if config.engine_args.disable_log_stats is False:
211
212
213
214
215
216
217
218
219
220
221
        # Register the dedicated dynamo_component registry callback
        # IMPORTANT: We do NOT use MultiProcessCollector for DYNAMO_COMPONENT_REGISTRY
        # because our gauges use in-memory values which work fine for single-process
        # and multi-process (each process has its own gauge with dp_rank label).
        # Using MultiProcessCollector would read from .db files which causes stale
        # values to accumulate across test runs.
        register_engine_metrics_callback(
            endpoint=generate_endpoint,
            registry=DYNAMO_COMPONENT_REGISTRY,
        )

222
223
224
225
226
227
228
229
230
        multiproc_dir = os.environ.get("PROMETHEUS_MULTIPROC_DIR")
        # After CRIU restore to another node, env still has the checkpoint pod's path
        # but that directory exists only on the checkpoint node; create it here if missing.
        if multiproc_dir and not os.path.isdir(multiproc_dir):
            try:
                os.makedirs(multiproc_dir, exist_ok=True)
            except OSError:
                pass
        if multiproc_dir and os.path.isdir(multiproc_dir):
231
232
233
234
235
236
237
238
            try:
                # MultiProcessCollector reads metrics from .db files in PROMETHEUS_MULTIPROC_DIR
                # Adding it to REGISTRY allows collecting both in-memory and .db file metrics
                multiprocess.MultiProcessCollector(REGISTRY)
                logger.debug("Added MultiProcessCollector to global REGISTRY")
                register_engine_metrics_callback(
                    endpoint=generate_endpoint,
                    registry=REGISTRY,
239
240
241
242
243
                    metric_prefix_filters=["vllm:", "lmcache:"],
                    namespace_name=config.namespace,
                    component_name=config.component,
                    endpoint_name=config.endpoint,
                    model_name=config.model,
244
245
246
247
248
249
250
251
252
253
254
                )
            except ValueError as e:
                # Conflict: metrics already in REGISTRY, MultiProcessCollector tries to add same metrics from .db files
                # Solution: Use separate registry that ONLY reads from .db files (no in-memory conflicts)
                logger.debug(
                    f"Could not add MultiProcessCollector to REGISTRY ({e}), using separate registry"
                )
                multiproc_registry = CollectorRegistry()
                multiprocess.MultiProcessCollector(multiproc_registry)

                # Register both registries to collect all metrics
255
                # Global REGISTRY has in-memory metrics (vllm)
256
257
258
                register_engine_metrics_callback(
                    endpoint=generate_endpoint,
                    registry=REGISTRY,
259
                    metric_prefix_filters=["vllm:"],
260
261
262
263
                    namespace_name=config.namespace,
                    component_name=config.component,
                    endpoint_name=config.endpoint,
                    model_name=config.model,
264
265
266
267
268
                )
                # Multiproc registry has .db file metrics (lmcache, possibly vllm duplicates)
                register_engine_metrics_callback(
                    endpoint=generate_endpoint,
                    registry=multiproc_registry,
269
270
271
272
273
                    metric_prefix_filters=["vllm:", "lmcache:"],
                    namespace_name=config.namespace,
                    component_name=config.component,
                    endpoint_name=config.endpoint,
                    model_name=config.model,
274
275
                )
        else:
276
277
278
279
280
            if multiproc_dir:
                logger.warning(
                    f"PROMETHEUS_MULTIPROC_DIR={multiproc_dir} is not a valid directory, "
                    "falling back to single-process metrics"
                )
281
282
283
284
285
            # No multiprocess mode
            register_engine_metrics_callback(
                endpoint=generate_endpoint,
                registry=REGISTRY,
                metric_prefix_filters=["vllm:", "lmcache:"],
286
287
288
289
                namespace_name=config.namespace,
                component_name=config.component,
                endpoint_name=config.endpoint,
                model_name=config.model,
290
291
292
            )


Yan Ru Pei's avatar
Yan Ru Pei committed
293
294
def setup_kv_event_publisher(
    config: Config,
295
296
    generate_endpoint: Endpoint,
    vllm_config: VllmConfig,
297
298
    consolidator_enabled: bool = False,
    consolidator_port: Optional[int] = 5558,
jh-nv's avatar
jh-nv committed
299
) -> Optional[list[KvEventPublisher]]:
Yan Ru Pei's avatar
Yan Ru Pei committed
300
    """
jh-nv's avatar
jh-nv committed
301
    list[KvEventPublisher] | None
Yan Ru Pei's avatar
Yan Ru Pei committed
302
303
    Set up KV event publishers for prefix caching if enabled.
    Creates one publisher per dp_rank since each dp_rank publishes to a different port.
304
305
306
307
308
309
310
    Args:
        config: Worker configuration
        generate_endpoint: Endpoint for worker ID
        vllm_config: vLLM configuration
        consolidator_enabled: If True, subscribe to kv eventconsolidator's ZMQ endpoint
        consolidator_port: Port where kv event consolidator publishes (default: 5558)

Yan Ru Pei's avatar
Yan Ru Pei committed
311
    Returns:
312
        List of KvEventPublisher instances (one per dp_rank) if prefix caching is enabled, None otherwise.
Yan Ru Pei's avatar
Yan Ru Pei committed
313
314
315
316
    """
    if not config.engine_args.enable_prefix_caching:
        return None

317
    # Skip KV event publishing for decode workers
318
    if config.disaggregation_mode == DisaggregationMode.DECODE:
319
320
321
        logger.info("Skipping KV event publisher setup for decode worker")
        return None

322
323
324
    if config.engine_args.kv_events_config is None:
        return None

325
326
327
328
329
330
331
    # Check if kv_cache_events are explicitly disabled
    if not config.engine_args.kv_events_config.enable_kv_cache_events:
        logger.info(
            "KV event publishing skipped: enable_kv_cache_events=False in kv_events_config"
        )
        return None

332
333
334
    # Get DP rank range managed by this worker to create publishers for corresponding dp_ranks,
    # all served workers should cover all ranks.
    dp_start, dp_size = get_dp_range_for_worker(vllm_config)
Yan Ru Pei's avatar
Yan Ru Pei committed
335
336
    kv_publishers = []

337
    for dp_rank in range(dp_start, dp_start + dp_size):
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
        if consolidator_enabled:
            # TODO: Use different port for each dp_rank once KVBM supports DP
            zmq_endpoint = f"tcp://127.0.0.1:{consolidator_port}"
            logger.info(
                f"KV event publisher for dp_rank={dp_rank} subscribing to consolidator at {zmq_endpoint}"
            )
        else:
            # Each dp_rank publishes to a different port
            zmq_endpoint = ZmqEventPublisher.offset_endpoint_port(
                config.engine_args.kv_events_config.endpoint,
                data_parallel_rank=dp_rank,
            ).replace("*", "127.0.0.1")
            logger.info(
                f"KV event publisher for dp_rank={dp_rank} subscribing to vLLM at {zmq_endpoint}"
            )
Yan Ru Pei's avatar
Yan Ru Pei committed
353

354
        kv_publisher = KvEventPublisher(
355
            endpoint=generate_endpoint,
Yan Ru Pei's avatar
Yan Ru Pei committed
356
357
            kv_block_size=vllm_config.cache_config.block_size,
            zmq_endpoint=zmq_endpoint,
358
            zmq_topic="",
359
            enable_local_indexer=config.enable_local_indexer,
360
            dp_rank=dp_rank,
Yan Ru Pei's avatar
Yan Ru Pei committed
361
362
        )
        kv_publishers.append(kv_publisher)
Yan Ru Pei's avatar
Yan Ru Pei committed
363

Yan Ru Pei's avatar
Yan Ru Pei committed
364
365
366
        logger.info(
            f"Worker reading KV events for dp_rank={dp_rank} from {zmq_endpoint}"
        )
Yan Ru Pei's avatar
Yan Ru Pei committed
367

Yan Ru Pei's avatar
Yan Ru Pei committed
368
    return kv_publishers if kv_publishers else None
Yan Ru Pei's avatar
Yan Ru Pei committed
369
370


371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def setup_fpm_relay(
    generate_endpoint: Endpoint,
    vllm_config: VllmConfig,
) -> Optional[list]:
    """
    Set up forward pass metrics relays for the event plane.

    Creates one FpmEventRelay per dp_rank. Each relay subscribes to the
    local raw ZMQ PUB from InstrumentedScheduler (in the EngineCore child
    process) and re-publishes to the Dynamo event plane with automatic
    discovery registration.

    Returns:
        List of FpmEventRelay instances, or None if FPM is not enabled.
    """
    if not envs.is_set("DYN_FORWARDPASS_METRIC_PORT"):
        return None

    try:
        from dynamo.llm import FpmEventRelay
    except ImportError:
        logger.warning(
            "FpmEventRelay not available (Rust bindings not built with FPM support). "
            "Forward pass metrics will not be relayed to the event plane."
        )
        return None

    dp_start, dp_size = get_dp_range_for_worker(vllm_config)
    relays = []

    for dp_rank in range(dp_start, dp_start + dp_size):
        base_port = envs.DYN_FORWARDPASS_METRIC_PORT
        zmq_endpoint = f"tcp://127.0.0.1:{base_port + dp_rank}"

        relay = FpmEventRelay(
            endpoint=generate_endpoint,
            zmq_endpoint=zmq_endpoint,
        )
        relays.append(relay)

        logger.info(f"FPM relay for dp_rank={dp_rank} subscribing to {zmq_endpoint}")

    return relays if relays else None


jh-nv's avatar
jh-nv committed
416
def setup_vllm_engine(
417
418
419
    config: Config,
    stat_logger: Optional[StatLoggerFactory] = None,
    fpm_worker_id: Optional[str] = None,
jh-nv's avatar
jh-nv committed
420
) -> tuple[AsyncLLM, VllmConfig, Any, Any, LLMBackendMetrics]:
421
422
423
    # vLLM v0.11.0 bug: vllm/v1.metrics/prometheus.py:79 passes TemporaryDirectory object
    # instead of .name string, causing false error on exit. Set PROMETHEUS_MULTIPROC_DIR
    # ourselves to avoid this and handle cleanup properly.
424
    prometheus_temp_dir = None
425
426
427
428
429
430
    existing_dir = os.environ.get("PROMETHEUS_MULTIPROC_DIR")
    if existing_dir and not os.path.isdir(existing_dir):
        logger.warning(
            f"PROMETHEUS_MULTIPROC_DIR={existing_dir} does not exist, recreating"
        )
        os.makedirs(existing_dir, exist_ok=True)
431
432
433
434
435
436
437
    if "PROMETHEUS_MULTIPROC_DIR" not in os.environ:
        prometheus_temp_dir = tempfile.TemporaryDirectory(prefix="vllm_prometheus_")
        os.environ["PROMETHEUS_MULTIPROC_DIR"] = prometheus_temp_dir.name
        logger.debug(
            f"Created PROMETHEUS_MULTIPROC_DIR at: {os.environ['PROMETHEUS_MULTIPROC_DIR']}"
        )

438
    setup_multiprocess_prometheus()  # call vLLM's library's function to setup multiprocess prometheus
439
440
441
442
    logger.debug(
        f"Prometheus multiproc dir set to: {os.environ.get('PROMETHEUS_MULTIPROC_DIR')}"
    )

443
444
445
446
447
448
449
450
451
452
453
454
455
    # Construct Prometheus gauges AFTER setup_multiprocess_prometheus() so Gauge objects
    # see the correct ValueClass (multiprocess vs in-memory).
    component_gauges = LLMBackendMetrics(
        registry=DYNAMO_COMPONENT_REGISTRY,
        model_name=config.served_model_name or "",
        component_name=config.component or "",
    )

    # If a StatLoggerFactory was provided, give it the gauges so the loggers
    # it creates can publish Prometheus metrics.
    if stat_logger is not None:
        stat_logger.component_gauges = component_gauges

Alec's avatar
Alec committed
456
457
458
459
    os.environ["VLLM_NO_USAGE_STATS"] = "1"  # Avoid internal HTTP requests
    os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

    engine_args = config.engine_args
460

461
462
463
464
465
    if engine_args.enable_lora:
        if "VLLM_ALLOW_RUNTIME_LORA_UPDATING" not in os.environ:
            os.environ["VLLM_ALLOW_RUNTIME_LORA_UPDATING"] = "True"
        if "VLLM_LORA_MODULES_LOADING_TIMEOUT" not in os.environ:
            os.environ["VLLM_LORA_MODULES_LOADING_TIMEOUT"] = "600"
466
467

    if engine_args.load_format == "gms":
468
        engine_args.worker_cls = "gpu_memory_service.integrations.vllm.worker.GMSWorker"
469

470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
        if config.gms_shadow_mode:
            from gpu_memory_service.integrations.vllm.utils import (
                configure_gms_lock_mode,
                validate_cudagraph_mode,
            )

            os.environ["DYN_GMS_SHADOW_MODE"] = "1"
            logger.info(
                "[Shadow] Enabled shadow mode: will skip KV cache allocation at startup"
            )
            # ENGINE_ID=0 writes weights, all others import (RO).
            # Prevents deadlock during TP>1 failover.
            configure_gms_lock_mode(engine_args)
            validate_cudagraph_mode(engine_args)

485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
    if engine_args.load_format in ("mx-source", "mx-target"):
        try:
            from modelexpress import register_modelexpress_loaders

            # Ensure the ModelExpress server URL env var is set for the model loader
            if config.model_express_url:
                os.environ["MODEL_EXPRESS_URL"] = config.model_express_url
            register_modelexpress_loaders()
            # Use wrapper worker to ensure loaders are registered in spawned worker processes
            engine_args.worker_cls = "modelexpress.vllm_worker.ModelExpressWorker"
        except ImportError as e:
            raise ImportError(
                f"ModelExpress package required for --load-format={engine_args.load_format}. "
                "Install with: pip install modelexpress"
            ) from e

Alec's avatar
Alec committed
501
502
503
504
505
    # Load default sampling params from `generation_config.json`
    default_sampling_params = (
        engine_args.create_model_config().get_diff_sampling_param()
    )

506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
    # Configure ec_both mode with DynamoMultimodalEmbeddingCacheConnector.
    # Must happen BEFORE engine setup so vLLM sees ec_transfer_config.
    if (
        not config.route_to_encoder
        and config.multimodal_embedding_cache_capacity_gb > 0
    ):
        from vllm.config import ECTransferConfig

        logger.info(
            "Configuring ec_both mode with DynamoMultimodalEmbeddingCacheConnector "
            "(capacity=%.2f GB)",
            config.multimodal_embedding_cache_capacity_gb,
        )
        instance_id = 0
        engine_id = f"{config.namespace}.{config.component}.backend.{instance_id}"
        engine_args.ec_transfer_config = ECTransferConfig(
            engine_id=engine_id,
            ec_role="ec_both",
            ec_connector="DynamoMultimodalEmbeddingCacheConnector",
            ec_connector_module_path="dynamo.vllm.multimodal_utils.multimodal_embedding_cache_connector",
            ec_connector_extra_config={
                "multimodal_embedding_cache_capacity_gb": config.multimodal_embedding_cache_capacity_gb,
            },
        )
        logger.info("Configured ec_both with engine_id=%s", engine_id)

Alec's avatar
Alec committed
532
533
534
535
    # Taken from build_async_engine_client_from_engine_args()
    usage_context = UsageContext.OPENAI_API_SERVER
    vllm_config = engine_args.create_engine_config(usage_context=usage_context)

536
    # Set up consolidator endpoints if KVBM (DynamoConnector) is enabled
537
    consolidator_endpoints = None
538
    if _uses_dynamo_connector(config.engine_args):
539
540
541
542
543
544
545
546
547
548
549
550
        try:
            from kvbm.vllm_integration.consolidator_config import (
                get_consolidator_endpoints,
            )

            consolidator_endpoints = get_consolidator_endpoints(vllm_config)
        except Exception as e:
            logger.warning(
                f"KVBM connector is enabled but failed to get consolidator endpoints: {e}. "
                "Continuing without KV event consolidation. "
                "Ensure 'kvbm' package is installed if this feature is needed."
            )
551
552
553
    # Store consolidator endpoints in additional_config (vLLM 0.16+ uses strict
    # dataclass fields; monkey-patching attributes onto VllmConfig is no longer safe).
    vllm_config.additional_config["consolidator_endpoints"] = consolidator_endpoints
554

555
556
557
558
    # Pass worker identity to InstrumentedScheduler via additional_config.
    if fpm_worker_id is not None:
        vllm_config.additional_config["fpm_worker_id"] = fpm_worker_id

Alec's avatar
Alec committed
559
560
561
562
    factory = []
    if stat_logger:
        factory.append(stat_logger)

563
564
    # Time engine initialization
    start_time = time.time()
Alec's avatar
Alec committed
565
566
567
568
    engine_client = AsyncLLM.from_vllm_config(
        vllm_config=vllm_config,
        usage_context=usage_context,
        stat_loggers=factory,
569
        enable_log_requests=engine_args.enable_log_requests,
Alec's avatar
Alec committed
570
571
        disable_log_stats=engine_args.disable_log_stats,
    )
572
573
574
575
    load_time = time.time() - start_time

    # Record model load time
    component_gauges.set_model_load_time(load_time)
576
577

    logger.info(f"VllmWorker for {config.served_model_name} has been initialized")
578

579
580
581
582
    # update block_size in vllm_config based on final engine cache info for later use
    runtime_values = get_engine_cache_info(engine_client)
    vllm_config.cache_config.block_size = runtime_values["block_size"]

583
584
585
586
587
588
589
    return (
        engine_client,
        vllm_config,
        default_sampling_params,
        prometheus_temp_dir,
        component_gauges,
    )
Alec's avatar
Alec committed
590
591


592
593
594
async def register_vllm_model(
    model_input: ModelInput,
    model_type: ModelType,
jh-nv's avatar
jh-nv committed
595
    generate_endpoint: Endpoint,
596
597
    config: Config,
    engine_client: AsyncLLM,
598
    vllm_config: VllmConfig,
jh-nv's avatar
jh-nv committed
599
) -> None:
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
    """
    Helper function to register a vLLM model with runtime configuration.

    Args:
        model_input: Input type for the model (e.g., ModelInput.Tokens)
        model_type: Type of model (e.g., ModelType.Chat, ModelType.Prefill)
        generate_endpoint: Endpoint to register
        config: Configuration object
        engine_client: vLLM engine client
        vllm_config: vLLM configuration
    """
    runtime_config = ModelRuntimeConfig()

    # Get runtime configuration from vLLM engine
    logging.info(
        f"Getting engine runtime configuration metadata from vLLM engine for {model_type}..."
    )
    runtime_values = get_engine_cache_info(engine_client)
618
619
620
621
622
623
624
625
626
627
628
629
630
631
    num_gpu_blocks = runtime_values["num_gpu_blocks"]
    if num_gpu_blocks is None:
        # TODO(upstream-vllm): remove this workaround once vLLM propagates
        # num_gpu_blocks from Ray DP workers back to the main-process vllm_config.
        # With Ray-based data-parallel backend, num_gpu_blocks is computed inside
        # Ray worker processes and is never written back to the main-process
        # vllm_config.  Use 0 as a sentinel so the Rust runtime can still register
        # the model; KV-cache capacity metrics will be unavailable in this mode.
        logging.warning(
            "num_gpu_blocks is None (expected when using --data-parallel-backend ray). "
            "Setting total_kv_blocks=0 for model registration."
        )
        num_gpu_blocks = 0
    runtime_config.total_kv_blocks = num_gpu_blocks
632
633
    runtime_config.max_num_seqs = runtime_values["max_num_seqs"]
    runtime_config.max_num_batched_tokens = runtime_values["max_num_batched_tokens"]
634
635
    # Decode workers don't create the WorkerKvQuery endpoint, so don't advertise local indexer
    runtime_config.enable_local_indexer = (
636
637
        config.enable_local_indexer
        and config.disaggregation_mode != DisaggregationMode.DECODE
638
    )
639
640
641

    # Add tool/reasoning parsers for decode models
    if model_type != ModelType.Prefill:
642
643
        runtime_config.tool_call_parser = config.dyn_tool_call_parser
        runtime_config.reasoning_parser = config.dyn_reasoning_parser
644
645
646
    runtime_config.exclude_tools_when_tool_choice_none = (
        config.exclude_tools_when_tool_choice_none
    )
647
648

    # Get data_parallel_size from vllm_config (defaults to 1)
649
650
651
    dp_range = get_dp_range_for_worker(vllm_config)
    runtime_config.data_parallel_start_rank = dp_range[0]
    runtime_config.data_parallel_size = dp_range[1]
652

653
654
655
656
657
658
659
660
661
662
    # Configure media decoder for frontend image decoding when enabled
    # This enables frontend to decode images and transfer via NIXL RDMA
    media_decoder = None
    media_fetcher = None
    if config.frontend_decoding:
        if not MEDIA_DECODER_AVAILABLE:
            raise RuntimeError(
                "--frontend-decoding requires MediaDecoder support. "
                "Ensure dynamo.llm module includes MediaDecoder and MediaFetcher."
            )
jh-nv's avatar
jh-nv committed
663
        assert MediaDecoder is not None and MediaFetcher is not None
664
665
666
667
668
669
        media_decoder = MediaDecoder()
        media_decoder.enable_image({"limits": {"max_alloc": 128 * 1024 * 1024}})
        # media_decoder.enable_video({})

        media_fetcher = MediaFetcher()
        media_fetcher.timeout_ms(30000)
670
        media_fetcher.allow_direct_port(True)
671

672
    await register_model(
673
674
675
676
677
        model_input,
        model_type,
        generate_endpoint,
        config.model,
        config.served_model_name,
678
        context_length=vllm_config.model_config.max_model_len,
679
        kv_cache_block_size=runtime_values["block_size"],
680
681
        runtime_config=runtime_config,
        custom_template_path=config.custom_jinja_template,
682
683
        media_decoder=media_decoder,
        media_fetcher=media_fetcher,
684
685
686
    )


jh-nv's avatar
jh-nv committed
687
def get_engine_cache_info(engine: AsyncLLM) -> dict[str, Any]:
688
689
690
691
692
693
    """Retrieve cache configuration information from [`AsyncLLM`] engine."""

    try:
        # Get values directly from vllm_config instead of collective_rpc
        cache_values = {
            "num_gpu_blocks": engine.vllm_config.cache_config.num_gpu_blocks,
694
            "block_size": engine.vllm_config.cache_config.block_size,
695
696
697
698
699
700
701
702
703
704
705
        }

        scheduler_values = {
            "max_num_seqs": engine.vllm_config.scheduler_config.max_num_seqs,
            "max_num_batched_tokens": engine.vllm_config.scheduler_config.max_num_batched_tokens,
        }

        logging.info(f"Cache config values: {cache_values}")
        logging.info(f"Scheduler config values: {scheduler_values}")
        return {
            "num_gpu_blocks": cache_values["num_gpu_blocks"],
706
            "block_size": cache_values["block_size"],
707
708
709
710
711
712
713
714
            "max_num_seqs": scheduler_values["max_num_seqs"],
            "max_num_batched_tokens": scheduler_values["max_num_batched_tokens"],
        }
    except Exception as e:
        logging.error(f"Failed to get configuration values from vLLM config: {e}")
        raise


jh-nv's avatar
jh-nv committed
715
def main() -> None:
Alec's avatar
Alec committed
716
717
718
    uvloop.run(worker())


Alec's avatar
Alec committed
719
if __name__ == "__main__":
Alec's avatar
Alec committed
720
    main()