Repository: vllm-project/vllm
Issue category: Feature Request

Issue:
Benchmarking script does not limit the maximum concurrency
The current benchmarking script if specified with `INF` arrivals, will not limit the maximum concurrency level as shown [here](https://github.com/vllm-project/vllm/blob/703e42ee4b3efed3c71e7ae7d15f0f96e05722d4/benchmarks/benchmark_serving.py#L191).

If we can change it to below, we can limit the maximum concurrency to have a fine controlled load level.

```
semaphore = asyncio.Semaphore(max_concurrency)  # Semaphore to limit concurrency

    async def make_request(request, sem):
        async with sem:  # Ensure only max_concurrency tasks run in parallel
            prompt, prompt_len, output_len = request
            request_func_input = RequestFuncInput(
                model=model_id,
                prompt=prompt,
                api_url=api_url,
                prompt_len=prompt_len,
                output_len=output_len,
                best_of=best_of,
                use_beam_search=use_beam_search,
            )
            # Call the request function directly here and return its result
            return await request_func(request_func_input=request_func_input, pbar=pbar)

    tasks = []
    for request in input_requests:  # Direct iteration may replace async iteration based on design
        # Enqueue task without immediately awaiting it
        tasks.append(make_request(request, semaphore))
        # Manage inter-arrival time
        if request_rate != float("inf"):
            await asyncio.sleep(1.0 / request_rate)

    outputs = await asyncio.gather(*tasks)  # Wait for all tasks to complete
```


Hints:
Good point. PR welcomed! 
@wangchen615 Please correct me if I misunderstood, but is this for testing the case where you have another layer on top of the model deployment with concurrency control?
I recently made https://github.com/vllm-project/vllm/pull/3194 to add prefix caching benchmark - @wangchen615 let me know if you want me to include changes to resolve this issue in that PR as well!
@ywang96 , thanks for offering help. Yes, please. Sorry for losing this thread for so long. 

Patch:
diff --git a/benchmarks/benchmark_serving.py b/benchmarks/benchmark_serving.py
index c1a396c81f666..4580729fa4767 100644
--- a/benchmarks/benchmark_serving.py
+++ b/benchmarks/benchmark_serving.py
@@ -397,6 +397,7 @@ async def benchmark(
     selected_percentile_metrics: List[str],
     selected_percentiles: List[str],
     ignore_eos: bool,
+    max_concurrency: Optional[int],
 ):
     if backend in ASYNC_REQUEST_FUNCS:
         request_func = ASYNC_REQUEST_FUNCS[backend]
@@ -445,9 +446,25 @@ async def benchmark(
             print("Profiler started")
 
     print(f"Traffic request rate: {request_rate}")
+    print(f"Maximum request concurrency: {max_concurrency}")
 
     pbar = None if disable_tqdm else tqdm(total=len(input_requests))
 
+    # This can be used once the minimum Python version is 3.10 or higher,
+    # and it will simplify the code in limited_request_func.
+    #    semaphore = (asyncio.Semaphore(max_concurrency)
+    #                 if max_concurrency else contextlib.nullcontext())
+    semaphore = (asyncio.Semaphore(max_concurrency)
+                 if max_concurrency else None)
+
+    async def limited_request_func(request_func_input, pbar):
+        if semaphore is None:
+            return await request_func(request_func_input=request_func_input,
+                                      pbar=pbar)
+        async with semaphore:
+            return await request_func(request_func_input=request_func_input,
+                                      pbar=pbar)
+
     benchmark_start_time = time.perf_counter()
     tasks: List[asyncio.Task] = []
     async for request in get_request(input_requests, request_rate):
@@ -463,8 +480,8 @@ async def benchmark(
                                               ignore_eos=ignore_eos)
         tasks.append(
             asyncio.create_task(
-                request_func(request_func_input=request_func_input,
-                             pbar=pbar)))
+                limited_request_func(request_func_input=request_func_input,
+                                     pbar=pbar)))
     outputs: List[RequestFuncOutput] = await asyncio.gather(*tasks)
 
     if profile:
@@ -680,6 +697,7 @@ def main(args: argparse.Namespace):
                 float(p) for p in args.metric_percentiles.split(",")
             ],
             ignore_eos=args.ignore_eos,
+            max_concurrency=args.max_concurrency,
         ))
 
     # Save config and results to json
@@ -709,13 +727,16 @@ def main(args: argparse.Namespace):
         # Traffic
         result_json["request_rate"] = (
             args.request_rate if args.request_rate < float("inf") else "inf")
+        result_json["max_concurrency"] = args.max_concurrency
 
         # Merge with benchmark result
         result_json = {**result_json, **benchmark_result}
 
         # Save to file
         base_model_id = model_id.split("/")[-1]
-        file_name = f"{backend}-{args.request_rate}qps-{base_model_id}-{current_dt}.json"  #noqa
+        max_concurrency_str = (f"-concurrency{args.max_concurrency}"
+                               if args.max_concurrency is not None else "")
+        file_name = f"{backend}-{args.request_rate}qps{max_concurrency_str}-{base_model_id}-{current_dt}.json"  #noqa
         if args.result_filename:
             file_name = args.result_filename
         if args.result_dir:
@@ -766,6 +787,19 @@ def main(args: argparse.Namespace):
                         default=None,
                         help="Path to the sharegpt/sonnet dataset. "
                         "Or the huggingface dataset ID if using HF dataset.")
+    parser.add_argument(
+        "--max-concurrency",
+        type=int,
+        default=None,
+        help="Maximum number of concurrent requests. This can be used "
+        "to help simulate an environment where a higher level component "
+        "is enforcing a maximum number of concurrent requests. While the "
+        "--request-rate argument controls the rate at which requests are "
+        "initiated, this argument will control how many are actually allowed "
+        "to execute at a time. This means that when used in combination, the "
+        "actual request rate may be lower than specified with --request-rate, "
+        "if the server is not processing requests fast enough to keep up.")
+
     parser.add_argument(
         "--model",
         type=str,


Test patch: