Move benchmarks to seperate crate "aquatic_bench"

aquatic_bench/Cargo.toml

@@ -0,0 +1,23 @@
+
+[package]
+name = "aquatic_bench"
+version = "0.1.0"
+authors = ["Joakim Frostegård <joakim.frostegard@gmail.com>"]
+edition = "2018"
+
+[lib]
+name = "aquatic_bench"
+
+[[bin]]
+name = "bench_handlers"
+
+[[bin]]
+name = "plot_pareto"
+
+[dependencies]
+aquatic = { path = "../aquatic" }
+mimalloc = "0.1"
+num-format = "0.4"
+plotly = "0.4"
+rand_distr = "0.2"
+rand = { version = "0.7", features = ["small_rng"] }

aquatic_bench/src/bin/bench_handlers/announce.rs

@@ -0,0 +1,98 @@
+use std::time::Instant;
+use std::net::SocketAddr;
+
+use rand::Rng;
+use rand_distr::Pareto;
+
+use aquatic::handlers::*;
+use aquatic::common::*;
+use aquatic_bench::*;
+
+use crate::common::*;
+
+
+const ANNOUNCE_REQUESTS: usize = 1_000_000;
+
+
+pub fn bench(
+    state: &State,
+    requests: Vec<(AnnounceRequest, SocketAddr)>,
+) -> (f64, f64) {
+    let mut responses = Vec::with_capacity(ANNOUNCE_REQUESTS);
+    let mut requests = requests;
+    let requests = requests.drain(..);
+
+    let now = Instant::now();
+
+    handle_announce_requests(
+        &state,
+        &mut responses,
+        requests,
+    );
+
+    let duration = Instant::now() - now;
+
+    let requests_per_second = ANNOUNCE_REQUESTS as f64 / (duration.as_millis() as f64 / 1000.0);
+    let time_per_request = duration.as_nanos() as f64 / ANNOUNCE_REQUESTS as f64;
+
+    // println!("\nrequests/second: {:.2}", requests_per_second);
+    // println!("time per request: {:.2}ns", time_per_request);
+
+    // let mut total_num_peers = 0.0f64;
+    let mut num_responses: usize = 0;
+    
+    for (response, _src) in responses.drain(..) {
+        if let Response::Announce(_response) = response {
+            // let n = response.peers.len() as f64;
+
+            // total_num_peers += n;
+            num_responses += 1;
+        }
+    }
+
+    if num_responses != ANNOUNCE_REQUESTS {
+        println!("ERROR: only {} responses received", num_responses);
+    }
+
+    // println!("avg num peers returned: {:.2}", total_num_peers / ANNOUNCE_REQUESTS as f64);
+
+    (requests_per_second, time_per_request)
+}
+
+
+
+pub fn create_requests(
+    rng: &mut impl Rng,
+    info_hashes: &Vec<InfoHash>
+) -> Vec<(AnnounceRequest, SocketAddr)> {
+    let pareto = Pareto::new(1., PARETO_SHAPE).unwrap();
+
+    let max_index = info_hashes.len() - 1;
+
+    let mut requests = Vec::new();
+
+    for _ in 0..ANNOUNCE_REQUESTS {
+        let info_hash_index = pareto_usize(rng, pareto, max_index);
+
+        let request = AnnounceRequest {
+            connection_id: ConnectionId(rng.gen()),
+            transaction_id: TransactionId(rng.gen()),
+            info_hash: info_hashes[info_hash_index],
+            peer_id: PeerId(rng.gen()),
+            bytes_downloaded: NumberOfBytes(rng.gen()),
+            bytes_uploaded: NumberOfBytes(rng.gen()),
+            bytes_left: NumberOfBytes(rng.gen()),
+            event: AnnounceEvent::Started,
+            ip_address: None, 
+            key: PeerKey(rng.gen()),
+            peers_wanted: NumberOfPeers(rng.gen()),
+            port: Port(rng.gen())
+        };
+
+        let src = SocketAddr::from(([rng.gen(), rng.gen(), rng.gen(), rng.gen()], rng.gen()));
+
+        requests.push((request, src));
+    }
+
+    requests
+}

