aquatic_udp: more work on splitting mio and glommio logic

aquatic_udp/src/lib/mio/mod.rs

@@ -1,2 +1,122 @@
+use std::sync::{
+    atomic::{AtomicUsize, Ordering},
+    Arc,
+};
+use std::thread::Builder;
+use std::time::Duration;
+
+use anyhow::Context;
+use crossbeam_channel::unbounded;
+use privdrop::PrivDrop;
+
 pub mod handlers;
 pub mod network;
+pub mod tasks;
+
+use crate::common::State;
+use crate::config::Config;
+use crate::update_access_list;
+
+pub fn run(config: Config) -> ::anyhow::Result<()> {
+    let state = State::default();
+
+    update_access_list(&config, &state.access_list);
+
+    let num_bound_sockets = Arc::new(AtomicUsize::new(0));
+
+    start_workers(config.clone(), state.clone(), num_bound_sockets.clone())?;
+
+    if config.privileges.drop_privileges {
+        let mut counter = 0usize;
+
+        loop {
+            let sockets = num_bound_sockets.load(Ordering::SeqCst);
+
+            if sockets == config.socket_workers {
+                PrivDrop::default()
+                    .chroot(config.privileges.chroot_path.clone())
+                    .user(config.privileges.user.clone())
+                    .apply()?;
+
+                break;
+            }
+
+            ::std::thread::sleep(Duration::from_millis(10));
+
+            counter += 1;
+
+            if counter == 500 {
+                panic!("Sockets didn't bind in time for privilege drop.");
+            }
+        }
+    }
+
+    loop {
+        ::std::thread::sleep(Duration::from_secs(config.cleaning.interval));
+
+        update_access_list(&config, &state.access_list);
+
+        state.torrents.lock().clean(&config, &state.access_list);
+    }
+}
+
+pub fn start_workers(
+    config: Config,
+    state: State,
+    num_bound_sockets: Arc<AtomicUsize>,
+) -> ::anyhow::Result<()> {
+    let (request_sender, request_receiver) = unbounded();
+    let (response_sender, response_receiver) = unbounded();
+
+    for i in 0..config.request_workers {
+        let state = state.clone();
+        let config = config.clone();
+        let request_receiver = request_receiver.clone();
+        let response_sender = response_sender.clone();
+
+        Builder::new()
+            .name(format!("request-{:02}", i + 1))
+            .spawn(move || {
+                handlers::run_request_worker(state, config, request_receiver, response_sender)
+            })
+            .with_context(|| "spawn request worker")?;
+    }
+
+    for i in 0..config.socket_workers {
+        let state = state.clone();
+        let config = config.clone();
+        let request_sender = request_sender.clone();
+        let response_receiver = response_receiver.clone();
+        let num_bound_sockets = num_bound_sockets.clone();
+
+        Builder::new()
+            .name(format!("socket-{:02}", i + 1))
+            .spawn(move || {
+                network::run_socket_worker(
+                    state,
+                    config,
+                    i,
+                    request_sender,
+                    response_receiver,
+                    num_bound_sockets,
+                )
+            })
+            .with_context(|| "spawn socket worker")?;
+    }
+
+    if config.statistics.interval != 0 {
+        let state = state.clone();
+        let config = config.clone();
+
+        Builder::new()
+            .name("statistics-collector".to_string())
+            .spawn(move || loop {
+                ::std::thread::sleep(Duration::from_secs(config.statistics.interval));
+
+                tasks::gather_and_print_statistics(&state, &config);
+            })
+            .with_context(|| "spawn statistics worker")?;
+    }
+
+    Ok(())
+}

aquatic_udp/src/lib/mio/network.rs

@@ -4,7 +4,7 @@ use std::sync::{
     atomic::{AtomicUsize, Ordering},
     Arc,
 };
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use std::vec::Drain;
 
 use crossbeam_channel::{Receiver, Sender};

aquatic_udp/src/lib/mio/tasks.rs

@@ -0,0 +1,76 @@
+use std::sync::atomic::Ordering;
+
+use histogram::Histogram;
+
+use crate::common::*;
+use crate::config::Config;
+
+pub fn gather_and_print_statistics(state: &State, config: &Config) {
+    let interval = config.statistics.interval;
+
+    let requests_received: f64 = state
+        .statistics
+        .requests_received
+        .fetch_and(0, Ordering::SeqCst) as f64;
+    let responses_sent: f64 = state
+        .statistics
+        .responses_sent
+        .fetch_and(0, Ordering::SeqCst) as f64;
+    let bytes_received: f64 = state
+        .statistics
+        .bytes_received
+        .fetch_and(0, Ordering::SeqCst) as f64;
+    let bytes_sent: f64 = state.statistics.bytes_sent.fetch_and(0, Ordering::SeqCst) as f64;
+
+    let requests_per_second = requests_received / interval as f64;
+    let responses_per_second: f64 = responses_sent / interval as f64;
+    let bytes_received_per_second: f64 = bytes_received / interval as f64;
+    let bytes_sent_per_second: f64 = bytes_sent / interval as f64;
+
+    println!(
+        "stats: {:.2} requests/second, {:.2} responses/second",
+        requests_per_second, responses_per_second
+    );
+
+    println!(
+        "bandwidth: {:7.2} Mbit/s in, {:7.2} Mbit/s out",
+        bytes_received_per_second * 8.0 / 1_000_000.0,
+        bytes_sent_per_second * 8.0 / 1_000_000.0,
+    );
+
+    let mut peers_per_torrent = Histogram::new();
+
+    {
+        let torrents = &mut state.torrents.lock();
+
+        for torrent in torrents.ipv4.values() {
+            let num_peers = (torrent.num_seeders + torrent.num_leechers) as u64;
+
+            if let Err(err) = peers_per_torrent.increment(num_peers) {
+                ::log::error!("error incrementing peers_per_torrent histogram: {}", err)
+            }
+        }
+        for torrent in torrents.ipv6.values() {
+            let num_peers = (torrent.num_seeders + torrent.num_leechers) as u64;
+
+            if let Err(err) = peers_per_torrent.increment(num_peers) {
+                ::log::error!("error incrementing peers_per_torrent histogram: {}", err)
+            }
+        }
+    }
+
+    if peers_per_torrent.entries() != 0 {
+        println!(
+            "peers per torrent: min: {}, p50: {}, p75: {}, p90: {}, p99: {}, p999: {}, max: {}",
+            peers_per_torrent.minimum().unwrap(),
+            peers_per_torrent.percentile(50.0).unwrap(),
+            peers_per_torrent.percentile(75.0).unwrap(),
+            peers_per_torrent.percentile(90.0).unwrap(),
+            peers_per_torrent.percentile(99.0).unwrap(),
+            peers_per_torrent.percentile(99.9).unwrap(),
+            peers_per_torrent.maximum().unwrap(),
+        );
+    }
+
+    println!();
+}