Use crossbeam channel insteaf of SegQueue

Reduces overutilization of CPU greatly

aquatic/Cargo.toml

@@ -14,8 +14,7 @@ name = "aquatic"
 [dependencies]
 bittorrent_udp = { path = "../bittorrent_udp" }
 cli_helpers = { path = "../cli_helpers" }
-crossbeam-queue = "0.2"
-crossbeam-utils = "0.7"
+crossbeam-channel = "0.4"
 dashmap = "3"
 histogram = "0.6"
 indexmap = "1"

aquatic/src/lib/common.rs

@@ -3,7 +3,6 @@ use std::sync::atomic::AtomicUsize;
 use std::net::{SocketAddr, IpAddr};
 use std::time::Instant;
 
-use crossbeam_queue::SegQueue;
 use dashmap::DashMap;
 use indexmap::IndexMap;
 
@@ -138,8 +137,6 @@ pub struct State {
     pub connections: Arc<ConnectionMap>,
     pub torrents: Arc<TorrentMap>,
     pub statistics: Arc<Statistics>,
-    pub request_queue: Arc<SegQueue<(Request, SocketAddr)>>,
-    pub response_queue: Arc<SegQueue<(Response, SocketAddr)>>,
 }
 
 impl State {
@@ -148,8 +145,6 @@ impl State {
             connections: Arc::new(DashMap::new()),
             torrents: Arc::new(DashMap::new()),
             statistics: Arc::new(Statistics::default()),
-            request_queue: Arc::new(SegQueue::new()),
-            response_queue: Arc::new(SegQueue::new()),
         }
     }
 }

aquatic/src/lib/handlers.rs

@@ -1,10 +1,10 @@
 use std::net::SocketAddr;
 use std::sync::atomic::Ordering;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 use std::vec::Drain;
 
+use crossbeam_channel::{Sender, Receiver};
 use rand::{SeedableRng, Rng, rngs::{SmallRng, StdRng}};
-use crossbeam_utils::Backoff;
 
 use bittorrent_udp::types::*;
 
@@ -15,6 +15,8 @@ use crate::config::Config;
 pub fn handle(
     state: State,
     config: Config,
+    request_receiver: Receiver<(Request, SocketAddr)>,
+    response_sender: Sender<(Response, SocketAddr)>,
 ){
     let mut connect_requests: Vec<(ConnectRequest, SocketAddr)> = Vec::new();
     let mut announce_requests: Vec<(AnnounceRequest, SocketAddr)> = Vec::new();
@@ -23,42 +25,53 @@ pub fn handle(
     let mut std_rng = StdRng::from_entropy();
     let mut small_rng = SmallRng::from_rng(&mut std_rng).unwrap();
 
-    let backoff = Backoff::new();
+    let timeout = Duration::from_millis(10);
 
     loop {
-        if state.request_queue.is_empty(){
-            backoff.snooze();
-        } else {
-            while let Ok((request, src)) = state.request_queue.pop(){
-                match request {
-                    Request::Connect(r) => {
-                        connect_requests.push((r, src))
-                    },
-                    Request::Announce(r) => {
-                        announce_requests.push((r, src))
-                    },
-                    Request::Scrape(r) => {
-                        scrape_requests.push((r, src))
-                    },
+        for i in 0..1000 {
+            let (request, src): (Request, SocketAddr) = if i == 0 {
+                match request_receiver.recv(){
+                    Ok(r) => r,
+                    Err(_) => break,
                 }
-            }
+            } else {
+                match request_receiver.recv_timeout(timeout){
+                    Ok(r) => r,
+                    Err(_) => break,
+                }
+            };
 
-            handle_connect_requests(
-                &state,
-                &mut std_rng,
-                connect_requests.drain(..)
-            );
-            handle_announce_requests(
-                &state,
-                &config,
-                &mut small_rng,
-                announce_requests.drain(..),
-            );
-            handle_scrape_requests(
-                &state,
-                scrape_requests.drain(..),
-            );
+            match request {
+                Request::Connect(r) => {
+                    connect_requests.push((r, src))
+                },
+                Request::Announce(r) => {
+                    announce_requests.push((r, src))
+                },
+                Request::Scrape(r) => {
+                    scrape_requests.push((r, src))
+                },
+            }
         }
+
+        handle_connect_requests(
+            &state,
+            &mut std_rng,
+            connect_requests.drain(..),
+            &response_sender
+        );
+        handle_announce_requests(
+            &state,
+            &config,
+            &mut small_rng,
+            announce_requests.drain(..),
+            &response_sender
+        );
+        handle_scrape_requests(
+            &state,
+            scrape_requests.drain(..),
+            &response_sender
+        );
     }
 }
 
@@ -68,6 +81,7 @@ pub fn handle_connect_requests(
     state: &State,
     rng: &mut StdRng,
     requests: Drain<(ConnectRequest, SocketAddr)>,
+    response_sender: &Sender<(Response, SocketAddr)>,
 ){
     let now = Time(Instant::now());
 
@@ -88,7 +102,7 @@ pub fn handle_connect_requests(
             }
         );
         
-        state.response_queue.push((response, src));
+        response_sender.send((response, src));
     }
 }
 
@@ -99,6 +113,7 @@ pub fn handle_announce_requests(
     config: &Config,
     rng: &mut SmallRng,
     requests: Drain<(AnnounceRequest, SocketAddr)>,
+    response_sender: &Sender<(Response, SocketAddr)>,
 ){
     for (request, src) in requests {
         let connection_key = ConnectionKey {
@@ -112,7 +127,7 @@ pub fn handle_announce_requests(
                 message: "Connection invalid or expired".to_string()
             };
 
-            state.response_queue.push((response.into(), src));
+            response_sender.send((response.into(), src));
         }
 
         let peer_key = PeerMapKey {
@@ -176,7 +191,7 @@ pub fn handle_announce_requests(
             peers: response_peers
         });
 
-        state.response_queue.push((response, src));
+        response_sender.send((response, src));
     }
 }
 
@@ -185,6 +200,7 @@ pub fn handle_announce_requests(
 pub fn handle_scrape_requests(
     state: &State,
     requests: Drain<(ScrapeRequest, SocketAddr)>,
+    response_sender: &Sender<(Response, SocketAddr)>,
 ){
     let empty_stats = create_torrent_scrape_statistics(0, 0);
 
@@ -200,7 +216,7 @@ pub fn handle_scrape_requests(
                 message: "Connection invalid or expired".to_string()
             };
 
-            state.response_queue.push((response.into(), src));
+            response_sender.send((response.into(), src));
         }
 
         let mut stats: Vec<TorrentScrapeStatistics> = Vec::with_capacity(
@@ -223,7 +239,7 @@ pub fn handle_scrape_requests(
             torrent_stats: stats,
         });
 
-        state.response_queue.push((response, src));
+        response_sender.send((response, src));
     };
 }
 

