main.py 29.6 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
    # 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,
149
150
151
152
        ) = snapshot_controller.reload_restore_identity(
            config.namespace,
            config.discovery_backend,
        )
153

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

160
    shutdown_event = asyncio.Event()
161
    runtime, loop = create_runtime(
162
        discovery_backend=config.discovery_backend,
163
164
        request_plane=config.request_plane,
        event_plane=config.event_plane,
165
166
    )

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

171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
    # 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,
    )
186
187

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


jh-nv's avatar
jh-nv committed
190
def setup_metrics_collection(
191
    config: "Config | OmniConfig", generate_endpoint: Endpoint, logger: logging.Logger
jh-nv's avatar
jh-nv committed
192
) -> None:
193
194
195
196
197
198
199
200
201
202
203
204
205
    """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.
206
207
208
209
210

    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.
211
212
    """
    if config.engine_args.disable_log_stats is False:
213
214
215
216
217
218
219
220
221
222
223
        # 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,
        )

224
225
226
227
228
229
230
231
232
        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):
233
234
235
236
237
238
239
240
            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,
241
242
243
244
245
                    metric_prefix_filters=["vllm:", "lmcache:"],
                    namespace_name=config.namespace,
                    component_name=config.component,
                    endpoint_name=config.endpoint,
                    model_name=config.model,
246
247
248
249
250
251
252
253
254
255
256
                )
            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
257
                # Global REGISTRY has in-memory metrics (vllm)
258
259
260
                register_engine_metrics_callback(
                    endpoint=generate_endpoint,
                    registry=REGISTRY,
261
                    metric_prefix_filters=["vllm:"],
262
263
264
265
                    namespace_name=config.namespace,
                    component_name=config.component,
                    endpoint_name=config.endpoint,
                    model_name=config.model,
266
267
268
269
270
                )
                # Multiproc registry has .db file metrics (lmcache, possibly vllm duplicates)
                register_engine_metrics_callback(
                    endpoint=generate_endpoint,
                    registry=multiproc_registry,
271
272
273
274
275
                    metric_prefix_filters=["vllm:", "lmcache:"],
                    namespace_name=config.namespace,
                    component_name=config.component,
                    endpoint_name=config.endpoint,
                    model_name=config.model,
276
277
                )
        else:
278
279
280
281
282
            if multiproc_dir:
                logger.warning(
                    f"PROMETHEUS_MULTIPROC_DIR={multiproc_dir} is not a valid directory, "
                    "falling back to single-process metrics"
                )
283
284
285
286
287
            # No multiprocess mode
            register_engine_metrics_callback(
                endpoint=generate_endpoint,
                registry=REGISTRY,
                metric_prefix_filters=["vllm:", "lmcache:"],
288
289
290
291
                namespace_name=config.namespace,
                component_name=config.component,
                endpoint_name=config.endpoint,
                model_name=config.model,
292
293
294
            )


Yan Ru Pei's avatar
Yan Ru Pei committed
295
296
def setup_kv_event_publisher(
    config: Config,
297
298
    generate_endpoint: Endpoint,
    vllm_config: VllmConfig,
299
300
    consolidator_enabled: bool = False,
    consolidator_port: Optional[int] = 5558,
jh-nv's avatar
jh-nv committed
301
) -> Optional[list[KvEventPublisher]]:
Yan Ru Pei's avatar
Yan Ru Pei committed
302
    """
jh-nv's avatar
jh-nv committed
303
    list[KvEventPublisher] | None
Yan Ru Pei's avatar
Yan Ru Pei committed
304
305
    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.
306
307
308
309
310
311
312
    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
313
    Returns:
314
        List of KvEventPublisher instances (one per dp_rank) if prefix caching is enabled, None otherwise.
Yan Ru Pei's avatar
Yan Ru Pei committed
315
316
317
318
    """
    if not config.engine_args.enable_prefix_caching:
        return None

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

324
325
326
    if config.engine_args.kv_events_config is None:
        return None

327
328
329
330
331
332
333
    # 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

334
335
336
    # 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
337
338
    kv_publishers = []

339
    for dp_rank in range(dp_start, dp_start + dp_size):
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
        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
355

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

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

Yan Ru Pei's avatar
Yan Ru Pei committed
370
    return kv_publishers if kv_publishers else None
Yan Ru Pei's avatar
Yan Ru Pei committed
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
416
417
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
418
def setup_vllm_engine(
419
420
421
    config: Config,
    stat_logger: Optional[StatLoggerFactory] = None,
    fpm_worker_id: Optional[str] = None,
jh-nv's avatar
jh-nv committed
422
) -> tuple[AsyncLLM, VllmConfig, Any, Any, LLMBackendMetrics]:
423
424
425
    # 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.
426
    prometheus_temp_dir = None
427
428
429
430
431
432
    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)
433
434
435
436
437
438
439
    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']}"
        )

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

445
446
447
448
449
450
451
452
453
454
455
456
457
    # 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
458
459
460
461
    os.environ["VLLM_NO_USAGE_STATS"] = "1"  # Avoid internal HTTP requests
    os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

    engine_args = config.engine_args
462

463
464
465
466
467
    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"
468
469

    if engine_args.load_format == "gms":
470
        engine_args.worker_cls = "gpu_memory_service.integrations.vllm.worker.GMSWorker"
471

472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
        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)

487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
    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
