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

Remove unused aquatic_udp_bench

Browse source
This commit is contained in:
Joakim Frostegård 2023-12-10 12:09:53 +01:00
parent 0c4140165b
commit a873dddb89
9 changed files with 2 additions and 546 deletions
Download patch file Download diff file Expand all files Collapse all files

90
Cargo.lock generated
View file

@ -261,24 +261,6 @@ dependencies = [
"tinytemplate",
]
[[package]]
name = "aquatic_udp_bench"
version = "0.8.0"
dependencies = [
"anyhow",
"aquatic_common",
"aquatic_toml_config",
"aquatic_udp",
"aquatic_udp_protocol",
"crossbeam-channel",
"indicatif",
"mimalloc",
"num-format",
"rand",
"rand_distr",
"serde",
]
[[package]]
name = "aquatic_udp_load_test"
version = "0.8.0"
@ -688,19 +670,6 @@ dependencies = [
"cache-padded",
]
[[package]]
name = "console"
version = "0.15.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c926e00cc70edefdc64d3a5ff31cc65bb97a3460097762bd23afb4d8145fccf8"
dependencies = [
"encode_unicode",
"lazy_static",
"libc",
"unicode-width",
"windows-sys 0.45.0",
]
[[package]]
name = "constant_time_eq"
version = "0.3.0"
@ -923,12 +892,6 @@ version = "1.1.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1056f553da426e9c025a662efa48b52e62e0a3a7648aa2d15aeaaf7f0d329357"
[[package]]
name = "encode_unicode"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a357d28ed41a50f9c765dbfe56cbc04a64e53e5fc58ba79fbc34c10ef3df831f"
[[package]]
name = "endian-type"
version = "0.1.2"
@ -1429,19 +1392,6 @@ dependencies = [
"hashbrown 0.14.2",
]
[[package]]
name = "indicatif"
version = "0.17.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fb28741c9db9a713d93deb3bb9515c20788cef5815265bee4980e87bde7e0f25"
dependencies = [
"console",
"instant",
"number_prefix",
"portable-atomic",
"unicode-width",
]
[[package]]
name = "instant"
version = "0.1.12"
@ -1905,12 +1855,6 @@ dependencies = [
"libc",
]
[[package]]
name = "number_prefix"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3"
[[package]]
name = "object"
version = "0.32.1"
@ -1979,7 +1923,7 @@ dependencies = [
"libc",
"redox_syscall",
"smallvec",
"windows-targets 0.48.5",
"windows-targets",
]
[[package]]
@ -2850,12 +2794,6 @@ dependencies = [
"tinyvec",
]
[[package]]
name = "unicode-width"
version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e51733f11c9c4f72aa0c160008246859e340b00807569a0da0e7a1079b27ba85"
[[package]]
name = "untrusted"
version = "0.9.0"
@ -3056,37 +2994,13 @@ dependencies = [
"windows_x86_64_msvc 0.42.2",
]
[[package]]
name = "windows-sys"
version = "0.45.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0"
dependencies = [
"windows-targets 0.42.2",
]
[[package]]
name = "windows-sys"
version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
dependencies = [
"windows-targets 0.48.5",
]
[[package]]
name = "windows-targets"
version = "0.42.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071"
dependencies = [
"windows_aarch64_gnullvm 0.42.2",
"windows_aarch64_msvc 0.42.2",
"windows_i686_gnu 0.42.2",
"windows_i686_msvc 0.42.2",
"windows_x86_64_gnu 0.42.2",
"windows_x86_64_gnullvm 0.42.2",
"windows_x86_64_msvc 0.42.2",
"windows-targets",
]
[[package]]

1
Cargo.toml
View file

