[Core] Add a random suffix to frontend-provided request IDs

Since #9550 and #10968 we support client's supplying a custom
request ID. The motivation for this is that it can be very helpful
when you need to correlate vLLM logs with logs of a related service.

Since the request ID is used ubiquitously across vLLM as a unique
key, it obviously is problematic if we ever have multiple in-flight
requests using the same client-provided request ID.

We saw this happening recently when `vllm serve bench` started
including a request ID and the request IDs from multiple concurrent
instances caused collisions. See #27723

We try to guard against request ID collisions currently in the
frontend in OutputProcessor:

```
    def add_request(...):
        if request_id in self.request_states:
            raise ValueError(f"Request id {request_id} already running.")
```

however, this is not always effective:

1) We can have abort race conditions where a request is no longer
   tracked by the frontend, but still not completed in the engine.
   See #15326 for an attempt to fix this.
2) We can have async scheduling race conditions where a request
   ID is removed from the output processor and being scheduled
   while the older request with that ID is still being completed
   by the model runner. See #29355
3) With P/D, a request will continue to be tracked by the prefill
   engine long after the prefill request has been completed in
   the frontend, while we wait for the decode side to fetch the
   KV blocks. See #20139

Let's instead ensure we use a unique request ID internally, even
when a client provides a custom request ID. We can do this simply
by appending a short random suffix to any request ID provided
by the frontend.

We need to ensure we track the external->internal request ID
mapping because abort() will be supplied an external request ID.
In the case where an external request ID maps to multiple running
requests, we assume the caller requires all of those requests
to be aborted. The caller can use EngineCoreRequest.request_id
as the request ID if they want to be more specific.

A full 32 character random UUID would be overkill as a suffix,
so how many random characters would be sufficient? 8 characters
gives us 32 bits of entropy, or 16^8 possible prefixes.

Using the collision probability approximation from
https://preshing.com/20110504/hash-collision-probabilities:

N = 16^8 and k is the number of generated suffixes, then the
probability of collision is (k^2)/(2N), so If a client somehow
caused vLLM to hold 10k requests that reuse the same client-provided
ID, then there would be a 1.16% chance of collision:

```
>>> (k**2)/(2*N)
0.011641532182693481
```