aquatic_bench/src/bin/bench_handlers/common.rs

@@ -0,0 +1,2 @@
+pub const PARETO_SHAPE: f64 = 0.1;
+pub const NUM_INFO_HASHES: usize = 10_000;

aquatic_bench/src/bin/bench_handlers/connect.rs

@@ -0,0 +1,74 @@
+use std::time::Instant;
+use std::net::SocketAddr;
+
+use rand::{Rng, thread_rng, rngs::SmallRng, SeedableRng};
+
+use aquatic::common::*;
+use aquatic::handlers::handle_connect_requests;
+
+
+const ITERATIONS: usize = 10_000_000;
+
+
+pub fn bench(
+    requests: Vec<(ConnectRequest, SocketAddr)>
+) -> (f64, f64){
+    let state = State::new();
+    let mut responses = Vec::with_capacity(ITERATIONS);
+    let mut requests = requests;
+    let requests = requests.drain(..);
+
+    let now = Instant::now();
+
+    handle_connect_requests(&state, &mut responses, requests);
+
+    let duration = Instant::now() - now;
+
+    let requests_per_second = ITERATIONS as f64 / (duration.as_millis() as f64 / 1000.0);
+    let time_per_request = duration.as_nanos() as f64 / ITERATIONS as f64;
+
+    // println!("\nrequests/second: {:.2}", requests_per_second);
+    // println!("time per request: {:.2}ns", time_per_request);
+
+    let mut dummy = 0usize;
+    let mut num_responses: usize = 0;
+    
+    for (response, _src) in responses {
+        if let Response::Connect(response) = response {
+            if response.connection_id.0 > 0 {
+                dummy += 1;
+            }
+
+            num_responses += 1;
+        }
+    }
+
+    if num_responses != ITERATIONS {
+        println!("ERROR: only {} responses received", num_responses);
+    }
+
+    if dummy == ITERATIONS {
+        println!("dummy test output: {}", dummy);
+    }
+
+    (requests_per_second, time_per_request)
+}
+
+
+pub fn create_requests() -> Vec<(ConnectRequest, SocketAddr)> {
+    let mut rng = SmallRng::from_rng(thread_rng()).unwrap();
+
+    let mut requests = Vec::new();
+
+    for _ in 0..ITERATIONS {
+        let request = ConnectRequest {
+            transaction_id: TransactionId(rng.gen()),
+        };
+
+        let src = SocketAddr::from(([rng.gen(), rng.gen(), rng.gen(), rng.gen()], rng.gen()));
+
+        requests.push((request, src));
+    }
+
+    requests
+}

aquatic_bench/src/bin/bench_handlers/main.rs