@ -9,7 +9,6 @@ members = [
"crates/toml_config",
"crates/toml_config_derive",
"crates/udp",
"crates/udp_bench",
"crates/udp_load_test",
"crates/udp_protocol",
"crates/ws",

27
crates/udp_bench/Cargo.toml
View file

@ -1,27 +0,0 @@
[package]
name = "aquatic_udp_bench"
version.workspace = true
authors.workspace = true
edition.workspace = true
license.workspace = true
repository.workspace = true
readme.workspace = true
rust-version.workspace = true
[[bin]]
name = "aquatic_udp_bench"
[dependencies]
aquatic_common.workspace = true
aquatic_toml_config.workspace = true
aquatic_udp.workspace = true
aquatic_udp_protocol.workspace = true
anyhow = "1"
crossbeam-channel = "0.5"
indicatif = "0.17"
mimalloc = { version = "0.1", default-features = false }
num-format = "0.4"
rand_distr = "0.4"
rand = { version = "0.8", features = ["small_rng"] }
serde = { version = "1", features = ["derive"] }

116
crates/udp_bench/src/announce.rs
View file

@ -1,116 +0,0 @@
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::time::{Duration, Instant};
use aquatic_common::CanonicalSocketAddr;
use crossbeam_channel::{Receiver, Sender};
use indicatif::ProgressIterator;
use rand::Rng;
use rand_distr::Gamma;
use aquatic_udp::common::*;
use aquatic_udp_protocol::*;
use crate::common::*;
use crate::config::BenchConfig;
pub fn bench_announce_handler(
bench_config: &BenchConfig,
request_sender: &Sender<(SocketWorkerIndex, ConnectedRequest, CanonicalSocketAddr)>,
response_receiver: &Receiver<(ConnectedResponse, CanonicalSocketAddr)>,
rng: &mut impl Rng,
info_hashes: &[InfoHash],
) -> (usize, Duration) {
let requests = create_requests(rng, info_hashes, bench_config.num_announce_requests);
let p = 10_000 * bench_config.num_threads; // FIXME: adjust to sharded workers
let mut num_responses = 0usize;
let mut dummy: u16 = rng.gen();
let pb = create_progress_bar("Announce", bench_config.num_rounds as u64);
// Start benchmark
let before = Instant::now();
for round in (0..bench_config.num_rounds).progress_with(pb) {
for request_chunk in requests.chunks(p) {
for (request, src) in request_chunk {
request_sender
.send((
SocketWorkerIndex(0),
ConnectedRequest::Announce(request.clone()),
*src,
))
.unwrap();
}
while let Ok((ConnectedResponse::AnnounceIpv4(r), _)) = response_receiver.try_recv() {
num_responses += 1;
if let Some(last_peer) = r.peers.last() {
dummy ^= last_peer.port.0.get();
}
}
}
let total = bench_config.num_announce_requests * (round + 1);
while num_responses < total {
if let Ok((ConnectedResponse::AnnounceIpv4(r), _)) = response_receiver.recv() {
num_responses += 1;
if let Some(last_peer) = r.peers.last() {
dummy ^= last_peer.port.0.get();
}
}
}
}
let elapsed = before.elapsed();
if dummy == 0 {
println!("dummy dummy");
}
(num_responses, elapsed)
}
pub fn create_requests(
rng: &mut impl Rng,
info_hashes: &[InfoHash],
number: usize,
) -> Vec<(AnnounceRequest, CanonicalSocketAddr)> {
let gamma = Gamma::new(GAMMA_SHAPE, GAMMA_SCALE).unwrap();
let max_index = info_hashes.len() - 1;
let mut requests = Vec::new();
for _ in 0..number {
let info_hash_index = gamma_usize(rng, gamma, max_index);
let request = AnnounceRequest {
connection_id: ConnectionId::new(0),
action_placeholder: Default::default(),
transaction_id: TransactionId::new(rng.gen()),
info_hash: info_hashes[info_hash_index],
peer_id: PeerId(rng.gen()),
bytes_downloaded: NumberOfBytes::new(rng.gen()),
bytes_uploaded: NumberOfBytes::new(rng.gen()),
bytes_left: NumberOfBytes::new(rng.gen()),
event: AnnounceEvent::Started.into(),
ip_address: Ipv4AddrBytes([0; 4]),
key: PeerKey::new(rng.gen()),
peers_wanted: NumberOfPeers::new(rng.gen()),
port: Port::new(rng.gen()),
};
requests.push((
request,
CanonicalSocketAddr::new(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 1))),
));
}
requests
}

24
crates/udp_bench/src/common.rs
View file

