YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-03-31 09:45:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add aquatic_load_tester: multi-run multi-implementation load tests

Browse source 
- Work in progress
- Only UDP is currently implemented so far
- Also includes some changes to other crates, notably deriving
  serde Serialize for Config structs and making udp_load_test
  a lib and a binary
This commit is contained in:
Joakim Frostegård 2023-12-17 21:57:11 +01:00
parent c7997d5aed
commit afc3deb656
18 changed files with 1666 additions and 312 deletions
Download patch file Download diff file Expand all files Collapse all files

3
crates/udp_load_test/Cargo.toml
View file

@ -13,6 +13,9 @@ rust-version.workspace = true
[features]
cpu-pinning = ["aquatic_common/hwloc"]
[lib]
name = "aquatic_udp_load_test"
[[bin]]
name = "aquatic_udp_load_test"

11
crates/udp_load_test/src/config.rs
View file

@ -1,6 +1,6 @@
use std::net::SocketAddr;
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use aquatic_common::cli::LogLevel;
#[cfg(feature = "cpu-pinning")]
@ -8,7 +8,7 @@ use aquatic_common::cpu_pinning::desc::CpuPinningConfigDesc;
use aquatic_toml_config::TomlConfig;
/// aquatic_udp_load_test configuration
#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct Config {
/// Server address
@ -42,7 +42,7 @@ impl Default for Config {
}
}
#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct NetworkConfig {
/// True means bind to one localhost IP per socket.
@ -56,8 +56,6 @@ pub struct NetworkConfig {
pub first_port: u16,
/// Socket worker poll timeout in microseconds
pub poll_timeout: u64,
/// Socket worker polling event number
pub poll_event_capacity: usize,
/// Size of socket recv buffer. Use 0 for OS default.
///
/// This setting can have a big impact on dropped packages. It might
@ -79,13 +77,12 @@ impl Default for NetworkConfig {
multiple_client_ipv4s: true,
first_port: 45_000,
poll_timeout: 276,
poll_event_capacity: 2_877,
recv_buffer: 6_000_000,
}
}
}
#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct RequestConfig {
/// Number of torrents to simulate

191
crates/udp_load_test/src/lib.rs Normal file
View file

