794 lines
26 KiB
Rust
794 lines
26 KiB
Rust
// src/bin/chat_load_tester_consistency.rs
|
|
use futures_util::{
|
|
SinkExt, StreamExt,
|
|
stream::{FuturesUnordered, SplitSink, SplitStream},
|
|
};
|
|
use std::sync::{
|
|
Arc,
|
|
atomic::{AtomicU64, Ordering},
|
|
};
|
|
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
|
use tokio::net::TcpStream;
|
|
use tokio::sync::Mutex;
|
|
use tokio::time::sleep;
|
|
use tokio_tungstenite::{
|
|
MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message,
|
|
};
|
|
mod shared_types;
|
|
use shared_types::{ClientToServerMsg, ServerToClientMsg};
|
|
|
|
|
|
static EARLY_EXIT_COUNT: AtomicU64 = AtomicU64::new(0);
|
|
|
|
// --- Configuration ---
|
|
const SERVER_URL: &str = "ws://localhost:8080/ws";
|
|
|
|
const TOTAL_USERS: u64 = 1000;
|
|
const USERS_PER_SECOND: u64 = 50; // Sustainable rate
|
|
const TEST_DURATION: Duration = Duration::from_secs(45);
|
|
|
|
|
|
// const TOTAL_USERS: u64 = 1000;
|
|
// const USERS_PER_SECOND: u64 = 100; // Push it a bit
|
|
// const TEST_DURATION: Duration = Duration::from_secs(30);
|
|
|
|
|
|
// const TOTAL_USERS: u64 = 2000;
|
|
// const USERS_PER_SECOND: u64 = 200; // Find limits
|
|
// const TEST_DURATION: Duration = Duration::from_secs(20);
|
|
|
|
|
|
const GROUP_NAME: &str = "global-chat";
|
|
const NUM_TEST_RUNS: usize = 5; // Number of consecutive tests
|
|
|
|
// --- Metrics ---
|
|
#[derive(Default)]
|
|
struct TestMetrics {
|
|
connections_successful: AtomicU64,
|
|
connections_failed: AtomicU64,
|
|
logins_successful: AtomicU64,
|
|
messages_sent: AtomicU64,
|
|
messages_received: AtomicU64,
|
|
round_trip_latency_us: AtomicU64,
|
|
round_trip_count: AtomicU64,
|
|
group_joins_attempted: AtomicU64,
|
|
group_joins_successful: AtomicU64,
|
|
group_joins_failed: AtomicU64,
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
struct TestResult {
|
|
run_number: usize,
|
|
duration_secs: f32,
|
|
connections_successful: u64,
|
|
connections_failed: u64,
|
|
logins_successful: u64,
|
|
group_joins_successful: u64, // Add this
|
|
group_joins_failed: u64, // Add this
|
|
messages_sent: u64,
|
|
messages_received: u64,
|
|
avg_latency_ms: f64,
|
|
connection_success_rate: f64,
|
|
login_success_rate: f64,
|
|
group_join_success_rate: f64, // Add this
|
|
msg_per_sec: f64,
|
|
message_amplification: f64,
|
|
}
|
|
|
|
impl TestResult {
|
|
fn from_metrics(run_number: usize, metrics: &TestMetrics, duration: Duration) -> Self {
|
|
let connections_successful = metrics.connections_successful.load(Ordering::Relaxed);
|
|
let connections_failed = metrics.connections_failed.load(Ordering::Relaxed);
|
|
let logins_successful = metrics.logins_successful.load(Ordering::Relaxed);
|
|
let messages_sent = metrics.messages_sent.load(Ordering::Relaxed);
|
|
let messages_received = metrics.messages_received.load(Ordering::Relaxed);
|
|
let rt_count = metrics.round_trip_count.load(Ordering::Relaxed);
|
|
let rt_total_us = metrics.round_trip_latency_us.load(Ordering::Relaxed);
|
|
|
|
let avg_latency_ms = if rt_count > 0 {
|
|
(rt_total_us as f64 / rt_count as f64) / 1000.0
|
|
} else {
|
|
0.0
|
|
};
|
|
|
|
let connection_success_rate = (connections_successful as f64 / TOTAL_USERS as f64) * 100.0;
|
|
let login_success_rate = if connections_successful > 0 {
|
|
(logins_successful as f64 / connections_successful as f64) * 100.0
|
|
} else {
|
|
0.0
|
|
};
|
|
|
|
let msg_per_sec = messages_sent as f64 / duration.as_secs_f64();
|
|
let message_amplification = if messages_sent > 0 {
|
|
messages_received as f64 / messages_sent as f64
|
|
} else {
|
|
0.0
|
|
};
|
|
|
|
let group_joins_successful = metrics.group_joins_successful.load(Ordering::Relaxed);
|
|
let group_joins_failed = metrics.group_joins_failed.load(Ordering::Relaxed);
|
|
let group_joins_attempted = metrics.group_joins_attempted.load(Ordering::Relaxed);
|
|
|
|
let group_join_success_rate = if group_joins_attempted > 0 {
|
|
(group_joins_successful as f64 / group_joins_attempted as f64) * 100.0
|
|
} else {
|
|
0.0
|
|
};
|
|
|
|
Self {
|
|
run_number,
|
|
duration_secs: duration.as_secs_f32(),
|
|
connections_successful,
|
|
connections_failed,
|
|
logins_successful,
|
|
messages_sent,
|
|
messages_received,
|
|
avg_latency_ms,
|
|
connection_success_rate,
|
|
login_success_rate,
|
|
msg_per_sec,
|
|
message_amplification,
|
|
group_joins_successful,
|
|
group_joins_failed,
|
|
group_join_success_rate,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|
println!("🚀 Starting Chat Server Consistency Load Test...");
|
|
println!(" Running {} consecutive tests", NUM_TEST_RUNS);
|
|
println!(
|
|
" Each test: {} users over {} seconds, connecting at {}/sec.",
|
|
TOTAL_USERS,
|
|
TEST_DURATION.as_secs(),
|
|
USERS_PER_SECOND
|
|
);
|
|
println!("{}", "=".repeat(60));
|
|
println!();
|
|
|
|
let mut results = Vec::new();
|
|
|
|
for run in 1..=NUM_TEST_RUNS {
|
|
println!("🔄 Starting Test Run {} of {}", run, NUM_TEST_RUNS);
|
|
|
|
if run > 1 {
|
|
if let Err(e) = reset_server().await {
|
|
println!("⚠️ Failed to reset server: {}", e);
|
|
}
|
|
// Wait a bit for reset to complete
|
|
sleep(Duration::from_secs(2)).await;
|
|
}
|
|
|
|
let result = run_single_test(run).await;
|
|
results.push(result.clone());
|
|
|
|
print_single_result(&result);
|
|
|
|
if run < NUM_TEST_RUNS {
|
|
// Much longer cooldown for 1000 users with high message volume
|
|
let cooldown_secs = if TOTAL_USERS >= 1000 { 30 } else { 10 };
|
|
println!("⏳ Waiting {} seconds before next test...\n", cooldown_secs);
|
|
sleep(Duration::from_secs(cooldown_secs)).await;
|
|
}
|
|
}
|
|
|
|
println!();
|
|
println!("{}", "=".repeat(60));
|
|
println!("📊 CONSISTENCY ANALYSIS");
|
|
println!("{}", "=".repeat(60));
|
|
|
|
analyze_consistency(&results);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn print_single_result(result: &TestResult) {
|
|
println!("📈 Test Run {} Results:", result.run_number);
|
|
println!(" Duration: {:.2}s", result.duration_secs);
|
|
println!(
|
|
" Connections: {} successful, {} failed ({:.1}% success)",
|
|
result.connections_successful, result.connections_failed, result.connection_success_rate
|
|
);
|
|
println!(
|
|
" Logins: {} successful ({:.1}% of connections)",
|
|
result.logins_successful, result.login_success_rate
|
|
);
|
|
println!(
|
|
" Messages: {} sent, {} received ({:.1}x amplification)",
|
|
result.messages_sent, result.messages_received, result.message_amplification
|
|
);
|
|
println!(" Throughput: {:.1} msg/sec", result.msg_per_sec);
|
|
println!(" Latency: {:.1}ms average", result.avg_latency_ms);
|
|
println!();
|
|
}
|
|
|
|
fn analyze_consistency(results: &[TestResult]) {
|
|
if results.is_empty() {
|
|
return;
|
|
}
|
|
|
|
// Calculate statistics
|
|
let connection_rates: Vec<f64> = results.iter().map(|r| r.connection_success_rate).collect();
|
|
let login_rates: Vec<f64> = results.iter().map(|r| r.login_success_rate).collect();
|
|
let latencies: Vec<f64> = results.iter().map(|r| r.avg_latency_ms).collect();
|
|
let throughputs: Vec<f64> = results.iter().map(|r| r.msg_per_sec).collect();
|
|
let amplifications: Vec<f64> = results.iter().map(|r| r.message_amplification).collect();
|
|
|
|
println!("📊 SUMMARY TABLE:");
|
|
println!("┌─────┬──────────┬────────┬─────────┬───────────┬──────────┐");
|
|
println!("│ Run │ Conn % │ Login %│ Lat(ms) │ Msg/sec │ Ampl │");
|
|
println!("├─────┼──────────┼────────┼─────────┼───────────┼──────────┤");
|
|
|
|
for result in results {
|
|
println!(
|
|
"│ {:3} │ {:8.1} │ {:6.1} │ {:7.1} │ {:9.1} │ {:8.1} │",
|
|
result.run_number,
|
|
result.connection_success_rate,
|
|
result.login_success_rate,
|
|
result.avg_latency_ms,
|
|
result.msg_per_sec,
|
|
result.message_amplification
|
|
);
|
|
}
|
|
println!("└─────┴──────────┴────────┴─────────┴───────────┴──────────┘");
|
|
|
|
println!("\n📈 CONSISTENCY METRICS:");
|
|
|
|
// Connection success rate
|
|
let (conn_avg, conn_std, conn_cv) = calculate_stats(&connection_rates);
|
|
println!("Connection Success Rate:");
|
|
println!(
|
|
" Average: {:.1}% StdDev: {:.2}% CV: {:.1}%",
|
|
conn_avg, conn_std, conn_cv
|
|
);
|
|
print_consistency_rating(conn_cv, "connection reliability");
|
|
|
|
// Login success rate
|
|
let (login_avg, login_std, login_cv) = calculate_stats(&login_rates);
|
|
println!("Login Success Rate:");
|
|
println!(
|
|
" Average: {:.1}% StdDev: {:.2}% CV: {:.1}%",
|
|
login_avg, login_std, login_cv
|
|
);
|
|
print_consistency_rating(login_cv, "login reliability");
|
|
|
|
// Latency
|
|
let (lat_avg, lat_std, lat_cv) = calculate_stats(&latencies);
|
|
println!("Average Latency:");
|
|
println!(
|
|
" Average: {:.1}ms StdDev: {:.2}ms CV: {:.1}%",
|
|
lat_avg, lat_std, lat_cv
|
|
);
|
|
print_consistency_rating(lat_cv, "latency consistency");
|
|
|
|
// Throughput
|
|
let (thr_avg, thr_std, thr_cv) = calculate_stats(&throughputs);
|
|
println!("Throughput:");
|
|
println!(
|
|
" Average: {:.1} msg/s StdDev: {:.2} msg/s CV: {:.1}%",
|
|
thr_avg, thr_std, thr_cv
|
|
);
|
|
print_consistency_rating(thr_cv, "throughput consistency");
|
|
|
|
// Message amplification
|
|
let (amp_avg, amp_std, amp_cv) = calculate_stats(&lifications);
|
|
println!("Message Amplification:");
|
|
println!(
|
|
" Average: {:.1}x StdDev: {:.2}x CV: {:.1}%",
|
|
amp_avg, amp_std, amp_cv
|
|
);
|
|
print_consistency_rating(amp_cv, "broadcast consistency");
|
|
|
|
println!("\n🎯 OVERALL ASSESSMENT:");
|
|
let overall_cv = (conn_cv + login_cv + lat_cv + thr_cv + amp_cv) / 5.0;
|
|
print_overall_rating(overall_cv);
|
|
|
|
// Identify problematic runs
|
|
identify_outliers(results);
|
|
}
|
|
|
|
fn calculate_stats(values: &[f64]) -> (f64, f64, f64) {
|
|
let n = values.len() as f64;
|
|
let mean = values.iter().sum::<f64>() / n;
|
|
let variance = values.iter().map(|x| (x - mean).powi(2)).sum::<f64>() / n;
|
|
let std_dev = variance.sqrt();
|
|
let cv = if mean != 0.0 {
|
|
(std_dev / mean.abs()) * 100.0
|
|
} else {
|
|
0.0
|
|
};
|
|
(mean, std_dev, cv)
|
|
}
|
|
|
|
fn print_consistency_rating(cv: f64, metric_name: &str) {
|
|
let rating = match cv {
|
|
cv if cv < 5.0 => "🟢 EXCELLENT",
|
|
cv if cv < 10.0 => "🟡 GOOD",
|
|
cv if cv < 20.0 => "🟠 FAIR",
|
|
_ => "🔴 POOR",
|
|
};
|
|
println!(" Consistency: {} ({})", rating, metric_name);
|
|
}
|
|
|
|
fn print_overall_rating(overall_cv: f64) {
|
|
let (rating, recommendation) = match overall_cv {
|
|
cv if cv < 5.0 => (
|
|
"🟢 EXCELLENT",
|
|
"System is highly consistent and production-ready",
|
|
),
|
|
cv if cv < 10.0 => ("🟡 GOOD", "System is stable with minor variations"),
|
|
cv if cv < 20.0 => (
|
|
"🟠 FAIR",
|
|
"System has noticeable inconsistencies, investigate bottlenecks",
|
|
),
|
|
_ => ("🔴 POOR", "System is unreliable, requires optimization"),
|
|
};
|
|
|
|
println!("Overall Consistency: {} (CV: {:.1}%)", rating, overall_cv);
|
|
println!("Recommendation: {}", recommendation);
|
|
}
|
|
|
|
fn identify_outliers(results: &[TestResult]) {
|
|
println!("\n🔍 OUTLIER ANALYSIS:");
|
|
|
|
// Find runs with significantly different performance
|
|
let latencies: Vec<f64> = results.iter().map(|r| r.avg_latency_ms).collect();
|
|
let (lat_mean, lat_std, _) = calculate_stats(&latencies);
|
|
|
|
let login_rates: Vec<f64> = results.iter().map(|r| r.login_success_rate).collect();
|
|
let (login_mean, login_std, _) = calculate_stats(&login_rates);
|
|
|
|
let mut outliers_found = false;
|
|
|
|
for result in results {
|
|
let mut issues = Vec::new();
|
|
|
|
if (result.avg_latency_ms - lat_mean).abs() > 2.0 * lat_std {
|
|
issues.push(format!(
|
|
"High latency variance ({:.1}ms vs {:.1}ms avg)",
|
|
result.avg_latency_ms, lat_mean
|
|
));
|
|
}
|
|
|
|
if (result.login_success_rate - login_mean).abs() > 2.0 * login_std {
|
|
issues.push(format!(
|
|
"Login rate variance ({:.1}% vs {:.1}% avg)",
|
|
result.login_success_rate, login_mean
|
|
));
|
|
}
|
|
|
|
if result.connection_success_rate < 95.0 {
|
|
issues.push("Low connection success rate".to_string());
|
|
}
|
|
|
|
if result.login_success_rate < 90.0 {
|
|
issues.push("Low login success rate".to_string());
|
|
}
|
|
|
|
if !issues.is_empty() {
|
|
outliers_found = true;
|
|
println!("⚠️ Run {}: {}", result.run_number, issues.join(", "));
|
|
}
|
|
}
|
|
|
|
if !outliers_found {
|
|
println!("✅ No significant outliers detected");
|
|
}
|
|
}
|
|
|
|
// Define aliases for the writer and reader types for clarity
|
|
type WsWriter = SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>;
|
|
type WsReader = SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>;
|
|
|
|
async fn simulate_user(user_id: u64, metrics: Arc<TestMetrics>) {
|
|
// --- 1. Connect ---
|
|
// Retry connection up to 3 times
|
|
let mut attempts = 0;
|
|
let ws_stream = loop {
|
|
attempts += 1;
|
|
match connect_async(SERVER_URL).await {
|
|
Ok((stream, _)) => {
|
|
metrics.connections_successful.fetch_add(1, Ordering::Relaxed);
|
|
break stream;
|
|
}
|
|
Err(e) => {
|
|
if attempts >= 3 {
|
|
metrics.connections_failed.fetch_add(1, Ordering::Relaxed);
|
|
return;
|
|
}
|
|
// Wait before retry with exponential backoff
|
|
sleep(Duration::from_millis(100 * attempts)).await;
|
|
}
|
|
}
|
|
};
|
|
|
|
let (write, mut read): (WsWriter, WsReader) = ws_stream.split();
|
|
let write = Arc::new(Mutex::new(write));
|
|
let write_clone = Arc::clone(&write);
|
|
|
|
let ping_task = tokio::spawn(async move {
|
|
let mut ping_interval = tokio::time::interval(Duration::from_secs(10));
|
|
loop {
|
|
ping_interval.tick().await;
|
|
|
|
let ping_msg = ClientToServerMsg::Ping {
|
|
timestamp: SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_micros(),
|
|
};
|
|
|
|
if let Ok(bytes) = postcard::to_allocvec(&ping_msg) {
|
|
if write_clone
|
|
.lock()
|
|
.await
|
|
.send(Message::Binary(bytes.into()))
|
|
.await
|
|
.is_err()
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
// --- 2. Login ---
|
|
let login_msg = ClientToServerMsg::Login {
|
|
username: format!("user_{}", user_id),
|
|
};
|
|
if let Ok(bytes) = postcard::to_allocvec(&login_msg) {
|
|
if write
|
|
.lock()
|
|
.await
|
|
.send(Message::Binary(bytes.into()))
|
|
.await
|
|
.is_err()
|
|
{
|
|
ping_task.abort();
|
|
return;
|
|
}
|
|
}
|
|
|
|
let login_confirmed = tokio::time::timeout(Duration::from_secs(5), async {
|
|
while let Some(Ok(Message::Binary(bin))) = read.next().await {
|
|
if let Ok(ServerToClientMsg::LoginSuccess { .. }) = postcard::from_bytes(&bin) {
|
|
return true;
|
|
}
|
|
}
|
|
false
|
|
})
|
|
.await
|
|
.unwrap_or(false);
|
|
|
|
if !login_confirmed {
|
|
ping_task.abort();
|
|
// Close the connection properly
|
|
let _ = write.lock().await.close().await;
|
|
return;
|
|
}
|
|
metrics.logins_successful.fetch_add(1, Ordering::Relaxed);
|
|
|
|
// --- 3. Join Group ---
|
|
let join_msg = ClientToServerMsg::JoinGroup {
|
|
group: GROUP_NAME.to_string(),
|
|
};
|
|
|
|
metrics.group_joins_attempted.fetch_add(1, Ordering::Relaxed);
|
|
|
|
if let Ok(bytes) = postcard::to_allocvec(&join_msg) {
|
|
if write
|
|
.lock()
|
|
.await
|
|
.send(Message::Binary(bytes.into()))
|
|
.await
|
|
.is_err()
|
|
{
|
|
metrics.group_joins_failed.fetch_add(1, Ordering::Relaxed);
|
|
ping_task.abort();
|
|
let _ = write.lock().await.close().await;
|
|
return;
|
|
}
|
|
}
|
|
|
|
// Wait for join confirmation
|
|
let join_confirmed = tokio::time::timeout(Duration::from_secs(5), async {
|
|
while let Some(Ok(Message::Binary(bin))) = read.next().await {
|
|
match postcard::from_bytes::<ServerToClientMsg>(&bin) {
|
|
Ok(ServerToClientMsg::JoinedGroup { group }) if group == GROUP_NAME => {
|
|
return true;
|
|
}
|
|
Ok(ServerToClientMsg::Error { .. }) => {
|
|
return false;
|
|
}
|
|
_ => continue,
|
|
}
|
|
}
|
|
false
|
|
})
|
|
.await
|
|
.unwrap_or(false);
|
|
|
|
if join_confirmed {
|
|
metrics.group_joins_successful.fetch_add(1, Ordering::Relaxed);
|
|
} else {
|
|
metrics.group_joins_failed.fetch_add(1, Ordering::Relaxed);
|
|
ping_task.abort();
|
|
let _ = write.lock().await.close().await;
|
|
return;
|
|
}
|
|
|
|
// --- 4. Messaging loop ---
|
|
let metrics_clone_reader = Arc::clone(&metrics);
|
|
let reader_task = tokio::spawn(async move {
|
|
while let Some(Ok(Message::Binary(bin))) = read.next().await {
|
|
metrics_clone_reader
|
|
.messages_received
|
|
.fetch_add(1, Ordering::Relaxed);
|
|
|
|
if let Ok(ServerToClientMsg::Pong { timestamp }) = postcard::from_bytes(&bin) {
|
|
let now_micros = SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_micros();
|
|
let rtt_us = now_micros.saturating_sub(timestamp);
|
|
if rtt_us < 1_000_000 {
|
|
metrics_clone_reader
|
|
.round_trip_latency_us
|
|
.fetch_add(rtt_us as u64, Ordering::Relaxed);
|
|
metrics_clone_reader
|
|
.round_trip_count
|
|
.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
let test_start_time = Instant::now();
|
|
let mut message_interval = tokio::time::interval(Duration::from_millis(500));
|
|
|
|
while test_start_time.elapsed() < TEST_DURATION {
|
|
message_interval.tick().await;
|
|
|
|
let msg = if metrics.messages_sent.load(Ordering::Relaxed) % 5 == 0 {
|
|
ClientToServerMsg::Ping {
|
|
timestamp: SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_micros(),
|
|
}
|
|
} else {
|
|
ClientToServerMsg::SendGroup {
|
|
group: GROUP_NAME.to_string(),
|
|
text: "hello group".to_string(),
|
|
}
|
|
};
|
|
|
|
if let Ok(bytes) = postcard::to_allocvec(&msg) {
|
|
if write
|
|
.lock()
|
|
.await
|
|
.send(Message::Binary(bytes.into()))
|
|
.await
|
|
.is_ok()
|
|
{
|
|
metrics.messages_sent.fetch_add(1, Ordering::Relaxed);
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
let messaging_duration = test_start_time.elapsed();
|
|
if messaging_duration < TEST_DURATION - Duration::from_secs(5) {
|
|
// User exited early - sample the logging to reduce spam
|
|
let count = EARLY_EXIT_COUNT.fetch_add(1, Ordering::Relaxed);
|
|
|
|
// Only log every 50th early exit, or the first one
|
|
if count == 0 || count % 50 == 0 {
|
|
eprintln!(
|
|
"⚠️ {} users exited early (user {} after {:.1}s, expected ~{}s)",
|
|
count + 1,
|
|
user_id,
|
|
messaging_duration.as_secs_f32(),
|
|
TEST_DURATION.as_secs()
|
|
);
|
|
}
|
|
}
|
|
|
|
|
|
// --- 5. PROPER CLEANUP ---
|
|
// Stop background tasks
|
|
ping_task.abort();
|
|
reader_task.abort();
|
|
|
|
// Send WebSocket close frame
|
|
{
|
|
let mut writer = write.lock().await;
|
|
|
|
// Send close frame
|
|
let _ = writer.send(Message::Close(None)).await;
|
|
|
|
// Flush and close
|
|
let _ = writer.flush().await;
|
|
let _ = writer.close().await;
|
|
}
|
|
}
|
|
|
|
async fn reset_server() -> Result<(), Box<dyn std::error::Error>> {
|
|
let client = reqwest::Client::builder()
|
|
.timeout(Duration::from_secs(60))
|
|
.build()?;
|
|
|
|
println!("🔄 Resetting server state...");
|
|
|
|
// Check health first
|
|
for attempt in 0..5 {
|
|
match client.get("http://localhost:8080/health")
|
|
.timeout(Duration::from_secs(2))
|
|
.send()
|
|
.await
|
|
{
|
|
Ok(resp) if resp.status().is_success() => break,
|
|
_ if attempt < 4 => {
|
|
println!(" Waiting for server to respond... ({}/5)", attempt + 1);
|
|
sleep(Duration::from_secs(2)).await;
|
|
}
|
|
_ => {
|
|
println!("❌ Server not responding after 10s");
|
|
return Err("Server unresponsive".into());
|
|
}
|
|
}
|
|
}
|
|
|
|
// Request reset
|
|
match client.post("http://localhost:8080/admin/reset").send().await {
|
|
Ok(resp) if resp.status().is_success() => {
|
|
println!(" Reset signal sent");
|
|
}
|
|
Ok(resp) => {
|
|
println!(" Reset returned status: {}", resp.status());
|
|
}
|
|
Err(e) => {
|
|
println!(" Reset request failed: {}", e);
|
|
// Server might be overwhelmed, wait and retry
|
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
|
client.post("http://localhost:8080/admin/reset").send().await?;
|
|
}
|
|
}
|
|
|
|
// Poll until connections reach 0
|
|
println!(" Waiting for connections to drain...");
|
|
for i in 0..60 {
|
|
sleep(Duration::from_secs(1)).await;
|
|
|
|
if let Ok(resp) = client.get("http://localhost:8080/stats")
|
|
.timeout(Duration::from_secs(2))
|
|
.send()
|
|
.await
|
|
{
|
|
if let Ok(stats) = resp.json::<serde_json::Value>().await {
|
|
let connections = stats["total_connections"].as_u64().unwrap_or(u64::MAX);
|
|
print!("\r Connections: {} ({}s elapsed) ", connections, i + 1);
|
|
std::io::Write::flush(&mut std::io::stdout()).unwrap();
|
|
|
|
if connections == 0 {
|
|
println!("\n✅ Reset complete");
|
|
// Extra wait for OS to release ports
|
|
sleep(Duration::from_secs(3)).await;
|
|
return Ok(());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
for i in 1..=30 { // 30 seconds max
|
|
tokio::time::sleep(Duration::from_secs(1)).await;
|
|
|
|
match client.get("http://localhost:8080/health").send().await {
|
|
Ok(resp) if resp.status().is_success() => {
|
|
// Check connection count from response or separate endpoint
|
|
println!(" Server healthy after {}s", i);
|
|
return Ok(());
|
|
}
|
|
Ok(_) => {
|
|
println!(" Waiting for server to respond... ({}/30)", i);
|
|
}
|
|
Err(_) => {
|
|
println!(" Server not responding... ({}/30)", i);
|
|
}
|
|
}
|
|
}
|
|
|
|
println!("\n⚠️ Connections still remaining, waiting longer...");
|
|
sleep(Duration::from_secs(10)).await;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn run_single_test(run_number: usize) -> TestResult {
|
|
|
|
EARLY_EXIT_COUNT.store(0, Ordering::Relaxed);
|
|
|
|
let metrics = Arc::new(TestMetrics::default());
|
|
let mut user_handles: Vec<tokio::task::JoinHandle<()>> = Vec::with_capacity(TOTAL_USERS as usize);
|
|
|
|
let connect_interval = if USERS_PER_SECOND > 0 {
|
|
Duration::from_millis(1000 / USERS_PER_SECOND)
|
|
} else {
|
|
Duration::from_secs(0)
|
|
};
|
|
|
|
let test_start_time = Instant::now();
|
|
|
|
// Start progress reporter
|
|
let metrics_clone = metrics.clone();
|
|
let progress_task = tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(Duration::from_secs(2));
|
|
loop {
|
|
interval.tick().await;
|
|
|
|
let attempted = metrics_clone.group_joins_attempted.load(Ordering::Relaxed);
|
|
let successful = metrics_clone.group_joins_successful.load(Ordering::Relaxed);
|
|
|
|
if attempted > 0 {
|
|
let success_rate = (successful as f64 / attempted as f64) * 100.0;
|
|
print!(
|
|
"\r⏳ Progress: {}/{} users joined ({:.1}% success rate)",
|
|
successful, TOTAL_USERS, success_rate
|
|
);
|
|
std::io::Write::flush(&mut std::io::stdout()).unwrap();
|
|
}
|
|
|
|
if attempted >= TOTAL_USERS {
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
|
|
// Spawn user simulation tasks
|
|
for i in 0..TOTAL_USERS {
|
|
let metrics_clone = Arc::clone(&metrics);
|
|
let handle = tokio::spawn(async move {
|
|
simulate_user(i, metrics_clone).await;
|
|
});
|
|
user_handles.push(handle);
|
|
|
|
if USERS_PER_SECOND > 0 {
|
|
sleep(connect_interval).await;
|
|
}
|
|
}
|
|
|
|
// Wait for ALL user tasks to complete (including cleanup)
|
|
for handle in user_handles {
|
|
let _ = handle.await;
|
|
}
|
|
|
|
// Stop progress reporter
|
|
progress_task.abort();
|
|
|
|
// Print summary
|
|
let total_attempted = metrics.group_joins_attempted.load(Ordering::Relaxed);
|
|
let total_successful = metrics.group_joins_successful.load(Ordering::Relaxed);
|
|
let total_failed = metrics.group_joins_failed.load(Ordering::Relaxed);
|
|
|
|
println!(
|
|
"\n✅ Join Summary: {}/{} successful, {} failed ({:.1}% success rate)",
|
|
total_successful,
|
|
total_attempted,
|
|
total_failed,
|
|
if total_attempted > 0 {
|
|
(total_successful as f64 / total_attempted as f64) * 100.0
|
|
} else {
|
|
0.0
|
|
}
|
|
);
|
|
|
|
// Wait for TCP cleanup
|
|
println!(" Waiting for TCP cleanup...");
|
|
sleep(Duration::from_secs(2)).await;
|
|
|
|
let elapsed = test_start_time.elapsed();
|
|
TestResult::from_metrics(run_number, &metrics, elapsed)
|
|
}
|