dev: rework torrent/repository

Author: da2ce7

Cargo.lock

@@ -39,6 +39,7 @@ binascii = "0"
 chrono = { version = "0", default-features = false, features = ["clock"] }
 config = "0"
 derive_more = "0"
+dashmap = "5"
 fern = "0"
 futures = "0"
 hyper = "1"

Cargo.toml

@@ -34,6 +34,7 @@
         "Containerfile",
         "curr",
         "Cyberneering",
+        "dashmap",
         "datagram",
         "datetime",
         "Deque",

cSpell.json

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 use clap::Parser;
 use futures::stream::FuturesUnordered;
-use torrust_tracker::core::torrent::repository_asyn::{RepositoryAsync, RepositoryTokioRwLock};
+use torrust_tracker::core::torrent::repository_asyn::RepositoryTokioRwLock;
 use torrust_tracker::core::torrent::UpdateTorrentAsync;
 use torrust_tracker::shared::bit_torrent::info_hash::InfoHash;
 
@@ -12,7 +12,8 @@ use crate::benches::utils::{generate_unique_info_hashes, get_average_and_adjuste
 
 pub async fn add_one_torrent<T>(samples: usize) -> (Duration, Duration)
 where
-    RepositoryTokioRwLock<T>: RepositoryAsync<T> + UpdateTorrentAsync,
+    T: Default,
+    RepositoryTokioRwLock<T>: UpdateTorrentAsync + Default,
 {
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
 
@@ -38,8 +39,8 @@ where
 // Add one torrent ten thousand times in parallel (depending on the set worker threads)
 pub async fn update_one_torrent_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
-    T: Send + Sync + 'static,
-    RepositoryTokioRwLock<T>: RepositoryAsync<T> + UpdateTorrentAsync,
+    T: Default + Send + Sync + 'static,
+    RepositoryTokioRwLock<T>: UpdateTorrentAsync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
@@ -88,8 +89,8 @@ where
 // Add ten thousand torrents in parallel (depending on the set worker threads)
 pub async fn add_multiple_torrents_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
-    T: Send + Sync + 'static,
-    RepositoryTokioRwLock<T>: RepositoryAsync<T> + UpdateTorrentAsync,
+    T: Default + Send + Sync + 'static,
+    RepositoryTokioRwLock<T>: UpdateTorrentAsync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
@@ -133,8 +134,8 @@ where
 // Async update ten thousand torrents in parallel (depending on the set worker threads)
 pub async fn update_multiple_torrents_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
-    T: Send + Sync + 'static,
-    RepositoryTokioRwLock<T>: RepositoryAsync<T> + UpdateTorrentAsync,
+    T: Default + Send + Sync + 'static,
+    RepositoryTokioRwLock<T>: UpdateTorrentAsync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);

packages/torrent-repository-benchmarks/src/benches/asyn.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 use clap::Parser;
 use futures::stream::FuturesUnordered;
-use torrust_tracker::core::torrent::repository_sync::{RepositoryStdRwLock, RepositorySync};
+use torrust_tracker::core::torrent::repository_sync::RepositoryStdRwLock;
 use torrust_tracker::core::torrent::UpdateTorrentSync;
 use torrust_tracker::shared::bit_torrent::info_hash::InfoHash;
 
@@ -14,7 +14,7 @@ use crate::benches::utils::{generate_unique_info_hashes, get_average_and_adjuste
 #[must_use]
 pub fn add_one_torrent<T>(samples: usize) -> (Duration, Duration)
 where
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentSync,
+    RepositoryStdRwLock<T>: UpdateTorrentSync + Default,
 {
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
 
@@ -39,7 +39,7 @@ where
 pub async fn update_one_torrent_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
     T: Send + Sync + 'static,
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentSync,
+    RepositoryStdRwLock<T>: UpdateTorrentSync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
@@ -85,7 +85,7 @@ where
 pub async fn add_multiple_torrents_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
     T: Send + Sync + 'static,
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentSync,
+    RepositoryStdRwLock<T>: UpdateTorrentSync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
@@ -128,7 +128,7 @@ where
 pub async fn update_multiple_torrents_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
     T: Send + Sync + 'static,
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentSync,
+    RepositoryStdRwLock<T>: UpdateTorrentSync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);

packages/torrent-repository-benchmarks/src/benches/sync.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 use clap::Parser;
 use futures::stream::FuturesUnordered;
-use torrust_tracker::core::torrent::repository_sync::{RepositoryStdRwLock, RepositorySync};
+use torrust_tracker::core::torrent::repository_sync::RepositoryStdRwLock;
 use torrust_tracker::core::torrent::UpdateTorrentAsync;
 use torrust_tracker::shared::bit_torrent::info_hash::InfoHash;
 
@@ -14,7 +14,7 @@ use crate::benches::utils::{generate_unique_info_hashes, get_average_and_adjuste
 #[must_use]
 pub async fn add_one_torrent<T>(samples: usize) -> (Duration, Duration)
 where
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentAsync,
+    RepositoryStdRwLock<T>: UpdateTorrentAsync + Default,
 {
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
 
@@ -41,7 +41,7 @@ where
 pub async fn update_one_torrent_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
     T: Send + Sync + 'static,
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentAsync,
+    RepositoryStdRwLock<T>: UpdateTorrentAsync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
@@ -91,7 +91,7 @@ where
 pub async fn add_multiple_torrents_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
     T: Send + Sync + 'static,
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentAsync,
+    RepositoryStdRwLock<T>: UpdateTorrentAsync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);
@@ -136,7 +136,7 @@ where
 pub async fn update_multiple_torrents_in_parallel<T>(runtime: &tokio::runtime::Runtime, samples: usize) -> (Duration, Duration)
 where
     T: Send + Sync + 'static,
-    RepositoryStdRwLock<T>: RepositorySync<T> + UpdateTorrentAsync,
+    RepositoryStdRwLock<T>: UpdateTorrentAsync + Default,
 {
     let args = Args::parse();
     let mut results: Vec<Duration> = Vec::with_capacity(samples);

packages/torrent-repository-benchmarks/src/benches/sync_asyn.rs

@@ -56,6 +56,8 @@ use self::error::Error;
 use crate::core::auth::{self, Key};
 use crate::shared::bit_torrent::info_hash::InfoHash;
 
+pub type PersistentTorrents = Vec<(InfoHash, u32)>;
+
 struct Builder<T>
 where
     T: Database,
@@ -125,7 +127,7 @@ pub trait Database: Sync + Send {
     /// # Errors
     ///
     /// Will return `Err` if unable to load.
-    async fn load_persistent_torrents(&self) -> Result<Vec<(InfoHash, u32)>, Error>;
+    async fn load_persistent_torrents(&self) -> Result<PersistentTorrents, Error>;
 
     /// It saves the torrent metrics data into the database.
     ///

src/core/databases/mod.rs

@@ -439,14 +439,13 @@ pub mod services;
 pub mod statistics;
 pub mod torrent;
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::HashMap;
 use std::net::IpAddr;
 use std::panic::Location;
 use std::sync::Arc;
 use std::time::Duration;
 
 use derive_more::Constructor;
-use futures::future::join_all;
 use log::debug;
 use tokio::sync::mpsc::error::SendError;
 use torrust_tracker_configuration::{AnnouncePolicy, Configuration};
@@ -455,8 +454,8 @@ use torrust_tracker_primitives::TrackerMode;
 use self::auth::Key;
 use self::error::Error;
 use self::peer::Peer;
-use self::torrent::repository_asyn::{RepositoryAsync, RepositoryTokioRwLock};
-use self::torrent::{Entry, UpdateTorrentAsync};
+use self::torrent::repository_asyn::RepositoryTokioRwLock;
+use self::torrent::{Entry, Repository, UpdateTorrentAsync};
 use crate::core::databases::Database;
 use crate::core::torrent::{SwarmMetadata, SwarmStats};
 use crate::shared::bit_torrent::info_hash::InfoHash;
@@ -685,9 +684,7 @@ impl Tracker {
 
     /// It returns the data for a `scrape` response.
     async fn get_swarm_metadata(&self, info_hash: &InfoHash) -> SwarmMetadata {
-        let torrents = self.torrents.get_torrents().await;
-
-        match torrents.get(info_hash) {
+        match self.torrents.get(info_hash).await {
             Some(torrent_entry) => torrent_entry.get_swarm_metadata(),
             None => SwarmMetadata::default(),
         }
@@ -704,29 +701,13 @@ impl Tracker {
     pub async fn load_torrents_from_database(&self) -> Result<(), databases::error::Error> {
         let persistent_torrents = self.database.load_persistent_torrents().await?;
 
-        let mut torrents = self.torrents.get_torrents_mut().await;
-
-        for (info_hash, completed) in persistent_torrents {
-            // Skip if torrent entry already exists
-            if torrents.contains_key(&info_hash) {
-                continue;
-            }
-
-            let torrent_entry = torrent::Entry {
-                peers: BTreeMap::default(),
-                completed,
-            };
-
-            torrents.insert(info_hash, torrent_entry);
-        }
+        self.torrents.import_persistent(&persistent_torrents).await;
 
         Ok(())
     }
 
     async fn get_torrent_peers_for_peer(&self, info_hash: &InfoHash, peer: &Peer) -> Vec<peer::Peer> {
-        let read_lock = self.torrents.get_torrents().await;
-
-        match read_lock.get(info_hash) {
+        match self.torrents.get(info_hash).await {
             None => vec![],
             Some(entry) => entry
                 .get_peers_for_peer(peer, TORRENT_PEERS_LIMIT)
@@ -740,9 +721,7 @@ impl Tracker {
     ///
     /// Get all torrent peers for a given torrent
     pub async fn get_torrent_peers(&self, info_hash: &InfoHash) -> Vec<peer::Peer> {
-        let read_lock = self.torrents.get_torrents().await;
-
-        match read_lock.get(info_hash) {
+        match self.torrents.get(info_hash).await {
             None => vec![],
             Some(entry) => entry.get_peers(TORRENT_PEERS_LIMIT).into_iter().copied().collect(),
         }
@@ -757,7 +736,7 @@ impl Tracker {
         // code-review: consider splitting the function in two (command and query segregation).
         // `update_torrent_with_peer` and `get_stats`
 
-        let (stats, stats_updated) = self.torrents.update_torrent_with_peer_and_get_stats(info_hash, peer).await;
+        let (stats_updated, stats) = self.torrents.update_torrent_with_peer_and_get_stats(info_hash, peer).await;
 
         if self.policy.persistent_torrent_completed_stat && stats_updated {
             let completed = stats.downloaded;
@@ -777,71 +756,18 @@ impl Tracker {
     /// # Panics
     /// Panics if unable to get the torrent metrics.
     pub async fn get_torrents_metrics(&self) -> TorrentsMetrics {
-        let arc_torrents_metrics = Arc::new(tokio::sync::Mutex::new(TorrentsMetrics {
-            seeders: 0,
-            completed: 0,
-            leechers: 0,
-            torrents: 0,
-        }));
-
-        let db = self.torrents.get_torrents().await.clone();
-
-        let futures = db
-            .values()
-            .map(|torrent_entry| {
-                let torrent_entry = torrent_entry.clone();
-                let torrents_metrics = arc_torrents_metrics.clone();
-
-                async move {
-                    tokio::spawn(async move {
-                        let (seeders, completed, leechers) = torrent_entry.get_stats();
-                        torrents_metrics.lock().await.seeders += u64::from(seeders);
-                        torrents_metrics.lock().await.completed += u64::from(completed);
-                        torrents_metrics.lock().await.leechers += u64::from(leechers);
-                        torrents_metrics.lock().await.torrents += 1;
-                    })
-                    .await
-                    .expect("Error torrent_metrics spawn");
-                }
-            })
-            .collect::<Vec<_>>();
-
-        join_all(futures).await;
-
-        let torrents_metrics = Arc::try_unwrap(arc_torrents_metrics).expect("Could not unwrap arc_torrents_metrics");
-
-        torrents_metrics.into_inner()
+        self.torrents.get_metrics().await
     }
 
     /// Remove inactive peers and (optionally) peerless torrents
     ///
     /// # Context: Tracker
     pub async fn cleanup_torrents(&self) {
-        let mut torrents_lock = self.torrents.get_torrents_mut().await;
-
         // If we don't need to remove torrents we will use the faster iter
         if self.policy.remove_peerless_torrents {
-            let mut cleaned_torrents_map: BTreeMap<InfoHash, torrent::Entry> = BTreeMap::new();
-
-            for (info_hash, torrent_entry) in &mut *torrents_lock {
-                torrent_entry.remove_inactive_peers(self.policy.max_peer_timeout);
-
-                if torrent_entry.peers.is_empty() {
-                    continue;
-                }
-
-                if self.policy.persistent_torrent_completed_stat && torrent_entry.completed == 0 {
-                    continue;
-                }
-
-                cleaned_torrents_map.insert(*info_hash, torrent_entry.clone());
-            }
-
-            *torrents_lock = cleaned_torrents_map;
+            self.torrents.remove_peerless_torrents(&self.policy).await;
         } else {
-            for torrent_entry in (*torrents_lock).values_mut() {
-                torrent_entry.remove_inactive_peers(self.policy.max_peer_timeout);
-            }
+            self.torrents.remove_inactive_peers(self.policy.max_peer_timeout).await;
         }
     }
 
@@ -1755,7 +1681,7 @@ mod tests {
             use aquatic_udp_protocol::AnnounceEvent;
 
             use crate::core::tests::the_tracker::{sample_info_hash, sample_peer, tracker_persisting_torrents_in_database};
-            use crate::core::torrent::repository_asyn::RepositoryAsync;
+            use crate::core::torrent::Repository;
 
             #[tokio::test]
             async fn it_should_persist_the_number_of_completed_peers_for_all_torrents_into_the_database() {
@@ -1774,14 +1700,15 @@ mod tests {
                 assert_eq!(swarm_stats.downloaded, 1);
 
                 // Remove the newly updated torrent from memory
-                tracker.torrents.get_torrents_mut().await.remove(&info_hash);
+                tracker.torrents.remove(&info_hash).await;
 
                 tracker.load_torrents_from_database().await.unwrap();
 
-                let torrents = tracker.torrents.get_torrents().await;
-                assert!(torrents.contains_key(&info_hash));
-
-                let torrent_entry = torrents.get(&info_hash).unwrap();
+                let torrent_entry = tracker
+                    .torrents
+                    .get(&info_hash)
+                    .await
+                    .expect("it should be able to get entry");
 
                 // It persists the number of completed peers.
                 assert_eq!(torrent_entry.completed, 1);