add check running queue length and waiting queue length

Signed-off-by: rongfu.leng <rongfu.leng@daocloud.io>

Author: lengrongfu

vllm/engine/protocol.py

@@ -60,7 +60,7 @@ def generate(
         ...
 
     @abstractmethod
-    def minimal_generation(self) -> str:
+    async def minimal_generation(self) -> str:
         """Generate outputs for a minimal spec prompt"""
         ...
 

vllm/entrypoints/openai/api_server.py

@@ -98,13 +98,10 @@
                                     log_non_default_args, with_cancellation)
 from vllm.logger import init_logger
 from vllm.reasoning import ReasoningParserManager
-from vllm.transformers_utils.config import (
-    maybe_register_config_serialize_by_value)
 from vllm.transformers_utils.tokenizer import MistralTokenizer
 from vllm.usage.usage_lib import UsageContext
 from vllm.utils import (Device, FlexibleArgumentParser, decorate_logs,
-                        get_open_zmq_ipc_path,is_valid_ipv6_address, random_uuid, 
-                        set_ulimit)
+                        is_valid_ipv6_address, set_ulimit)
 from vllm.v1.engine.exceptions import EngineDeadError
 from vllm.v1.metrics.prometheus import get_prometheus_registry
 from vllm.version import __version__ as VLLM_VERSION

vllm/v1/engine/async_llm.py

@@ -725,10 +725,14 @@ async def scale_elastic_ep(self,
             )
 
     async def minimal_generation(self) -> str:
-        prompt = "Hi"
+        prompt = "Ping"
         sampling_params = SamplingParams(temperature=0, max_tokens=2)
         request_id = random_uuid()
         result_text = ""
+        count = await self.engine_core.get_request_count()
+        num_running_reqs, num_waiting_reqs = count[0], count[1]
+        if num_running_reqs > 0 or num_waiting_reqs > 0:
+            return result_text
         async for output in self.generate(prompt, sampling_params, request_id):
             for completion in output.outputs:
                 result_text = completion.text

vllm/v1/engine/core.py

@@ -449,6 +449,9 @@ def preprocess_add_request(
             self.structured_output_manager.grammar_init(req)
         return req, request.current_wave
 
+    def get_request_count(self) -> tuple[int, int]:
+        return self.scheduler.get_request_counts()
+
 
 class EngineCoreProc(EngineCore):
     """ZMQ-wrapper for running EngineCore in background process."""

vllm/v1/engine/core_client.py

@@ -230,6 +230,9 @@ async def collective_rpc_async(
             kwargs: Optional[dict[str, Any]] = None) -> list[_R]:
         raise NotImplementedError
 
+    async def get_request_count(self) -> tuple[int, int]:
+        raise NotImplementedError
+
 
 class InprocClient(EngineCoreClient):
     """
@@ -311,6 +314,9 @@ def collective_rpc(self,
     def dp_engines_running(self) -> bool:
         return False
 
+    async def get_request_count(self) -> tuple[int, int]:
+        return self.engine_core.get_request_count()
+
 
 @dataclass
 class BackgroundResources:
@@ -755,6 +761,9 @@ def save_sharded_state(self,
                            max_size: Optional[int] = None) -> None:
         self.call_utility("save_sharded_state", path, pattern, max_size)
 
+    async def get_request_count(self) -> tuple[int, int]:
+        return self.call_utility("get_request_count")
+
 
 class AsyncMPClient(MPClient):
     """Asyncio-compatible client for multi-proc EngineCore."""
@@ -958,6 +967,9 @@ async def collective_rpc_async(
         return await self.call_utility_async("collective_rpc", method, timeout,
                                              args, kwargs)
 
+    async def get_request_count(self) -> tuple[int, int]:
+        return await self.call_utility_async("get_request_count")
+
 
 class DPAsyncMPClient(AsyncMPClient):
     """Asyncio-compatible client for multi-proc, multi-engine (data parallel)