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

Since vllm-project#9550 and vllm-project#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 vllm-project#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 vllm-project#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 vllm-project#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 vllm-project#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.

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].

The key changes to support this are:

1. `InputProcessor.process_inputs()` - we add some randomness to the
request ID just before creating an `EngineCoreRequest`, and store both
the random "internal" request ID (as `request_id`) and the supplied
"external" request ID (as `external_req_id`) in the
`EngineCoreRequest`.
2. `RequestState.make_request_output()` - we ensure that
`RequestOutput.request_id` continues to be the external request ID
(for backwards compat) and add `internal_request_id`.
3. `OutputProcessor.abort_requests()` - we make `OutputProcessor`
track a mapping from external request ID to internal request IDs, so
`abort_requests()` can abort based on either ID.
4. `AsyncLLM` - we use `RequestOutputCollector` to track the internal
request ID, so we can use the internal ID to abort an in-progress
request. We also add an `internal` boolean flag to `abort()` so API
users can abort based on either ID.
5. `ParentRequest` - in the case of parallel sampling, we need to
track both the internal and external ID for the later creation of
`RequestOutput` aggregating the child outputs.

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.

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

Author: markmc

tests/detokenizer/test_min_tokens.py

@@ -35,6 +35,7 @@ def test_min_tokens_with_stop(min_tokens: int, stop: str, truth: str):
     )
     request = EngineCoreRequest(
         request_id="",
+        external_req_id="",
         prompt_token_ids=prompt_token_ids,
         mm_features=None,
         sampling_params=params,

tests/detokenizer/test_stop_string_while_stop_model_terminates.py

@@ -31,6 +31,7 @@ def _make_request(stop, include_stop_str_in_output: bool, min_tokens: int = 0):
     # Keep other fields minimal for unit test purposes.
     req = EngineCoreRequest(
         request_id="test",
+        external_req_id="test-ext",
         prompt_token_ids=[],
         mm_features=None,
         sampling_params=params,

tests/entrypoints/openai/test_serving_chat.py

@@ -390,7 +390,9 @@ async def _fake_process_inputs(
         trace_headers,
         priority,
     ):
-        return dict(engine_prompt), {}
+        mock_request = MagicMock()
+        mock_request.request_id = request_id
+        return mock_request, {}
 
     serving_chat._process_inputs = AsyncMock(side_effect=_fake_process_inputs)
     return serving_chat
@@ -662,7 +664,11 @@ async def test_serving_chat_data_parallel_rank_extraction():
     mock_engine.get_tokenizer.return_value = get_tokenizer(MODEL_NAME)
     mock_engine.errored = False
     mock_engine.model_config = MockModelConfig()
+
+    mock_request = MagicMock()
+    mock_request.request_id = "test-request-internal"
     mock_engine.input_processor = MagicMock()
+    mock_engine.input_processor.process_inputs.return_value = mock_request
     mock_engine.io_processor = MagicMock()
 
     # Mock the generate method to return an async generator
@@ -689,7 +695,9 @@ async def mock_generate(*args, **kwargs):
             finished=True,
         )
 
-    mock_engine.generate = AsyncMock(side_effect=mock_generate)
+    mock_engine.generate = MagicMock(
+        side_effect=lambda *args, **kwargs: mock_generate()
+    )
 
     serving_chat = _build_serving_chat(mock_engine)
 

tests/tokenizers_/test_detokenize.py

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

tests/v1/engine/test_async_llm.py

@@ -253,7 +253,7 @@ async def test_multi_abort(output_kind: RequestOutputKind):
 
         # Use multi-abort to abort multiple requests at once
         abort_request_ids = [request_ids[i] for i in REQUEST_IDS_TO_ABORT]
-        await engine.abort(abort_request_ids)
+        await engine.abort(abort_request_ids, internal=False)
 
         # Wait for all tasks to complete
         results = await asyncio.gather(*tasks, return_exceptions=True)
@@ -548,7 +548,7 @@ async def test_abort_final_output(output_kind: RequestOutputKind):
         await asyncio.sleep(0.5)
 
         # Abort the request
-        await engine.abort(request_id)
+        await engine.abort(request_id, internal=False)
 
         # Wait for generation to complete and return final output
         final_output = await generated

tests/v1/engine/test_engine_core.py

@@ -40,10 +40,16 @@
 PROMPT = "I am Gyoubu Masataka Oniwa"
 PROMPT_TOKENS = TOKENIZER(PROMPT).input_ids
 
+_REQUEST_COUNTER = 0
+
 
 def make_request() -> EngineCoreRequest:
+    global _REQUEST_COUNTER
+    _REQUEST_COUNTER += 1
+    request_id = f"request-{_REQUEST_COUNTER}"
     return EngineCoreRequest(
-        request_id=str(uuid.uuid4()),
+        request_id=request_id,
+        external_req_id=f"{request_id}-{uuid.uuid4()}",
         prompt_token_ids=PROMPT_TOKENS,
         mm_features=None,
         sampling_params=SamplingParams(),

tests/v1/engine/test_engine_core_client.py

@@ -39,15 +39,21 @@
 PROMPT = "Hello my name is Robert and I love quantization kernels"
 PROMPT_TOKENS = TOKENIZER(PROMPT).input_ids
 
+_REQUEST_COUNTER = 0
+
 
 def make_request(
     params: SamplingParams, prompt_tokens_ids: list[int] | None = None
 ) -> EngineCoreRequest:
     if not prompt_tokens_ids:
         prompt_tokens_ids = PROMPT_TOKENS
 
+    global _REQUEST_COUNTER
+    _REQUEST_COUNTER += 1
+    request_id = f"request-{_REQUEST_COUNTER}"
     return EngineCoreRequest(
-        request_id=str(uuid.uuid4()),
+        request_id=request_id,
+        external_req_id=f"{request_id}-{uuid.uuid4()}",
         prompt_token_ids=prompt_tokens_ids,
         mm_features=None,
         sampling_params=params,

tests/v1/engine/test_fast_incdec_prefix_err.py

@@ -27,6 +27,7 @@ def test_fast_inc_detok_invalid_utf8_err_case():
     params = SamplingParams(skip_special_tokens=True)
     request = EngineCoreRequest(
         request_id="test",
+        external_req_id="test-ext",
         prompt_token_ids=prompt_token_ids,
         mm_features=None,
         sampling_params=params,