feat: [torrust#565] new torrent repository implementation usind DashMap

It's not enabled as the deafult repository becuase DashMap does not
return the items in order. Some tests fail:

``` output
failures:

---- core::services::torrent::tests::searching_for_torrents::should_return_torrents_ordered_by_info_hash stdout ----
thread 'core::services::torrent::tests::searching_for_torrents::should_return_torrents_ordered_by_info_hash' panicked at src/core/services/torrent.rs:303:13:
assertion `left == right` failed
  left: [BasicInfo { info_hash: InfoHash([158, 2, 23, 208, 250, 113, 200, 115, 50, 205, 139, 249, 219, 234, 188, 178, 194, 207, 60, 77]), seeders: 1, completed: 0, leechers: 0 }, BasicInfo { info_hash: InfoHash([3, 132, 5, 72, 100, 58, 242, 167, 182, 58, 159, 92, 188, 163, 72, 188, 113, 80, 202, 58]), seeders: 1, completed: 0, leechers: 0 }]
 right: [BasicInfo { info_hash: InfoHash([3, 132, 5, 72, 100, 58, 242, 167, 182, 58, 159, 92, 188, 163, 72, 188, 113, 80, 202, 58]), seeders: 1, completed: 0, leechers: 0 }, BasicInfo { info_hash: InfoHash([158, 2, 23, 208, 250, 113, 200, 115, 50, 205, 139, 249, 219, 234, 188, 178, 194, 207, 60, 77]), seeders: 1, completed: 0, leechers: 0 }]
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    core::services::torrent::tests::searching_for_torrents::should_return_torrents_ordered_by_info_hash

test result: FAILED. 212 passed; 1 failed; 0 ignored; 0 measured; 0 filtered out; finished in 1.18s

error: test failed, to rerun pass `--lib`
```

On the other hand, to use it, a new data strcuture should be added to the repo:

An Index with sorted torrents by Infohash

The API uses pagination returning torrents in alphabetically order by InfoHash. Adding such an Index would probably
decrease the performace of this repository implementation. And it's
performace looks similar to the current SkipMap implementation.

SkipMap performace with Aquatic UDP load test:

```
Requests out: 396788.68/second
Responses in: 357105.27/second
  - Connect responses:  176662.91
  - Announce responses: 176863.44
  - Scrape responses:   3578.91
  - Error responses:    0.00
Peers per announce response: 0.00
Announce responses per info hash:
  - p10: 1
  - p25: 1
  - p50: 1
  - p75: 1
  - p90: 2
  - p95: 3
  - p99: 105
  - p99.9: 287
  - p100: 351
```

DashMap performace with Aquatic UDP load test:

```
Requests out: 410658.38/second
Responses in: 365892.86/second
  - Connect responses:  181258.91
  - Announce responses: 181005.95
  - Scrape responses:   3628.00
  - Error responses:    0.00
Peers per announce response: 0.00
Announce responses per info hash:
  - p10: 1
  - p25: 1
  - p50: 1
  - p75: 1
  - p90: 2
  - p95: 3
  - p99: 104
  - p99.9: 295
  - p100: 363
```

Author: josecelano

packages/torrent-repository/benches/repository_benchmark.rs

@@ -4,8 +4,8 @@ mod helpers;
 
 use criterion::{criterion_group, criterion_main, Criterion};
 use torrust_tracker_torrent_repository::{
-    TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio, TorrentsRwLockTokio, TorrentsRwLockTokioMutexStd,
-    TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd,
+    TorrentsDashMapMutexStd, TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio, TorrentsRwLockTokio,
+    TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd,
 };
 
 use crate::helpers::{asyn, sync};
@@ -49,6 +49,10 @@ fn add_one_torrent(c: &mut Criterion) {
         b.iter_custom(sync::add_one_torrent::<TorrentsSkipMapMutexStd, _>);
     });
 
+    group.bench_function("DashMapMutexStd", |b| {
+        b.iter_custom(sync::add_one_torrent::<TorrentsDashMapMutexStd, _>);
+    });
+
     group.finish();
 }
 
@@ -98,6 +102,11 @@ fn add_multiple_torrents_in_parallel(c: &mut Criterion) {
             .iter_custom(|iters| sync::add_multiple_torrents_in_parallel::<TorrentsSkipMapMutexStd, _>(&rt, iters, None));
     });
 