@ -1,24 +0,0 @@
use indicatif::{ProgressBar, ProgressStyle};
use rand::Rng;
use rand_distr::Gamma;
pub const GAMMA_SHAPE: f64 = 0.2;
pub const GAMMA_SCALE: f64 = 100.0;
pub const NUM_INFO_HASHES: usize = 10_000;
pub fn create_progress_bar(name: &str, iterations: u64) -> ProgressBar {
let t = format!("{:<8} {}", name, "{wide_bar} {pos:>2}/{len:>2}");
let style = ProgressStyle::default_bar()
.template(&t)
.expect("setup indicatif template");
ProgressBar::new(iterations).with_style(style)
}
pub fn gamma_usize(rng: &mut impl Rng, gamma: Gamma<f64>, max: usize) -> usize {
let p: f64 = rng.sample(gamma);
let p = (p.min(101.0f64) - 1.0) / 100.0;
(p * max as f64) as usize
}

35
crates/udp_bench/src/config.rs
View file

@ -1,35 +0,0 @@
use aquatic_toml_config::TomlConfig;
use serde::Deserialize;
#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BenchConfig {
pub num_rounds: usize,
pub num_threads: usize,
pub num_connect_requests: usize,
pub num_announce_requests: usize,
pub num_scrape_requests: usize,
pub num_hashes_per_scrape_request: usize,
}
impl Default for BenchConfig {
fn default() -> Self {
Self {
num_rounds: 10,
num_threads: 2,
num_connect_requests: 5_000_000,
num_announce_requests: 2_000_000,
num_scrape_requests: 2_000_000,
num_hashes_per_scrape_request: 20,
}
}
}
impl aquatic_common::cli::Config for BenchConfig {}
#[cfg(test)]
mod tests {
use super::BenchConfig;
::aquatic_toml_config::gen_serialize_deserialize_test!(BenchConfig);
}

127
crates/udp_bench/src/main.rs
View file

@ -1,127 +0,0 @@
//! Benchmark announce and scrape handlers
//!
//! Example outputs:
//! ```
//! # Results over 10 rounds with 2 threads
//! Announce: 429 540 requests/second, 2328.07 ns/request
//! Scrape: 1 873 545 requests/second, 533.75 ns/request
//! ```
use aquatic_common::{PanicSentinelWatcher, ServerStartInstant};
use aquatic_udp::workers::swarm::run_swarm_worker;
use crossbeam_channel::unbounded;
use num_format::{Locale, ToFormattedString};
use rand::{rngs::SmallRng, thread_rng, Rng, SeedableRng};
use std::time::Duration;
use aquatic_common::cli::run_app_with_cli_and_config;
use aquatic_udp::common::*;
use aquatic_udp::config::Config;
use aquatic_udp_protocol::*;
use config::BenchConfig;
mod announce;
mod common;
mod config;
mod scrape;
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
fn main() {
run_app_with_cli_and_config::<BenchConfig>(
"aquatic_udp_bench: Run aquatic_udp benchmarks",
env!("CARGO_PKG_VERSION"),
run,
None,
)
}
pub fn run(bench_config: BenchConfig) -> ::anyhow::Result<()> {
// Setup common state, spawn request handlers
let mut aquatic_config = Config::default();
let (_, sentinel) = PanicSentinelWatcher::create_with_sentinel();
aquatic_config.cleaning.torrent_cleaning_interval = 60 * 60 * 24;
let (request_sender, request_receiver) = unbounded();
let (response_sender, response_receiver) = unbounded();
let response_sender = ConnectedResponseSender::new(vec![response_sender]);
let (statistics_sender, _) = unbounded();
let server_start_instant = ServerStartInstant::new();
{
let config = aquatic_config.clone();
let state = State::new(config.swarm_workers);
::std::thread::spawn(move || {
run_swarm_worker(
sentinel,
config,
state,
server_start_instant,
request_receiver,
response_sender,
statistics_sender,
SwarmWorkerIndex(0),
)
});
}
// Run benchmarks
let mut rng = SmallRng::from_rng(thread_rng()).unwrap();
let info_hashes = create_info_hashes(&mut rng);
let a = announce::bench_announce_handler(
&bench_config,
&request_sender,
&response_receiver,
&mut rng,
&info_hashes,
);
let s = scrape::bench_scrape_handler(
&bench_config,
&request_sender,
&response_receiver,
&mut rng,
&info_hashes,
);
println!(
"\n# Results over {} rounds with {} threads",
bench_config.num_rounds, bench_config.num_threads,
);
print_results("Announce:", a.0, a.1);
print_results("Scrape: ", s.0, s.1);
Ok(())
}
pub fn print_results(request_type: &str, num_responses: usize, duration: Duration) {
let per_second = ((num_responses as f64 / (duration.as_micros() as f64 / 1000000.0)) as usize)
.to_formatted_string(&Locale::se);
let time_per_request = duration.as_nanos() as f64 / (num_responses as f64);
println!(
"{} {:>10} requests/second, {:>8.2} ns/request",
request_type, per_second, time_per_request,
);
}
fn create_info_hashes(rng: &mut impl Rng) -> Vec<InfoHash> {
let mut info_hashes = Vec::new();
for _ in 0..common::NUM_INFO_HASHES {
info_hashes.push(InfoHash(rng.gen()));
}
info_hashes
}