@@ -0,0 +1,155 @@
+//! Benchmark announce and scrape handlers
+//! 
+//! Example summary output:
+//! ```
+//! # Average results over 100 rounds
+//! connect handler:   3 201 147 requests/second,   312.66 ns/request
+//! announce handler:    330 958 requests/second,  3029.13 ns/request
+//! scrape handler:    1 242 478 requests/second,   805.62 ns/request
+//! ```
+
+use std::time::{Duration, Instant};
+
+use num_format::{Locale, ToFormattedString};
+use rand::{Rng, thread_rng, rngs::SmallRng, SeedableRng};
+
+use aquatic::common::*;
+
+
+mod announce;
+mod common;
+mod connect;
+mod scrape;
+
+
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
+
+macro_rules! print_results {
+    ($request_type:expr, $num_rounds:expr, $data:expr) => {
+        let per_second = (
+            ($data.0 / ($num_rounds as f64)
+        ) as usize).to_formatted_string(&Locale::se);
+
+        println!(
+            "{} {:>10} requests/second, {:>8.2} ns/request",
+            $request_type,
+            per_second,
+            $data.1 / ($num_rounds as f64)
+        );
+    };
+}
+
+
+fn main(){
+    let num_rounds = 100;
+
+    let mut connect_data = (0.0, 0.0);
+    let mut announce_data = (0.0, 0.0);
+    let mut scrape_data = (0.0, 0.0);
+
+    {
+        let requests = connect::create_requests();
+
+        ::std::thread::sleep(Duration::from_secs(1));
+
+        for round in 0..num_rounds {
+            println!("# Round {}/{}\n", round + 1, num_rounds);
+
+            let d = connect::bench(requests.clone());
+            connect_data.0 += d.0;
+            connect_data.1 += d.1;
+        }
+    }
+
+    let mut rng = SmallRng::from_rng(thread_rng()).unwrap();
+    let info_hashes = create_info_hashes(&mut rng);
+
+    let state_for_scrape: State = {
+        let requests = announce::create_requests(
+            &mut rng,
+            &info_hashes
+        );
+
+        let mut state_for_scrape = State::new();
+
+        ::std::thread::sleep(Duration::from_secs(1));
+
+        for round in 0..num_rounds {
+            println!("# Round {}/{}\n", round + 1, num_rounds);
+
+            let state = State::new();
+
+            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 d = announce::bench(&state, requests.clone());
+            announce_data.0 += d.0;
+            announce_data.1 += d.1;
+
+            if round == num_rounds - 1 {
+                state_for_scrape = state.clone();
+            }
+        }
+
+        state_for_scrape
+    };
+
+    state_for_scrape.connections.clear();
+
+    {
+        let state = state_for_scrape;
+
+        let requests = scrape::create_requests(&mut rng, &info_hashes);
+
+        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);
+        }
+
+        ::std::thread::sleep(Duration::from_secs(1));
+
+        for round in 0..num_rounds {
+            println!("# Round {}/{}\n", round + 1, num_rounds);
+
+            let d = scrape::bench(&state, requests.clone());
+            scrape_data.0 += d.0;
+            scrape_data.1 += d.1;
+
+            println!();
+        }
+    }
+
+
+    println!("# Average results over {} rounds\n", num_rounds);
+
+    print_results!("connect handler: ", num_rounds, connect_data);
+    print_results!("announce handler:", num_rounds, announce_data);
+    print_results!("scrape handler:  ", num_rounds, scrape_data);
+}
+
+
+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
+}

aquatic_bench/src/bin/bench_handlers/scrape.rs