+    group.bench_function("DashMapMutexStd", |b| {
+        b.to_async(&rt)
+            .iter_custom(|iters| sync::add_multiple_torrents_in_parallel::<TorrentsDashMapMutexStd, _>(&rt, iters, None));
+    });
+
     group.finish();
 }
 
@@ -147,6 +156,11 @@ fn update_one_torrent_in_parallel(c: &mut Criterion) {
             .iter_custom(|iters| sync::update_one_torrent_in_parallel::<TorrentsSkipMapMutexStd, _>(&rt, iters, None));
     });
 
+    group.bench_function("DashMapMutexStd", |b| {
+        b.to_async(&rt)
+            .iter_custom(|iters| sync::update_one_torrent_in_parallel::<TorrentsDashMapMutexStd, _>(&rt, iters, None));
+    });
+
     group.finish();
 }
 
@@ -197,6 +211,11 @@ fn update_multiple_torrents_in_parallel(c: &mut Criterion) {
             .iter_custom(|iters| sync::update_multiple_torrents_in_parallel::<TorrentsSkipMapMutexStd, _>(&rt, iters, None));
     });
 
+    group.bench_function("DashMapMutexStd", |b| {
+        b.to_async(&rt)
+            .iter_custom(|iters| sync::update_multiple_torrents_in_parallel::<TorrentsDashMapMutexStd, _>(&rt, iters, None));
+    });
+
     group.finish();
 }
 

packages/torrent-repository/src/lib.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use repository::dash_map_mutex_std::XacrimonDashMap;
 use repository::rw_lock_std::RwLockStd;
 use repository::rw_lock_tokio::RwLockTokio;
 use repository::skip_map_mutex_std::CrossbeamSkipList;
@@ -20,6 +21,7 @@ pub type TorrentsRwLockTokioMutexStd = RwLockTokio<EntryMutexStd>;
 pub type TorrentsRwLockTokioMutexTokio = RwLockTokio<EntryMutexTokio>;
 
 pub type TorrentsSkipMapMutexStd = CrossbeamSkipList<EntryMutexStd>;
+pub type TorrentsDashMapMutexStd = XacrimonDashMap<EntryMutexStd>;
 
 /// This code needs to be copied into each crate.
 /// Working version, for production.

packages/torrent-repository/src/repository/dash_map_mutex_std.rs

@@ -0,0 +1,106 @@
+use std::collections::BTreeMap;
+use std::sync::Arc;
+
+use dashmap::DashMap;
+use torrust_tracker_configuration::TrackerPolicy;
+use torrust_tracker_primitives::info_hash::InfoHash;
+use torrust_tracker_primitives::pagination::Pagination;
+use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
+use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+
+use super::Repository;
+use crate::entry::{Entry, EntrySync};
+use crate::{EntryMutexStd, EntrySingle};
+
+#[derive(Default, Debug)]
+pub struct XacrimonDashMap<T> {
+    pub torrents: DashMap<InfoHash, T>,
+}
+
+impl Repository<EntryMutexStd> for XacrimonDashMap<EntryMutexStd>
+where
+    EntryMutexStd: EntrySync,
+    EntrySingle: Entry,
+{
+    fn update_torrent_with_peer_and_get_stats(&self, info_hash: &InfoHash, peer: &peer::Peer) -> (bool, SwarmMetadata) {
+        if let Some(entry) = self.torrents.get(info_hash) {
+            entry.insert_or_update_peer_and_get_stats(peer)
+        } else {
+            let _unused = self.torrents.insert(*info_hash, Arc::default());
+
+            match self.torrents.get(info_hash) {
+                Some(entry) => entry.insert_or_update_peer_and_get_stats(peer),
+                None => (false, SwarmMetadata::zeroed()),
+            }
+        }
+    }
+
+    fn get(&self, key: &InfoHash) -> Option<EntryMutexStd> {
+        let maybe_entry = self.torrents.get(key);
+        maybe_entry.map(|entry| entry.clone())
+    }
+
+    fn get_metrics(&self) -> TorrentsMetrics {
+        let mut metrics = TorrentsMetrics::default();
+
+        for entry in &self.torrents {
+            let stats = entry.value().lock().expect("it should get a lock").get_stats();
+            metrics.complete += u64::from(stats.complete);
+            metrics.downloaded += u64::from(stats.downloaded);
+            metrics.incomplete += u64::from(stats.incomplete);
+            metrics.torrents += 1;
+        }
+
+        metrics
+    }
+
+    fn get_paginated(&self, pagination: Option<&Pagination>) -> Vec<(InfoHash, EntryMutexStd)> {
+        match pagination {
+            Some(pagination) => self
+                .torrents
+                .iter()
+                .skip(pagination.offset as usize)
+                .take(pagination.limit as usize)
+                .map(|entry| (*entry.key(), entry.value().clone()))
+                .collect(),
+            None => self
+                .torrents
+                .iter()
+                .map(|entry| (*entry.key(), entry.value().clone()))
+                .collect(),
+        }
+    }
+
+    fn import_persistent(&self, persistent_torrents: &PersistentTorrents) {
+        for (info_hash, completed) in persistent_torrents {
+            if self.torrents.contains_key(info_hash) {
+                continue;
+            }
+
+            let entry = EntryMutexStd::new(
+                EntrySingle {
+                    peers: BTreeMap::default(),
+                    downloaded: *completed,
+                }
+                .into(),
+            );
+
+            self.torrents.insert(*info_hash, entry);
+        }
+    }
+
+    fn remove(&self, key: &InfoHash) -> Option<EntryMutexStd> {
+        self.torrents.remove(key).map(|(_key, value)| value.clone())
+    }
+
+    fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) {
+        for entry in &self.torrents {
+            entry.value().remove_inactive_peers(current_cutoff);
+        }
+    }
+
+    fn remove_peerless_torrents(&self, policy: &TrackerPolicy) {
+        self.torrents.retain(|_, entry| entry.is_good(policy));
+    }
+}

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

