From 12f54e703e78af677fbd4456c359f580e83b1c44 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Mon, 8 Apr 2024 14:07:15 +0100 Subject: [PATCH] test: add tests for new torrent repository using SkipMap --- .../src/repository/skip_map_mutex_std.rs | 2 +- .../torrent-repository/tests/common/repo.rs | 165 +++++++++++------- .../tests/repository/mod.rs | 108 ++++++++++-- 3 files changed, 193 insertions(+), 82 deletions(-) diff --git a/packages/torrent-repository/src/repository/skip_map_mutex_std.rs b/packages/torrent-repository/src/repository/skip_map_mutex_std.rs index aa1f4382..0c0127b1 100644 --- a/packages/torrent-repository/src/repository/skip_map_mutex_std.rs +++ b/packages/torrent-repository/src/repository/skip_map_mutex_std.rs @@ -15,7 +15,7 @@ use crate::{EntryMutexStd, EntrySingle}; #[derive(Default, Debug)] pub struct CrossbeamSkipList { - torrents: SkipMap, + pub torrents: SkipMap, } impl Repository for CrossbeamSkipList diff --git a/packages/torrent-repository/tests/common/repo.rs b/packages/torrent-repository/tests/common/repo.rs index 3a4b53d2..5a86aa3c 100644 --- a/packages/torrent-repository/tests/common/repo.rs +++ b/packages/torrent-repository/tests/common/repo.rs @@ -7,49 +7,54 @@ use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent use torrust_tracker_torrent_repository::repository::{Repository as _, RepositoryAsync as _}; use torrust_tracker_torrent_repository::{ EntrySingle, TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio, TorrentsRwLockTokio, - TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, + TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd, }; #[derive(Debug)] pub(crate) enum Repo { - Std(TorrentsRwLockStd), - StdMutexStd(TorrentsRwLockStdMutexStd), - StdMutexTokio(TorrentsRwLockStdMutexTokio), - Tokio(TorrentsRwLockTokio), - TokioMutexStd(TorrentsRwLockTokioMutexStd), - TokioMutexTokio(TorrentsRwLockTokioMutexTokio), + RwLockStd(TorrentsRwLockStd), + RwLockStdMutexStd(TorrentsRwLockStdMutexStd), + RwLockStdMutexTokio(TorrentsRwLockStdMutexTokio), + RwLockTokio(TorrentsRwLockTokio), + RwLockTokioMutexStd(TorrentsRwLockTokioMutexStd), + RwLockTokioMutexTokio(TorrentsRwLockTokioMutexTokio), + SkipMapMutexStd(TorrentsSkipMapMutexStd), } impl Repo { pub(crate) async fn get(&self, key: &InfoHash) -> Option { match self { - Repo::Std(repo) => repo.get(key), - Repo::StdMutexStd(repo) => Some(repo.get(key)?.lock().unwrap().clone()), - Repo::StdMutexTokio(repo) => Some(repo.get(key).await?.lock().await.clone()), - Repo::Tokio(repo) => repo.get(key).await, - Repo::TokioMutexStd(repo) => Some(repo.get(key).await?.lock().unwrap().clone()), - Repo::TokioMutexTokio(repo) => Some(repo.get(key).await?.lock().await.clone()), + Repo::RwLockStd(repo) => repo.get(key), + Repo::RwLockStdMutexStd(repo) => Some(repo.get(key)?.lock().unwrap().clone()), + Repo::RwLockStdMutexTokio(repo) => Some(repo.get(key).await?.lock().await.clone()), + Repo::RwLockTokio(repo) => repo.get(key).await, + 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()), } } + pub(crate) async fn get_metrics(&self) -> TorrentsMetrics { match self { - Repo::Std(repo) => repo.get_metrics(), - Repo::StdMutexStd(repo) => repo.get_metrics(), - Repo::StdMutexTokio(repo) => repo.get_metrics().await, - Repo::Tokio(repo) => repo.get_metrics().await, - Repo::TokioMutexStd(repo) => repo.get_metrics().await, - Repo::TokioMutexTokio(repo) => repo.get_metrics().await, + Repo::RwLockStd(repo) => repo.get_metrics(), + Repo::RwLockStdMutexStd(repo) => repo.get_metrics(), + Repo::RwLockStdMutexTokio(repo) => repo.get_metrics().await, + Repo::RwLockTokio(repo) => repo.get_metrics().await, + Repo::RwLockTokioMutexStd(repo) => repo.get_metrics().await, + Repo::RwLockTokioMutexTokio(repo) => repo.get_metrics().await, + Repo::SkipMapMutexStd(repo) => repo.get_metrics(), } } + pub(crate) async fn get_paginated(&self, pagination: Option<&Pagination>) -> Vec<(InfoHash, EntrySingle)> { match self { - Repo::Std(repo) => repo.get_paginated(pagination), - Repo::StdMutexStd(repo) => repo + Repo::RwLockStd(repo) => repo.get_paginated(pagination), + Repo::RwLockStdMutexStd(repo) => repo .get_paginated(pagination) .iter() .map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone())) .collect(), - Repo::StdMutexTokio(repo) => { + Repo::RwLockStdMutexTokio(repo) => { let mut v: Vec<(InfoHash, EntrySingle)> = vec![]; for (i, t) in repo.get_paginated(pagination).await { @@ -57,14 +62,14 @@ impl Repo { } v } - Repo::Tokio(repo) => repo.get_paginated(pagination).await, - Repo::TokioMutexStd(repo) => repo + Repo::RwLockTokio(repo) => repo.get_paginated(pagination).await, + Repo::RwLockTokioMutexStd(repo) => repo .get_paginated(pagination) .await .iter() .map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone())) .collect(), - Repo::TokioMutexTokio(repo) => { + Repo::RwLockTokioMutexTokio(repo) => { let mut v: Vec<(InfoHash, EntrySingle)> = vec![]; for (i, t) in repo.get_paginated(pagination).await { @@ -72,76 +77,102 @@ impl Repo { } v } + Repo::SkipMapMutexStd(repo) => repo + .get_paginated(pagination) + .iter() + .map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone())) + .collect(), } } + pub(crate) async fn import_persistent(&self, persistent_torrents: &PersistentTorrents) { match self { - Repo::Std(repo) => repo.import_persistent(persistent_torrents), - Repo::StdMutexStd(repo) => repo.import_persistent(persistent_torrents), - Repo::StdMutexTokio(repo) => repo.import_persistent(persistent_torrents).await, - Repo::Tokio(repo) => repo.import_persistent(persistent_torrents).await, - Repo::TokioMutexStd(repo) => repo.import_persistent(persistent_torrents).await, - Repo::TokioMutexTokio(repo) => repo.import_persistent(persistent_torrents).await, + Repo::RwLockStd(repo) => repo.import_persistent(persistent_torrents), + Repo::RwLockStdMutexStd(repo) => repo.import_persistent(persistent_torrents), + Repo::RwLockStdMutexTokio(repo) => repo.import_persistent(persistent_torrents).await, + Repo::RwLockTokio(repo) => repo.import_persistent(persistent_torrents).await, + 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), } } + pub(crate) async fn remove(&self, key: &InfoHash) -> Option { match self { - Repo::Std(repo) => repo.remove(key), - Repo::StdMutexStd(repo) => Some(repo.remove(key)?.lock().unwrap().clone()), - Repo::StdMutexTokio(repo) => Some(repo.remove(key).await?.lock().await.clone()), - Repo::Tokio(repo) => repo.remove(key).await, - Repo::TokioMutexStd(repo) => Some(repo.remove(key).await?.lock().unwrap().clone()), - Repo::TokioMutexTokio(repo) => Some(repo.remove(key).await?.lock().await.clone()), + Repo::RwLockStd(repo) => repo.remove(key), + Repo::RwLockStdMutexStd(repo) => Some(repo.remove(key)?.lock().unwrap().clone()), + Repo::RwLockStdMutexTokio(repo) => Some(repo.remove(key).await?.lock().await.clone()), + Repo::RwLockTokio(repo) => repo.remove(key).await, + 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()), } } + pub(crate) async fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) { match self { - Repo::Std(repo) => repo.remove_inactive_peers(current_cutoff), - Repo::StdMutexStd(repo) => repo.remove_inactive_peers(current_cutoff), - Repo::StdMutexTokio(repo) => repo.remove_inactive_peers(current_cutoff).await, - Repo::Tokio(repo) => repo.remove_inactive_peers(current_cutoff).await, - Repo::TokioMutexStd(repo) => repo.remove_inactive_peers(current_cutoff).await, - Repo::TokioMutexTokio(repo) => repo.remove_inactive_peers(current_cutoff).await, + Repo::RwLockStd(repo) => repo.remove_inactive_peers(current_cutoff), + Repo::RwLockStdMutexStd(repo) => repo.remove_inactive_peers(current_cutoff), + Repo::RwLockStdMutexTokio(repo) => repo.remove_inactive_peers(current_cutoff).await, + Repo::RwLockTokio(repo) => repo.remove_inactive_peers(current_cutoff).await, + 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), } } + pub(crate) async fn remove_peerless_torrents(&self, policy: &TrackerPolicy) { match self { - Repo::Std(repo) => repo.remove_peerless_torrents(policy), - Repo::StdMutexStd(repo) => repo.remove_peerless_torrents(policy), - Repo::StdMutexTokio(repo) => repo.remove_peerless_torrents(policy).await, - Repo::Tokio(repo) => repo.remove_peerless_torrents(policy).await, - Repo::TokioMutexStd(repo) => repo.remove_peerless_torrents(policy).await, - Repo::TokioMutexTokio(repo) => repo.remove_peerless_torrents(policy).await, + Repo::RwLockStd(repo) => repo.remove_peerless_torrents(policy), + Repo::RwLockStdMutexStd(repo) => repo.remove_peerless_torrents(policy), + Repo::RwLockStdMutexTokio(repo) => repo.remove_peerless_torrents(policy).await, + Repo::RwLockTokio(repo) => repo.remove_peerless_torrents(policy).await, + 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), } } + pub(crate) async fn update_torrent_with_peer_and_get_stats( &self, info_hash: &InfoHash, peer: &peer::Peer, ) -> (bool, SwarmMetadata) { match self { - Repo::Std(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer), - Repo::StdMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer), - Repo::StdMutexTokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await, - Repo::Tokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await, - Repo::TokioMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await, - Repo::TokioMutexTokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await, + Repo::RwLockStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer), + Repo::RwLockStdMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer), + Repo::RwLockStdMutexTokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await, + Repo::RwLockTokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await, + 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), } } + pub(crate) async fn insert(&self, info_hash: &InfoHash, torrent: EntrySingle) -> Option { match self { - Repo::Std(repo) => repo.write().insert(*info_hash, torrent), - Repo::StdMutexStd(repo) => Some(repo.write().insert(*info_hash, torrent.into())?.lock().unwrap().clone()), - Repo::StdMutexTokio(repo) => { - let r = repo.write().insert(*info_hash, torrent.into()); - match r { - Some(t) => Some(t.lock().await.clone()), - None => None, - } + Repo::RwLockStd(repo) => { + repo.write().insert(*info_hash, torrent); } - Repo::Tokio(repo) => repo.write().await.insert(*info_hash, torrent), - Repo::TokioMutexStd(repo) => Some(repo.write().await.insert(*info_hash, torrent.into())?.lock().unwrap().clone()), - Repo::TokioMutexTokio(repo) => Some(repo.write().await.insert(*info_hash, torrent.into())?.lock().await.clone()), - } + Repo::RwLockStdMutexStd(repo) => { + repo.write().insert(*info_hash, torrent.into()); + } + Repo::RwLockStdMutexTokio(repo) => { + repo.write().insert(*info_hash, torrent.into()); + } + Repo::RwLockTokio(repo) => { + repo.write().await.insert(*info_hash, torrent); + } + Repo::RwLockTokioMutexStd(repo) => { + repo.write().await.insert(*info_hash, torrent.into()); + } + Repo::RwLockTokioMutexTokio(repo) => { + repo.write().await.insert(*info_hash, torrent.into()); + } + Repo::SkipMapMutexStd(repo) => { + repo.torrents.insert(*info_hash, torrent.into()); + } + }; + self.get(info_hash).await } } diff --git a/packages/torrent-repository/tests/repository/mod.rs b/packages/torrent-repository/tests/repository/mod.rs index 117f3c0a..ab964858 100644 --- a/packages/torrent-repository/tests/repository/mod.rs +++ b/packages/torrent-repository/tests/repository/mod.rs @@ -10,6 +10,7 @@ use torrust_tracker_primitives::{NumberOfBytes, PersistentTorrents}; use torrust_tracker_torrent_repository::entry::Entry as _; 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; use torrust_tracker_torrent_repository::EntrySingle; use crate::common::repo::Repo; @@ -17,30 +18,37 @@ use crate::common::torrent_peer_builder::{a_completed_peer, a_started_peer}; #[fixture] fn standard() -> Repo { - Repo::Std(RwLockStd::default()) + Repo::RwLockStd(RwLockStd::default()) } + #[fixture] fn standard_mutex() -> Repo { - Repo::StdMutexStd(RwLockStd::default()) + Repo::RwLockStdMutexStd(RwLockStd::default()) } #[fixture] fn standard_tokio() -> Repo { - Repo::StdMutexTokio(RwLockStd::default()) + Repo::RwLockStdMutexTokio(RwLockStd::default()) } #[fixture] fn tokio_std() -> Repo { - Repo::Tokio(RwLockTokio::default()) + Repo::RwLockTokio(RwLockTokio::default()) } + #[fixture] fn tokio_mutex() -> Repo { - Repo::TokioMutexStd(RwLockTokio::default()) + Repo::RwLockTokioMutexStd(RwLockTokio::default()) } #[fixture] fn tokio_tokio() -> Repo { - Repo::TokioMutexTokio(RwLockTokio::default()) + Repo::RwLockTokioMutexTokio(RwLockTokio::default()) +} + +#[fixture] +fn skip_list_std() -> Repo { + Repo::SkipMapMutexStd(CrossbeamSkipList::default()) } type Entries = Vec<(InfoHash, EntrySingle)>; @@ -224,7 +232,16 @@ fn policy_remove_persist() -> TrackerPolicy { #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_get_a_torrent_entry( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, ) { make(&repo, &entries).await; @@ -247,7 +264,16 @@ async fn it_should_get_a_torrent_entry( #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_get_paginated_entries_in_a_stable_or_sorted_order( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, many_out_of_order: Entries, ) { @@ -280,7 +306,16 @@ async fn it_should_get_paginated_entries_in_a_stable_or_sorted_order( #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_get_paginated( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, #[values(paginated_limit_zero(), paginated_limit_one(), paginated_limit_one_offset_one())] paginated: Pagination, ) { @@ -328,7 +363,16 @@ async fn it_should_get_paginated( #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_get_metrics( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, ) { use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics; @@ -360,7 +404,16 @@ async fn it_should_get_metrics( #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_import_persistent_torrents( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, #[values(persistent_empty(), persistent_single(), persistent_three())] persistent_torrents: PersistentTorrents, ) { @@ -389,7 +442,16 @@ async fn it_should_import_persistent_torrents( #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_remove_an_entry( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, ) { make(&repo, &entries).await; @@ -416,7 +478,16 @@ async fn it_should_remove_an_entry( #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_remove_inactive_peers( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, ) { use std::ops::Sub as _; @@ -489,7 +560,16 @@ async fn it_should_remove_inactive_peers( #[case::in_order(many_hashed_in_order())] #[tokio::test] async fn it_should_remove_peerless_torrents( - #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo, + #[values( + standard(), + standard_mutex(), + standard_tokio(), + tokio_std(), + tokio_mutex(), + tokio_tokio(), + skip_list_std() + )] + repo: Repo, #[case] entries: Entries, #[values(policy_none(), policy_persist(), policy_remove(), policy_remove_persist())] policy: TrackerPolicy, ) {