Improve udp load test; use IndexMap::swap_remove explicitly; update deps

crates/bencher/src/main.rs

@@ -25,15 +25,15 @@ struct Args {
     #[arg(long, default_value_t = Priority::Medium)]
     min_priority: Priority,
     /// How long to run each load test for
-    #[arg(long, default_value_t = 90)]
+    #[arg(long, default_value_t = 30)]
     duration: usize,
     /// Only include data for last N seconds of load test runs.
     ///
     /// Useful if the tracker/load tester combination is slow at reaching
     /// maximum throughput
     ///
     /// 0 = use data for whole run
-    #[arg(long, default_value_t = 30)]
+    #[arg(long, default_value_t = 0)]
     summarize_last: usize,
     #[command(subcommand)]
     command: Command,

crates/bencher/src/protocols/udp.rs

@@ -443,6 +443,8 @@ impl ProcessRunner for AquaticUdpLoadTestRunner {
         c.duration = self.parameters.duration;
         c.summarize_last = self.parameters.summarize_last;
 
+        c.extra_statistics = false;
+
         c.requests.announce_peers_wanted = 30;
         c.requests.weight_connect = 0;
         c.requests.weight_announce = 100;

crates/http/src/workers/swarm/storage.rs

@@ -433,7 +433,7 @@ impl<I: Ip> LargePeerMap<I> {
     }
 
     fn remove_peer(&mut self, key: &ResponsePeer<I>) -> Option<Peer> {
-        let opt_removed_peer = self.peers.remove(key);
+        let opt_removed_peer = self.peers.swap_remove(key);
 
         if let Some(Peer {
             is_seeder: true, ..

crates/udp/src/workers/statistics/mod.rs

@@ -97,7 +97,7 @@ pub fn run_statistics_worker(
                             *count -= 1;
 
                             if *count == 0 {
-                                peers.remove(&peer_id);
+                                peers.swap_remove(&peer_id);
                             }
                         }
                     }

crates/udp/src/workers/swarm/storage.rs

@@ -407,7 +407,7 @@ impl<I: Ip> LargePeerMap<I> {
     }
 
     fn remove_peer(&mut self, key: &ResponsePeer<I>) -> Option<Peer> {
-        let opt_removed_peer = self.peers.remove(key);
+        let opt_removed_peer = self.peers.swap_remove(key);
 
         if let Some(Peer {
             is_seeder: true, ..

crates/udp_load_test/Cargo.toml

@@ -25,9 +25,9 @@ aquatic_toml_config.workspace = true
 aquatic_udp_protocol.workspace = true
 
 anyhow = "1"
-hashbrown = "0.14"
+crossbeam-channel = "0.5"
+hdrhistogram = "7"
 mimalloc = { version = "0.1", default-features = false }
-mio = { version = "0.8", features = ["net", "os-poll"] }
 rand_distr = "0.4"
 rand = { version = "0.8", features = ["small_rng"] }
 serde = { version = "1", features = ["derive"] }

crates/udp_load_test/src/common.rs

@@ -1,22 +1,16 @@
 use std::sync::{atomic::AtomicUsize, Arc};
 
-use hashbrown::HashMap;
-
+use aquatic_common::IndexMap;
 use aquatic_udp_protocol::*;
 
-#[derive(PartialEq, Eq, Clone)]
-pub struct TorrentPeer {
-    pub info_hash: InfoHash,
-    pub scrape_hash_indices: Box<[usize]>,
-    pub connection_id: ConnectionId,
-    pub peer_id: PeerId,
-    pub port: Port,
+#[derive(Clone)]
+pub struct LoadTestState {
+    pub info_hashes: Arc<[InfoHash]>,
+    pub statistics: Arc<SharedStatistics>,
 }
 
-pub type TorrentPeerMap = HashMap<TransactionId, TorrentPeer>;
-
 #[derive(Default)]
-pub struct Statistics {
+pub struct SharedStatistics {
     pub requests: AtomicUsize,
     pub response_peers: AtomicUsize,
     pub responses_connect: AtomicUsize,
@@ -25,25 +19,14 @@ pub struct Statistics {
     pub responses_error: AtomicUsize,
 }
 
-#[derive(Clone)]
-pub struct LoadTestState {
-    pub info_hashes: Arc<[InfoHash]>,
-    pub statistics: Arc<Statistics>,
+pub struct Peer {
+    pub announce_info_hash_index: usize,
+    pub announce_info_hash: InfoHash,
+    pub announce_port: Port,
+    pub scrape_info_hash_indices: Box<[usize]>,
+    pub socket_index: u8,
 }
 
-#[derive(PartialEq, Eq, Clone, Copy)]
-pub enum RequestType {
-    Announce,
-    Connect,
-    Scrape,
-}
-
-#[derive(Default)]
-pub struct SocketWorkerLocalStatistics {
-    pub requests: usize,
-    pub response_peers: usize,
-    pub responses_connect: usize,
-    pub responses_announce: usize,
-    pub responses_scrape: usize,
-    pub responses_error: usize,
+pub enum StatisticsMessage {
+    ResponsesPerInfoHash(IndexMap<usize, u64>),
 }

crates/udp_load_test/src/config.rs

@@ -25,6 +25,8 @@ pub struct Config {
     ///
     /// 0 = include whole run
     pub summarize_last: usize,
+    /// Display extra statistics
+    pub extra_statistics: bool,
     pub network: NetworkConfig,
     pub requests: RequestConfig,
     #[cfg(feature = "cpu-pinning")]
@@ -39,6 +41,7 @@ impl Default for Config {
             workers: 1,
             duration: 0,
             summarize_last: 0,
+            extra_statistics: true,
             network: NetworkConfig::default(),
             requests: RequestConfig::default(),
             #[cfg(feature = "cpu-pinning")]
@@ -47,6 +50,12 @@ impl Default for Config {
     }
 }
 
+impl aquatic_common::cli::Config for Config {
+    fn get_log_level(&self) -> Option<aquatic_common::cli::LogLevel> {
+        Some(self.log_level)
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize, Serialize)]
 #[serde(default, deny_unknown_fields)]
 pub struct NetworkConfig {
@@ -57,10 +66,8 @@ pub struct NetworkConfig {
     ///
     /// Setting this to true can cause issues on macOS.
     pub multiple_client_ipv4s: bool,
-    /// Number of first client port
-    pub first_port: u16,
-    /// Socket worker poll timeout in microseconds
-    pub poll_timeout: u64,
+    /// Number of sockets to open per worker
+    pub sockets_per_worker: u8,
     /// Size of socket recv buffer. Use 0 for OS default.
     ///
     /// This setting can have a big impact on dropped packages. It might
@@ -80,8 +87,7 @@ impl Default for NetworkConfig {
     fn default() -> Self {
         Self {
             multiple_client_ipv4s: true,
-            first_port: 45_000,
-            poll_timeout: 1,
+            sockets_per_worker: 4,
             recv_buffer: 8_000_000,
         }
     }
@@ -92,6 +98,8 @@ impl Default for NetworkConfig {
 pub struct RequestConfig {
     /// Number of torrents to simulate
     pub number_of_torrents: usize,
+    /// Number of peers to simulate
+    pub number_of_peers: usize,
     /// Maximum number of torrents to ask about in scrape requests
     pub scrape_max_torrents: usize,
     /// Ask for this number of peers in announce requests
@@ -105,30 +113,21 @@ pub struct RequestConfig {
     /// Probability that a generated request is a scrape request, as part
     /// of sum of the various weight arguments.
     pub weight_scrape: usize,
-    /// Peers choose torrents according to this Gamma distribution shape
-    pub torrent_gamma_shape: f64,
-    /// Peers choose torrents according to this Gamma distribution scale
-    pub torrent_gamma_scale: f64,
     /// Probability that a generated peer is a seeder
     pub peer_seeder_probability: f64,
-    /// Probability that an additional connect request will be sent for each
-    /// mio event
-    pub additional_request_probability: f32,
 }
 
 impl Default for RequestConfig {
     fn default() -> Self {
         Self {
-            number_of_torrents: 10_000,
+            number_of_torrents: 1_000_000,
+            number_of_peers: 2_000_000,
             scrape_max_torrents: 10,
             announce_peers_wanted: 30,
-            weight_connect: 0,
+            weight_connect: 1,
             weight_announce: 100,
             weight_scrape: 1,
-            torrent_gamma_shape: 0.2,
-            torrent_gamma_scale: 100.0,
             peer_seeder_probability: 0.75,
-            additional_request_probability: 0.5,
         }
     }
 }