refactor: use generics for perr list implementation with SkipMap

Author: josecelano

packages/torrent-repository/src/entry/mod.rs

@@ -2,16 +2,13 @@ use std::fmt::Debug;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use crossbeam_skiplist::SkipMap;
 use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
 
 pub mod mutex_std;
 pub mod mutex_tokio;
 pub mod single;
-pub mod skip_map;
-pub mod skip_map_mutex_std;
 
 pub trait Entry {
     /// It returns the swarm metadata (statistics) as a struct:
@@ -82,19 +79,10 @@ pub trait EntryAsync {
 /// that's the list of all the peers trying to download the same torrent.
 /// The tracker keeps one entry like this for every torrent.
 #[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub struct Torrent {
+pub struct Torrent<T> {
     /// The swarm: a network of peers that are all trying to download the torrent associated to this entry
     // #[serde(skip)]
-    pub(crate) peers: std::collections::BTreeMap<peer::Id, Arc<peer::Peer>>,
-    /// The number of peers that have ever completed downloading the torrent associated to this entry
-    pub(crate) downloaded: u32,
-}
-
-#[derive(Debug, Default)]
-pub struct SkipMapTorrent {
-    /// The swarm: a network of peers that are all trying to download the torrent associated to this entry
-    // #[serde(skip)]
-    pub(crate) peers: SkipMap<peer::Id, Arc<peer::Peer>>,
+    pub(crate) peers: T,
     /// The number of peers that have ever completed downloading the torrent associated to this entry
     pub(crate) downloaded: u32,
 }

packages/torrent-repository/src/entry/mutex_std.rs

@@ -6,9 +6,9 @@ use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
 
 use super::{Entry, EntrySync};
-use crate::{EntryMutexStd, EntrySingle};
+use crate::{BTreeMapPeerList, EntryMutexStd, EntrySingle, SkipMapPeerList};
 
-impl EntrySync for EntryMutexStd {
+impl EntrySync for EntryMutexStd<BTreeMapPeerList> {
     fn get_swarm_metadata(&self) -> SwarmMetadata {
         self.lock().expect("it should get a lock").get_swarm_metadata()
     }
@@ -44,8 +44,44 @@ impl EntrySync for EntryMutexStd {
     }
 }
 
-impl From<EntrySingle> for EntryMutexStd {
-    fn from(entry: EntrySingle) -> Self {
+impl EntrySync for EntryMutexStd<SkipMapPeerList> {
+    fn get_swarm_metadata(&self) -> SwarmMetadata {
+        self.lock().expect("it should get a lock").get_swarm_metadata()
+    }
+
+    fn is_good(&self, policy: &TrackerPolicy) -> bool {
+        self.lock().expect("it should get a lock").is_good(policy)
+    }
+
+    fn peers_is_empty(&self) -> bool {
+        self.lock().expect("it should get a lock").peers_is_empty()
+    }
+
+    fn get_peers_len(&self) -> usize {
+        self.lock().expect("it should get a lock").get_peers_len()
+    }
+
+    fn get_peers(&self, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+        self.lock().expect("it should get lock").get_peers(limit)
+    }
+
+    fn get_peers_for_client(&self, client: &SocketAddr, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+        self.lock().expect("it should get lock").get_peers_for_client(client, limit)
+    }
+
+    fn upsert_peer(&self, peer: &peer::Peer) -> bool {
+        self.lock().expect("it should lock the entry").upsert_peer(peer)
+    }
+
+    fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) {
+        self.lock()
+            .expect("it should lock the entry")
+            .remove_inactive_peers(current_cutoff);
+    }
+}
+
+impl<T> From<EntrySingle<T>> for EntryMutexStd<T> {
+    fn from(entry: EntrySingle<T>) -> Self {
         Arc::new(std::sync::Mutex::new(entry))
     }
 }

packages/torrent-repository/src/entry/mutex_tokio.rs

@@ -6,9 +6,9 @@ use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
 
 use super::{Entry, EntryAsync};
-use crate::{EntryMutexTokio, EntrySingle};
+use crate::{BTreeMapPeerList, EntryMutexTokio, EntrySingle, SkipMapPeerList};
 
-impl EntryAsync for EntryMutexTokio {
+impl EntryAsync for EntryMutexTokio<BTreeMapPeerList> {
     async fn get_swarm_metadata(&self) -> SwarmMetadata {
         self.lock().await.get_swarm_metadata()
     }
@@ -42,8 +42,42 @@ impl EntryAsync for EntryMutexTokio {
     }
 }
 
-impl From<EntrySingle> for EntryMutexTokio {
-    fn from(entry: EntrySingle) -> Self {
+impl EntryAsync for EntryMutexTokio<SkipMapPeerList> {
+    async fn get_swarm_metadata(&self) -> SwarmMetadata {
+        self.lock().await.get_swarm_metadata()
+    }
+
+    async fn check_good(self, policy: &TrackerPolicy) -> bool {
+        self.lock().await.is_good(policy)
+    }
+
+    async fn peers_is_empty(&self) -> bool {
+        self.lock().await.peers_is_empty()
+    }
+
+    async fn get_peers_len(&self) -> usize {
+        self.lock().await.get_peers_len()
+    }
+
+    async fn get_peers(&self, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+        self.lock().await.get_peers(limit)
+    }
+
+    async fn get_peers_for_client(&self, client: &SocketAddr, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+        self.lock().await.get_peers_for_client(client, limit)
+    }
+
+    async fn upsert_peer(self, peer: &peer::Peer) -> bool {
+        self.lock().await.upsert_peer(peer)
+    }
+
+    async fn remove_inactive_peers(self, current_cutoff: DurationSinceUnixEpoch) {
+        self.lock().await.remove_inactive_peers(current_cutoff);
+    }
+}
+
+impl<T> From<EntrySingle<T>> for EntryMutexTokio<T> {
+    fn from(entry: EntrySingle<T>) -> Self {
         Arc::new(tokio::sync::Mutex::new(entry))
     }
 }

packages/torrent-repository/src/entry/single.rs

@@ -3,14 +3,14 @@ use std::sync::Arc;
 
 use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::announce_event::AnnounceEvent;
-use torrust_tracker_primitives::peer::{self};
+use torrust_tracker_primitives::peer::{self, ReadInfo};
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::DurationSinceUnixEpoch;
 
 use super::Entry;
-use crate::EntrySingle;
+use crate::{BTreeMapPeerList, EntrySingle, SkipMapPeerList};
 
-impl Entry for EntrySingle {
+impl Entry for EntrySingle<BTreeMapPeerList> {
     #[allow(clippy::cast_possible_truncation)]
     fn get_swarm_metadata(&self) -> SwarmMetadata {
         let complete: u32 = self.peers.values().filter(|peer| peer.is_seeder()).count() as u32;
@@ -98,3 +98,108 @@ impl Entry for EntrySingle {
             .retain(|_, peer| peer::ReadInfo::get_updated(peer) > current_cutoff);
     }
 }
+
+impl Entry for EntrySingle<SkipMapPeerList> {
+    #[allow(clippy::cast_possible_truncation)]
+    fn get_swarm_metadata(&self) -> SwarmMetadata {
+        let complete: u32 = self.peers.iter().filter(|entry| entry.value().is_seeder()).count() as u32;
+        let incomplete: u32 = self.peers.len() as u32 - complete;
+
+        SwarmMetadata {
+            downloaded: self.downloaded,
+            complete,
+            incomplete,
+        }
+    }
+
+    fn is_good(&self, policy: &TrackerPolicy) -> bool {
+        if policy.persistent_torrent_completed_stat && self.downloaded > 0 {
+            return true;
+        }
+
+        if policy.remove_peerless_torrents && self.peers.is_empty() {
+            return false;
+        }
+
+        true
+    }
+
+    fn peers_is_empty(&self) -> bool {
+        self.peers.is_empty()
+    }
+
+    fn get_peers_len(&self) -> usize {
+        self.peers.len()
+    }
+    fn get_peers(&self, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+        match limit {
+            Some(limit) => self.peers.iter().take(limit).map(|entry| entry.value().clone()).collect(),
+            None => self.peers.iter().map(|entry| entry.value().clone()).collect(),
+        }
+    }
+
+    fn get_peers_for_client(&self, client: &SocketAddr, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+        match limit {
+            Some(limit) => self
+                .peers
+                .iter()
+                // Take peers which are not the client peer
+                .filter(|entry| peer::ReadInfo::get_address(entry.value().as_ref()) != *client)
+                // Limit the number of peers on the result
+                .take(limit)
+                .map(|entry| entry.value().clone())
+                .collect(),
+            None => self
+                .peers
+                .iter()
+                // Take peers which are not the client peer
+                .filter(|entry| peer::ReadInfo::get_address(entry.value().as_ref()) != *client)
+                .map(|entry| entry.value().clone())
+                .collect(),
+        }
+    }
+
+    fn upsert_peer(&mut self, peer: &peer::Peer) -> bool {
+        let mut downloaded_stats_updated: bool = false;
+
+        match peer::ReadInfo::get_event(peer) {
+            AnnounceEvent::Stopped => {
+                drop(self.peers.remove(&peer::ReadInfo::get_id(peer)));
+            }
+            AnnounceEvent::Completed => {
+                let previous = self.peers.get(&peer.get_id());
+
+                let increase_downloads = match previous {
+                    Some(entry) => {
+                        // Don't count if peer was already completed.
+                        entry.value().event != AnnounceEvent::Completed
+                    }
+                    None => {
+                        // Don't count if peer was not previously known
+                        false
+                    }
+                };
+
+                self.peers.insert(peer::ReadInfo::get_id(peer), Arc::new(*peer));
+
+                if increase_downloads {
+                    self.downloaded += 1;
+                    downloaded_stats_updated = true;
+                }
+            }
+            _ => {
+                drop(self.peers.insert(peer::ReadInfo::get_id(peer), Arc::new(*peer)));
+            }
+        }
+
+        downloaded_stats_updated
+    }
+
+    fn remove_inactive_peers(&mut self, current_cutoff: DurationSinceUnixEpoch) {
+        for entry in &self.peers {
+            if entry.value().get_updated() >= current_cutoff {
+                entry.remove();
+            }
+        }
+    }
+}