@@ -0,0 +1,98 @@
+use std::time::Instant;
+use std::net::SocketAddr;
+
+use rand::Rng;
+use rand_distr::Pareto;
+
+use aquatic::handlers::*;
+use aquatic::common::*;
+use aquatic_bench::*;
+
+use crate::common::*;
+
+
+const SCRAPE_REQUESTS: usize = 1_000_000;
+const SCRAPE_NUM_HASHES: usize = 10;
+
+
+pub fn bench(
+    state: &State,
+    requests: Vec<(ScrapeRequest, SocketAddr)>,
+) -> (f64, f64) {
+    let mut responses = Vec::with_capacity(SCRAPE_REQUESTS);
+    let mut scrape_requests = requests;
+    let scrape_requests = scrape_requests.drain(..);
+
+    let now = Instant::now();
+
+    println!("running benchmark..");
+
+    handle_scrape_requests(
+        &state,
+        &mut responses,
+        scrape_requests,
+    );
+
+    let duration = Instant::now() - now;
+
+    let requests_per_second = SCRAPE_REQUESTS as f64 / (duration.as_millis() as f64 / 1000.0);
+    let time_per_request = duration.as_nanos() as f64 / SCRAPE_REQUESTS as f64;
+
+    // println!("\nrequests/second: {:.2}", requests_per_second);
+    // println!("time per request: {:.2}ns", time_per_request);
+
+    // let mut total_num_peers = 0.0f64;
+    let mut num_responses: usize = 0;
+
+    for (response, _src) in responses.drain(..){
+        if let Response::Scrape(_response) = response {
+            // for stats in response.torrent_stats {
+            //     total_num_peers += f64::from(stats.seeders.0);
+            //     total_num_peers += f64::from(stats.leechers.0);
+            // }
+
+            num_responses += 1;
+        }
+    }
+
+    if num_responses != SCRAPE_REQUESTS {
+        println!("ERROR: only {} responses received", num_responses);
+    }
+
+    // println!("avg num peers reported: {:.2}", total_num_peers / (SCRAPE_REQUESTS as f64 * SCRAPE_NUM_HASHES as f64));
+
+    (requests_per_second, time_per_request)
+}
+
+
+pub fn create_requests(
+    rng: &mut impl Rng,
+    info_hashes: &Vec<InfoHash>
+) -> Vec<(ScrapeRequest, SocketAddr)> {
+    let pareto = Pareto::new(1., PARETO_SHAPE).unwrap();
+
+    let max_index = info_hashes.len() - 1;
+
+    let mut requests = Vec::new();
+
+    for _ in 0..SCRAPE_REQUESTS {
+        let mut request_info_hashes = Vec::new();
+
+        for _ in 0..SCRAPE_NUM_HASHES {
+            let info_hash_index = pareto_usize(rng, pareto, max_index);
+            request_info_hashes.push(info_hashes[info_hash_index])
+        }
+
+        let request = ScrapeRequest {
+            connection_id: ConnectionId(rng.gen()),
+            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));
+    }
+
+    requests
+}

aquatic_bench/src/bin/plot_pareto.rs

@@ -0,0 +1,44 @@
+use plotly::{Plot, Scatter, Layout};
+use plotly::common::Title;
+use plotly::layout::Axis;
+use rand::{thread_rng, rngs::SmallRng, SeedableRng};
+use rand_distr::Pareto;
+
+use aquatic_bench::pareto_usize;
+
+
+fn main(){
+    let mut plot = Plot::new();
+    let mut rng = SmallRng::from_rng(thread_rng()).unwrap();
+
+    const LEN: usize = 1_000;
+    const MAX_VAL: usize = LEN - 1;
+
+    for pareto_shape in [0.1, 0.2, 0.3, 0.4, 0.5].iter() {
+        let pareto = Pareto::new(1.0, *pareto_shape).unwrap();
+
+        let mut y_axis = [0; LEN];
+
+        for _ in 1..1_000_000 {
+            let index = pareto_usize(&mut rng, pareto, MAX_VAL);
+
+            y_axis[index] += 1;
+        }
+
+        let x_axis: Vec<usize> = (0..MAX_VAL).into_iter().collect();
+
+        let trace = Scatter::new(x_axis, y_axis.to_vec())
+            .name(&format!("pareto shape = {}", pareto_shape));
+
+        plot.add_trace(trace);
+    }
+
+    let layout = Layout::new()
+        .title(Title::new("Pareto distribution"))
+        .xaxis(Axis::new().title(Title::new("Info hash index")))
+        .yaxis(Axis::new().title(Title::new("Num requests")));
+    
+    plot.set_layout(layout);
+
+    plot.show();
+}

aquatic_bench/src/lib.rs

@@ -0,0 +1,14 @@
+use rand::Rng;
+use rand_distr::Pareto;
+
+
+pub fn pareto_usize(
+    rng: &mut impl Rng,
+    pareto: Pareto<f64>,
+    max: usize,
+) -> usize {
+    let p: f64 = rng.sample(pareto);
+    let p = (p.min(101.0f64) - 1.0) / 100.0;
+
+    (p * max as f64) as usize
+}