bench: add announce and scrape benchmarks

Only announce seems to be faster with more threads

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_announce_handler(
-
-
-pub fn bench(
     state: &State,
-    config: &Config,
+    bench_config: &BenchConfig,
-    requests: Arc<Vec<([u8; MAX_REQUEST_BYTES], SocketAddr)>>
+    aquatic_config: &Config,
-) -> (usize, Duration){
+    request_sender: &Sender<(Request, SocketAddr)>,
-    let mut buffer = [0u8; MAX_PACKET_SIZE];
+    response_receiver: &Receiver<(Response, SocketAddr)>,
-    let mut cursor = Cursor::new(buffer.as_mut());
+    rng: &mut impl Rng,
-    let mut num_responses: usize = 0;
+    info_hashes: &Vec<InfoHash>,
-    let mut dummy = 0u8;
+) -> (usize, Duration) {
-
+    let requests = create_requests(
-    let mut small_rng = SmallRng::from_rng(thread_rng()).unwrap();
+        state,
-
+        rng,
-    let now = Instant::now();
+        info_hashes,
-
+        bench_config.num_announce_requests
-    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,
     );
-    
+
-    while let Ok((response, _)) = state.response_queue.pop(){
+    let p = aquatic_config.handlers.max_requests_per_iter * bench_config.num_threads;
-        if let Response::Announce(_) = response {
+    let mut num_responses = 0usize;
-            num_responses += 1;
+
+    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 == 0 {
-
+        println!("dummy dummy");
-    if dummy == 123u8 {
-        println!("dummy info");
     }
 
-    (ANNOUNCE_REQUESTS, duration)
+    (num_responses, elapsed)
 }
 
 
-
 pub fn create_requests(
+    state: &State,
     rng: &mut impl Rng,
-    info_hashes: &Vec<InfoHash>
+    info_hashes: &Vec<InfoHash>,
+    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<ConnectionKey> = 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));
     }
 

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,
-    );
-}

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,
         }
     }
 }

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){
+pub fn bench_connect_handler(
-    // Setup common state, spawn request handlers
+    bench_config: &BenchConfig,
-
+    aquatic_config: &Config,
-    let state = State::new();
+    request_sender: &Sender<(Request, SocketAddr)>,
-    let aquatic_config = Config::default();
+    response_receiver: &Receiver<(Response, SocketAddr)>,
-
+) -> (usize, Duration) {
-    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
-
     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)
 }
 
 

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
+//! # Results over 20 rounds with 1 threads
-//! 
+//! Connect:   2 306 637 requests/second,   433.53 ns/request
-//! Connect handler:   2 543 896 requests/second,   393.10 ns/request
+//! Announce:    688 391 requests/second,  1452.66 ns/request
-//! Announce handler:    382 055 requests/second,  2617.42 ns/request
+//! Scrape:    1 505 700 requests/second,   664.14 ns/request
-//! Scrape handler:    1 168 651 requests/second,   855.69 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 crossbeam_channel::unbounded;
-use std::io::Cursor;
+use std::time::Duration;
-use std::net::SocketAddr;
-use std::sync::Arc;
-
-use indicatif::{ProgressBar, ProgressStyle, ProgressIterator};
 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;
+mod scrape;
-
-use common::*;
-use config::BenchConfig;
 
 
 #[global_allocator]
@@ -41,192 +40,98 @@ static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 fn main(){
     run_app_with_cli_and_config::<BenchConfig>(
         "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){
+    for _ in 0..bench_config.num_threads {
-    let mut connect_data = (0usize, Duration::new(0, 0));
+        let state = state.clone();
-    let mut announce_data = (0usize, Duration::new(0, 0));
+        let config = aquatic_config.clone();
-    let mut scrape_data = (0usize, Duration::new(0, 0));
+        let request_receiver = request_receiver.clone();
+        let response_sender = response_sender.clone();
 
-    let config = Config::default();
+        ::std::thread::spawn(move || {
-
+            handlers::run_request_worker(
-    println!("# Benchmarking request handlers\n");
+                state,
-
+                config,
-    // Benchmark connect handler
+                request_receiver,
-    {
+                response_sender
-        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;
-            }
-        }
     }
 
+    // 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 a = announce::bench_announce_handler(
-    let last_torrents: Option<Arc<TorrentMap>> = {
+        &state,
-        let requests = announce::create_requests(
+        &bench_config,
-            &mut rng,
+        &aquatic_config,
-            &info_hashes
+        &request_sender,
-        );
+        &response_receiver,
-
+        &mut rng,
-        // Create connections
+        &info_hashes
-
-        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
     );
 
-    print_results("Connect handler: ", &bench_config, connect_data.0, connect_data.1);
+    let s = scrape::bench_scrape_handler(
-    print_results("Announce handler:", &bench_config, announce_data.0, announce_data.1);
+        &state,
-    print_results("Scrape handler:  ", &bench_config, scrape_data.0, scrape_data.1);
+        &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<InfoHash> {
     }
 
     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)
-}
-
-*/

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 aquatic::config::Config;
-use bittorrent_udp::converters::*;
+
+use aquatic_bench::pareto_usize;
 
 use crate::common::*;
+use crate::config::BenchConfig;
 
 
-const SCRAPE_REQUESTS: usize = 1_000_000;
+pub fn bench_scrape_handler(
-const SCRAPE_NUM_HASHES: usize = 10;
-
-
-pub fn bench(
     state: &State,
-    requests: Arc<Vec<([u8; MAX_REQUEST_BYTES], SocketAddr)>>
+    bench_config: &BenchConfig,
-) -> (usize, Duration){
+    aquatic_config: &Config,
-    let mut buffer = [0u8; MAX_PACKET_SIZE];
+    request_sender: &Sender<(Request, SocketAddr)>,
-    let mut cursor = Cursor::new(buffer.as_mut());
+    response_receiver: &Receiver<(Response, SocketAddr)>,
-    let mut num_responses: usize = 0;
+    rng: &mut impl Rng,
-    let mut dummy = 0u8;
+    info_hashes: &Vec<InfoHash>,
-
+) -> (usize, Duration) {
-    let now = Instant::now();
+    let requests = create_requests(
-
+        state,
-    let mut requests: Vec<(ScrapeRequest, SocketAddr)> = requests.iter()
+        rng,
-        .map(|(request_bytes, src)| {
+        info_hashes,
-            if let Request::Scrape(r) = request_from_bytes(request_bytes, 255).unwrap() {
+        bench_config.num_scrape_requests,
-                (r, *src)
+        bench_config.num_hashes_per_scrape_request,
-            } else {
-                unreachable!()
-            }
-        })
-        .collect();
-    
-    let requests = requests.drain(..);
-
-    handle_scrape_requests(
-        &state,
-        requests,
     );
 
-    while let Ok((response, _)) = state.response_queue.pop(){
+    let p = aquatic_config.handlers.max_requests_per_iter * bench_config.num_threads;
-        if let Response::Scrape(_) = response {
+    let mut num_responses = 0usize;
-            num_responses += 1;
+
+    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 == 0 {
-
+        println!("dummy dummy");
-    if dummy == 123u8 {
-        println!("dummy info");
     }
 
-    (SCRAPE_REQUESTS, duration)
+    (num_responses, elapsed)
 }
 
 
+
 pub fn create_requests(
+    state: &State,
     rng: &mut impl Rng,
-    info_hashes: &Vec<InfoHash>
+    info_hashes: &Vec<InfoHash>,
+    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<ConnectionKey> = 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));
     }