@@ -5,6 +5,7 @@ use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
 use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
 
+pub mod dash_map_mutex_std;
 pub mod rw_lock_std;
 pub mod rw_lock_std_mutex_std;
 pub mod rw_lock_std_mutex_tokio;

packages/torrent-repository/tests/common/repo.rs

@@ -6,8 +6,8 @@ use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
 use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
 use torrust_tracker_torrent_repository::repository::{Repository as _, RepositoryAsync as _};
 use torrust_tracker_torrent_repository::{
-    EntrySingle, TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio, TorrentsRwLockTokio,
-    TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd,
+    EntrySingle, TorrentsDashMapMutexStd, TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio,
+    TorrentsRwLockTokio, TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd,
 };
 
 #[derive(Debug)]
@@ -19,6 +19,7 @@ pub(crate) enum Repo {
     RwLockTokioMutexStd(TorrentsRwLockTokioMutexStd),
     RwLockTokioMutexTokio(TorrentsRwLockTokioMutexTokio),
     SkipMapMutexStd(TorrentsSkipMapMutexStd),
+    DashMapMutexStd(TorrentsDashMapMutexStd),
 }
 
 impl Repo {
@@ -31,6 +32,7 @@ impl Repo {
             Repo::RwLockTokioMutexStd(repo) => Some(repo.get(key).await?.lock().unwrap().clone()),
             Repo::RwLockTokioMutexTokio(repo) => Some(repo.get(key).await?.lock().await.clone()),
             Repo::SkipMapMutexStd(repo) => Some(repo.get(key)?.lock().unwrap().clone()),
+            Repo::DashMapMutexStd(repo) => Some(repo.get(key)?.lock().unwrap().clone()),
         }
     }
 
@@ -43,6 +45,7 @@ impl Repo {
             Repo::RwLockTokioMutexStd(repo) => repo.get_metrics().await,
             Repo::RwLockTokioMutexTokio(repo) => repo.get_metrics().await,
             Repo::SkipMapMutexStd(repo) => repo.get_metrics(),
+            Repo::DashMapMutexStd(repo) => repo.get_metrics(),
         }
     }
 
@@ -82,6 +85,11 @@ impl Repo {
                 .iter()
                 .map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone()))
                 .collect(),
+            Repo::DashMapMutexStd(repo) => repo
+                .get_paginated(pagination)
+                .iter()
+                .map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone()))
+                .collect(),
         }
     }
 
@@ -94,6 +102,7 @@ impl Repo {
             Repo::RwLockTokioMutexStd(repo) => repo.import_persistent(persistent_torrents).await,
             Repo::RwLockTokioMutexTokio(repo) => repo.import_persistent(persistent_torrents).await,
             Repo::SkipMapMutexStd(repo) => repo.import_persistent(persistent_torrents),
+            Repo::DashMapMutexStd(repo) => repo.import_persistent(persistent_torrents),
         }
     }
 
