diff --git a/crates/bencher/src/protocols/udp.rs b/crates/bencher/src/protocols/udp.rs index e60d730..7c743b1 100644 --- a/crates/bencher/src/protocols/udp.rs +++ b/crates/bencher/src/protocols/udp.rs @@ -58,6 +58,12 @@ impl UdpCommand { indexmap::indexmap! { 1 => SetConfig { implementations: indexmap! { + UdpTracker::Aquatic => vec![ + AquaticUdpRunner::with_mio(1, Priority::High), + ], + UdpTracker::AquaticIoUring => vec![ + AquaticUdpRunner::with_io_uring(1, Priority::High), + ], UdpTracker::OpenTracker => vec![ OpenTrackerUdpRunner::new(0, Priority::Medium), // Handle requests within event loop OpenTrackerUdpRunner::new(1, Priority::High), @@ -74,12 +80,10 @@ impl UdpCommand { 2 => SetConfig { implementations: indexmap! { UdpTracker::Aquatic => vec![ - AquaticUdpRunner::with_mio(1, 1, Priority::Medium), - AquaticUdpRunner::with_mio(2, 1, Priority::High), + AquaticUdpRunner::with_mio(2, Priority::High), ], UdpTracker::AquaticIoUring => vec![ - AquaticUdpRunner::with_io_uring(1, 1, Priority::Medium), - AquaticUdpRunner::with_io_uring(2, 1, Priority::High), + AquaticUdpRunner::with_io_uring(2, Priority::High), ], UdpTracker::OpenTracker => vec![ OpenTrackerUdpRunner::new(2, Priority::High), @@ -97,12 +101,10 @@ impl UdpCommand { 4 => SetConfig { implementations: indexmap! { UdpTracker::Aquatic => vec![ - AquaticUdpRunner::with_mio(3, 1, Priority::High), - AquaticUdpRunner::with_mio(4, 1, Priority::Medium), + AquaticUdpRunner::with_mio(4, Priority::High), ], UdpTracker::AquaticIoUring => vec![ - AquaticUdpRunner::with_io_uring(3, 1, Priority::High), - AquaticUdpRunner::with_io_uring(4, 1, Priority::Medium), + AquaticUdpRunner::with_io_uring(4, Priority::High), ], UdpTracker::OpenTracker => vec![ OpenTrackerUdpRunner::new(4, Priority::High), @@ -119,10 +121,10 @@ impl UdpCommand { 6 => SetConfig { implementations: indexmap! { UdpTracker::Aquatic => vec![ - AquaticUdpRunner::with_mio(5, 1, Priority::High), + AquaticUdpRunner::with_mio(6, Priority::High), ], UdpTracker::AquaticIoUring => vec![ - AquaticUdpRunner::with_io_uring(5, 1, Priority::High), + AquaticUdpRunner::with_io_uring(6, Priority::High), ], UdpTracker::OpenTracker => vec![ OpenTrackerUdpRunner::new(6, Priority::High), @@ -139,10 +141,10 @@ impl UdpCommand { 8 => SetConfig { implementations: indexmap! { UdpTracker::Aquatic => vec![ - AquaticUdpRunner::with_mio(7, 1, Priority::High), + AquaticUdpRunner::with_mio(8, Priority::High), ], UdpTracker::AquaticIoUring => vec![ - AquaticUdpRunner::with_io_uring(7, 1, Priority::High), + AquaticUdpRunner::with_io_uring(8, Priority::High), ], UdpTracker::OpenTracker => vec![ OpenTrackerUdpRunner::new(8, Priority::High), @@ -159,12 +161,10 @@ impl UdpCommand { 12 => SetConfig { implementations: indexmap! { UdpTracker::Aquatic => vec![ - AquaticUdpRunner::with_mio(10, 2, Priority::High), - AquaticUdpRunner::with_mio(9, 3, Priority::Medium), + AquaticUdpRunner::with_mio(12, Priority::High), ], UdpTracker::AquaticIoUring => vec![ - AquaticUdpRunner::with_io_uring(10, 2, Priority::High), - AquaticUdpRunner::with_io_uring(9, 3, Priority::Medium), + AquaticUdpRunner::with_io_uring(12, Priority::High), ], UdpTracker::OpenTracker => vec![ OpenTrackerUdpRunner::new(12, Priority::High), @@ -181,10 +181,10 @@ impl UdpCommand { 16 => SetConfig { implementations: indexmap! { UdpTracker::Aquatic => vec![ - AquaticUdpRunner::with_mio(13, 3, Priority::High), + AquaticUdpRunner::with_mio(16, Priority::High), ], UdpTracker::AquaticIoUring => vec![ - AquaticUdpRunner::with_io_uring(13, 3, Priority::High), + AquaticUdpRunner::with_io_uring(16, Priority::High), ], UdpTracker::OpenTracker => vec![ OpenTrackerUdpRunner::new(16, Priority::High), @@ -211,7 +211,6 @@ impl UdpCommand { #[derive(Debug, Clone)] struct AquaticUdpRunner { socket_workers: usize, - swarm_workers: usize, use_io_uring: bool, priority: Priority, } @@ -219,24 +218,20 @@ struct AquaticUdpRunner { impl AquaticUdpRunner { fn with_mio( socket_workers: usize, - swarm_workers: usize, priority: Priority, ) -> Rc> { Rc::new(Self { socket_workers, - swarm_workers, use_io_uring: false, priority, }) } fn with_io_uring( socket_workers: usize, - swarm_workers: usize, priority: Priority, ) -> Rc> { Rc::new(Self { socket_workers, - swarm_workers, use_io_uring: true, priority, }) @@ -256,7 +251,6 @@ impl ProcessRunner for AquaticUdpRunner { let mut c = aquatic_udp::config::Config::default(); c.socket_workers = self.socket_workers; - c.swarm_workers = self.swarm_workers; c.network.address = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 3000)); c.network.use_io_uring = self.use_io_uring; c.protocol.max_response_peers = 30; @@ -283,7 +277,6 @@ impl ProcessRunner for AquaticUdpRunner { fn keys(&self) -> IndexMap { indexmap! { "socket workers".to_string() => self.socket_workers.to_string(), - "swarm workers".to_string() => self.swarm_workers.to_string(), } } } diff --git a/crates/udp/src/common.rs b/crates/udp/src/common.rs index c42aa35..dce58cb 100644 --- a/crates/udp/src/common.rs +++ b/crates/udp/src/common.rs @@ -94,11 +94,11 @@ pub struct State { pub server_start_instant: ServerStartInstant, } -impl State { - pub fn new(config: &Config) -> Self { +impl Default for State { + fn default() -> Self { Self { access_list: Arc::new(AccessListArcSwap::default()), - torrent_maps: TorrentMaps::new(config), + torrent_maps: TorrentMaps::default(), server_start_instant: ServerStartInstant::new(), } } diff --git a/crates/udp/src/lib.rs b/crates/udp/src/lib.rs index c0a7d71..e91027e 100644 --- a/crates/udp/src/lib.rs +++ b/crates/udp/src/lib.rs @@ -25,7 +25,7 @@ pub const APP_VERSION: &str = env!("CARGO_PKG_VERSION"); pub fn run(config: Config) -> ::anyhow::Result<()> { let mut signals = Signals::new([SIGUSR1])?; - let state = State::new(&config); + let state = State::default(); let statistics = Statistics::new(&config); let connection_validator = ConnectionValidator::new(&config)?; let priv_dropper = PrivilegeDropper::new(config.privileges.clone(), config.socket_workers); diff --git a/crates/udp/src/swarm.rs b/crates/udp/src/swarm.rs index f6cb336..d88b9d3 100644 --- a/crates/udp/src/swarm.rs +++ b/crates/udp/src/swarm.rs @@ -16,6 +16,7 @@ use aquatic_common::{CanonicalSocketAddr, IndexMap}; use aquatic_udp_protocol::*; use arrayvec::ArrayVec; use crossbeam_channel::Sender; +use hashbrown::HashMap; use hdrhistogram::Histogram; use parking_lot::RwLockUpgradableReadGuard; use rand::prelude::SmallRng; @@ -29,24 +30,24 @@ const SMALL_PEER_MAP_CAPACITY: usize = 2; use aquatic_udp_protocol::InfoHash; use parking_lot::RwLock; -type TorrentMapShard = IndexMap>>; - #[derive(Clone)] pub struct TorrentMaps { ipv4: TorrentMapShards, ipv6: TorrentMapShards, } -impl TorrentMaps { - pub fn new(config: &Config) -> Self { - let num_shards = 16usize; +impl Default for TorrentMaps { + fn default() -> Self { + const NUM_SHARDS: usize = 16; Self { - ipv4: TorrentMapShards::new(num_shards), - ipv6: TorrentMapShards::new(num_shards), + ipv4: TorrentMapShards::new(NUM_SHARDS), + ipv6: TorrentMapShards::new(NUM_SHARDS), } } +} +impl TorrentMaps { pub fn announce( &self, config: &Config, @@ -294,6 +295,9 @@ impl TorrentMapShards { } } +/// Use HashMap instead of IndexMap for better lookup performance +type TorrentMapShard = HashMap>>; + pub struct TorrentData { peer_map: RwLock>, pending_removal: AtomicBool,