dev: more torrent/repository work

Author: da2ce7

src/core/torrent/mod.rs

@@ -31,7 +31,7 @@ pub mod repository;
 
 use derive_more::Constructor;
 
-pub type Torrents = TorrentsRwLockStd; // Currently Used
+pub type Torrents = TorrentsRwLockStdMutexStd; // Currently Used
 
 pub type TorrentsRwLockStd = repository::RwLockStd<entry::Single>;
 pub type TorrentsRwLockStdMutexStd = repository::RwLockStd<entry::MutexStd>;

src/core/torrent/repository/rw_lock_std_mutex_std.rs

@@ -1,7 +1,6 @@
 use std::collections::BTreeMap;
 use std::sync::Arc;
 
-use futures::executor::block_on;
 use futures::future::join_all;
 
 use super::{Repository, UpdateTorrentSync};
@@ -59,6 +58,7 @@ impl Repository<entry::MutexStd> for TorrentsRwLockStdMutexStd {
     async fn get_metrics(&self) -> TorrentsMetrics {
         let metrics: Arc<tokio::sync::Mutex<TorrentsMetrics>> = Arc::default();
 
+        // todo:: replace with a ring buffer
         let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
         for e in self.get_torrents().values() {
@@ -114,19 +114,20 @@ impl Repository<entry::MutexStd> for TorrentsRwLockStdMutexStd {
     }
 
     async fn remove_inactive_peers(&self, max_peer_timeout: u32) {
-        let db = self.get_torrents();
+        // todo:: replace with a ring buffer
+        let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
-        let futures = db.values().map(|e| {
+        for e in self.get_torrents().values() {
             let entry = e.clone();
-            tokio::spawn(async move {
+            handles.push(tokio::task::spawn(async move {
                 entry
                     .lock()
                     .expect("it should get lock for entry")
                     .remove_inactive_peers(max_peer_timeout);
-            })
-        });
+            }));
+        }
 
-        block_on(join_all(futures));
+        join_all(handles).await;
     }
 
     async fn remove_peerless_torrents(&self, policy: &crate::core::TrackerPolicy) {

src/core/torrent/repository/rw_lock_std_mutex_tokio.rs

@@ -1,7 +1,6 @@
 use std::collections::BTreeMap;
 use std::sync::Arc;
 
-use futures::executor::block_on;
 use futures::future::join_all;
 
 use super::{Repository, UpdateTorrentAsync};
@@ -68,6 +67,7 @@ impl Repository<entry::MutexTokio> for TorrentsRwLockStdMutexTokio {
     async fn get_metrics(&self) -> TorrentsMetrics {
         let metrics: Arc<tokio::sync::Mutex<TorrentsMetrics>> = Arc::default();
 
+        // todo:: replace with a ring buffer
         let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
         for e in self.get_torrents().values() {
@@ -114,14 +114,18 @@ impl Repository<entry::MutexTokio> for TorrentsRwLockStdMutexTokio {
     }
 
     async fn remove_inactive_peers(&self, max_peer_timeout: u32) {
-        let db = self.get_torrents();
+        // todo:: replace with a ring buffer
+
+        let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
-        let futures = db.values().map(|e| {
+        for e in self.get_torrents().values() {
             let entry = e.clone();
-            tokio::spawn(async move { entry.lock().await.remove_inactive_peers(max_peer_timeout) })
-        });
+            handles.push(tokio::task::spawn(async move {
+                entry.lock().await.remove_inactive_peers(max_peer_timeout);
+            }));
+        }
 
-        block_on(join_all(futures));
+        join_all(handles).await;
     }
 
     async fn remove_peerless_torrents(&self, policy: &crate::core::TrackerPolicy) {

src/core/torrent/repository/rw_lock_tokio_mutex_std.rs

@@ -69,6 +69,8 @@ impl Repository<entry::MutexStd> for TorrentsRwLockTokioMutexStd {
     async fn get_metrics(&self) -> TorrentsMetrics {
         let metrics: Arc<tokio::sync::Mutex<TorrentsMetrics>> = Arc::default();
 
+        // todo:: replace with a ring buffer
+
         let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
         for e in self.get_torrents().await.values() {
@@ -115,19 +117,20 @@ impl Repository<entry::MutexStd> for TorrentsRwLockTokioMutexStd {
     }
 
     async fn remove_inactive_peers(&self, max_peer_timeout: u32) {
-        let db = self.get_torrents().await;
+        // todo:: replace with a ring buffer
+        let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
-        let futures = db.values().map(|e| {
+        for e in self.get_torrents().await.values() {
             let entry = e.clone();
-            tokio::spawn(async move {
+            handles.push(tokio::task::spawn(async move {
                 entry
                     .lock()
                     .expect("it should get lock for entry")
                     .remove_inactive_peers(max_peer_timeout);
-            })
-        });
+            }));
+        }
 
-        join_all(futures).await;
+        join_all(handles).await;
     }
 
     async fn remove_peerless_torrents(&self, policy: &TrackerPolicy) {

src/core/torrent/repository/rw_lock_tokio_mutex_tokio.rs

@@ -71,6 +71,7 @@ impl Repository<entry::MutexTokio> for TorrentsRwLockTokioMutexTokio {
     async fn get_metrics(&self) -> TorrentsMetrics {
         let metrics: Arc<tokio::sync::Mutex<TorrentsMetrics>> = Arc::default();
 
+        // todo:: replace with a ring buffer
         let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
         for e in self.get_torrents().await.values() {
@@ -117,14 +118,17 @@ impl Repository<entry::MutexTokio> for TorrentsRwLockTokioMutexTokio {
     }
 
     async fn remove_inactive_peers(&self, max_peer_timeout: u32) {
-        let db = self.get_torrents().await;
+        // todo:: replace with a ring buffer
+        let mut handles = Vec::<tokio::task::JoinHandle<()>>::default();
 
-        let futures = db.values().map(|e| {
+        for e in self.get_torrents().await.values() {
             let entry = e.clone();
-            tokio::spawn(async move { entry.lock().await.remove_inactive_peers(max_peer_timeout) })
-        });
+            handles.push(tokio::task::spawn(async move {
+                entry.lock().await.remove_inactive_peers(max_peer_timeout);
+            }));
+        }
 
-        join_all(futures).await;
+        join_all(handles).await;
     }
 
     async fn remove_peerless_torrents(&self, policy: &TrackerPolicy) {