[None][feat] serve: multi-process HTTP frontends on the classic IPC executor path

tensorrt_llm/commands/serve.py

@@ -9,6 +9,7 @@
 import socket
 import subprocess  # nosec B404
 import sys
+import tempfile
 import uuid
 from pathlib import Path
 from typing import Any, Dict, Literal, Mapping, Optional, Sequence
@@ -294,6 +295,64 @@ def get_llm_args(
     return llm_args, llm_args_extra_dict
 
 
+def _spawn_attached_frontends(llm, num_frontends: int) -> list:
+    """Spawn num_frontends - 1 attached serving frontend processes.
+
+    Multi-frontend serving (TLLM_SERVE_NUM_FRONTENDS > 1): each child
+    re-execs this trtllm-serve command line with env vars pointing at the
+    launcher executor's attach endpoints — the multi-frontend request
+    ingress plus per-frontend result lanes (see
+    GenerationExecutorProxy._setup_queues); the child's executor attaches to
+    the already-running worker instead of launching a new one (see
+    executor.py GenerationExecutor.create). All frontends bind the serving
+    port with SO_REUSEPORT, so the kernel load-balances accepted
+    connections across their independent processes (and GILs).
+    """
+    from tensorrt_llm.executor.proxy import GenerationExecutorProxy
+
+    executor = getattr(llm, "_executor", None)
+    if not isinstance(executor, GenerationExecutorProxy) or (
+            attach_info := executor.multi_frontend_attach_info()) is None:
+        raise ValueError(
+            "TLLM_SERVE_NUM_FRONTENDS > 1 requires the classic IPC executor "
+            f"proxy in multi-frontend mode, got {type(executor).__name__}")
+    # mkstemp creates the file 0600: it carries the executor HMAC keys.
+    fd, attach_info_path = tempfile.mkstemp(prefix="trtllm_frontend_",
+                                            suffix=".json")
+    with os.fdopen(fd, "w") as f:
+        json.dump(attach_info, f)
+
+    children = []
+    for frontend_id in range(1, num_frontends):
+        env = os.environ.copy()
+        env["TLLM_EXECUTOR_ATTACH_INFO"] = attach_info_path
+        env["TLLM_EXECUTOR_FRONTEND_ID"] = str(frontend_id)
+        # Attached frontends are plain RPC clients: keep them out of the
+        # launcher's MPI job (an inherited PMI rank identity would make
+        # mpi4py try to (re-)join it at import time).
+        env["TLLM_DISABLE_MPI"] = "1"
+        for key in list(env):
+            if key.startswith(("OMPI_", "PMIX_", "PMI_")):
+                del env[key]
+        child = subprocess.Popen([sys.executable] + sys.argv,
+                                 env=env)  # nosec B603
+        children.append(child)
+        logger.info(
+            f"Launched attached serving frontend {frontend_id} (pid {child.pid})"
+        )
+    return children
+
+
+def _terminate_attached_frontends(children: list) -> None:
+    for child in children:
+        child.terminate()
+    for child in children:
+        try:
+            child.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            child.kill()
+
+
 def launch_server(
         host: str,
         port: int,
@@ -309,13 +368,44 @@ def launch_server(
 
     backend = llm_args["backend"]
     model = served_model_name or llm_args["model"]
+
+    # Multi-frontend serving (prototype): TLLM_SERVE_NUM_FRONTENDS=K runs K
+    # HTTP frontend processes against ONE executor. The launcher (this
+    # process, frontend 0) launches the worker as usual and spawns K-1
+    # attached frontends; attached frontends (TLLM_EXECUTOR_ATTACH_INFO set)
+    # skip the spawning. Supported only on the classic IPC executor path
+    # (the default orchestrator).
+    num_frontends = int(os.getenv("TLLM_SERVE_NUM_FRONTENDS", "1") or "1")
+    if not 0 < num_frontends <= (1 << 16):
+        raise ValueError(
+            f"TLLM_SERVE_NUM_FRONTENDS out of range: {num_frontends}")
+    is_attached_frontend = os.getenv("TLLM_EXECUTOR_ATTACH_INFO") is not None
+    multi_frontend = num_frontends > 1 or is_attached_frontend
+    if multi_frontend and not is_attached_frontend:
+        if llm_args.get("orchestrator_type") is not None:
+            raise ValueError(
+                "TLLM_SERVE_NUM_FRONTENDS > 1 currently supports only the "
+                "default (classic IPC) executor path, not orchestrator_type="
+                f"{llm_args.get('orchestrator_type')!r}")
+        # Opt the launcher executor into multi-frontend mode BEFORE it is
+        # created: pre-generate the shared ipc directory and HMAC key so the
+        # launcher proxy, the rank0 worker and the attached frontends agree
+        # on deterministic endpoints (GenerationExecutorProxy._setup_queues).
+        os.environ["TLLM_MULTI_FRONTEND_IPC_DIR"] = tempfile.mkdtemp(
+            prefix="trtllm_frontends_")
+        os.environ["TLLM_MULTI_FRONTEND_HMAC"] = os.urandom(32).hex()
+
     addr_info = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
                                    socket.SOCK_STREAM)
     address_family = socket.AF_INET6 if all(
         [info[0] == socket.AF_INET6 for info in addr_info]) else socket.AF_INET
     with socket.socket(address_family, socket.SOCK_STREAM) as s:
         # If disagg cluster config is provided and port is not specified, try to find a free port, otherwise try to bind to the specified port
         assert port > 0 or disagg_cluster_config is not None, "Port must be specified if disagg cluster config is not provided"
+        if multi_frontend:
+            # Every frontend process binds its own listening socket on the
+            # same port; the kernel load-balances accepts across them.
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
         try:
             s.bind((host, port))
             if port == 0:
@@ -340,6 +430,10 @@ def launch_server(
                 f"{backend} is not a known backend, check help for available options.",
                 param_hint="backend")
 
+        frontend_children = []
+        if multi_frontend and not is_attached_frontend:
+            frontend_children = _spawn_attached_frontends(llm, num_frontends)
+
         server = OpenAIServer(generator=llm,
                               model=model,
                               tool_parser=tool_parser,
@@ -354,7 +448,11 @@ def launch_server(
         if os.getenv("TRTLLM_SERVER_DISABLE_GC", "0") == "1":
             gc.disable()
 
-        asyncio.run(server(host, port, sockets=[s]))
+        try:
+            asyncio.run(server(host, port, sockets=[s]))
+        finally:
+            if frontend_children:
+                _terminate_attached_frontends(frontend_children)
 
 
 def launch_grpc_server(host: str,

tensorrt_llm/executor/base_worker.py

@@ -36,6 +36,7 @@
 from .result import (GenerationResult, LogProbsResult, ResponseWrapper,
                      compute_logprobs)
 from .utils import (ErrorResponse, IntraProcessQueue, RequestError,
+                    bucket_responses_by_frontend, get_frontend_id,
                     is_llm_response)
 
 if TYPE_CHECKING:
@@ -105,6 +106,9 @@ def __init__(
         self.engine = None
         self.result_queue: Optional[IpcQueue] = None
         self.postproc_queues: Optional[List[IpcQueue]] = None
+        # Multi-frontend serving: one result lane per frontend process,
+        # selected by the frontend id in client_id's top bits.
+        self.frontend_result_queues: Optional[List[IpcQueue]] = None
         self.rank = mpi_rank()
         self.global_rank = global_mpi_rank()
         # mapping: client_id -> GenerationResult
@@ -329,13 +333,24 @@ def fetch_kv_cache_events(self) -> list:
     def set_result_queue(self, queue):
         """In multi-gpu mode, result_queue will be set here to communicate between the proxy and the worker 0 process."""
         assert self.postproc_queues is None
+        assert self.frontend_result_queues is None
         self.result_queue = queue
 
     def set_postproc_queues(self, queues: List["IpcQueue"]):
         """ Set the IPC queues for feeding post-processing processes. """
         assert self.result_queue is None
+        assert self.frontend_result_queues is None
         self.postproc_queues = queues
 
+    def set_frontend_result_queues(self, queues: List["IpcQueue"]):
+        """Multi-frontend serving: one result lane per frontend process.
+
+        The lane is selected by the frontend id in client_id's top bits.
+        """
+        assert self.result_queue is None
+        assert self.postproc_queues is None
+        self.frontend_result_queues = queues
+
     def _set_iteration_result_queue(self, it_result_queue: IterationResultQueue,
                                     queue: Union[Queue, FusedIpcQueue,
                                                  IntraProcessQueue]):
@@ -769,20 +784,20 @@ def responses_handler(self, responses: List[tllm.Response]):
         HandlerKind = AwaitResponseHelper.HandlerKind
 
         if self.handler_kind is HandlerKind.unknown:
-            if not (self.worker.result_queue is not None
-                    or self.worker.postproc_queues is not None):
+            has_ipc_queues = (self.worker.result_queue is not None
+                              or self.worker.postproc_queues is not None
+                              or self.worker.frontend_result_queues is not None)
+            if not has_ipc_queues:
                 logger_debug(f"creating await_response helper for Worker\n",
                              color="yellow")
                 # When ExecutorBindingWorker is used in the main process
                 # aka the single process mode
                 self.handler_kind = HandlerKind.single_process_worker
-            elif self.worker.result_queue is not None or self.worker.postproc_queues is not None:
+            else:
                 # The ExecutorBindingProxy is used
                 logger_debug(f"creating await_response helper for IPC\n",
                              color="yellow")
                 self.handler_kind = HandlerKind.ipc_batched
-            else:
-                raise NotImplementedError
 
         match self.handler_kind:
             case HandlerKind.single_process_worker:
@@ -877,7 +892,15 @@ def handle_for_ipc_batched(self, responses: List[tllm.Response]) -> None:
                     self.worker.postproc_queues[wid].put(batch)
 
         if rsp_batch:
-            self.worker.result_queue.put(rsp_batch)
+            if (lanes := self.worker.frontend_result_queues) is not None:
+                # Multi-frontend serving: route each response to its origin
+                # frontend's lane by the id in client_id's top bits.
+                for frontend_id, sub_batch in enumerate(
+                        bucket_responses_by_frontend(rsp_batch, len(lanes))):
+                    if sub_batch:
+                        lanes[frontend_id].put(sub_batch)
+            else:
+                self.worker.result_queue.put(rsp_batch)
 
 
 def _get_params_for_first_rsp(
@@ -981,7 +1004,19 @@ def _send_rsp(
         rsp_batch: Optional[List[tllm.Response]] = None):
     # if postproc_batches is set, append to batch instead of putting to IpcQueue
 
-    if worker.result_queue is not None:
+    if worker.frontend_result_queues is not None:
+        # Multi-frontend serving: route to the origin frontend's result lane
+        # (client_id may be None for e.g. ADP dummy requests -- those go to
+        # lane 0, the launcher, which silently discards them like today).
+        if rsp_batch is not None:
+            rsp_batch.append(response)
+        else:
+            lanes = worker.frontend_result_queues
+            frontend_id = get_frontend_id(getattr(response, "client_id", None))
+            if frontend_id >= len(lanes):
+                frontend_id = 0
+            lanes[frontend_id].put(response)
+    elif worker.result_queue is not None:
         if rsp_batch is not None:
             rsp_batch.append(response)
         else:

tensorrt_llm/executor/executor.py

@@ -1,6 +1,8 @@
 import atexit
 import faulthandler
+import json
 import multiprocessing
+import os
 import platform
 import signal
 import traceback
@@ -543,6 +545,28 @@ def create(
                 f"Using {postproc_worker_config.num_postprocess_workers} postprocess parallel processes.\n",
                 "green")
 
+        # Multi-frontend serving: attach to an already-running executor
+        # instead of launching one. Set by trtllm-serve for the attached
+        # frontend processes (see commands/serve.py); the frontend id is
+        # picked up from TLLM_EXECUTOR_FRONTEND_ID.
+        attach_info_path = os.getenv("TLLM_EXECUTOR_ATTACH_INFO")
+        if attach_info_path:
+            with open(attach_info_path) as f:
+                attach_info = json.load(f)
+            if attach_info.get("mode") != "classic":
+                raise ValueError(
+                    "TLLM_EXECUTOR_ATTACH_INFO only supports the classic IPC "
+                    f"executor path, got mode={attach_info.get('mode')!r}")
+            from .proxy import GenerationExecutorFrontendProxy
+            frontend_id = int(os.environ["TLLM_EXECUTOR_FRONTEND_ID"])
+            logger.info(f"Attaching executor frontend {frontend_id} to the "
+                        f"running classic IPC worker via {attach_info_path}")
+            return GenerationExecutorFrontendProxy(
+                attach_info,
+                frontend_id=frontend_id,
+                postproc_worker_config=postproc_worker_config,
+                is_llm_executor=is_llm_executor)
+
         worker_kwargs = {
             "engine": engine,
             "executor_config": executor_config,

tensorrt_llm/executor/postproc_worker.py

@@ -14,7 +14,7 @@
 from ..logger import logger
 from ..sampling_params import SamplingParams
 from .ipc import ZeroMqQueue
-from .utils import ErrorResponse, is_llm_response
+from .utils import ErrorResponse, bucket_responses_by_frontend, is_llm_response
 
 if TYPE_CHECKING:
     from ..disaggregated_params import DisaggregatedParams
@@ -78,15 +78,19 @@ class Output(NamedTuple):
     def __init__(
         self,
         pull_pipe_addr: tuple[str, Optional[bytes]],
-        push_pipe_addr: tuple[str, Optional[bytes]],
+        push_pipe_addr: Union[tuple[str, Optional[bytes]],
+                              List[tuple[str, Optional[bytes]]]],
         tokenizer_dir: str,
         record_creator: Callable[
             ["PostprocWorker.Input", TransformersTokenizer], Any],
     ):
         '''
         Args:
             pull_pipe_addr (tuple[str, Optional[bytes]]): The address and HMAC key of the input IPC.
-            push_pipe_addr (tuple[str, Optional[bytes]]): The address and HMAC key of the output IPC.
+            push_pipe_addr: The address and HMAC key of the output IPC, or a
+                list of them with multi-frontend serving -- one result lane
+                per frontend, selected by the frontend id in the output's
+                client_id top bits.
             tokenizer_dir (str): The directory to load tokenizer.
             record_creator (Callable[["ResponsePostprocessWorker.Input"], Any]): A creator for creating a record for a request.
             result_handler (Optional[Callable[[GenerationResultBase], Any]]): A callback handles the final result.
@@ -98,11 +102,16 @@ def __init__(
                                       is_async=True,
                                       is_server=False,
                                       name="postprocess_pull_pipe")
-        self._push_pipe = ZeroMqQueue(address=push_pipe_addr,
-                                      is_async=True,
-                                      is_server=False,
-                                      socket_type=zmq.PUSH,
-                                      name="postprocess_push_pipe")
+        push_pipe_addrs = (push_pipe_addr if isinstance(push_pipe_addr, list)
+                           else [push_pipe_addr])
+        self._push_pipes = [
+            ZeroMqQueue(address=addr,
+                        is_async=True,
+                        is_server=False,
+                        socket_type=zmq.PUSH,
+                        name=f"postprocess_push_pipe_{i}")
+            for i, addr in enumerate(push_pipe_addrs)
+        ]
         self._to_stop = asyncio.Event()
 
         self._q = deque()
@@ -172,11 +181,21 @@ async def _batched_put(self):
         ''' Batched IPC send. '''
         async for batch in self._mainloop():
             if batch is None:
-                # notify dispatch_result corountine to quit
-                await self._push_pipe.put_async(None)
+                # notify the dispatch_result coroutine in every frontend to
+                # quit
+                for pipe in self._push_pipes:
+                    await pipe.put_async(None)
                 break
             assert isinstance(batch, list)
-            await self._push_pipe.put_async(batch)
+            if len(self._push_pipes) == 1:
+                await self._push_pipes[0].put_async(batch)
+                continue
+            # Multi-frontend serving: route each output to its origin
+            # frontend's result lane by the id in client_id's top bits.
+            for frontend_id, sub_batch in enumerate(
+                    bucket_responses_by_frontend(batch, len(self._push_pipes))):
+                if sub_batch:
+                    await self._push_pipes[frontend_id].put_async(sub_batch)
 
     async def _mainloop(self):
         ''' The loop for handle_response and keep producing outputs. '''
@@ -257,7 +276,9 @@ async def main():
 
 @print_traceback_on_error
 def postproc_worker_main(feedin_ipc_addr: tuple[str, Optional[bytes]],
-                         feedout_ipc_addr: tuple[str, Optional[bytes]],
+                         feedout_ipc_addr: Union[tuple[str, Optional[bytes]],
+                                                 List[tuple[str,
+                                                            Optional[bytes]]]],
                          tokenizer_dir: str, record_creator: Callable):
     worker = PostprocWorker(feedin_ipc_addr,
                             feedout_ipc_addr,