[Core] Combine per-engine utility results like the Rust client (#59240)
Signed-off-by: aoshen02 <aoshen02@users.noreply.github.com> Co-authored-by: aoshen02 <aoshen02@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
aoshen02 committed
Sep 30, 2026 at 06:20 UTC
a7a9ac2745f47e55fa21ffe6bb5c94d2e086efa1
1 file changed
+25
-13
vllm/v1/engine/core_client.py
+25
-13
@@ -1239,6 +1239,23 @@ class AsyncMPClient(MPClient):
1239
async def call_utility_async(self, method: str, *args) -> Any:
1240
return await self._call_utility_async(method, *args, engine=self.core_engine)
1241
1242
+ async def call_utility_all_async(self, method: str, *args) -> list[Any]:
1243
+ # Like the Rust client's call_utility: one result per managed engine.
1244
+ return await asyncio.gather(
1245
+ *[
1246
+ self._call_utility_async(method, *args, engine=engine)
1247
+ for engine in self.core_engines
1248
+ ]
1249
+ )
1250
+
1251
+ async def call_utility_consensus_async(self, method: str, *args) -> Any:
1252
+ results = await self.call_utility_all_async(method, *args)
1253
+ if any(result != results[0] for result in results):
1254
+ raise RuntimeError(
1255
+ f"Engines returned different {method} results: {results}"
1256
+ )
1257
+ return results[0]
1258
+
1259
async def _call_utility_async(
1260
self, method: str, *args, engine: EngineIdentity
1261
) -> Any:
@@ -1274,7 +1291,7 @@ class AsyncMPClient(MPClient):
1291
await self.call_utility_async("resume_scheduler")
1292
1293
async def is_scheduler_paused_async(self) -> bool:
1277
- return await self.call_utility_async("is_scheduler_paused")
1294
+ return await self.call_utility_consensus_async("is_scheduler_paused")
1295
1296
async def profile_async(
1297
self, is_start: bool = True, profile_prefix: str | None = None
@@ -1287,8 +1304,10 @@ class AsyncMPClient(MPClient):
1304
async def reset_prefix_cache_async(
1305
self, reset_running_requests: bool = False, reset_connector: bool = False
1306
) -> bool:
1290
- return await self.call_utility_async(
1291
- "reset_prefix_cache", reset_running_requests, reset_connector
1307
+ return all(
1308
+ await self.call_utility_all_async(
1309
+ "reset_prefix_cache", reset_running_requests, reset_connector
1310
+ )
1311
)
1312
1313
async def reset_encoder_cache_async(self) -> None:
@@ -1304,7 +1323,7 @@ class AsyncMPClient(MPClient):
1323
return await self.call_utility_async("wake_up", tags)
1324
1325
async def is_sleeping_async(self) -> bool:
1307
- return await self.call_utility_async("is_sleeping")
1326
+ return await self.call_utility_consensus_async("is_sleeping")
1327
1328
async def execute_dummy_batch_async(self) -> None:
1329
await self.call_utility_async("execute_dummy_batch")
@@ -1313,7 +1332,7 @@ class AsyncMPClient(MPClient):
1332
await self.call_utility_async("set_weight_version", weight_version)
1333
1334
async def get_weight_version_async(self) -> str:
1316
- return await self.call_utility_async("get_weight_version")
1335
+ return await self.call_utility_consensus_async("get_weight_version")
1336
1337
async def add_lora_async(self, lora_request: LoRARequest) -> bool:
1338
return await self.call_utility_async("add_lora", lora_request)
@@ -1645,14 +1664,7 @@ class DPLBAsyncMPClient(DPAsyncMPClient):
1664
1665
async def call_utility_async(self, method: str, *args) -> Any:
1666
# Only the result from the first engine is returned.
1648
- return (
1649
- await asyncio.gather(
1650
- *[
1651
- self._call_utility_async(method, *args, engine=engine)
1652
- for engine in self.core_engines
1653
- ]
1654
- )
1655
- )[0]
1667
+ return (await self.call_utility_all_async(method, *args))[0]
1668
1669
@staticmethod
1670
async def process_engine_outputs(