503
504
505
506
507
    # Load default sampling params from `generation_config.json`
    default_sampling_params = (
        engine_args.create_model_config().get_diff_sampling_param()
    )

508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
    # 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
534
535
536
537
    # 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)

538
    # Set up consolidator endpoints if KVBM (DynamoConnector) is enabled
539
    consolidator_endpoints = None
540
    if _uses_dynamo_connector(config.engine_args):
541
542
543
544
545
546
547
548
549
550
551
552
        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."
            )
553
554
555
    # 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
556

557
558
559
560
    # 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

561
562
563
564
565
566
567
568
569
    # Pass benchmark config to InstrumentedScheduler via additional_config.
    if hasattr(config, "_benchmark_additional_config"):
        bench = config._benchmark_additional_config
        if fpm_worker_id and bench["output_path"] == "/tmp/benchmark_results.json":
            short_id = fpm_worker_id[-8:]
            bench["output_path"] = f"/tmp/benchmark_results_{short_id}.json"
        vllm_config.additional_config["benchmark"] = bench
        logger.info("Benchmark config injected into additional_config")

Alec's avatar
Alec committed
570
571
572
573
    factory = []
    if stat_logger:
        factory.append(stat_logger)

574
575
    # Time engine initialization
    start_time = time.time()
Alec's avatar
Alec committed
576
577
578
579
    engine_client = AsyncLLM.from_vllm_config(
        vllm_config=vllm_config,
        usage_context=usage_context,
        stat_loggers=factory,
580
        enable_log_requests=engine_args.enable_log_requests,
Alec's avatar
Alec committed
581
582
        disable_log_stats=engine_args.disable_log_stats,
    )
583
584
585
586
    load_time = time.time() - start_time

    # Record model load time
    component_gauges.set_model_load_time(load_time)
587
588

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

590
591
592
593
    # 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"]

594
595
596
597
598
599
600
    return (
        engine_client,
        vllm_config,
        default_sampling_params,
        prometheus_temp_dir,
        component_gauges,
    )
Alec's avatar
Alec committed
601
602


603
604
605
async def register_vllm_model(
    model_input: ModelInput,
    model_type: ModelType,
jh-nv's avatar
jh-nv committed
606
    generate_endpoint: Endpoint,
607
608
    config: Config,
    engine_client: AsyncLLM,
609
    vllm_config: VllmConfig,
jh-nv's avatar
jh-nv committed
610
) -> None:
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
    """
    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)
629
630
631
632
633
634
635
636
637
638
639
640
641
642
    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
643
644
    runtime_config.max_num_seqs = runtime_values["max_num_seqs"]
    runtime_config.max_num_batched_tokens = runtime_values["max_num_batched_tokens"]
645
646
    # Decode workers don't create the WorkerKvQuery endpoint, so don't advertise local indexer
    runtime_config.enable_local_indexer = (
647
648
        config.enable_local_indexer
        and config.disaggregation_mode != DisaggregationMode.DECODE
649
    )
650
651
652

    # Add tool/reasoning parsers for decode models
    if model_type != ModelType.Prefill:
653
654
        runtime_config.tool_call_parser = config.dyn_tool_call_parser
        runtime_config.reasoning_parser = config.dyn_reasoning_parser
655
656
657
    runtime_config.exclude_tools_when_tool_choice_none = (
        config.exclude_tools_when_tool_choice_none
    )
658

659
660
661
662
663
664
665
    # Propagate stream_interval so the frontend can respect --stream-interval.
    # set_engine_specific requires a JSON-encoded string (the Rust binding
    # parses it with serde_json::from_str); str(int) happens to be valid JSON.
    stream_interval = getattr(config.engine_args, "stream_interval", None)
    if stream_interval is not None:
        runtime_config.set_engine_specific("stream_interval", str(stream_interval))

666
    # Get data_parallel_size from vllm_config (defaults to 1)
667
668
669
    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]
670

671
672
673
674
675
676
677
678
679
680
    # 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
681
        assert MediaDecoder is not None and MediaFetcher is not None
682
683
684
685
686
687
        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)
688
689
690
        allow_internal = os.getenv("DYN_MM_ALLOW_INTERNAL", "0") == "1"
        media_fetcher.allow_direct_ip(allow_internal)
        media_fetcher.allow_direct_port(allow_internal)
691

692
    await register_model(
693
694
695
696
697
        model_input,
        model_type,
        generate_endpoint,
        config.model,
        config.served_model_name,
698
        context_length=vllm_config.model_config.max_model_len,
699
        kv_cache_block_size=runtime_values["block_size"],
700
701
        runtime_config=runtime_config,
        custom_template_path=config.custom_jinja_template,
702
703
        media_decoder=media_decoder,
        media_fetcher=media_fetcher,
704
705
706
    )


jh-nv's avatar
jh-nv committed
707
def get_engine_cache_info(engine: AsyncLLM) -> dict[str, Any]:
708
709
710
711
712
713
    """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,
714
            "block_size": engine.vllm_config.cache_config.block_size,
715
716
717
718
719
720
721
722
723
724
725
        }

        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"],
726
            "block_size": cache_values["block_size"],
727
728
729
730
731
732
733
734
            "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
735
def main() -> None:
Alec's avatar
Alec committed
736
737
738
    uvloop.run(worker())


Alec's avatar
Alec committed
739
if __name__ == "__main__":
Alec's avatar
Alec committed
740
    main()