[Pooling] Report input throughput for batched requests (#53213)
Signed-off-by: Taneem Ibrahim <taneem.ibrahim@gmail.com> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Taneem Ibrahim committed
Aug 22, 2026 at 08:01 UTC
704f12aa1ed66db6307659f2996fd905a73570d4
12 files changed
+93
-8
rust/src/bench/src/backends/mod.rs
+2
@@ -101,6 +101,7 @@ pub struct RequestFuncOutput {
101
pub prompt_len: usize,
102
pub error: String,
103
pub start_time: f64,
104
+ pub num_input_sequences: usize,
105
}
106
107
impl Default for RequestFuncOutput {
@@ -116,6 +117,7 @@ impl Default for RequestFuncOutput {
117
prompt_len: 0,
118
error: String::new(),
119
start_time: 0.0,
120
+ num_input_sequences: 1,
121
}
122
}
123
}
rust/src/bench/src/backends/pooling.rs
+13
-2
@@ -39,9 +39,22 @@ impl PoolingBackend {
39
input: &RequestFuncInput,
40
client: &reqwest::Client,
41
) -> Result<RequestFuncOutput> {
42
+ let payload = self.build_payload(input);
43
+
44
// Preserve client-side prompt_len as fallback if server doesn't report usage.
45
let mut output = RequestFuncOutput {
46
prompt_len: input.prompt_len,
47
+ num_input_sequences: match self.kind {
48
+ BackendKind::OpenaiEmbeddings | BackendKind::VllmPooling => {
49
+ input.prompt_list.as_ref().map_or(1, |list| list.len())
50
+ }
51
+ BackendKind::OpenaiEmbeddingsChat => 1,
52
+ BackendKind::VllmRerank => payload
53
+ .get("documents")
54
+ .and_then(|documents| documents.as_array())
55
+ .map_or(0, |documents| documents.len()),
56
+ _ => unreachable!("PoolingBackend with non-pooling kind"),
57
+ },
58
..Default::default()
59
};
60
@@ -51,8 +64,6 @@ impl PoolingBackend {
64
&input.request_id,
65
);
66
54
- let payload = self.build_payload(input);
55
-
67
let mut request = client.post(&input.api_url);
68
for (k, v) in &headers {
69
request = request.header(k, v);
rust/src/bench/src/compare.rs
+5
@@ -16,6 +16,11 @@ const METRICS: &[MetricDef] = &[
16
key: "request_throughput",
17
lower_is_better: false,
18
},
19
+ MetricDef {
20
+ label: "Input throughput (inputs/s)",
21
+ key: "input_sequence_throughput",
22
+ lower_is_better: false,
23
+ },
24
MetricDef {
25
label: "Output throughput (tok/s)",
26
key: "output_throughput",
rust/src/bench/src/metrics/calculator.rs
+6
@@ -184,8 +184,10 @@ pub fn calculate_metrics(
184
completed,
185
failed,
186
total_input,
187
+ total_input_sequences: completed,
188
total_output,
189
request_throughput: completed as f64 / dur_s,
190
+ input_sequence_throughput: completed as f64 / dur_s,
191
request_goodput,
192
input_throughput: total_input as f64 / dur_s,
193
output_throughput: total_output as f64 / dur_s,
@@ -291,6 +293,7 @@ pub fn calculate_embedding_metrics(
293
selected_percentiles: &[f64],
294
) -> BenchmarkMetrics {
295
let mut total_input: usize = 0;
296
+ let mut total_input_sequences: usize = 0;
297
let mut completed: usize = 0;
298
let mut e2els: Vec<f64> = Vec::new();
299
@@ -299,6 +302,7 @@ pub fn calculate_embedding_metrics(
302
e2els.push(output.latency);
303
completed += 1;
304
total_input += output.prompt_len;
305
+ total_input_sequences += output.num_input_sequences;
306
}
307
}
308
@@ -344,8 +348,10 @@ pub fn calculate_embedding_metrics(
348
completed,
349
failed,
350
total_input,
351
+ total_input_sequences,
352
total_output: 0,
353
request_throughput: completed as f64 / dur_s,
354
+ input_sequence_throughput: total_input_sequences as f64 / dur_s,
355
request_goodput: 0.0,
356
input_throughput: total_input as f64 / dur_s,
357
output_throughput: 0.0,
rust/src/bench/src/metrics/mod.rs
+2
@@ -25,8 +25,10 @@ pub struct BenchmarkMetrics {
25
pub completed: usize,
26
pub failed: usize,
27
pub total_input: usize,
28
+ pub total_input_sequences: usize,
29
pub total_output: usize,
30
pub request_throughput: f64,
31
+ pub input_sequence_throughput: f64,
32
pub request_goodput: f64,
33
pub input_throughput: f64,
34
pub output_throughput: f64,
rust/src/bench/src/multi_run.rs
+17
-3
@@ -7,6 +7,7 @@ use crate::error::Result;
7
/// Key metrics extracted from a single run's JSON result.
8
struct RunMetrics {
9
request_throughput: f64,
10
+ input_sequence_throughput: Option<f64>,
11
output_throughput: f64,
12
total_token_throughput: f64,
13
mean_ttft_ms: f64,
@@ -38,6 +39,7 @@ impl RunMetrics {
39
fn from_json(json: &serde_json::Value) -> Self {
40
Self {
41
request_throughput: get(json, "request_throughput"),
42
+ input_sequence_throughput: get_opt(json, "input_sequence_throughput"),
43
output_throughput: get(json, "output_throughput"),
44
total_token_throughput: get(json, "total_token_throughput"),
45
mean_ttft_ms: get(json, "mean_ttft_ms"),
@@ -71,6 +73,10 @@ fn get(json: &serde_json::Value, key: &str) -> f64 {
73
json.get(key).and_then(|v| v.as_f64()).unwrap_or(0.0)
74
}
75
76
+fn get_opt(json: &serde_json::Value, key: &str) -> Option<f64> {
77
+ json.get(key).and_then(|v| v.as_f64())
78
+}
79
+
80
fn get_ss_opt(json: &serde_json::Value, key: &str) -> Option<f64> {
81
json.get("steady_state")
82
.and_then(|ss| if ss.is_null() { None } else { Some(ss) })
@@ -119,8 +125,15 @@ fn print_multi_run_summary(runs: &[RunMetrics]) {
125
let n = runs.len();
126
127
// Collect each metric into a series, compute stats
122
- let stats = vec![
123
- compute_stats("Request throughput (req/s)", runs, |r| r.request_throughput),
128
+ let mut stats = vec![compute_stats("Request throughput (req/s)", runs, |r| {
129
+ r.request_throughput
130
+ })];
131
+ if runs.iter().all(|r| r.input_sequence_throughput.is_some()) {
132
+ stats.push(compute_stats("Input throughput (inputs/s)", runs, |r| {
133
+ r.input_sequence_throughput.unwrap_or_default()
134
+ }));
135
+ }
136
+ stats.extend([
137
compute_stats("Output throughput (tok/s)", runs, |r| r.output_throughput),
138
compute_stats("Total token throughput (tok/s)", runs, |r| {
139
r.total_token_throughput
@@ -138,7 +151,7 @@ fn print_multi_run_summary(runs: &[RunMetrics]) {
151
compute_stats("Completed requests", runs, |r| r.completed),
152
compute_stats("Failed requests", runs, |r| r.failed),
153
compute_stats("Duration (s)", runs, |r| r.duration),
141
- ];
154
+ ]);
155
156
println!("{:=^80}", format!(" Multi-Run Summary ({n} runs) "));
157
println!(
@@ -307,6 +320,7 @@ mod tests {
320
fn mk_run(ss: Option<f64>) -> RunMetrics {
321
RunMetrics {
322
request_throughput: 0.0,
323
+ input_sequence_throughput: None,
324
output_throughput: 0.0,
325
total_token_throughput: 0.0,
326
mean_ttft_ms: 0.0,
rust/src/bench/src/output/console.rs
+6
@@ -51,6 +51,12 @@ pub fn print_results(
51
"{:<40} {:<10.2}",
52
"Request throughput (req/s):", metrics.request_throughput
53
);
54
+ if is_pooling {
55
+ println!(
56
+ "{:<40} {:<10.2}",
57
+ "Input throughput (inputs/s):", metrics.input_sequence_throughput
58
+ );
59
+ }
60
if metrics.request_goodput > 0.0 {
61
println!(
62
"{:<40} {:<10.2}",
rust/src/bench/src/output/json.rs
+10
@@ -98,6 +98,16 @@ pub fn build_result_json(
98
"request_throughput".into(),
99
serde_json::json!(metrics.request_throughput),
100
);
101
+ if is_pooling {
102
+ result.insert(
103
+ "total_input_sequences".into(),
104
+ serde_json::json!(metrics.total_input_sequences),
105
+ );
106
+ result.insert(
107
+ "input_sequence_throughput".into(),
108
+ serde_json::json!(metrics.input_sequence_throughput),
109
+ );
110
+ }
111
if config.goodput.is_empty() {
112
result.insert("request_goodput".into(), Value::Null);
113
} else {
rust/src/bench/src/sweep.rs
+3
@@ -295,6 +295,7 @@ fn print_sweep_summary(param_name: &str, points: &[SweepPoint], summary_percenti
295
fn build_summary_columns(summary_percentiles: &[f64]) -> Vec<SummaryColumn> {
296
let mut columns = vec![
297
SummaryColumn::new("Req/s", "request_throughput", 10),
298
+ SummaryColumn::new("Inputs/s", "input_sequence_throughput", 10),
299
SummaryColumn::new("Tok/s", "output_throughput", 10),
300
SummaryColumn::new("Total tok/s", "total_token_throughput", 12),
301
SummaryColumn::new("SS req/s", SS_REQUEST_THROUGHPUT_KEY, 10),
@@ -452,6 +453,7 @@ mod tests {
453
headers,
454
vec![
455
"Req/s",
456
+ "Inputs/s",
457
"Tok/s",
458
"Total tok/s",
459
"SS req/s",
@@ -495,6 +497,7 @@ mod tests {
497
headers,
498
vec![
499
"Req/s",
500
+ "Inputs/s",
501
"Tok/s",
502
"Total tok/s",
503
"SS req/s",
vllm/benchmarks/datasets/datasets.py
+1
-1
@@ -79,7 +79,7 @@ class SampleRequest:
79
Represents a single inference request for benchmarking.
80
"""
81
82
- prompt: str | list[str] | list[int] | list[dict]
82
+ prompt: str | list[str] | list[int] | list[list[int]] | list[dict]
83
prompt_len: int
84
expected_output_len: int = 0
85
multi_modal_data: MultiModalDataDict | dict | list[dict] | None = None
vllm/benchmarks/lib/endpoint_request_func.py
+14
-2
@@ -66,7 +66,7 @@ class StreamedResponseHandler:
66
class RequestFuncInput:
67
"""The input for the request function."""
68
69
- prompt: str | list[str] | list[int] | list[dict[str, Any]]
69
+ prompt: str | list[str] | list[int] | list[list[int]] | list[dict[str, Any]]
70
api_url: str
71
prompt_len: int
72
output_len: int
@@ -100,6 +100,7 @@ class RequestFuncOutput:
100
error: str = ""
101
start_time: float = 0.0
102
input_audio_duration: float = 0.0 # in seconds
103
+ num_input_sequences: int = 1
104
105
106
class RequestFunc(Protocol):
@@ -586,8 +587,9 @@ async def _run_pooling_request(
587
payload: dict[str, Any],
588
headers: dict[str, Any],
589
pbar: tqdm | None = None,
590
+ num_input_sequences: int = 1,
591
) -> RequestFuncOutput:
590
- output = RequestFuncOutput()
592
+ output = RequestFuncOutput(num_input_sequences=num_input_sequences)
593
st = time.perf_counter()
594
output.start_time = st
595
try:
@@ -617,6 +619,12 @@ async def _run_pooling_request(
619
return output
620
621
622
+def _get_num_input_sequences(prompt: Any) -> int:
623
+ if prompt and isinstance(prompt, list) and isinstance(prompt[0], (str, list)):
624
+ return len(prompt)
625
+ return 1
626
+
627
+
628
async def async_request_openai_embeddings(
629
request_func_input: RequestFuncInput,
630
session: aiohttp.ClientSession,
@@ -645,6 +653,7 @@ async def async_request_openai_embeddings(
653
payload=payload,
654
headers=headers,
655
pbar=pbar,
656
+ num_input_sequences=_get_num_input_sequences(request_func_input.prompt),
657
)
658
659
@@ -681,6 +690,7 @@ async def async_request_vllm_rerank(
690
payload=payload,
691
headers=headers,
692
pbar=pbar,
693
+ num_input_sequences=len(request_func_input.prompt) - 1,
694
)
695
696
@@ -818,6 +828,7 @@ async def async_request_infinity_embeddings(
828
payload=payload,
829
headers=headers,
830
pbar=pbar,
831
+ num_input_sequences=_get_num_input_sequences(request_func_input.prompt),
832
)
833
834
@@ -866,6 +877,7 @@ async def async_request_vllm_pooling(
877
payload=payload,
878
headers=headers,
879
pbar=pbar,
880
+ num_input_sequences=_get_num_input_sequences(request_func_input.prompt),
881
)
882
883
vllm/benchmarks/serve.py
+14
@@ -357,7 +357,9 @@ class EmbedBenchmarkMetrics:
357
completed: int
358
failed: int
359
total_input: int
360
+ total_input_sequences: int
361
request_throughput: float
362
+ input_sequence_throughput: float
363
total_token_throughput: float
364
mean_e2el_ms: float
365
std_e2el_ms: float
@@ -520,6 +522,7 @@ def calculate_metrics_for_embeddings(
522
The calculated benchmark metrics.
523
"""
524
total_input = 0
525
+ total_input_sequences = 0
526
completed = 0
527
failed = 0
528
e2els: list[float] = []
@@ -528,6 +531,7 @@ def calculate_metrics_for_embeddings(
531
e2els.append(outputs[i].latency)
532
completed += 1
533
total_input += outputs[i].prompt_len
534
+ total_input_sequences += outputs[i].num_input_sequences
535
else:
536
failed += 1
537
@@ -541,7 +545,9 @@ def calculate_metrics_for_embeddings(
545
completed=completed,
546
failed=failed,
547
total_input=total_input,
548
+ total_input_sequences=total_input_sequences,
549
request_throughput=completed / dur_s,
550
+ input_sequence_throughput=total_input_sequences / dur_s,
551
total_token_throughput=total_input / dur_s,
552
mean_e2el_ms=np.mean(e2els or 0) * 1000,
553
std_e2el_ms=np.std(e2els or 0) * 1000,
@@ -1181,6 +1187,12 @@ async def benchmark(
1187
"Request throughput (req/s):", metrics.request_throughput
1188
)
1189
)
1190
+ if isinstance(metrics, EmbedBenchmarkMetrics):
1191
+ print(
1192
+ "{:<40} {:<10.2f}".format(
1193
+ "Input throughput (inputs/s):", metrics.input_sequence_throughput
1194
+ )
1195
+ )
1196
if goodput_config_dict and isinstance(metrics, BenchmarkMetrics):
1197
print(
1198
"{:<40} {:<10.2f}".format(
@@ -1284,7 +1296,9 @@ async def benchmark(
1296
"duration": benchmark_duration,
1297
"completed": metrics.completed,
1298
"total_input_tokens": metrics.total_input,
1299
+ "total_input_sequences": metrics.total_input_sequences,
1300
"request_throughput": metrics.request_throughput,
1301
+ "input_sequence_throughput": metrics.input_sequence_throughput,
1302
"total_token_throughput": metrics.total_token_throughput,
1303
"input_lens": [output.prompt_len for output in outputs],
1304
"errors": [output.error for output in outputs],