Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 78 additions & 45 deletions tensorrt_llm/commands/serve.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,20 +138,25 @@ def _report_observed_child_failure(return_code: int, component: str,
def _apply_effective_telemetry_config(llm_args: dict,
*,
telemetry: bool,
component: str = "server") -> None:
"""Apply a parsed config opt-out before later setup can fail."""
component: str = "server") -> dict:
"""Apply a parsed config opt-out before later setup can fail.

Returns the mapping to use; never modifies ``llm_args``. The caller
rebinds its name to the result.
"""
telemetry_config = llm_args.get("telemetry_config")
if telemetry_config is not None:
if not telemetry:
telemetry_config = telemetry_config.model_copy(
update={"disabled": True})
llm_args["telemetry_config"] = telemetry_config
llm_args = {**llm_args, "telemetry_config": telemetry_config}
apply_usage_session_config(
telemetry_config,
default_usage_context="cli_serve",
component=component,
lifecycle_phase="config_validation",
)
return llm_args


def _pop_bool_config_option(config: dict[str, Any], key: str) -> bool:
Expand Down Expand Up @@ -518,7 +523,7 @@ def _reserve_multi_frontend_port(host: str, port: int,

@contextlib.contextmanager
def _provision_kv_cache_pool(llm_args: dict,
owns_engine: bool = True) -> Iterator[None]:
owns_engine: bool = True) -> Iterator[dict]:
"""Bring up the shared cache this server joins, for its lifetime.

A connector backed by a cluster-wide pool needs that pool reachable before
Expand All @@ -529,9 +534,12 @@ def _provision_kv_cache_pool(llm_args: dict,
Only the process that owns the engine does this. An attached frontend
re-execs this command line but shares the launcher's executor, so it would
otherwise render a second client config over the first.

Yields the mapping to pass to the LLM constructor; never modifies
``llm_args``. The caller binds the yielded mapping (``as llm_args``).
"""
if not owns_engine:
yield
yield llm_args
return

from tensorrt_llm._torch.pyexecutor.connectors.mooncake_store import \
Expand All @@ -543,10 +551,10 @@ def _provision_kv_cache_pool(llm_args: dict,
# before the LLM constructor would coerce it. The validated model is
# handed on so it is not parsed twice.
connector_config = KvCacheConnectorConfig(**connector_config)
llm_args["kv_connector_config"] = connector_config
llm_args = {**llm_args, "kv_connector_config": connector_config}

with maybe_provision_pool(connector_config):
yield
yield llm_args


def _bind_launcher_uds(llm) -> Tuple[socket.socket, str]:
Expand Down Expand Up @@ -589,15 +597,18 @@ def _share_prometheus_multiproc_dir(llm, ipc_dir: str) -> None:
def _resolve_default_num_serve_frontends(llm_args: dict, *, requested: bool,
port: int, grpc: bool,
report_addr: Optional[str],
enable_sleep_mode: bool) -> None:
enable_sleep_mode: bool) -> dict:
"""Fall back to one frontend where the default count cannot run.

An explicit count (CLI, --config YAML or --set) is left alone, so the
guard that owns the conflict still rejects it.

Returns the mapping to use; never modifies ``llm_args``. The caller
rebinds its name to the result.
"""
num_frontends = llm_args.get("num_serve_frontends", 1)
if requested or num_frontends <= 1:
return
return llm_args
if grpc:
# OpenEngine supports several frontends, but only when asked to.
reason = "--grpc"
Expand All @@ -610,12 +621,12 @@ def _resolve_default_num_serve_frontends(llm_args: dict, *, requested: bool,
elif enable_sleep_mode:
reason = "--enable_sleep_mode"
else:
return
return llm_args
logger.info(
f"num_serve_frontends defaults to {num_frontends} for the OpenAI HTTP "
f"server only; running 1 frontend because of {reason}. Pass "
"--num_serve_frontends explicitly to override.")
llm_args["num_serve_frontends"] = 1
return {**llm_args, "num_serve_frontends": 1}


def launch_server(
Expand All @@ -639,6 +650,9 @@ def launch_server(
enable_runtime_control_endpoints: bool = False,
runtime_control_api_key: Optional[str] = None):

# Never write into the caller's mapping; every edit below is on this copy.
llm_args = dict(llm_args)

# Publish an explicit serving role before constructing/spawning the model.
# Remote-tail context workers use it to omit generation-only weights.
if server_role is not None:
Expand All @@ -649,6 +663,10 @@ def launch_server(
model = served_model_name or llm_args["model"]

multi_frontend = _init_multi_frontend_mode(llm_args, multi_frontend_enabled)
if not multi_frontend_enabled:
# Entry points that disable multi-frontend mode must not hand the knob
# to the LLM constructor, which rejects unknown keys.
llm_args.pop("num_serve_frontends", None)
# An attached frontend gets the launcher's socket from its spawn env; the
# launcher is its parent.
multi_frontend_serving = None
Expand Down Expand Up @@ -732,8 +750,8 @@ def launch_server(
_diagnose_port_in_use) as
(affinity_socket, attached_affinity_sockets),
_provision_kv_cache_pool(
llm_args,
owns_engine=not multi_frontend.is_attached_frontend)):
llm_args, owns_engine=not multi_frontend.is_attached_frontend)
as llm_args):
if backend == 'pytorch':
llm_args.pop("build_config", None)
llm = PyTorchLLM(**llm_args)
Expand Down Expand Up @@ -812,6 +830,8 @@ def launch_mm_encoder_server(
metadata_server_cfg: Optional[MetadataServerConfig] = None,
allow_request_chat_template: bool = False,
):
# Never write into the caller's mapping; every edit below is on this copy.
encoder_args = dict(encoder_args)
model = encoder_args["model"]
encoder_args.pop("build_config", None)
mm_encoder = MultimodalEncoder(**encoder_args)
Expand Down Expand Up @@ -908,6 +928,8 @@ def launch_embedding_server(
max_queue_size: int,
metadata_server_cfg: Optional[MetadataServerConfig] = None,
):
# Never write into the caller's mapping; every edit below is on this copy.
llm_args = dict(llm_args)
model = llm_args["model"]
llm_args.pop("build_config", None)
# encode_only is forced on below; drop any value coming from --config to avoid
Expand Down Expand Up @@ -1660,9 +1682,9 @@ def _serve_llm():
# even when an explicit convenience flag targets the same nested field.
llm_args = _apply_config_overrides(llm_args, parsed_config_overrides)
if enable_sleep_mode and llm_args.get("sleep_config") is None:
llm_args["sleep_config"] = SleepConfig()
llm_args = {**llm_args, "sleep_config": SleepConfig()}