123
crates/udp_bench/src/scrape.rs
View file

@ -1,123 +0,0 @@
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::time::{Duration, Instant};
use aquatic_common::CanonicalSocketAddr;
use crossbeam_channel::{Receiver, Sender};
use indicatif::ProgressIterator;
use rand::Rng;
use rand_distr::Gamma;
use aquatic_udp::common::*;
use aquatic_udp_protocol::*;
use crate::common::*;
use crate::config::BenchConfig;
pub fn bench_scrape_handler(
bench_config: &BenchConfig,
request_sender: &Sender<(SocketWorkerIndex, ConnectedRequest, CanonicalSocketAddr)>,
response_receiver: &Receiver<(ConnectedResponse, CanonicalSocketAddr)>,
rng: &mut impl Rng,
info_hashes: &[InfoHash],
) -> (usize, Duration) {
let requests = create_requests(
rng,
info_hashes,
bench_config.num_scrape_requests,
bench_config.num_hashes_per_scrape_request,
);
let p = 10_000 * bench_config.num_threads; // FIXME: adjust to sharded workers
let mut num_responses = 0usize;
let mut dummy: i32 = rng.gen();
let pb = create_progress_bar("Scrape", bench_config.num_rounds as u64);
// Start benchmark
let before = Instant::now();
for round in (0..bench_config.num_rounds).progress_with(pb) {
for request_chunk in requests.chunks(p) {
for (request, src) in request_chunk {
let request = ConnectedRequest::Scrape(PendingScrapeRequest {
slab_key: 0,
info_hashes: request
.info_hashes
.clone()
.into_iter()
.enumerate()
.collect(),
});
request_sender
.send((SocketWorkerIndex(0), request, *src))
.unwrap();
}
while let Ok((ConnectedResponse::Scrape(response), _)) = response_receiver.try_recv() {
num_responses += 1;
if let Some(stat) = response.torrent_stats.values().last() {
dummy ^= stat.leechers.0.get();
}
}
}
let total = bench_config.num_scrape_requests * (round + 1);
while num_responses < total {
if let Ok((ConnectedResponse::Scrape(response), _)) = response_receiver.recv() {
num_responses += 1;
if let Some(stat) = response.torrent_stats.values().last() {
dummy ^= stat.leechers.0.get();
}
}
}
}
let elapsed = before.elapsed();
if dummy == 0 {
println!("dummy dummy");
}
(num_responses, elapsed)
}
pub fn create_requests(
rng: &mut impl Rng,
info_hashes: &[InfoHash],
number: usize,
hashes_per_request: usize,
) -> Vec<(ScrapeRequest, CanonicalSocketAddr)> {
let gamma = Gamma::new(GAMMA_SHAPE, GAMMA_SCALE).unwrap();
let max_index = info_hashes.len() - 1;
let mut requests = Vec::new();
for _ in 0..number {
let mut request_info_hashes = Vec::new();
for _ in 0..hashes_per_request {
let info_hash_index = gamma_usize(rng, gamma, max_index);
request_info_hashes.push(info_hashes[info_hash_index])
}
let request = ScrapeRequest {
connection_id: ConnectionId::new(0),
transaction_id: TransactionId::new(rng.gen()),
info_hashes: request_info_hashes,
};
requests.push((
request,
CanonicalSocketAddr::new(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 1))),
));
}
requests
}

5
scripts/bench-udp-handlers.sh
View file

@ -1,5 +0,0 @@
#!/bin/sh
. ./scripts/env-native-cpu-without-avx-512
cargo run --profile "release-debug" -p aquatic_udp_bench -- $@