aquatic/src/lib/lib.rs

@@ -1,5 +1,7 @@
 use std::time::Duration;
 
+use crossbeam_channel::unbounded;
+
 pub mod common;
 pub mod config;
 pub mod handlers;
@@ -13,21 +15,28 @@ use common::State;
 pub fn run(config: Config){
     let state = State::new();
 
+    let (request_sender, request_receiver) = unbounded();
+    let (response_sender, response_receiver) = unbounded();
+
     for _ in 0..config.response_workers {
         let state = state.clone();
         let config = config.clone();
+        let request_receiver = request_receiver.clone();
+        let response_sender = response_sender.clone();
 
         ::std::thread::spawn(move || {
-            handlers::handle(state, config);
+            handlers::handle(state, config, request_receiver, response_sender);
         });
     }
 
     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();
 
         ::std::thread::spawn(move || {
-            network::run_event_loop(state, config, i);
+            network::run_event_loop(state, config, i, request_sender, response_receiver);
         });
     }
 

aquatic/src/lib/network.rs

@@ -1,7 +1,9 @@
 use std::sync::atomic::Ordering;
 use std::io::{Cursor, ErrorKind};
+use std::net::SocketAddr;
 use std::time::Duration;
 
+use crossbeam_channel::{Sender, Receiver};
 use mio::{Events, Poll, Interest, Token};
 use mio::net::UdpSocket;
 use net2::{UdpSocketExt, UdpBuilder};
@@ -18,6 +20,8 @@ pub fn run_event_loop(
     state: State,
     config: Config,
     token_num: usize,
+    request_sender: Sender<(Request, SocketAddr)>,
+    response_receiver: Receiver<(Response, SocketAddr)>,
 ){
     let mut buffer = [0u8; MAX_PACKET_SIZE];
 
@@ -32,7 +36,7 @@ pub fn run_event_loop(
 
     let mut events = Events::with_capacity(config.network.poll_event_capacity);
 
-    let timeout = Duration::from_millis(1);
+    let timeout = Duration::from_millis(50);
 
     loop {
         poll.poll(&mut events, Some(timeout))
@@ -48,6 +52,7 @@ pub fn run_event_loop(
                         &config,
                         &mut socket,
                         &mut buffer,
+                        &request_sender,
                     );
 
                     state.statistics.readable_events.fetch_add(1, Ordering::SeqCst);
@@ -64,6 +69,7 @@ pub fn run_event_loop(
             &config,
             &mut socket,
             &mut buffer,
+            &response_receiver,
         );
     }
 }
@@ -107,6 +113,7 @@ fn read_requests(
     config: &Config,
     socket: &mut UdpSocket,
     buffer: &mut [u8],
+    request_sender: &Sender<(Request, SocketAddr)>,
 ){
     let mut requests_received: usize = 0;
     let mut bytes_received: usize = 0;
@@ -127,7 +134,7 @@ fn read_requests(
 
                 match request {
                     Ok(request) => {
-                        state.request_queue.push((request, src));
+                        request_sender.try_send((request, src));
                     },
                     Err(err) => {
                         eprintln!("request_from_bytes error: {:?}", err);
@@ -178,13 +185,14 @@ fn send_responses(
     config: &Config,
     socket: &mut UdpSocket,
     buffer: &mut [u8],
+    response_receiver: &Receiver<(Response, SocketAddr)>,
 ){
     let mut responses_sent: usize = 0;
     let mut bytes_sent: usize = 0;
 
     let mut cursor = Cursor::new(buffer);
 
-    while let Ok((response, src)) = state.response_queue.pop(){
+    for (response, src) in response_receiver.try_iter(){
         cursor.set_position(0);
 
         response_to_bytes(&mut cursor, response, IpVersion::IPv4).unwrap();