@ -0,0 +1,191 @@
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::AtomicUsize;
use std::sync::{atomic::Ordering, Arc};
use std::thread::{self, Builder};
use std::time::{Duration, Instant};
#[cfg(feature = "cpu-pinning")]
use aquatic_common::cpu_pinning::{pin_current_if_configured_to, WorkerIndex};
use rand_distr::Gamma;
pub mod common;
pub mod config;
pub mod utils;
pub mod worker;
use common::*;
use config::Config;
use utils::*;
use worker::*;
impl aquatic_common::cli::Config for Config {
fn get_log_level(&self) -> Option<aquatic_common::cli::LogLevel> {
Some(self.log_level)
}
}
pub fn run(config: Config) -> ::anyhow::Result<()> {
if config.requests.weight_announce
+ config.requests.weight_connect
+ config.requests.weight_scrape
== 0
{
panic!("Error: at least one weight must be larger than zero.");
}
println!("Starting client with config: {:#?}", config);
let mut info_hashes = Vec::with_capacity(config.requests.number_of_torrents);
for _ in 0..config.requests.number_of_torrents {
info_hashes.push(generate_info_hash());
}
let state = LoadTestState {
info_hashes: Arc::new(info_hashes),
statistics: Arc::new(Statistics::default()),
};
let gamma = Gamma::new(
config.requests.torrent_gamma_shape,
config.requests.torrent_gamma_scale,
)
.unwrap();
// Start workers
for i in 0..config.workers {
let port = config.network.first_port + (i as u16);
let ip = if config.server_address.is_ipv6() {
Ipv6Addr::LOCALHOST.into()
} else {
if config.network.multiple_client_ipv4s {
Ipv4Addr::new(127, 0, 0, 1 + i).into()
} else {
Ipv4Addr::LOCALHOST.into()
}
};
let addr = SocketAddr::new(ip, port);
let config = config.clone();
let state = state.clone();
Builder::new().name("load-test".into()).spawn(move || {
#[cfg(feature = "cpu-pinning")]
pin_current_if_configured_to(
&config.cpu_pinning,
config.workers as usize,
0,
WorkerIndex::SocketWorker(i as usize),
);
run_worker_thread(state, gamma, &config, addr)
})?;
}
#[cfg(feature = "cpu-pinning")]
pin_current_if_configured_to(
&config.cpu_pinning,
config.workers as usize,
0,
WorkerIndex::Util,
);
monitor_statistics(state, &config);
Ok(())
}
fn monitor_statistics(state: LoadTestState, config: &Config) {
let mut report_avg_connect: Vec<f64> = Vec::new();
let mut report_avg_announce: Vec<f64> = Vec::new();
let mut report_avg_scrape: Vec<f64> = Vec::new();
let mut report_avg_error: Vec<f64> = Vec::new();
let interval = 5;
let start_time = Instant::now();
let duration = Duration::from_secs(config.duration as u64);
let mut last = start_time;
let time_elapsed = loop {
thread::sleep(Duration::from_secs(interval));
let requests = fetch_and_reset(&state.statistics.requests);
let response_peers = fetch_and_reset(&state.statistics.response_peers);
let responses_connect = fetch_and_reset(&state.statistics.responses_connect);
let responses_announce = fetch_and_reset(&state.statistics.responses_announce);
let responses_scrape = fetch_and_reset(&state.statistics.responses_scrape);
let responses_error = fetch_and_reset(&state.statistics.responses_error);
let now = Instant::now();
let elapsed = (now - last).as_secs_f64();
last = now;
let peers_per_announce_response = response_peers / responses_announce;
let avg_requests = requests / elapsed;
let avg_responses_connect = responses_connect / elapsed;
let avg_responses_announce = responses_announce / elapsed;
let avg_responses_scrape = responses_scrape / elapsed;
let avg_responses_error = responses_error / elapsed;
let avg_responses = avg_responses_connect
+ avg_responses_announce
+ avg_responses_scrape
+ avg_responses_error;
report_avg_connect.push(avg_responses_connect);
report_avg_announce.push(avg_responses_announce);
report_avg_scrape.push(avg_responses_scrape);
report_avg_error.push(avg_responses_error);
println!();
println!("Requests out: {:.2}/second", avg_requests);
println!("Responses in: {:.2}/second", avg_responses);
println!(" - Connect responses: {:.2}", avg_responses_connect);
println!(" - Announce responses: {:.2}", avg_responses_announce);
println!(" - Scrape responses: {:.2}", avg_responses_scrape);
println!(" - Error responses: {:.2}", avg_responses_error);
println!(
"Peers per announce response: {:.2}",
peers_per_announce_response
);
let time_elapsed = start_time.elapsed();
if config.duration != 0 && time_elapsed >= duration {
break time_elapsed;
}
};
let len = report_avg_connect.len() as f64;
let avg_connect: f64 = report_avg_connect.into_iter().sum::<f64>() / len;
let avg_announce: f64 = report_avg_announce.into_iter().sum::<f64>() / len;
let avg_scrape: f64 = report_avg_scrape.into_iter().sum::<f64>() / len;
let avg_error: f64 = report_avg_error.into_iter().sum::<f64>() / len;
let avg_total = avg_connect + avg_announce + avg_scrape + avg_error;
println!();
println!("# aquatic load test report");
println!();
println!("Test ran for {} seconds", time_elapsed.as_secs());
println!("Average responses per second: {:.2}", avg_total);
println!(" - Connect responses: {:.2}", avg_connect);
println!(" - Announce responses: {:.2}", avg_announce);
println!(" - Scrape responses: {:.2}", avg_scrape);
println!(" - Error responses: {:.2}", avg_error);
println!();
println!("Config: {:#?}", config);
println!();
}
fn fetch_and_reset(atomic_usize: &AtomicUsize) -> f64 {
atomic_usize.fetch_and(0, Ordering::Relaxed) as f64
}

194
crates/udp_load_test/src/main.rs
View file

