From b9a9a822079275c94f38776116174c7cfe8a6ff8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Mon, 13 Apr 2020 03:23:17 +0200 Subject: [PATCH] bench: add announce and scrape benchmarks Only announce seems to be faster with more threads --- .../src/bin/bench_handlers/announce.rs | 136 ++++---- .../src/bin/bench_handlers/common.rs | 22 -- .../src/bin/bench_handlers/config.rs | 8 +- .../src/bin/bench_handlers/connect.rs | 44 +-- aquatic_bench/src/bin/bench_handlers/main.rs | 295 ++++++------------ .../src/bin/bench_handlers/scrape.rs | 131 +++++--- 6 files changed, 269 insertions(+), 367 deletions(-) diff --git a/aquatic_bench/src/bin/bench_handlers/announce.rs b/aquatic_bench/src/bin/bench_handlers/announce.rs index b88420e..f05d4e2 100644 --- a/aquatic_bench/src/bin/bench_handlers/announce.rs +++ b/aquatic_bench/src/bin/bench_handlers/announce.rs @@ -1,84 +1,93 @@ -use std::io::Cursor; -use std::time::{Duration, Instant}; use std::net::SocketAddr; -use std::sync::Arc; +use std::time::{Duration, Instant}; -use rand::{Rng, SeedableRng, thread_rng, rngs::SmallRng}; +use crossbeam_channel::{Sender, Receiver}; +use indicatif::ProgressIterator; +use rand::Rng; use rand_distr::Pareto; -use aquatic::handlers::*; use aquatic::common::*; use aquatic::config::Config; -use aquatic_bench::*; -use bittorrent_udp::converters::*; + +use aquatic_bench::pareto_usize; use crate::common::*; +use crate::config::BenchConfig; -const ANNOUNCE_REQUESTS: usize = 1_000_000; - - -pub fn bench( +pub fn bench_announce_handler( state: &State, - config: &Config, - requests: Arc> -) -> (usize, Duration){ - let mut buffer = [0u8; MAX_PACKET_SIZE]; - let mut cursor = Cursor::new(buffer.as_mut()); - let mut num_responses: usize = 0; - let mut dummy = 0u8; - - let mut small_rng = SmallRng::from_rng(thread_rng()).unwrap(); - - let now = Instant::now(); - - let mut requests: Vec<(AnnounceRequest, SocketAddr)> = requests.iter() - .map(|(request_bytes, src)| { - if let Request::Announce(r) = request_from_bytes(request_bytes, 255).unwrap() { - (r, *src) - } else { - unreachable!() - } - }) - .collect(); - - let requests = requests.drain(..); - - handle_announce_requests( - &state, - config, - &mut small_rng, - requests, + bench_config: &BenchConfig, + aquatic_config: &Config, + request_sender: &Sender<(Request, SocketAddr)>, + response_receiver: &Receiver<(Response, SocketAddr)>, + rng: &mut impl Rng, + info_hashes: &Vec, +) -> (usize, Duration) { + let requests = create_requests( + state, + rng, + info_hashes, + bench_config.num_announce_requests ); - - while let Ok((response, _)) = state.response_queue.pop(){ - if let Response::Announce(_) = response { - num_responses += 1; + + let p = aquatic_config.handlers.max_requests_per_iter * bench_config.num_threads; + 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((request.clone().into(), *src)).unwrap(); + } + + while let Ok((Response::Announce(r), _)) = response_receiver.try_recv() { + num_responses += 1; + + if let Some(last_peer) = r.peers.last(){ + dummy ^= last_peer.port.0; + } + } } - cursor.set_position(0); + let total = bench_config.num_announce_requests * (round + 1); - response_to_bytes(&mut cursor, response, IpVersion::IPv4).unwrap(); + while num_responses < total { + match response_receiver.recv(){ + Ok((Response::Announce(r), _)) => { + num_responses += 1; - dummy ^= cursor.get_ref()[0]; + if let Some(last_peer) = r.peers.last(){ + dummy ^= last_peer.port.0; + } + }, + _ => {} + } + } } - let duration = Instant::now() - now; + let elapsed = before.elapsed(); - assert_eq!(num_responses, ANNOUNCE_REQUESTS); - - if dummy == 123u8 { - println!("dummy info"); + if dummy == 0 { + println!("dummy dummy"); } - (ANNOUNCE_REQUESTS, duration) + (num_responses, elapsed) } - pub fn create_requests( + state: &State, rng: &mut impl Rng, - info_hashes: &Vec + info_hashes: &Vec, + number: usize, ) -> Vec<(AnnounceRequest, SocketAddr)> { let pareto = Pareto::new(1., PARETO_SHAPE).unwrap(); @@ -86,11 +95,22 @@ pub fn create_requests( let mut requests = Vec::new(); - for _ in 0..ANNOUNCE_REQUESTS { + let d = state.handler_data.lock(); + + let connection_keys: Vec = d.connections.keys() + .take(number) + .cloned() + .collect(); + + for i in 0..number { let info_hash_index = pareto_usize(rng, pareto, max_index); + // Will panic if less connection requests than announce requests + let connection_id = connection_keys[i].connection_id; + let src = connection_keys[i].socket_addr; + let request = AnnounceRequest { - connection_id: ConnectionId(rng.gen()), + connection_id, transaction_id: TransactionId(rng.gen()), info_hash: info_hashes[info_hash_index], peer_id: PeerId(rng.gen()), @@ -104,8 +124,6 @@ pub fn create_requests( port: Port(rng.gen()) }; - let src = SocketAddr::from(([rng.gen(), rng.gen(), rng.gen(), rng.gen()], rng.gen())); - requests.push((request, src)); } diff --git a/aquatic_bench/src/bin/bench_handlers/common.rs b/aquatic_bench/src/bin/bench_handlers/common.rs index 391172b..b573c17 100644 --- a/aquatic_bench/src/bin/bench_handlers/common.rs +++ b/aquatic_bench/src/bin/bench_handlers/common.rs @@ -1,7 +1,5 @@ -use std::time::Duration; use indicatif::{ProgressBar, ProgressStyle}; -use num_format::{Locale, ToFormattedString}; pub const PARETO_SHAPE: f64 = 0.1; @@ -14,23 +12,3 @@ pub fn create_progress_bar(name: &str, iterations: u64) -> ProgressBar { ProgressBar::new(iterations).with_style(style) } - - -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, - ); -} diff --git a/aquatic_bench/src/bin/bench_handlers/config.rs b/aquatic_bench/src/bin/bench_handlers/config.rs index f54e698..a457663 100644 --- a/aquatic_bench/src/bin/bench_handlers/config.rs +++ b/aquatic_bench/src/bin/bench_handlers/config.rs @@ -6,15 +6,21 @@ 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: 20, + 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, } } } \ No newline at end of file diff --git a/aquatic_bench/src/bin/bench_handlers/connect.rs b/aquatic_bench/src/bin/bench_handlers/connect.rs index dd61225..e227a4e 100644 --- a/aquatic_bench/src/bin/bench_handlers/connect.rs +++ b/aquatic_bench/src/bin/bench_handlers/connect.rs @@ -1,11 +1,10 @@ -use std::time::Instant; +use std::time::{Duration, Instant}; -use crossbeam_channel::unbounded; +use crossbeam_channel::{Sender, Receiver}; use indicatif::ProgressIterator; use rand::{Rng, SeedableRng, thread_rng, rngs::SmallRng}; use std::net::SocketAddr; -use aquatic::handlers; use aquatic::common::*; use aquatic::config::Config; @@ -13,33 +12,12 @@ use crate::common::*; use crate::config::BenchConfig; -pub fn bench_connect_handler(bench_config: BenchConfig){ - // Setup common state, spawn request handlers - - let state = State::new(); - let aquatic_config = Config::default(); - - let (request_sender, request_receiver) = unbounded(); - let (response_sender, response_receiver) = unbounded(); - - for _ in 0..bench_config.num_threads { - let state = state.clone(); - let config = aquatic_config.clone(); - let request_receiver = request_receiver.clone(); - let response_sender = response_sender.clone(); - - ::std::thread::spawn(move || { - handlers::run_request_worker( - state, - config, - request_receiver, - response_sender - ) - }); - } - - // Setup connect benchmark data - +pub fn bench_connect_handler( + bench_config: &BenchConfig, + aquatic_config: &Config, + request_sender: &Sender<(Request, SocketAddr)>, + response_receiver: &Receiver<(Response, SocketAddr)>, +) -> (usize, Duration) { let requests = create_requests( bench_config.num_connect_requests ); @@ -49,7 +27,7 @@ pub fn bench_connect_handler(bench_config: BenchConfig){ let mut dummy: i64 = thread_rng().gen(); - let pb = create_progress_bar("Connect handler", bench_config.num_rounds as u64); + let pb = create_progress_bar("Connect", bench_config.num_rounds as u64); // Start connect benchmark @@ -82,11 +60,11 @@ pub fn bench_connect_handler(bench_config: BenchConfig){ let elapsed = before.elapsed(); - print_results("Connect handler:", num_responses, elapsed); - if dummy == 0 { println!("dummy dummy"); } + + (num_responses, elapsed) } diff --git a/aquatic_bench/src/bin/bench_handlers/main.rs b/aquatic_bench/src/bin/bench_handlers/main.rs index 9540901..1a31690 100644 --- a/aquatic_bench/src/bin/bench_handlers/main.rs +++ b/aquatic_bench/src/bin/bench_handlers/main.rs @@ -1,37 +1,36 @@ //! Benchmark announce and scrape handlers //! -//! Example summary output: +//! Example outputs: //! ``` -//! ## Average results over 20 rounds in 4 threads -//! -//! Connect handler: 2 543 896 requests/second, 393.10 ns/request -//! Announce handler: 382 055 requests/second, 2617.42 ns/request -//! Scrape handler: 1 168 651 requests/second, 855.69 ns/request +//! # Results over 20 rounds with 1 threads +//! Connect: 2 306 637 requests/second, 433.53 ns/request +//! Announce: 688 391 requests/second, 1452.66 ns/request +//! Scrape: 1 505 700 requests/second, 664.14 ns/request +//! ``` +//! ``` +//! # Results over 20 rounds with 2 threads +//! Connect: 3 472 434 requests/second, 287.98 ns/request +//! Announce: 739 371 requests/second, 1352.50 ns/request +//! Scrape: 1 845 253 requests/second, 541.93 ns/request //! ``` -use std::time::{Duration, Instant}; -use std::io::Cursor; -use std::net::SocketAddr; -use std::sync::Arc; - -use indicatif::{ProgressBar, ProgressStyle, ProgressIterator}; +use crossbeam_channel::unbounded; +use std::time::Duration; use num_format::{Locale, ToFormattedString}; use rand::{Rng, thread_rng, rngs::SmallRng, SeedableRng}; use aquatic::common::*; use aquatic::config::Config; -use bittorrent_udp::converters::*; +use aquatic::handlers; use cli_helpers::run_app_with_cli_and_config; +use config::BenchConfig; -// mod announce; +mod announce; mod common; mod config; mod connect; -// mod scrape; - -use common::*; -use config::BenchConfig; +mod scrape; #[global_allocator] @@ -41,192 +40,98 @@ static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; fn main(){ run_app_with_cli_and_config::( "aquatic benchmarker", - connect::bench_connect_handler + run ) } +pub fn run(bench_config: BenchConfig){ + // Setup common state, spawn request handlers + let state = State::new(); + let aquatic_config = Config::default(); -/* + let (request_sender, request_receiver) = unbounded(); + let (response_sender, response_receiver) = unbounded(); -fn run(bench_config: BenchConfig){ - let mut connect_data = (0usize, Duration::new(0, 0)); - let mut announce_data = (0usize, Duration::new(0, 0)); - let mut scrape_data = (0usize, Duration::new(0, 0)); + for _ in 0..bench_config.num_threads { + let state = state.clone(); + let config = aquatic_config.clone(); + let request_receiver = request_receiver.clone(); + let response_sender = response_sender.clone(); - let config = Config::default(); - - println!("# Benchmarking request handlers\n"); - - // Benchmark connect handler - { - let requests = connect::create_requests(); - - let requests: Vec<([u8; MAX_REQUEST_BYTES], SocketAddr)> = requests.into_iter() - .map(|(request, src)| { - let mut buffer = [0u8; MAX_REQUEST_BYTES]; - let mut cursor = Cursor::new(buffer.as_mut()); - - request_to_bytes(&mut cursor, Request::Connect(request)).unwrap(); - - (buffer, src) - }) - .collect(); - - let requests = Arc::new(requests); - - let pb = create_progress_bar("Connect handler", bench_config.num_rounds as u64); - - for _ in (0..bench_config.num_rounds).progress_with(pb){ - let state = State::new(); - - let handles: Vec<_> = (0..bench_config.num_threads).map(|_| { - let requests = requests.clone(); - let state = state.clone(); - - ::std::thread::spawn(|| connect::bench(state, requests)) - }).collect(); - - for handle in handles { - let (iterations, duration) = handle.join().unwrap(); - - connect_data.0 += iterations; - connect_data.1 += duration; - } - } + ::std::thread::spawn(move || { + handlers::run_request_worker( + state, + config, + request_receiver, + response_sender + ) + }); } + // Run benchmarks + + let c = connect::bench_connect_handler( + &bench_config, + &aquatic_config, + &request_sender, + &response_receiver, + ); + let mut rng = SmallRng::from_rng(thread_rng()).unwrap(); let info_hashes = create_info_hashes(&mut rng); - // Benchmark announce handler - let last_torrents: Option> = { - let requests = announce::create_requests( - &mut rng, - &info_hashes - ); - - // Create connections - - let connections = Arc::new(DashMap::new()); - let time = Time(Instant::now()); - - for (request, src) in requests.iter() { - let key = ConnectionKey { - connection_id: request.connection_id, - socket_addr: *src, - }; - - connections.insert(key, time); - } - - let requests: Vec<([u8; MAX_REQUEST_BYTES], SocketAddr)> = requests.into_iter() - .map(|(request, src)| { - let mut buffer = [0u8; MAX_REQUEST_BYTES]; - let mut cursor = Cursor::new(buffer.as_mut()); - - request_to_bytes(&mut cursor, Request::Announce(request)).unwrap(); - - (buffer, src) - }) - .collect(); - - let requests = Arc::new(requests); - - let pb = create_progress_bar("Announce handler", bench_config.num_rounds); - - let mut last_torrents = None; - - for i in (0..bench_config.num_rounds).progress_with(pb){ - let mut state = State::new(); - - state.connections = connections.clone(); - - let handles: Vec<_> = (0..bench_config.num_threads).map(|_| { - let requests = requests.clone(); - let state = state.clone(); - let config = config.clone(); - - ::std::thread::spawn(move || announce::bench(&state, &config, requests)) - }).collect(); - - for handle in handles { - let (iterations, duration) = handle.join().unwrap(); - - announce_data.0 += iterations; - announce_data.1 += duration; - } - - if i + 1 == bench_config.num_rounds { - last_torrents = Some(state.torrents); - } - } - - last_torrents - }; - - // Benchmark scrape handler - { - let mut state = State::new(); - state.torrents = last_torrents.unwrap(); - - let requests = scrape::create_requests(&mut rng, &info_hashes); - - // Create connections in state - - let time = Time(Instant::now()); - - for (request, src) in requests.iter() { - let key = ConnectionKey { - connection_id: request.connection_id, - socket_addr: *src, - }; - - state.connections.insert(key, time); - } - - let requests: Vec<([u8; MAX_REQUEST_BYTES], SocketAddr)> = requests.into_iter() - .map(|(request, src)| { - let mut buffer = [0u8; MAX_REQUEST_BYTES]; - let mut cursor = Cursor::new(buffer.as_mut()); - - request_to_bytes(&mut cursor, Request::Scrape(request)).unwrap(); - - (buffer, src) - }) - .collect(); - - let requests = Arc::new(requests); - - let pb = create_progress_bar("Scrape handler", bench_config.num_rounds); - - for _ in (0..bench_config.num_rounds).progress_with(pb){ - let handles: Vec<_> = (0..bench_config.num_threads).map(|_| { - let requests = requests.clone(); - let state = state.clone(); - - ::std::thread::spawn(move || scrape::bench(&state, requests)) - }).collect(); - - for handle in handles { - let (iterations, duration) = handle.join().unwrap(); - - scrape_data.0 += iterations; - scrape_data.1 += duration; - } - } - } - - println!( - "\n## Average results over {} rounds in {} threads\n", - bench_config.num_rounds, - bench_config.num_threads + let a = announce::bench_announce_handler( + &state, + &bench_config, + &aquatic_config, + &request_sender, + &response_receiver, + &mut rng, + &info_hashes ); - print_results("Connect handler: ", &bench_config, connect_data.0, connect_data.1); - print_results("Announce handler:", &bench_config, announce_data.0, announce_data.1); - print_results("Scrape handler: ", &bench_config, scrape_data.0, scrape_data.1); + let s = scrape::bench_scrape_handler( + &state, + &bench_config, + &aquatic_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("Connect: ", c.0, c.1); + print_results("Announce:", a.0, a.1); + print_results("Scrape: ", s.0, s.1); +} + + + +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, + ); } @@ -238,14 +143,4 @@ fn create_info_hashes(rng: &mut impl Rng) -> Vec { } info_hashes -} - - -fn create_progress_bar(name: &str, iterations: u64) -> ProgressBar { - let t = format!("{:<16} {}", name, "{wide_bar} {pos:>2}/{len:>2}"); - let style = ProgressStyle::default_bar().template(&t); - - ProgressBar::new(iterations).with_style(style) -} - -*/ \ No newline at end of file +} \ No newline at end of file diff --git a/aquatic_bench/src/bin/bench_handlers/scrape.rs b/aquatic_bench/src/bin/bench_handlers/scrape.rs index 8a85b9b..190db68 100644 --- a/aquatic_bench/src/bin/bench_handlers/scrape.rs +++ b/aquatic_bench/src/bin/bench_handlers/scrape.rs @@ -1,101 +1,128 @@ -use std::io::Cursor; use std::net::SocketAddr; use std::time::{Duration, Instant}; -use std::sync::Arc; +use crossbeam_channel::{Sender, Receiver}; +use indicatif::ProgressIterator; use rand::Rng; use rand_distr::Pareto; -use aquatic::handlers::*; use aquatic::common::*; -use aquatic_bench::*; -use bittorrent_udp::converters::*; +use aquatic::config::Config; + +use aquatic_bench::pareto_usize; use crate::common::*; +use crate::config::BenchConfig; -const SCRAPE_REQUESTS: usize = 1_000_000; -const SCRAPE_NUM_HASHES: usize = 10; - - -pub fn bench( +pub fn bench_scrape_handler( state: &State, - requests: Arc> -) -> (usize, Duration){ - let mut buffer = [0u8; MAX_PACKET_SIZE]; - let mut cursor = Cursor::new(buffer.as_mut()); - let mut num_responses: usize = 0; - let mut dummy = 0u8; - - let now = Instant::now(); - - let mut requests: Vec<(ScrapeRequest, SocketAddr)> = requests.iter() - .map(|(request_bytes, src)| { - if let Request::Scrape(r) = request_from_bytes(request_bytes, 255).unwrap() { - (r, *src) - } else { - unreachable!() - } - }) - .collect(); - - let requests = requests.drain(..); - - handle_scrape_requests( - &state, - requests, + bench_config: &BenchConfig, + aquatic_config: &Config, + request_sender: &Sender<(Request, SocketAddr)>, + response_receiver: &Receiver<(Response, SocketAddr)>, + rng: &mut impl Rng, + info_hashes: &Vec, +) -> (usize, Duration) { + let requests = create_requests( + state, + rng, + info_hashes, + bench_config.num_scrape_requests, + bench_config.num_hashes_per_scrape_request, ); - while let Ok((response, _)) = state.response_queue.pop(){ - if let Response::Scrape(_) = response { - num_responses += 1; + let p = aquatic_config.handlers.max_requests_per_iter * bench_config.num_threads; + 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 { + request_sender.send((request.clone().into(), *src)).unwrap(); + } + + while let Ok((Response::Scrape(r), _)) = response_receiver.try_recv() { + num_responses += 1; + + if let Some(stat) = r.torrent_stats.last(){ + dummy ^= stat.leechers.0; + } + } } - cursor.set_position(0); + let total = bench_config.num_scrape_requests * (round + 1); - response_to_bytes(&mut cursor, response, IpVersion::IPv4).unwrap(); + while num_responses < total { + match response_receiver.recv(){ + Ok((Response::Scrape(r), _)) => { + num_responses += 1; - dummy ^= cursor.get_ref()[0]; + if let Some(stat) = r.torrent_stats.last(){ + dummy ^= stat.leechers.0; + } + }, + _ => {} + } + } } - let duration = Instant::now() - now; + let elapsed = before.elapsed(); - assert_eq!(num_responses, SCRAPE_REQUESTS); - - if dummy == 123u8 { - println!("dummy info"); + if dummy == 0 { + println!("dummy dummy"); } - (SCRAPE_REQUESTS, duration) + (num_responses, elapsed) } + pub fn create_requests( + state: &State, rng: &mut impl Rng, - info_hashes: &Vec + info_hashes: &Vec, + number: usize, + hashes_per_request: usize, ) -> Vec<(ScrapeRequest, SocketAddr)> { let pareto = Pareto::new(1., PARETO_SHAPE).unwrap(); let max_index = info_hashes.len() - 1; + let d = state.handler_data.lock(); + + let connection_keys: Vec = d.connections.keys() + .take(number) + .cloned() + .collect(); + let mut requests = Vec::new(); - for _ in 0..SCRAPE_REQUESTS { + for i in 0..number { let mut request_info_hashes = Vec::new(); - for _ in 0..SCRAPE_NUM_HASHES { + for _ in 0..hashes_per_request { let info_hash_index = pareto_usize(rng, pareto, max_index); request_info_hashes.push(info_hashes[info_hash_index]) } + // Will panic if less connection requests than scrape requests + let connection_id = connection_keys[i].connection_id; + let src = connection_keys[i].socket_addr; + let request = ScrapeRequest { - connection_id: ConnectionId(rng.gen()), + connection_id, transaction_id: TransactionId(rng.gen()), info_hashes: request_info_hashes, }; - let src = SocketAddr::from(([rng.gen(), rng.gen(), rng.gen(), rng.gen()], rng.gen())); - requests.push((request, src)); }