@@ -106,6 +115,7 @@ impl Repo {
             Repo::RwLockTokioMutexStd(repo) => Some(repo.remove(key).await?.lock().unwrap().clone()),
             Repo::RwLockTokioMutexTokio(repo) => Some(repo.remove(key).await?.lock().await.clone()),
             Repo::SkipMapMutexStd(repo) => Some(repo.remove(key)?.lock().unwrap().clone()),
+            Repo::DashMapMutexStd(repo) => Some(repo.remove(key)?.lock().unwrap().clone()),
         }
     }
 
@@ -118,6 +128,7 @@ impl Repo {
             Repo::RwLockTokioMutexStd(repo) => repo.remove_inactive_peers(current_cutoff).await,
             Repo::RwLockTokioMutexTokio(repo) => repo.remove_inactive_peers(current_cutoff).await,
             Repo::SkipMapMutexStd(repo) => repo.remove_inactive_peers(current_cutoff),
+            Repo::DashMapMutexStd(repo) => repo.remove_inactive_peers(current_cutoff),
         }
     }
 
@@ -130,6 +141,7 @@ impl Repo {
             Repo::RwLockTokioMutexStd(repo) => repo.remove_peerless_torrents(policy).await,
             Repo::RwLockTokioMutexTokio(repo) => repo.remove_peerless_torrents(policy).await,
             Repo::SkipMapMutexStd(repo) => repo.remove_peerless_torrents(policy),
+            Repo::DashMapMutexStd(repo) => repo.remove_peerless_torrents(policy),
         }
     }
 
@@ -146,6 +158,7 @@ impl Repo {
             Repo::RwLockTokioMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await,
             Repo::RwLockTokioMutexTokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await,
             Repo::SkipMapMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer),
+            Repo::DashMapMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer),
         }
     }
 
@@ -172,6 +185,9 @@ impl Repo {
             Repo::SkipMapMutexStd(repo) => {
                 repo.torrents.insert(*info_hash, torrent.into());
             }
+            Repo::DashMapMutexStd(repo) => {
+                repo.torrents.insert(*info_hash, torrent.into());
+            }
         };
         self.get(info_hash).await
     }

packages/torrent-repository/tests/repository/mod.rs

@@ -8,6 +8,7 @@ use torrust_tracker_primitives::info_hash::InfoHash;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::{NumberOfBytes, PersistentTorrents};
 use torrust_tracker_torrent_repository::entry::Entry as _;
+use torrust_tracker_torrent_repository::repository::dash_map_mutex_std::XacrimonDashMap;
 use torrust_tracker_torrent_repository::repository::rw_lock_std::RwLockStd;
 use torrust_tracker_torrent_repository::repository::rw_lock_tokio::RwLockTokio;
 use torrust_tracker_torrent_repository::repository::skip_map_mutex_std::CrossbeamSkipList;
@@ -51,6 +52,11 @@ fn skip_list_std() -> Repo {
     Repo::SkipMapMutexStd(CrossbeamSkipList::default())
 }
 
+#[fixture]
+fn dash_map_std() -> Repo {
+    Repo::DashMapMutexStd(XacrimonDashMap::default())
+}
+
 type Entries = Vec<(InfoHash, EntrySingle)>;
 
 #[fixture]
@@ -239,7 +245,8 @@ async fn it_should_get_a_torrent_entry(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,
@@ -271,7 +278,8 @@ async fn it_should_get_paginated_entries_in_a_stable_or_sorted_order(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,
@@ -313,7 +321,8 @@ async fn it_should_get_paginated(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,
@@ -370,7 +379,8 @@ async fn it_should_get_metrics(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,
@@ -411,7 +421,8 @@ async fn it_should_import_persistent_torrents(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,
@@ -449,7 +460,8 @@ async fn it_should_remove_an_entry(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,
@@ -485,7 +497,8 @@ async fn it_should_remove_inactive_peers(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,
@@ -567,7 +580,8 @@ async fn it_should_remove_peerless_torrents(
         tokio_std(),
         tokio_mutex(),
         tokio_tokio(),
-        skip_list_std()
+        skip_list_std(),
+        dash_map_std()
     )]
     repo: Repo,
     #[case] entries: Entries,