http, http_private: rename request workers to swarm workers

aquatic_http_private/src/workers/swarm/common.rs

@@ -0,0 +1,122 @@
+use std::net::{Ipv4Addr, Ipv6Addr};
+use std::time::Instant;
+
+use aquatic_common::{AmortizedIndexMap, ValidUntil};
+use aquatic_http_protocol::common::{AnnounceEvent, InfoHash, PeerId};
+use aquatic_http_protocol::response::ResponsePeer;
+
+pub trait Ip: ::std::fmt::Debug + Copy + Eq + ::std::hash::Hash {}
+
+impl Ip for Ipv4Addr {}
+impl Ip for Ipv6Addr {}
+
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+pub enum PeerStatus {
+    Seeding,
+    Leeching,
+    Stopped,
+}
+
+impl PeerStatus {
+    /// Determine peer status from announce event and number of bytes left.
+    ///
+    /// Likely, the last branch will be taken most of the time.
+    #[inline]
+    pub fn from_event_and_bytes_left(event: AnnounceEvent, opt_bytes_left: Option<usize>) -> Self {
+        if let AnnounceEvent::Stopped = event {
+            Self::Stopped
+        } else if let Some(0) = opt_bytes_left {
+            Self::Seeding
+        } else {
+            Self::Leeching
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct Peer<I: Ip> {
+    pub ip_address: I,
+    pub port: u16,
+    pub status: PeerStatus,
+    pub valid_until: ValidUntil,
+}
+
+impl<I: Ip> Peer<I> {
+    pub fn to_response_peer(&self) -> ResponsePeer<I> {
+        ResponsePeer {
+            ip_address: self.ip_address,
+            port: self.port,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct PeerMapKey<I: Ip> {
+    pub peer_id: PeerId,
+    pub ip_address: I,
+}
+
+pub type PeerMap<I> = AmortizedIndexMap<PeerMapKey<I>, Peer<I>>;
+
+pub struct TorrentData<I: Ip> {
+    pub peers: PeerMap<I>,
+    pub num_seeders: usize,
+    pub num_leechers: usize,
+}
+
+impl<I: Ip> Default for TorrentData<I> {
+    #[inline]
+    fn default() -> Self {
+        Self {
+            peers: Default::default(),
+            num_seeders: 0,
+            num_leechers: 0,
+        }
+    }
+}
+
+pub type TorrentMap<I> = AmortizedIndexMap<InfoHash, TorrentData<I>>;
+
+#[derive(Default)]
+pub struct TorrentMaps {
+    pub ipv4: TorrentMap<Ipv4Addr>,
+    pub ipv6: TorrentMap<Ipv6Addr>,
+}
+
+impl TorrentMaps {
+    pub fn clean(&mut self) {
+        Self::clean_torrent_map(&mut self.ipv4);
+        Self::clean_torrent_map(&mut self.ipv6);
+    }
+
+    fn clean_torrent_map<I: Ip>(torrent_map: &mut TorrentMap<I>) {
+        let now = Instant::now();
+
+        torrent_map.retain(|_, torrent_data| {
+            let num_seeders = &mut torrent_data.num_seeders;
+            let num_leechers = &mut torrent_data.num_leechers;
+
+            torrent_data.peers.retain(|_, peer| {
+                if peer.valid_until.0 >= now {
+                    true
+                } else {
+                    match peer.status {
+                        PeerStatus::Seeding => {
+                            *num_seeders -= 1;
+                        }
+                        PeerStatus::Leeching => {
+                            *num_leechers -= 1;
+                        }
+                        _ => (),
+                    };
+
+                    false
+                }
+            });
+
+            !torrent_data.peers.is_empty()
+        });
+
+        torrent_map.shrink_to_fit();
+    }
+}

aquatic_http_private/src/workers/swarm/mod.rs

@@ -0,0 +1,211 @@
+mod common;
+
+use std::cell::RefCell;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+use std::rc::Rc;
+
+use aquatic_http_protocol::request::AnnounceRequest;
+use rand::prelude::SmallRng;
+use rand::SeedableRng;
+use tokio::sync::mpsc::Receiver;
+use tokio::task::LocalSet;
+use tokio::time;
+
+use aquatic_common::{extract_response_peers, CanonicalSocketAddr, PanicSentinel, ValidUntil};
+use aquatic_http_protocol::response::{
+    AnnounceResponse, Response, ResponsePeer, ResponsePeerListV4, ResponsePeerListV6,
+};
+
+use crate::common::ChannelAnnounceRequest;
+use crate::config::Config;
+
+use common::*;
+
+pub fn run_swarm_worker(
+    _sentinel: PanicSentinel,
+    config: Config,
+    request_receiver: Receiver<ChannelAnnounceRequest>,
+) -> anyhow::Result<()> {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
+
+    runtime.block_on(run_inner(config, request_receiver))?;
+
+    Ok(())
+}
+
+async fn run_inner(
+    config: Config,
+    mut request_receiver: Receiver<ChannelAnnounceRequest>,
+) -> anyhow::Result<()> {
+    let torrents = Rc::new(RefCell::new(TorrentMaps::default()));
+    let mut rng = SmallRng::from_entropy();
+
+    LocalSet::new().spawn_local(periodically_clean_torrents(
+        config.clone(),
+        torrents.clone(),
+    ));
+
+    loop {
+        let request = request_receiver
+            .recv()
+            .await
+            .ok_or_else(|| anyhow::anyhow!("request channel closed"))?;
+
+        let valid_until = ValidUntil::new(config.cleaning.max_peer_age);
+
+        let response = handle_announce_request(
+            &config,
+            &mut rng,
+            &mut torrents.borrow_mut(),
+            valid_until,
+            request.source_addr,
+            request.request.into(),
+        );
+
+        let _ = request.response_sender.send(Response::Announce(response));
+    }
+}
+
+async fn periodically_clean_torrents(config: Config, torrents: Rc<RefCell<TorrentMaps>>) {
+    let mut interval = time::interval(time::Duration::from_secs(
+        config.cleaning.torrent_cleaning_interval,
+    ));
+
+    loop {
+        interval.tick().await;
+
+        torrents.borrow_mut().clean();
+    }
+}
+
+fn handle_announce_request(
+    config: &Config,
+    rng: &mut SmallRng,
+    torrent_maps: &mut TorrentMaps,
+    valid_until: ValidUntil,
+    source_addr: CanonicalSocketAddr,
+    request: AnnounceRequest,
+) -> AnnounceResponse {
+    match source_addr.get().ip() {
+        IpAddr::V4(source_ip) => {
+            let torrent_data: &mut TorrentData<Ipv4Addr> =
+                torrent_maps.ipv4.entry(request.info_hash).or_default();
+
+            let (seeders, leechers, response_peers) = upsert_peer_and_get_response_peers(
+                config,
+                rng,
+                torrent_data,
+                source_ip,
+                request,
+                valid_until,
+            );
+
+            let response = AnnounceResponse {
+                complete: seeders,
+                incomplete: leechers,
+                announce_interval: config.protocol.peer_announce_interval,
+                peers: ResponsePeerListV4(response_peers),
+                peers6: ResponsePeerListV6(vec![]),
+                warning_message: None,
+            };
+
+            response
+        }
+        IpAddr::V6(source_ip) => {
+            let torrent_data: &mut TorrentData<Ipv6Addr> =
+                torrent_maps.ipv6.entry(request.info_hash).or_default();
+
+            let (seeders, leechers, response_peers) = upsert_peer_and_get_response_peers(
+                config,
+                rng,
+                torrent_data,
+                source_ip,
+                request,
+                valid_until,
+            );
+
+            let response = AnnounceResponse {
+                complete: seeders,
+                incomplete: leechers,
+                announce_interval: config.protocol.peer_announce_interval,
+                peers: ResponsePeerListV4(vec![]),
+                peers6: ResponsePeerListV6(response_peers),
+                warning_message: None,
+            };
+
+            response
+        }
+    }
+}
+
+/// Insert/update peer. Return num_seeders, num_leechers and response peers
+pub fn upsert_peer_and_get_response_peers<I: Ip>(
+    config: &Config,
+    rng: &mut SmallRng,
+    torrent_data: &mut TorrentData<I>,
+    source_ip: I,
+    request: AnnounceRequest,
+    valid_until: ValidUntil,
+) -> (usize, usize, Vec<ResponsePeer<I>>) {
+    // Insert/update/remove peer who sent this request
+
+    let peer_status =
+        PeerStatus::from_event_and_bytes_left(request.event, Some(request.bytes_left));
+
+    let peer = Peer {
+        ip_address: source_ip,
+        port: request.port,
+        status: peer_status,
+        valid_until,
+    };
+
+    let peer_map_key = PeerMapKey {
+        peer_id: request.peer_id,
+        ip_address: source_ip,
+    };
+
+    let opt_removed_peer = match peer_status {
+        PeerStatus::Leeching => {
+            torrent_data.num_leechers += 1;
+
+            torrent_data.peers.insert(peer_map_key.clone(), peer)
+        }
+        PeerStatus::Seeding => {
+            torrent_data.num_seeders += 1;
+
+            torrent_data.peers.insert(peer_map_key.clone(), peer)
+        }
+        PeerStatus::Stopped => torrent_data.peers.remove(&peer_map_key),
+    };
+
+    match opt_removed_peer.map(|peer| peer.status) {
+        Some(PeerStatus::Leeching) => {
+            torrent_data.num_leechers -= 1;
+        }
+        Some(PeerStatus::Seeding) => {
+            torrent_data.num_seeders -= 1;
+        }
+        _ => {}
+    }
+
+    let max_num_peers_to_take = match request.numwant {
+        Some(0) | None => config.protocol.max_peers,
+        Some(numwant) => numwant.min(config.protocol.max_peers),
+    };
+
+    let response_peers: Vec<ResponsePeer<I>> = extract_response_peers(
+        rng,
+        &torrent_data.peers,
+        max_num_peers_to_take,
+        peer_map_key,
+        Peer::to_response_peer,
+    );
+
+    (
+        torrent_data.num_seeders,
+        torrent_data.num_leechers,
+        response_peers,
+    )
+}