use clap::{Args, ValueEnum};
use colored::*;
use ferrum_bench_core::env::HttpRequestSampling;
use ferrum_bench_core::{
arrivals::poisson_arrival_times, compute_metrics, BenchReport, BenchmarkPhase,
BenchmarkRequestCorrelation, Env, ItlEvidenceSource, OutputTokenCountSource,
QualityIssueCounts, RequestItlEvidence, RequestRecord, RunRecord, Scenario, Slo,
TokenLengthStats, WarmupSummary, BENCHMARK_CELL_ID_HEADER, BENCHMARK_PHASE_HEADER,
BENCHMARK_REPEAT_INDEX_HEADER, BENCHMARK_REQUEST_INDEX_HEADER, BENCHMARK_RUN_ID_HEADER,
};
use ferrum_server::chat_template::ReasoningEffort;
use ferrum_types::Result;
use rand::{rngs::StdRng, Rng, SeedableRng};
use serde::Deserialize;
use sha2::{Digest, Sha256};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Semaphore;
use tokio_stream::StreamExt;
use uuid::Uuid;
use crate::config::CliConfig;
mod decode_isolation;
mod sampling;
use decode_isolation::{BenchServeWorkload, DecodeIsolationArgs};
use sampling::BenchSamplingArgs;
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
pub enum BenchTargetBackend {
Cpu,
Metal,
Cuda,
}
impl BenchTargetBackend {
fn as_str(self) -> &'static str {
match self {
Self::Cpu => "cpu",
Self::Metal => "metal",
Self::Cuda => "cuda",
}
}
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, ValueEnum)]
pub enum BenchHttpConnectionMode {
#[default]
Pooled,
Fresh,
}
impl BenchHttpConnectionMode {
fn as_str(self) -> &'static str {
match self {
Self::Pooled => "pooled",
Self::Fresh => "fresh",
}
}
}
#[derive(Args, Clone)]
pub struct BenchServeCommand {
#[arg(long)]
pub base_url: String,
#[arg(long)]
pub model: String,
#[arg(long)]
pub tokenizer: PathBuf,
#[arg(long, value_enum)]
pub target_backend: Option<BenchTargetBackend>,
#[arg(long, value_enum, default_value = "pooled")]
pub http_connection_mode: BenchHttpConnectionMode,
#[arg(long, value_enum, default_value = "standard")]
pub scenario: BenchServeWorkload,
#[command(flatten)]
pub decode_isolation: DecodeIsolationArgs,
#[command(flatten)]
pub sampling: BenchSamplingArgs,
#[arg(long, default_value_t = 32, alias = "max-concurrency")]
pub concurrency: u32,
#[arg(long, value_delimiter = ',')]
pub concurrency_sweep: Vec<u32>,
#[arg(long)]
pub request_rate: Option<f64>,
#[arg(long, default_value = "random")]
pub dataset: String,
#[arg(long, default_value_t = 256)]
pub random_input_len: usize,
#[arg(long, default_value_t = 128)]
pub random_output_len: usize,
#[arg(long)]
pub ignore_eos: bool,
#[arg(long, action = clap::ArgAction::Set)]
pub enable_thinking: Option<bool>,
#[arg(long, value_name = "LEVEL")]
pub reasoning_effort: Option<ReasoningEffort>,
#[arg(long)]
pub sharegpt_path: Option<PathBuf>,
#[arg(long, default_value_t = 1024)]
pub shared_prefix_len: usize,
#[arg(long, default_value_t = 64)]
pub shared_suffix_len: usize,
#[arg(long, default_value_t = 100)]
pub num_prompts: u32,
#[arg(long, default_value_t = 10)]
pub warmup_requests: u32,
#[arg(long, default_value_t = 1)]
pub n_repeats: u32,
#[arg(long, value_parser = parse_slo)]
pub goodput: Option<Slo>,
#[arg(long, default_value_t = 600.0)]
pub timeout: f64,
#[arg(long)]
pub fail_on_error: bool,
#[arg(long)]
pub max_error_rate: Option<f64>,
#[arg(long)]
pub require_ci: bool,
#[arg(long)]
pub seed: Option<u64>,
#[arg(long, default_value = "json")]
pub output: String,
#[arg(long, alias = "result-file")]
pub out: Option<PathBuf>,
#[arg(long)]
pub hw_id: Option<String>,
#[arg(long)]
pub commit_sha: Option<String>,
#[arg(long)]
pub tag: Option<String>,
}
pub(super) fn parse_slo(s: &str) -> std::result::Result<Slo, String> {
let mut ttft: Option<f64> = None;
let mut tpot: Option<f64> = None;
let mut e2el: Option<f64> = None;
for tok in s.split(|c: char| c == ',' || c.is_whitespace()) {
if tok.is_empty() {
continue;
}
let (k, v) = tok
.split_once(':')
.ok_or_else(|| format!("bad SLO token '{tok}', expected key:value"))?;
let v: f64 = v.parse().map_err(|e| format!("bad SLO value '{v}': {e}"))?;
if !v.is_finite() || v <= 0.0 {
return Err(format!(
"bad SLO value '{v}': expected a positive finite number"
));
}
match k {
"ttft" => ttft = Some(v),
"tpot" => tpot = Some(v),
"e2el" | "e2e" => e2el = Some(v),
other => return Err(format!("unknown SLO key '{other}'")),
}
}
Ok(Slo {
ttft_p99_ms: ttft.ok_or("missing ttft in --goodput")?,
tpot_p99_ms: tpot.ok_or("missing tpot in --goodput")?,
e2e_p99_ms: e2el.ok_or("missing e2el in --goodput")?,
})
}
#[derive(Debug, Deserialize)]
struct OpenAiStreamChunk {
id: Option<String>,
choices: Option<Vec<OpenAiStreamChoice>>,
usage: Option<OpenAiUsage>,
error: Option<OpenAiStreamError>,
}
#[derive(Debug, Deserialize)]
struct OpenAiStreamError {
message: String,
#[serde(rename = "type")]
error_type: Option<String>,
code: Option<serde_json::Value>,
}
#[derive(Debug, Deserialize)]
struct OpenAiStreamChoice {
delta: Option<OpenAiStreamDelta>,
}
#[derive(Debug, Deserialize)]
struct OpenAiStreamDelta {
content: Option<String>,
reasoning: Option<String>,
reasoning_content: Option<String>,
}
#[derive(Debug, Deserialize)]
struct OpenAiUsage {
prompt_tokens: Option<u32>,
completion_tokens: Option<u32>,
}
#[derive(Clone)]
struct PromptCase {
text: String,
input_tokens: u32,
sha256: String,
}
async fn stream_one(
client: &reqwest::Client,
base_url: &str,
model: &str,
prompt: PromptCase,
max_tokens: usize,
ignore_eos: bool,
enable_thinking: Option<bool>,
reasoning_effort: Option<ReasoningEffort>,
sampling: HttpRequestSampling,
timeout_s: f64,
correlation: BenchmarkRequestCorrelation,
) -> RequestRecord {
stream_one_observed(
client,
base_url,
model,
prompt,
max_tokens,
ignore_eos,
enable_thinking,
reasoning_effort,
sampling,
timeout_s,
correlation,
None,
)
.await
.record
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub(super) struct DecodeStreamProgress {
pub output_events: u32,
pub finished: bool,
pub first_output_at: Option<Instant>,
pub last_output_at: Option<Instant>,
}
pub(super) struct ObservedRequest {
pub record: RequestRecord,
pub started_at: Instant,
pub output_event_times: Vec<Instant>,
}
async fn stream_one_observed(
client: &reqwest::Client,
base_url: &str,
model: &str,
prompt: PromptCase,
max_tokens: usize,
ignore_eos: bool,
enable_thinking: Option<bool>,
reasoning_effort: Option<ReasoningEffort>,
sampling: HttpRequestSampling,
timeout_s: f64,
correlation: BenchmarkRequestCorrelation,
progress: Option<tokio::sync::watch::Sender<DecodeStreamProgress>>,
) -> ObservedRequest {
let PromptCase {
text,
input_tokens,
sha256: prompt_sha256,
} = prompt;
let body = chat_completion_body(
model,
&text,
max_tokens,
ignore_eos,
enable_thinking,
reasoning_effort,
sampling,
);
let start = Instant::now();
let mut state = StreamState::for_prompt(
start,
input_tokens,
prompt_sha256.clone(),
Some(correlation.clone()),
);
let resp = match client
.post(format!("{}/v1/chat/completions", base_url))
.header(BENCHMARK_RUN_ID_HEADER, &correlation.benchmark_run_id)
.header(BENCHMARK_CELL_ID_HEADER, &correlation.cell_id)
.header(
BENCHMARK_REPEAT_INDEX_HEADER,
correlation.repeat_index.to_string(),
)
.header(BENCHMARK_PHASE_HEADER, correlation.phase.as_str())
.header(
BENCHMARK_REQUEST_INDEX_HEADER,
correlation.request_index.to_string(),
)
.json(&body)
.timeout(Duration::from_secs_f64(timeout_s))
.send()
.await
{
Ok(r) => r,
Err(e) => {
eprintln!(
"[err] post prompt_sha256={prompt_sha256} input_tokens={} \
timeout={} connect={} request={} body={} decode={}: display={e}; debug={e:?}",
input_tokens,
e.is_timeout(),
e.is_connect(),
e.is_request(),
e.is_body(),
e.is_decode(),
);
let mut quality_issues = QualityIssueCounts::default();
quality_issues.malformed_stream = 1;
publish_stream_finished(&progress, 0);
return ObservedRequest {
record: failed_record(input_tokens, start, quality_issues, correlation),
started_at: start,
output_event_times: vec![],
};
}
};
if !resp.status().is_success() {
let status = resp.status();
let txt = resp.text().await.unwrap_or_default();
eprintln!(
"[err] http prompt_sha256={prompt_sha256} {status}: {}",
clipped_debug_text(&txt, 200)
);
let mut quality_issues = QualityIssueCounts::default();
if status.as_u16() == 500 {
quality_issues.http_500 = 1;
}
if looks_like_panic(&txt) {
quality_issues.panic = 1;
}
publish_stream_finished(&progress, 0);
return ObservedRequest {
record: failed_record(input_tokens, start, quality_issues, correlation),
started_at: start,
output_event_times: vec![],
};
}
let mut stream = resp.bytes_stream();
let mut sse = SseLineBuffer::default();
while let Some(chunk) = stream.next().await {
let chunk = match chunk {
Ok(c) => c,
Err(e) => {
eprintln!("[err] stream prompt_sha256={prompt_sha256}: display={e}; debug={e:?}");
state.stream_error = Some(e.to_string());
state.quality_issues.malformed_stream = 1;
break;
}
};
let before_output_events = state.output_delta_events;
sse.push(&chunk, &mut state);
if let Some(progress) = &progress {
progress.send_replace(DecodeStreamProgress {
output_events: state.output_delta_events,
finished: false,
first_output_at: state.output_event_times.first().copied(),
last_output_at: state.output_event_times.last().copied(),
});
}
state.note_transport_chunk(
state
.output_delta_events
.saturating_sub(before_output_events),
);
}
sse.finish(&mut state);
let output_events = state.output_delta_events;
let observed = state.finish_observed();
if let Some(progress) = &progress {
progress.send_replace(DecodeStreamProgress {
output_events,
finished: true,
first_output_at: observed.output_event_times.first().copied(),
last_output_at: observed.output_event_times.last().copied(),
});
}
observed
}
fn publish_stream_finished(
progress: &Option<tokio::sync::watch::Sender<DecodeStreamProgress>>,
output_events: u32,
) {
if let Some(progress) = progress {
progress.send_replace(DecodeStreamProgress {
output_events,
finished: true,
first_output_at: None,
last_output_at: None,
});
}
}
#[derive(Default)]
struct SseLineBuffer {
pending: Vec<u8>,
}
impl SseLineBuffer {
fn push(&mut self, chunk: &[u8], state: &mut StreamState) {
self.pending.extend_from_slice(chunk);
let mut consumed = 0;
while let Some(relative_newline) = self.pending[consumed..]
.iter()
.position(|byte| *byte == b'\n')
{
let newline = consumed + relative_newline;
let line = self.pending[consumed..newline].to_vec();
consumed = newline + 1;
Self::process_line(&line, state);
}
if consumed > 0 {
self.pending.drain(..consumed);
}
}
fn finish(&mut self, state: &mut StreamState) {
if !self.pending.is_empty() {
let line = std::mem::take(&mut self.pending);
Self::process_line(&line, state);
}
}
fn process_line(raw_line: &[u8], state: &mut StreamState) {
let raw_line = raw_line.strip_suffix(b"\r").unwrap_or(raw_line);
let line = match std::str::from_utf8(raw_line) {
Ok(line) => line.trim(),
Err(error) => {
let message = format!("invalid UTF-8 in SSE line: {error}");
eprintln!("[err] prompt_sha256={} {message}", state.prompt_sha256);
state.stream_error = Some(message);
state.quality_issues.malformed_stream = 1;
state.quality_issues.bad_output = 1;
return;
}
};
let Some(payload) = line.strip_prefix("data:").map(str::trim) else {
return;
};
if payload == "[DONE]" {
state.done_count = state
.done_count
.checked_add(1)
.expect("SSE done count overflow");
if state.done_count > 1 {
state.quality_issues.duplicate_done = 1;
}
return;
}
if let Err(error) = state.handle_payload(payload) {
eprintln!(
"[err] malformed sse json prompt_sha256={}: {error}",
state.prompt_sha256
);
state.stream_error = Some(error);
state.quality_issues.malformed_stream = 1;
}
}
}
fn chat_completion_body(
model: &str,
prompt_text: &str,
max_tokens: usize,
ignore_eos: bool,
enable_thinking: Option<bool>,
reasoning_effort: Option<ReasoningEffort>,
sampling: HttpRequestSampling,
) -> serde_json::Value {
let mut body = serde_json::json!({
"model": model,
"messages": [{"role": "user", "content": prompt_text}],
"max_tokens": max_tokens,
"stream": true,
"stream_options": {"include_usage": true},
});
let serde_json::Value::Object(sampling) =
serde_json::to_value(sampling).expect("validated HTTP sampling must serialize")
else {
unreachable!("HTTP sampling serializes as an object");
};
body.as_object_mut()
.expect("request body is an object")
.extend(sampling);
let mut chat_template_kwargs = serde_json::Map::new();
if let Some(enable_thinking) = enable_thinking {
chat_template_kwargs.insert(
"enable_thinking".to_string(),
serde_json::json!(enable_thinking),
);
}
if let Some(reasoning_effort) = reasoning_effort {
chat_template_kwargs.insert(
"reasoning_effort".to_string(),
serde_json::json!(reasoning_effort),
);
}
if !chat_template_kwargs.is_empty() {
body["chat_template_kwargs"] = serde_json::Value::Object(chat_template_kwargs);
}
if ignore_eos {
body["ignore_eos"] = serde_json::json!(true);
}
body
}
fn failed_record(
input_tokens: u32,
start: Instant,
quality_issues: QualityIssueCounts,
benchmark_correlation: BenchmarkRequestCorrelation,
) -> RequestRecord {
RequestRecord {
benchmark_correlation: Some(benchmark_correlation),
server_request_id: None,
success: false,
ttft_ms: 0.0,
e2e_ms: start.elapsed().as_secs_f64() * 1000.0,
input_tokens,
server_input_tokens: None,
output_tokens: 0,
output_token_count_source: OutputTokenCountSource::None,
itl_evidence: RequestItlEvidence::failed(ItlEvidenceSource::SseDeltaEvents),
quality_issues,
itl_ms: vec![],
}
}
fn join_failed_record(
input_tokens: u32,
benchmark_correlation: BenchmarkRequestCorrelation,
) -> RequestRecord {
let mut quality_issues = QualityIssueCounts::default();
quality_issues.panic = 1;
RequestRecord {
benchmark_correlation: Some(benchmark_correlation),
server_request_id: None,
success: false,
ttft_ms: 0.0,
e2e_ms: 0.0,
input_tokens,
server_input_tokens: None,
output_tokens: 0,
output_token_count_source: OutputTokenCountSource::None,
itl_evidence: RequestItlEvidence::failed(ItlEvidenceSource::SseDeltaEvents),
quality_issues,
itl_ms: vec![],
}
}
async fn collect_measured_handles(
handles: Vec<(
u32,
BenchmarkRequestCorrelation,
tokio::task::JoinHandle<RequestRecord>,
)>,
) -> Vec<RequestRecord> {
let mut records = Vec::with_capacity(handles.len());
for (input_tokens, correlation, handle) in handles {
match handle.await {
Ok(record) => records.push(record),
Err(error) => {
eprintln!("[err] measured request task: {error}");
records.push(join_failed_record(input_tokens, correlation));
}
}
}
records
}
struct StreamState {
start: Instant,
input_tokens: u32,
usage_prompt_tokens: Option<u32>,
first_token_time: Option<Instant>,
last_token_time: Option<Instant>,
output_delta_events: u32,
usage_completion_tokens: Option<u32>,
itl_ms: Vec<f64>,
transport_coalesced_output_chunks: u32,
done_count: u32,
stream_error: Option<String>,
quality_issues: QualityIssueCounts,
prompt_sha256: String,
benchmark_correlation: Option<BenchmarkRequestCorrelation>,
server_request_id: Option<String>,
output_event_times: Vec<Instant>,
}
impl StreamState {
#[cfg(test)]
fn new(start: Instant, input_tokens: u32) -> Self {
Self::for_prompt(start, input_tokens, "unknown".to_string(), None)
}
fn for_prompt(
start: Instant,
input_tokens: u32,
prompt_sha256: String,
benchmark_correlation: Option<BenchmarkRequestCorrelation>,
) -> Self {
Self {
start,
input_tokens,
usage_prompt_tokens: None,
first_token_time: None,
last_token_time: None,
output_delta_events: 0,
usage_completion_tokens: None,
itl_ms: Vec::new(),
transport_coalesced_output_chunks: 0,
done_count: 0,
stream_error: None,
quality_issues: QualityIssueCounts::default(),
prompt_sha256,
benchmark_correlation,
server_request_id: None,
output_event_times: Vec::new(),
}
}
fn handle_payload(&mut self, payload: &str) -> std::result::Result<(), String> {
let chunk: OpenAiStreamChunk =
serde_json::from_str(payload).map_err(|e| format!("{e}: {payload}"))?;
if let Some(id) = chunk.id.filter(|id| !id.is_empty()) {
if self
.server_request_id
.as_ref()
.is_some_and(|observed| observed != &id)
{
return Err(format!(
"OpenAI stream response changed request id from {} to {id}",
self.server_request_id.as_deref().unwrap_or_default()
));
}
self.server_request_id = Some(id);
}
if let Some(error) = chunk.error {
return Err(format!(
"OpenAI stream error type={} code={} message={}",
error.error_type.as_deref().unwrap_or("unknown"),
error
.code
.as_ref()
.map(serde_json::Value::to_string)
.unwrap_or_else(|| "null".to_owned()),
error.message
));
}
if let Some(usage) = chunk.usage {
if let Some(tokens) = usage.prompt_tokens {
self.usage_prompt_tokens = Some(tokens);
}
if let Some(tokens) = usage.completion_tokens {
self.usage_completion_tokens = Some(tokens);
}
}
if let Some(choices) = chunk.choices {
if let Some(first) = choices.into_iter().next() {
if let Some(delta) = first.delta {
if let Some(text) = first_non_empty_delta_text(&delta) {
let now = Instant::now();
if self.first_token_time.is_none() {
self.first_token_time = Some(now);
} else if let Some(prev) = self.last_token_time {
self.itl_ms.push((now - prev).as_secs_f64() * 1000.0);
}
self.last_token_time = Some(now);
self.output_event_times.push(now);
self.output_delta_events = self
.output_delta_events
.checked_add(1)
.expect("stream output event count overflow");
if let Some(reason) = bad_output_reason(text) {
eprintln!(
"[err] bad output prompt_sha256={} {reason}: {}",
self.prompt_sha256,
clipped_debug_text(text, 160)
);
self.quality_issues.bad_output = 1;
}
if looks_like_panic(text) {
self.quality_issues.panic = 1;
}
}
}
}
}
Ok(())
}
fn note_transport_chunk(&mut self, output_events: u32) {
if output_events > 1 {
self.transport_coalesced_output_chunks = self
.transport_coalesced_output_chunks
.checked_add(1)
.expect("coalesced output transport chunk count overflow");
}
}
fn finish(mut self) -> RequestRecord {
let (output_tokens, source) = match self.usage_completion_tokens {
Some(tokens) => (tokens, OutputTokenCountSource::Usage),
None if self.output_delta_events > 0 => (
self.output_delta_events,
OutputTokenCountSource::StreamChunks,
),
None => (0, OutputTokenCountSource::None),
};
if self.done_count == 0 {
self.quality_issues.missing_done = 1;
} else if self.done_count > 1 {
self.quality_issues.duplicate_done = 1;
}
if output_tokens == 0 {
self.quality_issues.zero_output_tokens = 1;
}
let e2e_ms = self.start.elapsed().as_secs_f64() * 1000.0;
let ttft_ms = self
.first_token_time
.map(|t| t.duration_since(self.start).as_secs_f64() * 1000.0)
.unwrap_or(e2e_ms);
let success = self.done_count == 1
&& output_tokens > 0
&& self.stream_error.is_none()
&& self.quality_issues.request_error_count() == 0;
let observed_intervals =
u32::try_from(self.itl_ms.len()).expect("stream ITL interval count overflow");
let itl_evidence = RequestItlEvidence::sse(
success,
self.output_delta_events,
self.usage_completion_tokens,
observed_intervals,
self.transport_coalesced_output_chunks,
);
RequestRecord {
benchmark_correlation: self.benchmark_correlation,
server_request_id: self.server_request_id,
success,
ttft_ms,
e2e_ms,
input_tokens: self.input_tokens,
server_input_tokens: self.usage_prompt_tokens,
output_tokens,
output_token_count_source: source,
itl_evidence,
quality_issues: self.quality_issues,
itl_ms: self.itl_ms,
}
}
fn finish_observed(mut self) -> ObservedRequest {
let started_at = self.start;
let output_event_times = std::mem::take(&mut self.output_event_times);
let record = self.finish();
ObservedRequest {
record,
started_at,
output_event_times,
}
}
}
fn first_non_empty_delta_text(delta: &OpenAiStreamDelta) -> Option<&str> {
delta
.content
.as_deref()
.filter(|s| !s.is_empty())
.or_else(|| delta.reasoning.as_deref().filter(|s| !s.is_empty()))
.or_else(|| delta.reasoning_content.as_deref().filter(|s| !s.is_empty()))
}
#[cfg(test)]
fn has_bad_output_text(text: &str) -> bool {
bad_output_reason(text).is_some()
}
fn bad_output_reason(text: &str) -> Option<&'static str> {
const BAD_FRAGMENTS: &[(&str, &str)] = &[
("<unk>", "reserved-token"),
("[PAD", "reserved-token"),
("<pad>", "reserved-token"),
("<|endoftext|>", "reserved-token"),
("<|im_start|>", "reserved-token"),
("<|im_end|>", "reserved-token"),
("<|reserved_special_token", "reserved-token"),
("\u{fffd}", "invalid-utf8"),
];
for (fragment, reason) in BAD_FRAGMENTS {
if text.contains(fragment) {
return Some(reason);
}
}
contains_mojibake_sequence(text).then_some("mojibake")
}
fn contains_mojibake_sequence(text: &str) -> bool {
let mut chars = text.chars().peekable();
while let Some(ch) = chars.next() {
match ch {
'\u{00c2}' | '\u{00c3}' => {
if chars.peek().is_some_and(|next| !next.is_ascii()) {
return true;
}
}
'\u{00e2}' => {
if chars.peek().is_some_and(|next| *next == '\u{20ac}') {
return true;
}
}
_ => {}
}
}
false
}
fn clipped_debug_text(text: &str, max_chars: usize) -> String {
text.escape_debug().take(max_chars).collect()
}
fn sha256_hex(bytes: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(bytes);
format!("{:x}", hasher.finalize())
}
fn looks_like_panic(text: &str) -> bool {
let lower = text.to_ascii_lowercase();
lower.contains("panicked at") || lower.contains("thread '") && lower.contains("panicked")
}
fn gen_random_prompt(
tok: &tokenizers::Tokenizer,
n_tokens: usize,
rng: &mut (impl Rng + ?Sized),
) -> String {
if let Some(text) = gen_random_prompt_target_len(tok, n_tokens, rng) {
return text;
}
" x".repeat(n_tokens)
}
fn generated_prompt_is_safe(text: &str) -> bool {
!text.is_empty()
&& !text.contains("<|")
&& bad_output_reason(text).is_none()
&& text
.chars()
.all(|ch| !ch.is_control() || matches!(ch, '\n' | '\r' | '\t'))
}
fn sample_safe_token_ids(
tok: &tokenizers::Tokenizer,
count: usize,
rng: &mut (impl Rng + ?Sized),
lo: u32,
hi: u32,
) -> Option<Vec<u32>> {
let mut ids = Vec::with_capacity(count);
for _ in 0..count {
let mut selected = None;
for _ in 0..256 {
let id = rng.random_range(lo..=hi);
let Ok(piece) = tok.decode(&[id], false) else {
continue;
};
if generated_prompt_is_safe(&piece) {
selected = Some(id);
break;
}
}
ids.push(selected?);
}
Some(ids)
}
fn gen_random_prompt_target_len(
tok: &tokenizers::Tokenizer,
target_len: usize,
rng: &mut (impl Rng + ?Sized),
) -> Option<String> {
if target_len == 0 {
return Some(String::new());
}
let vocab_size = tok.get_vocab_size(false) as u32;
let lo: u32 = 256.min(vocab_size.saturating_sub(1));
let hi: u32 = vocab_size.saturating_sub(1);
if hi < lo {
return None;
}
let mut sample_len = target_len;
let mut best_under: Option<(usize, String)> = None;
let mut best_any: Option<(usize, String)> = None;
for _ in 0..64 {
let ids = sample_safe_token_ids(tok, sample_len, rng, lo, hi)?;
let text = match tok.decode(&ids, false) {
Ok(text) if generated_prompt_is_safe(&text) => text,
_ => continue,
};
let len = match token_count(tok, &text) {
Some(len) => len,
None => continue,
};
if len == target_len {
return Some(text);
}
let delta = len.abs_diff(target_len);
if best_any
.as_ref()
.map(|(best_len, _)| delta < best_len.abs_diff(target_len))
.unwrap_or(true)
{
best_any = Some((len, text.clone()));
}
if len < target_len
&& best_under
.as_ref()
.map(|(best_len, _)| len > *best_len)
.unwrap_or(true)
{
best_under = Some((len, text));
}
sample_len = if len > target_len {
sample_len.saturating_sub(len - target_len).max(1)
} else {
sample_len + (target_len - len).max(1)
};
}
if let Some((len, text)) = best_under {
return fill_random_prompt_to_len(tok, text, len, target_len, rng, lo, hi);
}
best_any.map(|(_, text)| text)
}
fn token_count(tok: &tokenizers::Tokenizer, text: &str) -> Option<usize> {
tok.encode(text, false).ok().map(|enc| enc.len())
}
fn fill_random_prompt_to_len(
tok: &tokenizers::Tokenizer,
mut text: String,
mut len: usize,
target_len: usize,
rng: &mut (impl Rng + ?Sized),
lo: u32,
hi: u32,
) -> Option<String> {
for _ in 0..target_len.saturating_mul(8).max(16) {
if len == target_len {
return Some(text);
}
let next = random_one_token_extension(tok, &text, len, rng, lo, hi)?;
text = next.0;
len = next.1;
}
(len == target_len).then_some(text)
}
fn random_one_token_extension(
tok: &tokenizers::Tokenizer,
base: &str,
base_len: usize,
rng: &mut (impl Rng + ?Sized),
lo: u32,
hi: u32,
) -> Option<(String, usize)> {
for _ in 0..128 {
let id = rng.random_range(lo..=hi);
if let Some(candidate) = append_decoded_piece(tok, base, base_len, id) {
return Some(candidate);
}
}
for piece in [" x", " y", " z", ".", ",", "\n"] {
let candidate = format!("{base}{piece}");
if token_count(tok, &candidate) == Some(base_len + 1) {
return Some((candidate, base_len + 1));
}
}
None
}
fn append_decoded_piece(
tok: &tokenizers::Tokenizer,
base: &str,
base_len: usize,
id: u32,
) -> Option<(String, usize)> {
let piece = tok.decode(&[id], false).ok()?;
if !generated_prompt_is_safe(&piece) {
return None;
}
let candidate = format!("{base}{piece}");
(generated_prompt_is_safe(&candidate) && token_count(tok, &candidate) == Some(base_len + 1))
.then_some((candidate, base_len + 1))
}
fn build_prompts(
cmd: &BenchServeCommand,
tok: &tokenizers::Tokenizer,
rng: &mut (impl Rng + ?Sized),
count: usize,
) -> Result<Vec<PromptCase>> {
match cmd.dataset.as_str() {
"random" => (0..count)
.map(|_| {
let text = gen_random_prompt(tok, cmd.random_input_len, rng);
prompt_case(tok, text)
})
.collect(),
"shared-prefix" => gen_shared_prefix_prompts(
tok,
count,
cmd.shared_prefix_len,
cmd.shared_suffix_len,
rng,
),
"sharegpt" => {
let p = cmd.sharegpt_path.as_ref().ok_or_else(|| {
ferrum_types::FerrumError::model("--dataset sharegpt requires --sharegpt-path PATH")
})?;
load_sharegpt_prompts(p, tok, count, rng)
}
other => Err(ferrum_types::FerrumError::model(format!(
"unknown --dataset '{}': allowed values are random, sharegpt, shared-prefix",
other
))),
}
}
fn prompt_case(tok: &tokenizers::Tokenizer, text: String) -> Result<PromptCase> {
let encoding = tok
.encode(text.as_str(), false)
.map_err(|e| ferrum_types::FerrumError::model(format!("tokenize generated prompt: {e}")))?;
let input_tokens = u32::try_from(encoding.len()).map_err(|_| {
ferrum_types::FerrumError::model("generated prompt token count exceeds report capacity")
})?;
let sha256 = sha256_hex(text.as_bytes());
Ok(PromptCase {
text,
input_tokens,
sha256,
})
}
fn gen_shared_prefix_prompts(
tok: &tokenizers::Tokenizer,
count: usize,
prefix_len: usize,
suffix_len: usize,
rng: &mut (impl Rng + ?Sized),
) -> Result<Vec<PromptCase>> {
let prefix = gen_random_prompt(tok, prefix_len, rng);
(0..count)
.map(|_| {
let suffix = gen_random_prompt(tok, suffix_len, rng);
prompt_case(tok, format!("{prefix}\n{suffix}"))
})
.collect()
}
fn load_sharegpt_prompts(
path: &std::path::Path,
tok: &tokenizers::Tokenizer,
count: usize,
rng: &mut (impl Rng + ?Sized),
) -> Result<Vec<PromptCase>> {
use std::io::BufRead;
let f = std::fs::File::open(path).map_err(|e| {
ferrum_types::FerrumError::model(format!("open sharegpt {}: {e}", path.display()))
})?;
let mut prompts: Vec<String> = Vec::new();
for (idx, line) in std::io::BufReader::new(f).lines().enumerate() {
let line = line.map_err(|e| {
ferrum_types::FerrumError::model(format!("read line {idx} of {}: {e}", path.display()))
})?;
if line.trim().is_empty() {
continue;
}
let v: serde_json::Value = match serde_json::from_str(&line) {
Ok(v) => v,
Err(e) => {
eprintln!("[warn] sharegpt line {idx}: skip (parse error: {e})");
continue;
}
};
let prompt: Option<String> = v
.get("conversations")
.and_then(|c| c.as_array())
.and_then(|arr| {
arr.iter()
.find(|t| t.get("from").and_then(|f| f.as_str()) == Some("human"))
.and_then(|t| {
t.get("value")
.and_then(|x| x.as_str())
.map(|s| s.to_string())
})
})
.or_else(|| {
v.get("input")
.and_then(|s| s.as_str())
.map(|s| s.to_string())
});
if let Some(p) = prompt {
if !p.is_empty() {
prompts.push(p);
}
}
}
if prompts.is_empty() {
return Err(ferrum_types::FerrumError::model(format!(
"sharegpt {}: no usable prompts found",
path.display()
)));
}
let mut out = Vec::with_capacity(count);
for _ in 0..count {
let idx = rng.random_range(0..prompts.len());
out.push(prompt_case(tok, prompts[idx].clone())?);
}
Ok(out)
}
struct RunContext {
client: Arc<reqwest::Client>,
base_url: Arc<String>,
model: Arc<String>,
max_out: usize,
ignore_eos: bool,
enable_thinking: Option<bool>,
reasoning_effort: Option<ReasoningEffort>,
sampling: HttpRequestSampling,
timeout_s: f64,
benchmark_run_id: Arc<String>,
}
fn benchmark_request_correlation(
benchmark_run_id: &str,
cell_id: &str,
repeat_index: u32,
phase: BenchmarkPhase,
request_index: usize,
) -> BenchmarkRequestCorrelation {
BenchmarkRequestCorrelation::new(
benchmark_run_id.to_string(),
cell_id.to_string(),
repeat_index,
phase,
u32::try_from(request_index).expect("benchmark request index overflow"),
)
.expect("generated benchmark correlation must be valid")
}
fn summarize_warmup(
expected: usize,
records: &[RequestRecord],
join_errors: usize,
) -> WarmupSummary {
assert_eq!(records.len() + join_errors, expected);
let completed = records.iter().filter(|record| record.success).count() as u32;
let mut quality_issues = QualityIssueCounts::default();
for record in records {
quality_issues.add_assign(&record.quality_issues);
}
quality_issues.panic = quality_issues
.panic
.checked_add(u32::try_from(join_errors).expect("warmup join error count overflow"))
.expect("warmup panic count overflow");
let expected = u32::try_from(expected).expect("warmup expected count overflow");
WarmupSummary {
expected,
completed,
errored: expected
.checked_sub(completed)
.expect("warmup completed count exceeds expected"),
quality_issues,
}
}
async fn run_closed_loop(
ctx: &RunContext,
prompts: Vec<PromptCase>,
warmup_requests: u32,
concurrency: u32,
cell_id: &str,
repeat_index: u32,
) -> RunRecord {
let n_warmup = warmup_requests as usize;
let total = prompts.len();
assert!(
total > n_warmup,
"num_prompts ({total}) must exceed warmup_requests ({n_warmup})"
);
let warmup = {
let sem = Arc::new(Semaphore::new(concurrency as usize));
let mut handles = Vec::new();
for (request_index, prompt) in prompts.iter().take(n_warmup).enumerate() {
let permit = sem.clone().acquire_owned().await.expect("semaphore");
let ctx_c = ctx.clone_inner();
let p = prompt.clone();
let correlation = benchmark_request_correlation(
&ctx.benchmark_run_id,
cell_id,
repeat_index,
BenchmarkPhase::Warmup,
request_index,
);
handles.push(tokio::spawn(async move {
let _g = permit;
stream_one(
&ctx_c.client,
&ctx_c.base_url,
&ctx_c.model,
p,
ctx_c.max_out,
ctx_c.ignore_eos,
ctx_c.enable_thinking,
ctx_c.reasoning_effort,
ctx_c.sampling,
ctx_c.timeout_s,
correlation,
)
.await
}));
}
let mut records = Vec::with_capacity(n_warmup);
let mut join_errors = 0;
for handle in handles {
match handle.await {
Ok(record) => records.push(record),
Err(_) => join_errors += 1,
}
}
summarize_warmup(n_warmup, &records, join_errors)
};
let sem = Arc::new(Semaphore::new(concurrency as usize));
let start = Instant::now();
let mut handles = Vec::with_capacity(total - n_warmup);
for (request_index, prompt) in prompts.into_iter().skip(n_warmup).enumerate() {
let input_tokens = prompt.input_tokens;
let permit = sem.clone().acquire_owned().await.expect("semaphore");
let ctx_c = ctx.clone_inner();
let correlation = benchmark_request_correlation(
&ctx.benchmark_run_id,
cell_id,
repeat_index,
BenchmarkPhase::Measured,
request_index,
);
handles.push((
input_tokens,
correlation.clone(),
tokio::spawn(async move {
let _g = permit;
stream_one(
&ctx_c.client,
&ctx_c.base_url,
&ctx_c.model,
prompt,
ctx_c.max_out,
ctx_c.ignore_eos,
ctx_c.enable_thinking,
ctx_c.reasoning_effort,
ctx_c.sampling,
ctx_c.timeout_s,
correlation,
)
.await
}),
));
}
let records = collect_measured_handles(handles).await;
let duration_s = start.elapsed().as_secs_f64();
RunRecord {
records,
expected_requests: u32::try_from(total - n_warmup)
.expect("measured request count overflow"),
duration_s,
warmup,
}
}
async fn run_open_loop(
ctx: &RunContext,
prompts: Vec<PromptCase>,
warmup_requests: u32,
rate: f64,
cell_id: &str,
repeat_index: u32,
) -> RunRecord {
let n_warmup = warmup_requests as usize;
let total = prompts.len();
assert!(total > n_warmup);
let mut warmup_records = Vec::with_capacity(n_warmup);
for (request_index, prompt) in prompts.iter().take(n_warmup).enumerate() {
warmup_records.push(
stream_one(
&ctx.client,
&ctx.base_url,
&ctx.model,
prompt.clone(),
ctx.max_out,
ctx.ignore_eos,
ctx.enable_thinking,
ctx.reasoning_effort,
ctx.sampling,
ctx.timeout_s,
benchmark_request_correlation(
&ctx.benchmark_run_id,
cell_id,
repeat_index,
BenchmarkPhase::Warmup,
request_index,
),
)
.await,
);
}
let warmup = summarize_warmup(n_warmup, &warmup_records, 0);
let mut rng = rand::rng();
let measurement_count = total - n_warmup;
let schedule = poisson_arrival_times(rate, measurement_count, &mut rng);
let start = Instant::now();
let mut handles = Vec::with_capacity(measurement_count);
for (i, prompt) in prompts.into_iter().skip(n_warmup).enumerate() {
let target = schedule[i];
let now = start.elapsed().as_secs_f64();
if target > now {
tokio::time::sleep(Duration::from_secs_f64(target - now)).await;
}
let ctx_c = ctx.clone_inner();
let input_tokens = prompt.input_tokens;
let correlation = benchmark_request_correlation(
&ctx.benchmark_run_id,
cell_id,
repeat_index,
BenchmarkPhase::Measured,
i,
);
handles.push((
input_tokens,
correlation.clone(),
tokio::spawn(async move {
stream_one(
&ctx_c.client,
&ctx_c.base_url,
&ctx_c.model,
prompt,
ctx_c.max_out,
ctx_c.ignore_eos,
ctx_c.enable_thinking,
ctx_c.reasoning_effort,
ctx_c.sampling,
ctx_c.timeout_s,
correlation,
)
.await
}),
));
}
let records = collect_measured_handles(handles).await;
let duration_s = start.elapsed().as_secs_f64();
RunRecord {
records,
expected_requests: u32::try_from(measurement_count)
.expect("measured request count overflow"),
duration_s,
warmup,
}
}
impl RunContext {
fn clone_inner(&self) -> Self {
Self {
client: self.client.clone(),
base_url: self.base_url.clone(),
model: self.model.clone(),
max_out: self.max_out,
ignore_eos: self.ignore_eos,
enable_thinking: self.enable_thinking,
reasoning_effort: self.reasoning_effort,
sampling: self.sampling,
timeout_s: self.timeout_s,
benchmark_run_id: self.benchmark_run_id.clone(),
}
}
}
fn build_env(cmd: &BenchServeCommand, features: Vec<String>) -> Env {
let commit_sha = cmd
.commit_sha
.clone()
.or_else(|| {
std::process::Command::new("git")
.args(["rev-parse", "--short", "HEAD"])
.output()
.ok()
.and_then(|o| String::from_utf8(o.stdout).ok())
.map(|s| s.trim().to_string())
})
.unwrap_or_else(|| "unknown".to_string());
let mut env = Env::capture_minimal(commit_sha, features);
env.http_connection_mode = Some(cmd.http_connection_mode.as_str().to_string());
env.http_request_sampling = Some(cmd.sampling.request_sampling());
if let Some(hw) = cmd.hw_id.clone() {
env.hw_id = hw;
}
env
}
fn detect_features() -> Vec<String> {
let mut v = Vec::new();
#[cfg(feature = "metal")]
v.push("metal".to_string());
#[cfg(feature = "cuda")]
v.push("cuda".to_string());
#[cfg(feature = "vllm-marlin")]
v.push("vllm-marlin".to_string());
#[cfg(feature = "vllm-moe-marlin")]
v.push("vllm-moe-marlin".to_string());
#[cfg(feature = "vllm-paged-attn-v2")]
v.push("vllm-paged-attn-v2".to_string());
#[cfg(feature = "triton-kernels")]
v.push("triton-kernels".to_string());
v.sort();
v.dedup();
v
}
#[derive(Clone, Copy)]
enum Cell {
Closed(u32),
Open(f64),
}
async fn execute_cell(
cmd: &BenchServeCommand,
ctx: &RunContext,
cell: Cell,
cell_id: &str,
) -> Result<BenchReport> {
let backend = cmd
.target_backend
.map(BenchTargetBackend::as_str)
.unwrap_or_else(|| {
if cfg!(feature = "cuda") {
"cuda"
} else if cfg!(feature = "metal") {
"metal"
} else {
"cpu"
}
});
let tokenizer_path = cmd.tokenizer.join("tokenizer.json");
let tok = tokenizers::Tokenizer::from_file(&tokenizer_path).map_err(|e| {
ferrum_types::FerrumError::model(format!(
"Load tokenizer at {}: {}",
tokenizer_path.display(),
e
))
})?;
let total_prompts_u32 = cmd
.num_prompts
.checked_add(cmd.warmup_requests)
.ok_or_else(|| {
ferrum_types::FerrumError::model("num_prompts + warmup_requests overflow")
})?;
let total_prompts = usize::try_from(total_prompts_u32)
.map_err(|_| ferrum_types::FerrumError::model("prompt count exceeds platform capacity"))?;
let mut runs: Vec<RunRecord> = Vec::with_capacity(cmd.n_repeats as usize);
let mut actual_input_lengths: Vec<u32> = Vec::new();
let mut actual_input_tokens_per_request: Vec<Vec<u32>> =
Vec::with_capacity(cmd.n_repeats as usize);
for repeat_idx in 0..cmd.n_repeats {
let mut seeded_rng;
let mut thread_rng;
let rng: &mut dyn rand::RngCore = if let Some(seed) = cmd.seed {
seeded_rng =
StdRng::seed_from_u64(seed ^ ((repeat_idx as u64) << 32) ^ cell_seed(cell));
&mut seeded_rng
} else {
thread_rng = rand::rng();
&mut thread_rng
};
let prompts = build_prompts(cmd, &tok, rng, total_prompts)?;
let measured_input_lengths: Vec<u32> = prompts
.iter()
.skip(cmd.warmup_requests as usize)
.map(|p| p.input_tokens)
.collect();
actual_input_lengths.extend(measured_input_lengths.iter().copied());
actual_input_tokens_per_request.push(measured_input_lengths);
eprintln!(
"{}",
format!(
" cell {} — repeat {}/{} ({} prompts including {} warmup)",
cell_label(cell),
repeat_idx + 1,
cmd.n_repeats,
total_prompts,
cmd.warmup_requests
)
.dimmed()
);
let run = match cell {
Cell::Closed(c) => {
run_closed_loop(ctx, prompts, cmd.warmup_requests, c, cell_id, repeat_idx).await
}
Cell::Open(r) => {
run_open_loop(ctx, prompts, cmd.warmup_requests, r, cell_id, repeat_idx).await
}
};
eprintln!(
" {} completed / {} errored / {:.1}s",
run.n_completed(),
run.n_errored(),
run.duration_s
);
runs.push(run);
}
let requested_input_len = requested_input_len(cmd)?;
let requested_output_len = u32::try_from(cmd.random_output_len).map_err(|_| {
ferrum_types::FerrumError::model("random output length exceeds report capacity")
})?;
let actual_input_tokens = input_token_stats(&actual_input_lengths, requested_input_len);
let token_count_source = output_token_count_source_from_runs(&runs);
let env = build_env(cmd, detect_features());
let slo = cmd.goodput.unwrap_or_else(Slo::unbounded);
let model_field = match &cmd.tag {
Some(t) => format!("{}#{}", cmd.model, t),
None => cmd.model.clone(),
};
let (scenario, concurrency, request_rate) = match cell {
Cell::Closed(c) => (Scenario::ClosedLoop, Some(c), None),
Cell::Open(r) => (Scenario::OpenLoop, None, Some(r)),
};
let mut report = compute_metrics(
model_field,
backend.to_string(),
scenario,
concurrency,
request_rate,
requested_input_len,
requested_output_len,
cmd.warmup_requests,
slo,
runs,
env,
);
report.actual_input_tokens = Some(actual_input_tokens);
report.actual_input_tokens_per_request = Some(actual_input_tokens_per_request);
report.output_token_count_source = Some(token_count_source);
Ok(report)
}
fn requested_input_len(cmd: &BenchServeCommand) -> Result<u32> {
match cmd.dataset.as_str() {
"shared-prefix" => cmd
.shared_prefix_len
.checked_add(cmd.shared_suffix_len)
.and_then(|value| u32::try_from(value).ok())
.ok_or_else(|| {
ferrum_types::FerrumError::model(
"shared prefix + suffix length exceeds report capacity",
)
}),
_ => u32::try_from(cmd.random_input_len).map_err(|_| {
ferrum_types::FerrumError::model("random input length exceeds report capacity")
}),
}
}
fn input_token_stats(lengths: &[u32], requested: u32) -> TokenLengthStats {
let min = lengths.iter().copied().min().unwrap_or(0);
let max = lengths.iter().copied().max().unwrap_or(0);
let mean = if lengths.is_empty() {
0.0
} else {
lengths.iter().map(|&n| n as f64).sum::<f64>() / lengths.len() as f64
};
TokenLengthStats {
requested,
min,
max,
mean,
}
}
fn output_token_count_source_from_runs(runs: &[RunRecord]) -> String {
let mut saw_usage = false;
let mut saw_stream_chunks = false;
let mut saw_none = false;
for record in runs.iter().flat_map(|run| &run.records) {
match record.output_token_count_source {
OutputTokenCountSource::Usage => saw_usage = true,
OutputTokenCountSource::StreamChunks => saw_stream_chunks = true,
OutputTokenCountSource::None => saw_none = true,
}
}
match (saw_usage, saw_stream_chunks, saw_none) {
(true, false, false) => "usage".to_string(),
(false, true, false) => "stream_chunks".to_string(),
(false, false, true) => "none".to_string(),
_ => "mixed".to_string(),
}
}
fn cell_seed(cell: Cell) -> u64 {
match cell {
Cell::Closed(c) => 0xC10C_ED00_0000_0000u64 ^ c as u64,
Cell::Open(r) => 0x0FEE_D000_0000_0000u64 ^ r.to_bits(),
}
}
fn cell_label(cell: Cell) -> String {
match cell {
Cell::Closed(c) => format!("closed_loop c={c}"),
Cell::Open(r) => format!("open_loop rate={r}"),
}
}
fn benchmark_cell_id(index: usize, cell: Cell) -> String {
match cell {
Cell::Closed(concurrency) => format!("cell-{index}-closed-c{concurrency}"),
Cell::Open(rate) => format!("cell-{index}-open-rate-bits-{:016x}", rate.to_bits()),
}
}
pub async fn execute(cmd: BenchServeCommand, _cfg: CliConfig) -> Result<()> {
validate_command(&cmd)?;
let banner = if cmd.scenario == BenchServeWorkload::DecodeIsolation {
format!(
"ferrum bench-serve — scenario=decode-isolation dataset=random warmup={} n_repeats={} fixed_output=true",
cmd.warmup_requests, cmd.n_repeats
)
} else {
format!(
"ferrum bench-serve — dataset={} num_prompts={} warmup={} n_repeats={}",
cmd.dataset, cmd.num_prompts, cmd.warmup_requests, cmd.n_repeats
)
};
eprintln!("{}", banner.dimmed());
if cmd.n_repeats < 3 {
eprintln!(
"{}",
" [warn] n_repeats < 3 — emitting mean only, no stddev/CI95 (PLAYBOOK § 0.4)".yellow()
);
}
let cells: Vec<Cell> = if let Some(rate) = cmd.request_rate {
vec![Cell::Open(rate)]
} else if !cmd.concurrency_sweep.is_empty() {
cmd.concurrency_sweep
.iter()
.copied()
.map(Cell::Closed)
.collect()
} else {
vec![Cell::Closed(cmd.concurrency)]
};
let pool_max_idle_per_host = match cmd.http_connection_mode {
BenchHttpConnectionMode::Pooled => 64,
BenchHttpConnectionMode::Fresh => 0,
};
let client = Arc::new(
reqwest::Client::builder()
.pool_max_idle_per_host(pool_max_idle_per_host)
.build()
.map_err(|e| ferrum_types::FerrumError::model(format!("reqwest client: {e}")))?,
);
let ctx = RunContext {
client,
base_url: Arc::new(cmd.base_url.clone()),
model: Arc::new(cmd.model.clone()),
max_out: cmd.random_output_len,
ignore_eos: cmd.ignore_eos,
enable_thinking: cmd.enable_thinking,
reasoning_effort: cmd.reasoning_effort,
sampling: cmd.sampling.request_sampling(),
timeout_s: cmd.timeout,
benchmark_run_id: Arc::new(format!("bench-{}", Uuid::new_v4())),
};
if cmd.scenario == BenchServeWorkload::DecodeIsolation {
return decode_isolation::execute(&cmd, &ctx).await;
}
let mut reports: Vec<BenchReport> = Vec::with_capacity(cells.len());
for (cell_index, cell) in cells.into_iter().enumerate() {
eprintln!("{}", format!("→ {}", cell_label(cell)).bold());
let cell_id = benchmark_cell_id(cell_index, cell);
let r = execute_cell(&cmd, &ctx, cell, &cell_id).await?;
emit_summary_line(&r);
reports.push(r);
}
emit_then_enforce_error_policy(&cmd, &reports)
}
fn emit_then_enforce_error_policy(cmd: &BenchServeCommand, reports: &[BenchReport]) -> Result<()> {
emit_reports(cmd, reports)?;
ferrum_bench_core::trace::flush_global_trace();
enforce_error_policy(cmd, reports)
}
fn emit_reports(cmd: &BenchServeCommand, reports: &[BenchReport]) -> Result<()> {
match cmd.output.as_str() {
"json" => emit_json(cmd, reports)?,
"jsonl" => emit_jsonl(cmd, reports)?,
"md" => emit_markdown(cmd, reports)?,
other => {
return Err(ferrum_types::FerrumError::model(format!(
"unknown --output '{other}': allowed values are json, jsonl, md"
)))
}
}
Ok(())
}
fn validate_command(cmd: &BenchServeCommand) -> Result<()> {
cmd.sampling.request_sampling().validate()?;
if cmd.scenario == BenchServeWorkload::DecodeIsolation {
if cmd.dataset != "random" {
return Err(ferrum_types::FerrumError::model(
"--scenario decode-isolation requires --dataset random",
));
}
if cmd.sharegpt_path.is_some()
|| cmd.shared_prefix_len != 1024
|| cmd.shared_suffix_len != 64
{
return Err(ferrum_types::FerrumError::model(
"--scenario decode-isolation cannot use ShareGPT or shared-prefix options",
));
}
if cmd.num_prompts != 100 || cmd.concurrency != 32 {
return Err(ferrum_types::FerrumError::model(
"--scenario decode-isolation derives request count and concurrency; do not set --num-prompts or --concurrency",
));
}
if cmd.goodput.is_some() {
return Err(ferrum_types::FerrumError::model(
"--scenario decode-isolation does not produce goodput",
));
}
if cmd.target_backend.is_none() {
return Err(ferrum_types::FerrumError::model(
"--scenario decode-isolation requires --target-backend",
));
}
}
if cmd.scenario == BenchServeWorkload::DecodeIsolation
&& (cmd.request_rate.is_some() || !cmd.concurrency_sweep.is_empty())
{
return Err(ferrum_types::FerrumError::model(
"--scenario decode-isolation cannot be combined with --request-rate or --concurrency-sweep",
));
}
if let Some(rate) = cmd.request_rate {
if rate <= 0.0 || !rate.is_finite() {
return Err(ferrum_types::FerrumError::model(
"--request-rate must be a positive finite number",
));
}
}
if cmd.timeout <= 0.0
|| !cmd.timeout.is_finite()
|| Duration::try_from_secs_f64(cmd.timeout).is_err()
{
return Err(ferrum_types::FerrumError::model(
"--timeout must be a positive finite duration",
));
}
if cmd.concurrency == 0 {
return Err(ferrum_types::FerrumError::model(
"--concurrency must be > 0",
));
}
if cmd.num_prompts == 0 {
return Err(ferrum_types::FerrumError::model(
"--num-prompts must be > 0",
));
}
if cmd.n_repeats == 0 {
return Err(ferrum_types::FerrumError::model("--n-repeats must be > 0"));
}
if cmd.random_input_len == 0 {
return Err(ferrum_types::FerrumError::model(
"--random-input-len must be > 0",
));
}
if cmd.random_output_len == 0 {
return Err(ferrum_types::FerrumError::model(
"--random-output-len must be > 0",
));
}
u32::try_from(cmd.random_output_len).map_err(|_| {
ferrum_types::FerrumError::model("--random-output-len exceeds report capacity")
})?;
requested_input_len(cmd)?;
if cmd.goodput.is_some_and(|slo| !slo.is_valid()) {
return Err(ferrum_types::FerrumError::model(
"--goodput values must be positive finite numbers",
));
}
if cmd.require_ci && cmd.n_repeats < 3 {
return Err(ferrum_types::FerrumError::model(
"--require-ci requires --n-repeats >= 3",
));
}
if let Some(max_error_rate) = cmd.max_error_rate {
if !(0.0..=1.0).contains(&max_error_rate) || !max_error_rate.is_finite() {
return Err(ferrum_types::FerrumError::model(
"--max-error-rate must be in [0.0, 1.0]",
));
}
}
if !matches!(cmd.output.as_str(), "json" | "jsonl" | "md") {
return Err(ferrum_types::FerrumError::model(format!(
"unknown --output '{}': allowed values are json, jsonl, md",
cmd.output
)));
}
if cmd.output == "jsonl" && cmd.out.is_none() {
return Err(ferrum_types::FerrumError::model(
"--output jsonl requires --out PATH (append-mode log)",
));
}
for cell in &cmd.concurrency_sweep {
if *cell == 0 {
return Err(ferrum_types::FerrumError::model(
"--concurrency-sweep values must be > 0",
));
}
}
let _total_prompts = cmd
.num_prompts
.checked_add(cmd.warmup_requests)
.ok_or_else(|| {
ferrum_types::FerrumError::model("num_prompts + warmup_requests overflow")
})?;
Ok(())
}
fn enforce_error_policy(cmd: &BenchServeCommand, reports: &[BenchReport]) -> Result<()> {
let max_error_rate = cmd.max_error_rate.unwrap_or(0.0);
for report in reports {
let warmup_errored: u64 = report
.repeat_metrics
.iter()
.map(|repeat| repeat.warmup_errored as u64)
.sum();
let warmup_has_quality_issue = report
.repeat_metrics
.iter()
.any(|repeat| repeat.warmup_quality_issues != QualityIssueCounts::default());
if warmup_errored > 0 || warmup_has_quality_issue {
return Err(ferrum_types::FerrumError::model(format!(
"bench-serve warmup failed for {}: {} errored request(s)",
report.model, warmup_errored
)));
}
if !cmd.fail_on_error && cmd.max_error_rate.is_none() {
continue;
}
let completed: u64 = report
.completed_per_run
.iter()
.map(|value| *value as u64)
.sum();
let errored: u64 = report
.errored_per_run
.iter()
.map(|value| *value as u64)
.sum();
let total = completed
.checked_add(errored)
.ok_or_else(|| ferrum_types::FerrumError::model("measured request count overflow"))?;
let error_rate = if total == 0 {
1.0
} else {
errored as f64 / total as f64
};
if error_rate > max_error_rate {
return Err(ferrum_types::FerrumError::model(format!(
"bench-serve error rate {:.4} exceeds max {:.4} for {}",
error_rate, max_error_rate, report.model
)));
}
}
Ok(())
}
fn emit_jsonl(cmd: &BenchServeCommand, reports: &[BenchReport]) -> Result<()> {
use std::io::Write as _;
let out_path = cmd.out.as_ref().ok_or_else(|| {
ferrum_types::FerrumError::model("--output jsonl requires --out PATH (append-mode log)")
})?;
let mut f = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(out_path)
.map_err(|e| {
ferrum_types::FerrumError::model(format!("open {}: {e}", out_path.display()))
})?;
for r in reports {
let line = serde_json::to_string(r).expect("serialize");
writeln!(f, "{}", line)
.map_err(|e| ferrum_types::FerrumError::model(format!("write jsonl: {e}")))?;
}
eprintln!(
"\n→ appended {} record(s) to {}",
reports.len(),
out_path.display()
);
Ok(())
}
fn emit_markdown(cmd: &BenchServeCommand, reports: &[BenchReport]) -> Result<()> {
use ferrum_bench_core::report::{render_single, render_sweep};
let md = if reports.len() == 1 {
render_single(&reports[0])
} else {
render_sweep(reports)
};
if let Some(out) = cmd.out.as_ref() {
std::fs::write(out, &md).map_err(|e| {
ferrum_types::FerrumError::model(format!("write {}: {e}", out.display()))
})?;
eprintln!("\n→ wrote {}", out.display());
} else {
println!("{}", md);
}
Ok(())
}
fn emit_summary_line(r: &BenchReport) {
let scenario_str = match r.scenario {
Scenario::ClosedLoop => format!("c={}", r.concurrency.unwrap_or(0)),
Scenario::OpenLoop => format!("rate={}", r.request_rate.unwrap_or(0.0)),
Scenario::SharedPrefix => "shared_prefix".to_string(),
Scenario::Cli => "cli".to_string(),
Scenario::DecodeIsolation => "decode_isolation".to_string(),
};
let ci = if r.n_repeats >= 3 {
format!(" (n_repeats={}, ± = ci95_hw)", r.n_repeats)
} else {
format!(" (n_repeats={}, no CI)", r.n_repeats)
};
eprintln!(" {} {}{}", "summary".bold(), scenario_str, ci);
fmt_metric("TTFT_ms ", &r.ttft_ms, r.n_repeats);
fmt_metric("TPOT_ms ", &r.tpot_ms, r.n_repeats);
if r.has_complete_itl_evidence() {
fmt_metric("ITL_ms ", &r.itl_ms, r.n_repeats);
} else {
eprintln!(" ITL_ms unavailable");
}
let thr = &r.output_throughput_tps;
let good = &r.goodput_rps;
if r.n_repeats >= 3 {
eprintln!(
" throughput {:.1} ± {:.1} tok/s",
thr.mean, thr.ci95_hw
);
eprintln!(
" goodput {:.2} ± {:.2} req/s",
good.mean, good.ci95_hw
);
} else {
eprintln!(" throughput {:.1} tok/s", thr.mean);
eprintln!(" goodput {:.2} req/s", good.mean);
}
}
fn fmt_metric(name: &str, m: &ferrum_bench_core::MetricSet, n_repeats: u32) {
if n_repeats >= 3 {
eprintln!(
" {} p50={:.1}±{:.1} p95={:.1}±{:.1} p99={:.1}±{:.1}",
name, m.p50.mean, m.p50.ci95_hw, m.p95.mean, m.p95.ci95_hw, m.p99.mean, m.p99.ci95_hw
);
} else {
eprintln!(
" {} p50={:.1} p95={:.1} p99={:.1}",
name, m.p50.mean, m.p95.mean, m.p99.mean
);
}
}
fn emit_json(cmd: &BenchServeCommand, reports: &[BenchReport]) -> Result<()> {
let value = if reports.len() == 1 {
serde_json::to_value(&reports[0]).expect("serialize")
} else {
serde_json::to_value(reports).expect("serialize")
};
let pretty = serde_json::to_string_pretty(&value).expect("pretty");
if let Some(out) = cmd.out.as_ref() {
std::fs::write(out, &pretty).map_err(|e| {
ferrum_types::FerrumError::model(format!("write {}: {e}", out.display()))
})?;
eprintln!("\n→ wrote {}", out.display());
} else {
println!("{}", pretty);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use ferrum_bench_core::ItlEligibility;
fn parse_sse_chunks<'a>(chunks: impl IntoIterator<Item = &'a [u8]>) -> RequestRecord {
let mut state = StreamState::new(Instant::now(), 7);
let mut parser = SseLineBuffer::default();
for chunk in chunks {
let before_output_events = state.output_delta_events;
parser.push(chunk, &mut state);
let output_events = state
.output_delta_events
.saturating_sub(before_output_events);
state.note_transport_chunk(output_events);
}
parser.finish(&mut state);
state.finish()
}
#[test]
fn stream_preserves_server_prompt_usage_separately_from_content_length() {
for (usage, expected) in [
(r#"{"prompt_tokens":23,"completion_tokens":1}"#, Some(23)),
(r#"{"completion_tokens":1}"#, None),
] {
let stream = format!(
"data: {{\"choices\":[{{\"delta\":{{\"content\":\"OK\"}}}}]}}\n\ndata: {{\"choices\":[],\"usage\":{usage}}}\n\ndata: [DONE]\n\n"
);
let record = parse_sse_chunks(stream.as_bytes().chunks(3));
assert!(record.success);
assert_eq!(record.input_tokens, 7);
assert_eq!(record.server_input_tokens, expected);
assert_eq!(record.output_tokens, 1);
}
}
fn stream_semantics(
record: &RequestRecord,
) -> (
bool,
u32,
OutputTokenCountSource,
QualityIssueCounts,
RequestItlEvidence,
) {
(
record.success,
record.output_tokens,
record.output_token_count_source,
record.quality_issues.clone(),
record.itl_evidence.clone(),
)
}
#[test]
fn slo_parses_space_separated() {
let s = parse_slo("ttft:500 tpot:50 e2el:30000").unwrap();
assert_eq!(s.ttft_p99_ms, 500.0);
assert_eq!(s.tpot_p99_ms, 50.0);
assert_eq!(s.e2e_p99_ms, 30000.0);
}
#[test]
fn slo_parses_comma_separated() {
let s = parse_slo("ttft:500,tpot:50,e2el:30000").unwrap();
assert_eq!(s.ttft_p99_ms, 500.0);
}
#[test]
fn slo_accepts_e2e_alias() {
let s = parse_slo("ttft:1 tpot:2 e2e:3").unwrap();
assert_eq!(s.e2e_p99_ms, 3.0);
}
#[test]
fn slo_rejects_missing_key() {
assert!(parse_slo("ttft:500 tpot:50").is_err());
}
#[test]
fn slo_rejects_unknown_key() {
assert!(parse_slo("ttft:500 tpot:50 e2el:30000 bogus:1").is_err());
}
#[test]
fn slo_rejects_nonfinite_and_nonpositive_values() {
for value in ["NaN", "inf", "0", "-1"] {
assert!(
parse_slo(&format!("ttft:{value} tpot:50 e2el:30000")).is_err(),
"SLO value {value} should be rejected"
);
}
}
#[test]
fn stream_record_preserves_openai_response_request_id() {
let record = parse_sse_chunks([
b"data: {\"id\":\"chatcmpl-123\",\"choices\":[{\"delta\":{\"content\":\"ok\"}}],\"usage\":null}\n\n"
.as_slice(),
b"data: {\"id\":\"chatcmpl-123\",\"choices\":[],\"usage\":{\"completion_tokens\":1}}\n\ndata: [DONE]\n\n"
.as_slice(),
]);
assert!(record.success);
assert_eq!(record.server_request_id.as_deref(), Some("chatcmpl-123"));
}
#[test]
fn chat_completion_body_omits_ignore_eos_by_default() {
let body = chat_completion_body(
"model",
"prompt",
128,
false,
None,
None,
HttpRequestSampling::default(),
);
assert_eq!(body["model"], serde_json::json!("model"));
assert_eq!(body["max_tokens"], serde_json::json!(128));
assert_eq!(body["stream"], serde_json::json!(true));
assert!(body.get("chat_template_kwargs").is_none());
assert_eq!(
body["stream_options"]["include_usage"],
serde_json::json!(true)
);
assert!(body.get("ignore_eos").is_none());
}
#[test]
fn chat_completion_body_sends_ignore_eos_when_requested() {
let body = chat_completion_body(
"model",
"prompt",
128,
true,
None,
None,
HttpRequestSampling::default(),
);
assert_eq!(body["ignore_eos"], serde_json::json!(true));
}
#[test]
fn chat_completion_body_sends_typed_thinking_values() {
let disabled = chat_completion_body(
"model",
"prompt",
128,
false,
Some(false),
None,
HttpRequestSampling::default(),
);
assert_eq!(
disabled["chat_template_kwargs"]["enable_thinking"],
serde_json::json!(false)
);
let enabled = chat_completion_body(
"model",
"prompt",
128,
false,
Some(true),
None,
HttpRequestSampling::default(),
);
assert_eq!(
enabled["chat_template_kwargs"]["enable_thinking"],
serde_json::json!(true)
);
}
#[test]
fn chat_completion_body_sends_typed_reasoning_effort() {
let body = chat_completion_body(
"model",
"prompt",
128,
false,
Some(false),
Some(ReasoningEffort::Low),
HttpRequestSampling::default(),
);
assert_eq!(
body["chat_template_kwargs"]["enable_thinking"],
serde_json::json!(false)
);
assert_eq!(
body["chat_template_kwargs"]["reasoning_effort"],
serde_json::json!("low")
);
}
#[test]
fn thinking_and_reasoning_effort_cli_are_typed() {
use clap::Parser as _;
#[derive(clap::Parser)]
struct TestCli {
#[command(flatten)]
command: BenchServeCommand,
}
let base = [
"test",
"--base-url",
"http://127.0.0.1:8000",
"--model",
"model",
"--tokenizer",
".",
];
let absent = TestCli::parse_from(base).command;
assert_eq!(absent.enable_thinking, None);
assert_eq!(absent.reasoning_effort, None);
assert_eq!(absent.http_connection_mode, BenchHttpConnectionMode::Pooled);
let disabled =
TestCli::parse_from(base.into_iter().chain(["--enable-thinking", "false"])).command;
assert_eq!(disabled.enable_thinking, Some(false));
let enabled =
TestCli::parse_from(base.into_iter().chain(["--enable-thinking", "true"])).command;
assert_eq!(enabled.enable_thinking, Some(true));
let low =
TestCli::parse_from(base.into_iter().chain(["--reasoning-effort", "low"])).command;
assert_eq!(low.reasoning_effort, Some(ReasoningEffort::Low));
let xhigh =
TestCli::parse_from(base.into_iter().chain(["--reasoning-effort", "xhigh"])).command;
assert_eq!(xhigh.reasoning_effort, Some(ReasoningEffort::XHigh));
let metal =
TestCli::parse_from(base.into_iter().chain(["--target-backend", "metal"])).command;
assert_eq!(metal.target_backend, Some(BenchTargetBackend::Metal));
let fresh =
TestCli::parse_from(base.into_iter().chain(["--http-connection-mode", "fresh"]))
.command;
assert_eq!(fresh.http_connection_mode, BenchHttpConnectionMode::Fresh);
let isolation = TestCli::parse_from(base.into_iter().chain([
"--scenario",
"decode-isolation",
"--decode-isolation-incumbents",
"6",
"--decode-isolation-prefill-tokens",
"4096",
]))
.command;
assert_eq!(isolation.scenario, BenchServeWorkload::DecodeIsolation);
assert_eq!(
isolation.decode_isolation.decode_isolation_incumbents,
Some(6)
);
assert_eq!(
isolation.decode_isolation.decode_isolation_prefill_tokens,
Some(4096)
);
}
#[test]
fn bench_env_locks_http_connection_mode() {
let pooled_cmd = test_command();
let pooled = build_env(&pooled_cmd, vec![]);
assert_eq!(pooled.http_connection_mode.as_deref(), Some("pooled"));
let mut fresh_cmd = pooled_cmd;
fresh_cmd.http_connection_mode = BenchHttpConnectionMode::Fresh;
let fresh = build_env(&fresh_cmd, vec![]);
assert_eq!(fresh.http_connection_mode.as_deref(), Some("fresh"));
assert_ne!(pooled.hash(), fresh.hash());
}
#[test]
fn incremental_sse_utf8_semantics_are_invariant_to_every_byte_cut() {
let stream = concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"你好🙂\"}}],\"usage\":null}\n\n",
"data: {\"choices\":[],\"usage\":{\"completion_tokens\":1}}\n\n",
"data: [DONE]\n\n",
)
.as_bytes();
let baseline = parse_sse_chunks([stream]);
assert!(baseline.success);
assert_eq!(baseline.quality_issues, QualityIssueCounts::default());
for cut in 0..=stream.len() {
let split = parse_sse_chunks([&stream[..cut], &stream[cut..]]);
assert_eq!(
stream_semantics(&split),
stream_semantics(&baseline),
"SSE semantics changed at byte cut {cut}"
);
}
let bytewise = parse_sse_chunks(stream.chunks(1));
assert_eq!(stream_semantics(&bytewise), stream_semantics(&baseline));
}
#[test]
fn transport_coalescing_is_diagnostic_not_stream_bulk_failure() {
const FIRST: &[u8] =
b"data: {\"choices\":[{\"delta\":{\"content\":\"a\"}}],\"usage\":null}\n\n";
const SECOND: &[u8] =
b"data: {\"choices\":[{\"delta\":{\"content\":\"b\"}}],\"usage\":null}\n\n";
const TAIL: &[u8] =
b"data: {\"choices\":[],\"usage\":{\"completion_tokens\":2}}\n\ndata: [DONE]\n\n";
let together_bytes = [FIRST, SECOND, TAIL].concat();
let together = parse_sse_chunks([together_bytes.as_slice()]);
assert!(together.success);
assert_eq!(together.quality_issues.stream_bulk_flush, 0);
assert_eq!(together.itl_evidence.transport_coalesced_output_chunks, 1);
assert_eq!(
together.itl_evidence.eligibility,
ItlEligibility::TransportCoalesced
);
let separated = parse_sse_chunks([FIRST, SECOND, TAIL]);
assert!(separated.success);
assert_eq!(separated.quality_issues, QualityIssueCounts::default());
assert_eq!(separated.itl_evidence.transport_coalesced_output_chunks, 0);
assert_eq!(separated.itl_evidence.eligibility, ItlEligibility::Eligible);
}
#[test]
fn stream_done_with_usage_succeeds() {
let mut state = StreamState::new(Instant::now(), 7);
state
.handle_payload(r#"{"choices":[{"delta":{"content":"hello"}}],"usage":null}"#)
.unwrap();
state
.handle_payload(r#"{"choices":[],"usage":{"completion_tokens":3}}"#)
.unwrap();
state.done_count = 1;
let record = state.finish();
assert!(record.success);
assert_eq!(record.output_tokens, 3);
assert_eq!(
record.output_token_count_source,
OutputTokenCountSource::Usage
);
assert_eq!(
record.itl_evidence.eligibility,
ItlEligibility::EventUsageMismatch
);
assert_eq!(record.itl_evidence.output_events, 1);
assert_eq!(record.itl_evidence.usage_output_tokens, Some(3));
}
#[test]
fn matching_sse_events_usage_and_intervals_are_itl_eligible() {
let mut state = StreamState::new(Instant::now(), 7);
for content in ["a", "b", "c"] {
state
.handle_payload(&format!(
r#"{{"choices":[{{"delta":{{"content":"{content}"}}}}]}}"#
))
.unwrap();
}
state
.handle_payload(r#"{"choices":[],"usage":{"completion_tokens":3}}"#)
.unwrap();
state.done_count = 1;
let record = state.finish();
assert!(record.success);
assert_eq!(record.itl_ms.len(), 2);
assert_eq!(record.itl_evidence.eligibility, ItlEligibility::Eligible);
}
#[tokio::test]
async fn measured_join_error_becomes_failed_evidence() {
let correlation = |request_index| {
BenchmarkRequestCorrelation::new(
"bench-test".to_string(),
"cell-test".to_string(),
0,
BenchmarkPhase::Measured,
request_index,
)
.unwrap()
};
let first_correlation = correlation(0);
let second_correlation = correlation(1);
let good_record_correlation = first_correlation.clone();
let good = tokio::spawn(async move {
RequestRecord {
benchmark_correlation: Some(good_record_correlation),
server_request_id: None,
success: true,
ttft_ms: 1.0,
e2e_ms: 2.0,
input_tokens: 7,
server_input_tokens: None,
output_tokens: 1,
output_token_count_source: OutputTokenCountSource::Usage,
itl_evidence: RequestItlEvidence::sse(true, 1, Some(1), 0, 0),
quality_issues: QualityIssueCounts::default(),
itl_ms: vec![],
}
});
let panic_record_correlation = second_correlation.clone();
let panicked = tokio::spawn(async move {
if true {
panic!("measured task panic");
}
join_failed_record(0, panic_record_correlation)
});
let records = collect_measured_handles(vec![
(7, first_correlation, good),
(11, second_correlation, panicked),
])
.await;
assert_eq!(records.len(), 2);
assert!(records[0].success);
assert!(!records[1].success);
assert_eq!(records[1].input_tokens, 11);
assert_eq!(records[1].quality_issues.panic, 1);
assert_eq!(
records[1].output_token_count_source,
OutputTokenCountSource::None
);
}
#[test]
fn stream_reasoning_chunk_counts_as_token_event() {
let mut state = StreamState::new(Instant::now(), 7);
state
.handle_payload(r#"{"choices":[{"delta":{"reasoning":"thinking"}}]}"#)
.unwrap();
state.done_count = 1;
let record = state.finish();
assert!(record.success);
assert_eq!(record.output_tokens, 1);
assert_eq!(
record.output_token_count_source,
OutputTokenCountSource::StreamChunks
);
}
#[test]
fn stream_error_after_chunk_fails() {
let mut state = StreamState::new(Instant::now(), 7);
state
.handle_payload(r#"{"choices":[{"delta":{"content":"hello"}}]}"#)
.unwrap();
state.stream_error = Some("broken stream".into());
state.done_count = 1;
let record = state.finish();
assert!(!record.success);
assert_eq!(
record.output_token_count_source,
OutputTokenCountSource::StreamChunks
);
}
#[test]
fn openai_error_sse_event_fails_even_when_done_follows() {
let record = parse_sse_chunks([concat!(
"data: {\"error\":{\"message\":\"decode failed\",",
"\"type\":\"internal_server_error\",\"param\":null,\"code\":null}}\n\n",
"data: [DONE]\n\n"
)
.as_bytes()]);
assert!(!record.success);
assert_eq!(record.quality_issues.malformed_stream, 1);
assert_eq!(
record.itl_evidence.eligibility,
ItlEligibility::RequestFailed
);
}
#[test]
fn eof_before_done_after_chunk_fails() {
let mut state = StreamState::new(Instant::now(), 7);
state
.handle_payload(r#"{"choices":[{"delta":{"content":"hello"}}]}"#)
.unwrap();
let record = state.finish();
assert!(!record.success);
assert_eq!(record.output_tokens, 1);
assert_eq!(record.quality_issues.missing_done, 1);
}
#[test]
fn malformed_sse_json_fails() {
let mut state = StreamState::new(Instant::now(), 7);
assert!(state.handle_payload("{bad json}").is_err());
state.stream_error = Some("malformed json".into());
state.quality_issues.malformed_stream = 1;
state.done_count = 1;
let record = state.finish();
assert!(!record.success);
assert_eq!(record.output_tokens, 0);
assert_eq!(record.quality_issues.malformed_stream, 1);
}
#[test]
fn done_with_zero_content_tokens_fails() {
let mut state = StreamState::new(Instant::now(), 7);
state.done_count = 1;
let record = state.finish();
assert!(!record.success);
assert_eq!(record.output_tokens, 0);
assert_eq!(record.quality_issues.zero_output_tokens, 1);
assert_eq!(
record.output_token_count_source,
OutputTokenCountSource::None
);
}
#[test]
fn duplicate_done_fails() {
let mut state = StreamState::new(Instant::now(), 7);
state
.handle_payload(r#"{"choices":[{"delta":{"content":"hello"}}]}"#)
.unwrap();
state.done_count = 2;
let record = state.finish();
assert!(!record.success);
assert_eq!(record.quality_issues.duplicate_done, 1);
}
#[test]
fn bad_output_text_fails() {
let mut state = StreamState::new(Instant::now(), 7);
state
.handle_payload(r#"{"choices":[{"delta":{"content":"<unk>"}}]}"#)
.unwrap();
state.done_count = 1;
let record = state.finish();
assert!(!record.success);
assert_eq!(record.quality_issues.bad_output, 1);
}
#[test]
fn mojibake_sequences_fail_but_standalone_leads_do_not() {
assert!(has_bad_output_text("caf\u{00c3}\u{00a9}"));
assert!(has_bad_output_text("copyright \u{00c2}\u{00a9}"));
assert!(has_bad_output_text("quote\u{00e2}\u{20ac}\u{2122}"));
assert!(!has_bad_output_text("\u{00c2}"));
assert!(!has_bad_output_text("\u{00c3}"));
assert!(!has_bad_output_text("Grade \u{00c2} report"));
}
#[test]
fn random_prompt_generation_targets_reencoded_length_when_fixture_is_set() {
let Some(path) = ferrum_env_value("FERRUM_BENCH_TOKENIZER_FIXTURE") else {
return;
};
let tok = tokenizers::Tokenizer::from_file(path).expect("load tokenizer fixture");
let mut rng = StdRng::seed_from_u64(9271);
for _ in 0..16 {
let text = gen_random_prompt(&tok, 256, &mut rng);
assert_eq!(token_count(&tok, &text), Some(256));
assert!(generated_prompt_is_safe(&text));
}
let prompts = gen_shared_prefix_prompts(&tok, 4, 128, 16, &mut rng)
.expect("generate shared-prefix prompts");
assert_eq!(prompts.len(), 4);
assert!(prompts
.iter()
.all(|prompt| generated_prompt_is_safe(&prompt.text)));
}
#[test]
fn generated_prompt_safety_rejects_transport_poisoning_text() {
assert!(generated_prompt_is_safe("hello world\nnext"));
assert!(!generated_prompt_is_safe("bad \u{fffd} text"));
assert!(!generated_prompt_is_safe("<|im_start|>"));
assert!(!generated_prompt_is_safe("control \u{7}"));
}
fn ferrum_env_value(key: &str) -> Option<String> {
ferrum_types::RuntimeConfigSnapshot::capture_current()
.entries
.into_iter()
.find(|entry| entry.key == key)
.map(|entry| entry.effective_value)
}
pub(super) fn test_command() -> BenchServeCommand {
BenchServeCommand {
base_url: "http://127.0.0.1:9".to_string(),
model: "test-model".to_string(),
tokenizer: std::path::PathBuf::from("."),
target_backend: None,
http_connection_mode: BenchHttpConnectionMode::Pooled,
scenario: BenchServeWorkload::Standard,
decode_isolation: DecodeIsolationArgs::default(),
sampling: BenchSamplingArgs::default(),
concurrency: 1,
concurrency_sweep: vec![],
request_rate: None,
dataset: "random".to_string(),
random_input_len: 2,
random_output_len: 3,
ignore_eos: false,
enable_thinking: None,
reasoning_effort: None,
sharegpt_path: None,
shared_prefix_len: 1024,
shared_suffix_len: 64,
num_prompts: 1,
warmup_requests: 0,
n_repeats: 1,
goodput: None,
timeout: 1.0,
fail_on_error: false,
max_error_rate: None,
require_ci: false,
seed: Some(9271),
output: "json".to_string(),
out: None,
hw_id: None,
commit_sha: None,
tag: None,
}
}
fn policy_report(
completed: u32,
errored: u32,
warmup_completed: u32,
warmup_errored: u32,
) -> BenchReport {
let mut records = Vec::with_capacity((completed + errored) as usize);
for _ in 0..completed {
records.push(RequestRecord {
benchmark_correlation: None,
server_request_id: None,
success: true,
ttft_ms: 10.0,
e2e_ms: 30.0,
input_tokens: 2,
server_input_tokens: None,
output_tokens: 3,
output_token_count_source: OutputTokenCountSource::Usage,
itl_evidence: RequestItlEvidence::sse(true, 3, Some(3), 2, 0),
quality_issues: QualityIssueCounts::default(),
itl_ms: vec![10.0, 10.0],
});
}
for _ in 0..errored {
let mut quality = QualityIssueCounts::default();
quality.missing_done = 1;
records.push(RequestRecord {
benchmark_correlation: None,
server_request_id: None,
success: false,
ttft_ms: 0.0,
e2e_ms: 30.0,
input_tokens: 2,
server_input_tokens: None,
output_tokens: 0,
output_token_count_source: OutputTokenCountSource::None,
itl_evidence: RequestItlEvidence::failed(ItlEvidenceSource::SseDeltaEvents),
quality_issues: quality,
itl_ms: vec![],
});
}
let warmup_expected = warmup_completed.checked_add(warmup_errored).unwrap();
let mut warmup_quality = QualityIssueCounts::default();
warmup_quality.missing_done = warmup_errored;
compute_metrics(
"test-model".to_string(),
"test-backend".to_string(),
Scenario::ClosedLoop,
Some(1),
None,
2,
3,
warmup_expected,
Slo::default(),
vec![RunRecord {
expected_requests: completed.checked_add(errored).unwrap(),
records,
duration_s: 1.0,
warmup: WarmupSummary {
expected: warmup_expected,
completed: warmup_completed,
errored: warmup_errored,
quality_issues: warmup_quality,
},
}],
Env::default(),
)
}
#[test]
fn validate_command_rejects_zero_repeats_and_invalid_programmatic_slo() {
let mut cmd = test_command();
cmd.n_repeats = 0;
assert!(validate_command(&cmd).is_err());
cmd.n_repeats = 1;
cmd.goodput = Some(Slo {
ttft_p99_ms: f64::NAN,
..Slo::default()
});
assert!(validate_command(&cmd).is_err());
cmd.goodput = None;
cmd.timeout = f64::MAX;
assert!(validate_command(&cmd).is_err());
}
#[test]
fn decode_isolation_rejects_non_random_dataset() {
let mut cmd = test_command();
cmd.scenario = BenchServeWorkload::DecodeIsolation;
cmd.target_backend = Some(BenchTargetBackend::Cuda);
cmd.dataset = "sharegpt".to_string();
let error = validate_command(&cmd).expect_err("dataset must match actual workload");
assert!(error
.to_string()
.contains("decode-isolation requires --dataset random"));
}
#[test]
fn decode_isolation_rejects_ignored_standard_workload_options() {
let mut cmd = test_command();
cmd.scenario = BenchServeWorkload::DecodeIsolation;
cmd.target_backend = Some(BenchTargetBackend::Cuda);
cmd.num_prompts = 12;
let error = validate_command(&cmd).expect_err("num-prompts is not part of this shape");
assert!(error.to_string().contains("derives request count"));
cmd.num_prompts = 100;
cmd.sharegpt_path = Some(PathBuf::from("unused.jsonl"));
let error = validate_command(&cmd).expect_err("unused dataset option must be rejected");
assert!(error.to_string().contains("cannot use ShareGPT"));
}
#[test]
fn measured_error_rate_excludes_successful_warmups() {
let report = policy_report(99, 1, 10, 0);
let mut cmd = test_command();
cmd.max_error_rate = Some(0.0095);
let err = enforce_error_policy(&cmd, &[report]).expect_err("measured rate is one percent");
assert!(err.to_string().contains("bench-serve error rate"));
}
#[test]
fn warmup_failure_is_independent_of_measured_error_allowance() {
let report = policy_report(100, 0, 9, 1);
let mut cmd = test_command();
cmd.max_error_rate = Some(1.0);
let err = enforce_error_policy(&cmd, &[report]).expect_err("warmup must be perfect");
assert!(err.to_string().contains("bench-serve warmup failed"));
}
#[test]
fn fail_on_error_still_writes_json_report() {
let out = std::env::temp_dir().join(format!(
"ferrum-bench-serve-failed-report-{}-{}.json",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let _ = std::fs::remove_file(&out);
let mut failed_quality = QualityIssueCounts::default();
failed_quality.http_500 = 1;
let run = RunRecord {
records: vec![
RequestRecord {
benchmark_correlation: None,
server_request_id: None,
success: true,
ttft_ms: 10.0,
e2e_ms: 30.0,
input_tokens: 4,
server_input_tokens: None,
output_tokens: 3,
output_token_count_source: OutputTokenCountSource::Usage,
itl_evidence: RequestItlEvidence::sse(true, 3, Some(3), 2, 0),
quality_issues: QualityIssueCounts::default(),
itl_ms: vec![10.0, 10.0],
},
RequestRecord {
benchmark_correlation: None,
server_request_id: None,
success: false,
ttft_ms: 0.0,
e2e_ms: 50.0,
input_tokens: 4,
server_input_tokens: None,
output_tokens: 0,
output_token_count_source: OutputTokenCountSource::None,
itl_evidence: RequestItlEvidence::failed(ItlEvidenceSource::SseDeltaEvents),
quality_issues: failed_quality,
itl_ms: vec![],
},
],
expected_requests: 2,
duration_s: 1.0,
warmup: Default::default(),
};
let report = compute_metrics(
"test-model".to_string(),
"test-backend".to_string(),
Scenario::ClosedLoop,
Some(2),
None,
2,
3,
0,
Slo::default(),
vec![run],
Env::default(),
);
let cmd = BenchServeCommand {
base_url: "http://127.0.0.1:9".to_string(),
model: "test-model".to_string(),
tokenizer: std::path::PathBuf::from("."),
target_backend: None,
http_connection_mode: BenchHttpConnectionMode::Pooled,
scenario: BenchServeWorkload::Standard,
decode_isolation: DecodeIsolationArgs::default(),
sampling: BenchSamplingArgs::default(),
concurrency: 2,
concurrency_sweep: vec![],
request_rate: None,
dataset: "random".to_string(),
random_input_len: 4,
random_output_len: 3,
ignore_eos: false,
enable_thinking: None,
reasoning_effort: None,
sharegpt_path: None,
shared_prefix_len: 1024,
shared_suffix_len: 64,
num_prompts: 2,
warmup_requests: 0,
n_repeats: 1,
goodput: None,
timeout: 1.0,
fail_on_error: true,
max_error_rate: None,
require_ci: false,
seed: Some(9271),
output: "json".to_string(),
out: Some(out.clone()),
hw_id: None,
commit_sha: None,
tag: None,
};
let err = emit_then_enforce_error_policy(&cmd, &[report]).expect_err("error policy");
assert!(
err.to_string().contains("bench-serve error rate"),
"unexpected error: {err}"
);
let raw = std::fs::read_to_string(&out).expect("report written before error");
let json: serde_json::Value = serde_json::from_str(&raw).expect("json report");
assert_eq!(json["completed_per_run"], serde_json::json!([1]));
assert_eq!(json["errored_per_run"], serde_json::json!([1]));
assert_eq!(json["http_500_per_run"], serde_json::json!([1]));
assert_eq!(
json["output_tokens_per_request"],
serde_json::json!([[3, 0]])
);
let _ = std::fs::remove_file(out);
}
#[test]
fn fail_on_error_rejects_warmup_only_failure() {
let mut warmup_quality = QualityIssueCounts::default();
warmup_quality.missing_done = 1;
let report = compute_metrics(
"test-model".to_string(),
"test-backend".to_string(),
Scenario::ClosedLoop,
Some(1),
None,
2,
3,
1,
Slo::default(),
vec![RunRecord {
records: vec![RequestRecord {
benchmark_correlation: None,
server_request_id: None,
success: true,
ttft_ms: 10.0,
e2e_ms: 30.0,
input_tokens: 2,
server_input_tokens: None,
output_tokens: 3,
output_token_count_source: OutputTokenCountSource::Usage,
itl_evidence: RequestItlEvidence::sse(true, 3, Some(3), 2, 0),
quality_issues: QualityIssueCounts::default(),
itl_ms: vec![10.0, 10.0],
}],
expected_requests: 1,
duration_s: 1.0,
warmup: WarmupSummary {
expected: 1,
completed: 0,
errored: 1,
quality_issues: warmup_quality,
},
}],
Env::default(),
);
let cmd = BenchServeCommand {
base_url: "http://127.0.0.1:9".to_string(),
model: "test-model".to_string(),
tokenizer: std::path::PathBuf::from("."),
target_backend: None,
http_connection_mode: BenchHttpConnectionMode::Pooled,
scenario: BenchServeWorkload::Standard,
decode_isolation: DecodeIsolationArgs::default(),
sampling: BenchSamplingArgs::default(),
concurrency: 1,
concurrency_sweep: vec![],
request_rate: None,
dataset: "random".to_string(),
random_input_len: 2,
random_output_len: 3,
ignore_eos: false,
enable_thinking: None,
reasoning_effort: None,
sharegpt_path: None,
shared_prefix_len: 1024,
shared_suffix_len: 64,
num_prompts: 1,
warmup_requests: 1,
n_repeats: 1,
goodput: None,
timeout: 1.0,
fail_on_error: true,
max_error_rate: None,
require_ci: false,
seed: Some(9271),
output: "json".to_string(),
out: None,
hw_id: None,
commit_sha: None,
tag: None,
};
let err = enforce_error_policy(&cmd, &[report]).expect_err("warmup failure");
assert!(err.to_string().contains("bench-serve warmup failed"));
}
}