@ -1,22 +1,4 @@
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::AtomicUsize;
use std::sync::{atomic::Ordering, Arc};
use std::thread::{self, Builder};
use std::time::{Duration, Instant};
#[cfg(feature = "cpu-pinning")]
use aquatic_common::cpu_pinning::{pin_current_if_configured_to, WorkerIndex};
use rand_distr::Gamma;
mod common;
mod config;
mod utils;
mod worker;
use common::*;
use config::Config;
use utils::*;
use worker::*;
use aquatic_udp_load_test::config::Config;
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
@ -25,179 +7,7 @@ pub fn main() {
aquatic_common::cli::run_app_with_cli_and_config::<Config>(
"aquatic_udp_load_test: BitTorrent load tester",
env!("CARGO_PKG_VERSION"),
run,
aquatic_udp_load_test::run,
None,
)
}
impl aquatic_common::cli::Config for Config {
fn get_log_level(&self) -> Option<aquatic_common::cli::LogLevel> {
Some(self.log_level)
}
}
fn run(config: Config) -> ::anyhow::Result<()> {
if config.requests.weight_announce
+ config.requests.weight_connect
+ config.requests.weight_scrape
== 0
{
panic!("Error: at least one weight must be larger than zero.");
}
println!("Starting client with config: {:#?}", config);
let mut info_hashes = Vec::with_capacity(config.requests.number_of_torrents);
for _ in 0..config.requests.number_of_torrents {
info_hashes.push(generate_info_hash());
}
let state = LoadTestState {
info_hashes: Arc::new(info_hashes),
statistics: Arc::new(Statistics::default()),
};
let gamma = Gamma::new(
config.requests.torrent_gamma_shape,
config.requests.torrent_gamma_scale,
)
.unwrap();
// Start workers
for i in 0..config.workers {
let port = config.network.first_port + (i as u16);
let ip = if config.server_address.is_ipv6() {
Ipv6Addr::LOCALHOST.into()
} else {
if config.network.multiple_client_ipv4s {
Ipv4Addr::new(127, 0, 0, 1 + i).into()
} else {
Ipv4Addr::LOCALHOST.into()
}
};
let addr = SocketAddr::new(ip, port);
let config = config.clone();
let state = state.clone();
Builder::new().name("load-test".into()).spawn(move || {
#[cfg(feature = "cpu-pinning")]
pin_current_if_configured_to(
&config.cpu_pinning,
config.workers as usize,
0,
WorkerIndex::SocketWorker(i as usize),
);
run_worker_thread(state, gamma, &config, addr)
})?;
}
#[cfg(feature = "cpu-pinning")]
pin_current_if_configured_to(
&config.cpu_pinning,
config.workers as usize,
0,
WorkerIndex::Util,
);
monitor_statistics(state, &config);
Ok(())
}
fn monitor_statistics(state: LoadTestState, config: &Config) {
let mut report_avg_connect: Vec<f64> = Vec::new();
let mut report_avg_announce: Vec<f64> = Vec::new();
let mut report_avg_scrape: Vec<f64> = Vec::new();
let mut report_avg_error: Vec<f64> = Vec::new();
let interval = 5;
let start_time = Instant::now();
let duration = Duration::from_secs(config.duration as u64);
let mut last = start_time;
let time_elapsed = loop {
thread::sleep(Duration::from_secs(interval));
let requests = fetch_and_reset(&state.statistics.requests);
let response_peers = fetch_and_reset(&state.statistics.response_peers);
let responses_connect = fetch_and_reset(&state.statistics.responses_connect);
let responses_announce = fetch_and_reset(&state.statistics.responses_announce);
let responses_scrape = fetch_and_reset(&state.statistics.responses_scrape);
let responses_error = fetch_and_reset(&state.statistics.responses_error);
let now = Instant::now();
let elapsed = (now - last).as_secs_f64();
last = now;
let peers_per_announce_response = response_peers / responses_announce;
let avg_requests = requests / elapsed;
let avg_responses_connect = responses_connect / elapsed;
let avg_responses_announce = responses_announce / elapsed;
let avg_responses_scrape = responses_scrape / elapsed;
let avg_responses_error = responses_error / elapsed;
let avg_responses = avg_responses_connect
+ avg_responses_announce
+ avg_responses_scrape
+ avg_responses_error;
report_avg_connect.push(avg_responses_connect);
report_avg_announce.push(avg_responses_announce);
report_avg_scrape.push(avg_responses_scrape);
report_avg_error.push(avg_responses_error);
println!();
println!("Requests out: {:.2}/second", avg_requests);
println!("Responses in: {:.2}/second", avg_responses);
println!(" - Connect responses: {:.2}", avg_responses_connect);
println!(" - Announce responses: {:.2}", avg_responses_announce);
println!(" - Scrape responses: {:.2}", avg_responses_scrape);
println!(" - Error responses: {:.2}", avg_responses_error);
println!(
"Peers per announce response: {:.2}",
peers_per_announce_response
);
let time_elapsed = start_time.elapsed();
if config.duration != 0 && time_elapsed >= duration {
break time_elapsed;
}
};
let len = report_avg_connect.len() as f64;
let avg_connect: f64 = report_avg_connect.into_iter().sum::<f64>() / len;
let avg_announce: f64 = report_avg_announce.into_iter().sum::<f64>() / len;
let avg_scrape: f64 = report_avg_scrape.into_iter().sum::<f64>() / len;
let avg_error: f64 = report_avg_error.into_iter().sum::<f64>() / len;
let avg_total = avg_connect + avg_announce + avg_scrape + avg_error;
println!();
println!("# aquatic load test report");
println!();
println!("Test ran for {} seconds", time_elapsed.as_secs());
println!("Average responses per second: {:.2}", avg_total);
println!(" - Connect responses: {:.2}", avg_connect);
println!(" - Announce responses: {:.2}", avg_announce);
println!(" - Scrape responses: {:.2}", avg_scrape);
println!(" - Error responses: {:.2}", avg_error);
println!();
println!("Config: {:#?}", config);
println!();
}
fn fetch_and_reset(atomic_usize: &AtomicUsize) -> f64 {
atomic_usize.fetch_and(0, Ordering::Relaxed) as f64
}

2
crates/udp_load_test/src/worker/mod.rs
View file

@ -41,7 +41,7 @@ pub fn run_worker_thread(
.register(&mut socket, token, interests)
.unwrap();
let mut events = Events::with_capacity(config.network.poll_event_capacity);
let mut events = Events::with_capacity(1);
let mut statistics = SocketWorkerLocalStatistics::default();