_resolve_default_num_serve_frontends(
llm_args = _resolve_default_num_serve_frontends(
llm_args,
requested=("num_serve_frontends" in explicit_cli_keys
or "num_serve_frontends" in llm_args_extra_dict),
Expand All @@ -1671,7 +1693,8 @@ def _serve_llm():
report_addr=report_addr,
enable_sleep_mode=enable_sleep_mode)

_apply_effective_telemetry_config(llm_args, telemetry=telemetry)
llm_args = _apply_effective_telemetry_config(llm_args,
telemetry=telemetry)
# Preserve requested settings for failures before entering the LLM constructor.
try:
from tensorrt_llm.usage.usage_lib import _capture_startup_context
Expand All @@ -1684,7 +1707,8 @@ def _serve_llm():

# Specify disagg_cluster_config in config file or through command line "--disagg_cluster_uri",
# but disagg_cluster_uri takes precedence over cluster uri in config file
disagg_cluster_config = llm_args.pop("disagg_cluster", None)
disagg_cluster_config = llm_args.get("disagg_cluster")
llm_args = {k: v for k, v in llm_args.items() if k != "disagg_cluster"}
if disagg_cluster_config:
disagg_cluster_config = extract_disagg_cluster_config(
disagg_cluster_config, disagg_cluster_uri)
Expand Down Expand Up @@ -1758,9 +1782,9 @@ def _serve_llm():

from tensorrt_llm.grpc.smg.server import launch_smg_server

launch_smg_server(host,
port,
llm_args,
launch_smg_server(host=host,
port=port,
llm_args=llm_args,
served_model_name=served_model_name)
else:
try:
Expand All @@ -1778,11 +1802,11 @@ def _serve_llm():
with _provision_kv_cache_pool(
llm_args,
owns_engine=not grpc_multi_frontend.is_attached_frontend
):
) as llm_args:
launch_grpc_server(
host,
port,
llm_args,
host=host,
port=port,
llm_args=llm_args,
served_model_name=served_model_name,
report_failure=_report_observed_child_failure,
enable_load_metrics=openengine_enable_load_metrics,
Expand All @@ -1793,16 +1817,16 @@ def _serve_llm():
else:
# Default: launch OpenAI HTTP server
launch_server(
host,
port,
llm_args,
tool_parser,
middleware,
chat_template,
metadata_server_cfg,
server_role,
disagg_cluster_config,
multimodal_server_config,
host=host,
port=port,
llm_args=llm_args,
tool_parser=tool_parser,
middleware=middleware,
chat_template=chat_template,
metadata_server_cfg=metadata_server_cfg,
server_role=server_role,
disagg_cluster_config=disagg_cluster_config,
multimodal_server_config=multimodal_server_config,
served_model_name=served_model_name,
allow_request_chat_template=allow_request_chat_template,
num_input_processor_workers=num_input_processor_workers,
Expand All @@ -1821,8 +1845,12 @@ def _serve_visual_gen():
metadata_server_cfg = parse_metadata_server_config_file(
metadata_server_config_file)

launch_visual_gen_server(host, port, model, parsed_visual_gen_args,
metadata_server_cfg, middleware)
launch_visual_gen_server(host=host,
port=port,
model=model,
visual_gen_args=parsed_visual_gen_args,
metadata_server_cfg=metadata_server_cfg,
middleware=middleware)

# Only the OpenAI HTTP path publishes the bound address. Fail loudly rather
# than leaving a launcher waiting forever on a file nobody writes.
Expand Down Expand Up @@ -1990,16 +2018,17 @@ def serve_encoder(
encoder_args = update_llm_args_with_extra_dict(
llm_args, encoder_args_extra_dict, explicit_cli_keys=explicit_cli_keys)

_apply_effective_telemetry_config(encoder_args, telemetry=telemetry)
encoder_args = _apply_effective_telemetry_config(encoder_args,
telemetry=telemetry)

metadata_server_cfg = parse_metadata_server_config_file(
metadata_server_config_file)

launch_mm_encoder_server(
host,
port,
encoder_args,
metadata_server_cfg,
host=host,
port=port,
encoder_args=encoder_args,
metadata_server_cfg=metadata_server_cfg,
allow_request_chat_template=allow_request_chat_template)


Expand Down Expand Up @@ -2119,7 +2148,7 @@ def serve_embedding(
llm_args = update_llm_args_with_extra_dict(
llm_args, extra_dict, explicit_cli_keys=explicit_cli_keys)

_apply_effective_telemetry_config(llm_args, telemetry=telemetry)
llm_args = _apply_effective_telemetry_config(llm_args, telemetry=telemetry)

# The CLI does not expose TP/PP/CP, but a --config YAML could still set them. Reject
# that explicitly rather than hang: the in-process encode-only path cannot shard.
Expand All @@ -2138,8 +2167,12 @@ def serve_embedding(
metadata_server_cfg = parse_metadata_server_config_file(
metadata_server_config_file)

launch_embedding_server(host, port, llm_args, max_queue_delay,
max_queue_size, metadata_server_cfg)
launch_embedding_server(host=host,
port=port,
llm_args=llm_args,
max_queue_delay=max_queue_delay,
max_queue_size=max_queue_size,
metadata_server_cfg=metadata_server_cfg)


@click.command("disaggregated")
Expand Down Expand Up @@ -2752,7 +2785,7 @@ def disaggregated_mpi_worker(config_file: Optional[str], log_level: str,
llm_args = _build_llm_args_from_disagg_server_cfg(server_cfg.other_args)

# Ignore the non-LLM args
llm_args.pop("router", None)
llm_args = {k: v for k, v in llm_args.items() if k != "router"}
_launch_disaggregated_server(config_file, llm_args)
return

Expand Down
3 changes: 3 additions & 0 deletions tensorrt_llm/grpc/openengine/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -342,6 +342,9 @@ def launch_server(
metrics_host: HTTP bind address; defaults to the inference host.
metrics_port: HTTP port when performance-metrics configuration enables Prometheus.
"""
# Never write into the caller's mapping; the nested serve() below edits
# this copy.
llm_args = dict(llm_args)
_warn_if_python_protobuf()

async def serve() -> None:
Expand Down
5 changes: 4 additions & 1 deletion tensorrt_llm/grpc/smg/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,9 @@ def launch_smg_server(
served_model_name: Model name returned by discovery RPCs. Defaults to
the model path.
"""
# Never write into the caller's mapping; the nested serve_grpc_async()
# below edits this copy (or the mapping the pool provisioning yields).
llm_args = dict(llm_args)
_warn_if_python_protobuf()
try:
from grpc_reflection.v1alpha import reflection
Expand Down Expand Up @@ -133,7 +136,7 @@ def signal_handler() -> None:
# server.
from tensorrt_llm.commands.serve import _provision_kv_cache_pool

with _provision_kv_cache_pool(llm_args):
with _provision_kv_cache_pool(llm_args) as llm_args:
uvloop.run(serve_grpc_async())


Expand Down
3 changes: 3 additions & 0 deletions tensorrt_llm/llmapi/disagg_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -523,6 +523,9 @@ def extract_disagg_cluster_config(
Use the default value of DisaggClusterConfig and MinimalInstances if the corresponding fields are not provided.
If cluster_uri is provided, it will override the cluster_uri in the cluster_config_dict.
"""
# Never write into the caller's mapping (it is nested in the user's YAML);
# replace minimal_instances on this copy only.
cluster_config_dict = dict(cluster_config_dict)

def update_dataclass(obj, data_dict: Dict[str, Any]):
for key, value in data_dict.items():
Expand Down
4 changes: 3 additions & 1 deletion tensorrt_llm/llmapi/llm_args.py
Original file line number Diff line number Diff line change
Expand Up @@ -8239,7 +8239,9 @@ def update_llm_args_with_extra_dict(
base_tc = base_tc.model_dump(exclude_unset=True)
if isinstance(yaml_tc, TelemetryConfig):
yaml_tc = yaml_tc.model_dump(exclude_unset=True)
yaml_tc.pop('usage_context', None)
# Drop on a copy: the dict branch aliases the caller's nested
# telemetry_config mapping.
yaml_tc = {k: v for k, v in yaml_tc.items() if k != 'usage_context'}
merged = base_tc | yaml_tc
if "telemetry" in explicit_cli_keys and 'disabled' in base_tc:
merged['disabled'] = base_tc['disabled']
Expand Down
5 changes: 4 additions & 1 deletion tensorrt_llm/serve/_frontend_processes.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,12 @@ def _init_multi_frontend_mode(llm_args: dict, enabled: bool) -> MultiFrontendMod
executor: the launcher (frontend 0) owns the engine and spawns K-1
attached frontends (classic IPC executor path only). enabled=False
entry points (e.g. disaggregated MPI workers) never honor the knob.

Never modifies ``llm_args``; a caller that disables multi-frontend mode
drops ``num_serve_frontends`` itself.
"""
if not enabled:
if llm_args.pop("num_serve_frontends", 1) > 1:
if llm_args.get("num_serve_frontends", 1) > 1:
logger.warning(
"num_serve_frontends is only supported on plain "
"trtllm-serve; ignored on this entry point."
Expand Down
7 changes: 5 additions & 2 deletions tensorrt_llm/serve/router.py
Original file line number Diff line number Diff line change
Expand Up @@ -2243,8 +2243,11 @@ def create_router(
if router_class is None:
raise ValueError(f"Unsupported router type: {router_type}. "
f"Supported types are: {list(router_map.keys())}")
extra_args = router_config.args if router_config else {}
extra_args["disagg_node_id"] = disagg_node_id
extra_args = {
**(router_config.args if router_config else {}),
"disagg_node_id":
disagg_node_id,
}

return router_class(router_config.server_role if router_config else None,
servers,
Expand Down
1 change: 1 addition & 0 deletions tests/integration/test_lists/test-db/l0_cpu.yml
Original file line number Diff line number Diff line change
Expand Up @@ -316,6 +316,7 @@ l0_cpu:
- unittest/llmapi/test_runtime_memory_api.py
- unittest/llmapi/test_sampling_params.py
- unittest/llmapi/test_serialization.py
- unittest/llmapi/test_serve_cli_no_mutation.py
- unittest/llmapi/test_serve_frontend_affinity.py
- unittest/llmapi/test_serve_multi_frontend_helpers.py
- unittest/llmapi/test_serve_multi_frontend_route_gate.py
Expand Down
Loading
Loading