That seems (super good enough)[https://hownot2.com/products/hownot2-super-good-enough-t-shirt].

Signed-off-by: Mark McLoughlin <markmc@redhat.com>

Author: markmc

tests/tokenizers_/test_detokenize.py

@@ -62,6 +62,7 @@ def _run_incremental_decode(
     )
     request = EngineCoreRequest(
         request_id="",
+        external_request_id="",
         prompt_token_ids=prompt_token_ids,
         mm_features=None,
         sampling_params=params,

tests/v1/engine/test_process_multi_modal_uuids.py

@@ -6,6 +6,7 @@
 from vllm.assets.image import ImageAsset
 from vllm.assets.video import VideoAsset
 from vllm.config import CacheConfig, DeviceConfig, ModelConfig, VllmConfig
+from vllm.multimodal import MultiModalUUIDDict
 from vllm.sampling_params import SamplingParams
 from vllm.v1.engine import input_processor as input_processor_mod
 from vllm.v1.engine.input_processor import InputProcessor
@@ -166,7 +167,7 @@ def test_multi_modal_uuids_ignored_when_caching_disabled(monkeypatch):
         monkeypatch, mm_cache_gb=0.0, enable_prefix_caching=False
     )
 
-    captured: dict[str, object] = {}
+    captured: dict[str, MultiModalUUIDDict] = {}
 
     def fake_preprocess(
         prompt, *, tokenization_kwargs=None, lora_request=None, mm_uuids=None
@@ -196,7 +197,16 @@ def fake_preprocess(
     )
 
     # Expect request-id-based overrides are passed through
-    assert captured["mm_uuids"] == {
-        "image": [f"{request_id}-image-0", f"{request_id}-image-1"],
-        "video": [f"{request_id}-video-0"],
-    }
+    mm_uuids = captured["mm_uuids"]
+    assert set(mm_uuids.keys()) == {"image", "video"}
+    assert len(mm_uuids["image"]) == 2
+    assert len(mm_uuids["video"]) == 1
+    assert mm_uuids["image"][0].startswith(f"{request_id}-") and mm_uuids["image"][
+        0
+    ].endswith("-image-0")
+    assert mm_uuids["image"][1].startswith(f"{request_id}-") and mm_uuids["image"][
+        1
+    ].endswith("-image-1")
+    assert mm_uuids["video"][0].startswith(f"{request_id}-") and mm_uuids["video"][
+        0
+    ].endswith("-video-0")

vllm/entrypoints/llm.py

@@ -1700,15 +1700,30 @@ def _add_request(
         )
 
         self.llm_engine.add_request(
-            request_id,
+            engine_request.request_id,
             engine_request,
             params,
             lora_request=lora_request,
             tokenization_kwargs=tokenization_kwargs,
             priority=priority,
             prompt_text=prompt_text,
         )
-        return request_id
+        return engine_request.request_id
+
+    @staticmethod
+    def _sort_outputs(
+        outputs: list[RequestOutput | PoolingRequestOutput],
+    ) -> list[RequestOutput | PoolingRequestOutput]:
+        # Sort the outputs by request ID.
+        # This is necessary because some requests may be finished earlier than
+        # its previous requests.
+
+        # Extract the original request ID prefix for sorting.
+        # See how InputProcessor._generate_request_id() adds a random suffix
+        def extract_request_id_prefix(request_id: str) -> int:
+            return int(request_id.rsplit("-", 1)[0])
+
+        return sorted(outputs, key=lambda x: extract_request_id_prefix(x.request_id))
 
     def _run_engine(
         self, *, use_tqdm: bool | Callable[..., tqdm] = True
@@ -1756,7 +1771,5 @@ def _run_engine(
 
         if use_tqdm:
             pbar.close()
-        # Sort the outputs by request ID.
-        # This is necessary because some requests may be finished earlier than
-        # its previous requests.
-        return sorted(outputs, key=lambda x: int(x.request_id))
+
+        return self._sort_outputs(outputs)

vllm/entrypoints/openai/serving_chat.py

@@ -341,7 +341,7 @@ async def create_chat_completion(
                     generator = self.engine_client.generate(
                         engine_request,
                         sampling_params,
-                        sub_request_id,
+                        engine_request.request_id,
                         lora_request=lora_request,
                         trace_headers=trace_headers,
                         priority=request.priority,

vllm/entrypoints/openai/serving_completion.py

@@ -231,7 +231,7 @@ async def create_completion(
                     generator = self.engine_client.generate(
                         engine_request,
                         sampling_params,
-                        request_id_item,
+                        engine_request.request_id,
                         lora_request=lora_request,
                         trace_headers=trace_headers,
                         priority=request.priority,

vllm/entrypoints/openai/serving_engine.py

@@ -1260,7 +1260,7 @@ async def _generate_with_builtin_tools(
             generator = self.engine_client.generate(
                 engine_request,
                 sampling_params,
-                sub_request_id,
+                engine_request.request_id,
                 lora_request=lora_request,
                 priority=priority,
                 prompt_text=prompt_text,

vllm/v1/engine/__init__.py

@@ -49,6 +49,7 @@ class EngineCoreRequest(
     gc=False,
 ):  # type: ignore[call-arg]
     request_id: str
+    external_req_id: str
     prompt_token_ids: list[int] | None
     mm_features: list[MultiModalFeatureSpec] | None
     sampling_params: SamplingParams | None

vllm/v1/engine/async_llm.py

@@ -304,6 +304,12 @@ async def add_request(
         # Convert Input --> Request.
         if isinstance(prompt, EngineCoreRequest):
             request = prompt
+            if request_id != request.request_id:
+                logger.warning_once(
+                    "AsyncLLM.add_request() was passed a request_id parameter that "
+                    "does not match the EngineCoreRequest.request_id attribute. The "
+                    "latter will be used, and the former will be ignored."
+                )
         else:
             assert prompt_text is None
             request = self.input_processor.process_inputs(
@@ -333,7 +339,7 @@ async def add_request(
         assert isinstance(parent_params, SamplingParams)
 
         # Fan out child requests (for n>1).
-        parent_request = ParentRequest(request_id, parent_params)
+        parent_request = ParentRequest(request.request_id, parent_params)
         for idx in range(parent_params.n):
             request_id, child_params = parent_request.get_child_info(idx)
             child_request = request if idx == parent_params.n - 1 else copy(request)

vllm/v1/engine/input_processor.py

@@ -20,7 +20,7 @@
 from vllm.pooling_params import PoolingParams
 from vllm.sampling_params import SamplingParams
 from vllm.tokenizers import MistralTokenizer, TokenizerLike
-from vllm.utils import length_from_prompt_token_ids_or_embeds
+from vllm.utils import length_from_prompt_token_ids_or_embeds, random_uuid
 from vllm.v1.engine import EngineCoreRequest
 from vllm.v1.metrics.stats import MultiModalCacheStats
 from vllm.v1.structured_output.backend_guidance import validate_guidance_grammar
@@ -382,6 +382,12 @@ def _extract_mm_data(p: PromptType):
             mm_uuids[modality] = [f"{request_id}-{modality}-{i}" for i in range(n)]
         return mm_uuids
 
+    def _generate_request_id(self, request_id: str):
+        """Construct an internal request ID by adding 8 random characters
+        to the supplied request ID in order to ensure uniquness.
+        """
+        return f"{request_id}-{random_uuid()[:8]}"
+
     def process_inputs(
         self,
         request_id: str,
@@ -409,6 +415,9 @@ def process_inputs(
         if arrival_time is None:
             arrival_time = time.time()
 
+        external_req_id = request_id
+        request_id = self._generate_request_id(request_id)
+
         # Optionally generate multimodal hash overrides to avoid hashing
         # multimodal data items by their content as their identifiers.
 
@@ -509,6 +518,7 @@ def process_inputs(
 
         return EngineCoreRequest(
             request_id=request_id,
+            external_req_id=external_req_id,
             prompt_token_ids=prompt_token_ids,
             prompt_embeds=prompt_embeds,
             mm_features=mm_features,

vllm/v1/engine/llm_engine.py

@@ -239,6 +239,12 @@ def add_request(
         # Process raw inputs into the request.
         if isinstance(prompt, EngineCoreRequest):
             request = prompt
+            if request_id != request.request_id:
+                logger.warning_once(
+                    "AsyncLLM.add_request() was passed a request_id parameter that "
+                    "does not match the EngineCoreRequest.request_id attribute. The "
+                    "latter will be used, and the former will be ignored."
+                )
         else:
             assert prompt_text is None
             request = self.input_processor.process_inputs(
@@ -269,7 +275,7 @@ def add_request(
             return
 
         # Fan out child requests (for n>1).
-        parent_req = ParentRequest(request_id, params)
+        parent_req = ParentRequest(request.request_id, params)
         for idx in range(n):
             request_id, child_params = parent_req.get_child_info(idx)
             child_request = request if idx == n - 1 else copy(request)