Skip to content

Commit 9258ac0

Browse files
josecelanojosecelano
committed
feat: new torrent repo implementation using parking_lot RwLock
1 parent 5750e2c commit 9258ac0

9 files changed

Lines changed: 238 additions & 28 deletions

File tree

‎packages/torrent-repository/benches/repository_benchmark.rs‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ mod helpers;
55
use criterion::{criterion_group, criterion_main, Criterion};
66
use torrust_tracker_torrent_repository::{
77
TorrentsDashMapMutexStd, TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio, TorrentsRwLockTokio,
8-
TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd,
8+
TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd, TorrentsSkipMapRwLockParkingLot,
99
};
1010

1111
use crate::helpers::{asyn, sync};
@@ -49,6 +49,10 @@ fn add_one_torrent(c: &mut Criterion) {
4949
b.iter_custom(sync::add_one_torrent::<TorrentsSkipMapMutexStd, _>);
5050
});
5151

52+
group.bench_function("SkipMapRwLockParkingLot", |b| {
53+
b.iter_custom(sync::add_one_torrent::<TorrentsSkipMapRwLockParkingLot, _>);
54+
});
55+
5256
group.bench_function("DashMapMutexStd", |b| {
5357
b.iter_custom(sync::add_one_torrent::<TorrentsDashMapMutexStd, _>);
5458
});
@@ -102,6 +106,11 @@ fn add_multiple_torrents_in_parallel(c: &mut Criterion) {
102106
.iter_custom(|iters| sync::add_multiple_torrents_in_parallel::<TorrentsSkipMapMutexStd, _>(&rt, iters, None));
103107
});
104108

109+
group.bench_function("SkipMapRwLockParkingLot", |b| {
110+
b.to_async(&rt)
111+
.iter_custom(|iters| sync::add_multiple_torrents_in_parallel::<TorrentsSkipMapRwLockParkingLot, _>(&rt, iters, None));
112+
});
113+
105114
group.bench_function("DashMapMutexStd", |b| {
106115
b.to_async(&rt)
107116
.iter_custom(|iters| sync::add_multiple_torrents_in_parallel::<TorrentsDashMapMutexStd, _>(&rt, iters, None));
@@ -156,6 +165,11 @@ fn update_one_torrent_in_parallel(c: &mut Criterion) {
156165
.iter_custom(|iters| sync::update_one_torrent_in_parallel::<TorrentsSkipMapMutexStd, _>(&rt, iters, None));
157166
});
158167

168+
group.bench_function("SkipMapRwLockParkingLot", |b| {
169+
b.to_async(&rt)
170+
.iter_custom(|iters| sync::update_one_torrent_in_parallel::<TorrentsSkipMapRwLockParkingLot, _>(&rt, iters, None));
171+
});
172+
159173
group.bench_function("DashMapMutexStd", |b| {
160174
b.to_async(&rt)
161175
.iter_custom(|iters| sync::update_one_torrent_in_parallel::<TorrentsDashMapMutexStd, _>(&rt, iters, None));
@@ -211,6 +225,12 @@ fn update_multiple_torrents_in_parallel(c: &mut Criterion) {
211225
.iter_custom(|iters| sync::update_multiple_torrents_in_parallel::<TorrentsSkipMapMutexStd, _>(&rt, iters, None));
212226
});
213227

228+
group.bench_function("SkipMapRwLockParkingLot", |b| {
229+
b.to_async(&rt).iter_custom(|iters| {
230+
sync::update_multiple_torrents_in_parallel::<TorrentsSkipMapRwLockParkingLot, _>(&rt, iters, None)
231+
});
232+
});
233+
214234
group.bench_function("DashMapMutexStd", |b| {
215235
b.to_async(&rt)
216236
.iter_custom(|iters| sync::update_multiple_torrents_in_parallel::<TorrentsDashMapMutexStd, _>(&rt, iters, None));

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
88

99
use self::peer_list::PeerList;
1010

11+
pub mod mutex_parking_lot;
1112
pub mod mutex_std;
1213
pub mod mutex_tokio;
1314
pub mod peer_list;

‎packages/torrent-repository/src/entry/mutex_parking_lot.rs‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
use std::net::SocketAddr;
2+
use std::sync::Arc;
3+
4+
use torrust_tracker_configuration::TrackerPolicy;
5+
use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
6+
use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
7+
8+
use super::{Entry, EntrySync};
9+
use crate::{EntryRwLockParkingLot, EntrySingle};
10+
11+
impl EntrySync for EntryRwLockParkingLot {
12+
fn get_swarm_metadata(&self) -> SwarmMetadata {
13+
self.read().get_swarm_metadata()
14+
}
15+
16+
fn is_good(&self, policy: &TrackerPolicy) -> bool {
17+
self.read().is_good(policy)
18+
}
19+
20+
fn peers_is_empty(&self) -> bool {
21+
self.read().peers_is_empty()
22+
}
23+
24+
fn get_peers_len(&self) -> usize {
25+
self.read().get_peers_len()
26+
}
27+
28+
fn get_peers(&self, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
29+
self.read().get_peers(limit)
30+
}
31+
32+
fn get_peers_for_client(&self, client: &SocketAddr, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
33+
self.read().get_peers_for_client(client, limit)
34+
}
35+
36+
fn upsert_peer(&self, peer: &peer::Peer) -> bool {
37+
self.write().upsert_peer(peer)
38+
}
39+
40+
fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) {
41+
self.write().remove_inactive_peers(current_cutoff);
42+
}
43+
}
44+
45+
impl From<EntrySingle> for EntryRwLockParkingLot {
46+
fn from(entry: EntrySingle) -> Self {
47+
Arc::new(parking_lot::RwLock::new(entry))
48+
}
49+
}

‎packages/torrent-repository/src/lib.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,9 +9,14 @@ use torrust_tracker_clock::clock;
99
pub mod entry;
1010
pub mod repository;
1111

12+
// Repo Entries
13+
1214
pub type EntrySingle = entry::Torrent;
1315
pub type EntryMutexStd = Arc<std::sync::Mutex<entry::Torrent>>;
1416
pub type EntryMutexTokio = Arc<tokio::sync::Mutex<entry::Torrent>>;
17+
pub type EntryRwLockParkingLot = Arc<parking_lot::RwLock<entry::Torrent>>;
18+
19+
// Repos
1520

1621
pub type TorrentsRwLockStd = RwLockStd<EntrySingle>;
1722
pub type TorrentsRwLockStdMutexStd = RwLockStd<EntryMutexStd>;
@@ -21,6 +26,8 @@ pub type TorrentsRwLockTokioMutexStd = RwLockTokio<EntryMutexStd>;
2126
pub type TorrentsRwLockTokioMutexTokio = RwLockTokio<EntryMutexTokio>;
2227

2328
pub type TorrentsSkipMapMutexStd = CrossbeamSkipList<EntryMutexStd>;
29+
pub type TorrentsSkipMapRwLockParkingLot = CrossbeamSkipList<EntryRwLockParkingLot>;
30+
2431
pub type TorrentsDashMapMutexStd = XacrimonDashMap<EntryMutexStd>;
2532

2633
/// This code needs to be copied into each crate.

‎packages/torrent-repository/src/repository/skip_map_mutex_std.rs‎

Lines changed: 92 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent
1111
use super::Repository;
1212
use crate::entry::peer_list::PeerList;
1313
use crate::entry::{Entry, EntrySync};
14-
use crate::{EntryMutexStd, EntrySingle};
14+
use crate::{EntryMutexStd, EntryRwLockParkingLot, EntrySingle};
1515

1616
#[derive(Default, Debug)]
1717
pub struct CrossbeamSkipList<T> {
@@ -108,3 +108,94 @@ where
108108
}
109109
}
110110
}
111+
112+
impl Repository<EntryRwLockParkingLot> for CrossbeamSkipList<EntryRwLockParkingLot>
113+
where
114+
EntryRwLockParkingLot: EntrySync,
115+
EntrySingle: Entry,
116+
{
117+
fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
118+
let entry = self.torrents.get_or_insert(*info_hash, Arc::default());
119+
entry.value().upsert_peer(peer);
120+
}
121+
122+
fn get_swarm_metadata(&self, info_hash: &InfoHash) -> Option<SwarmMetadata> {
123+
self.torrents.get(info_hash).map(|entry| entry.value().get_swarm_metadata())
124+
}
125+
126+
fn get(&self, key: &InfoHash) -> Option<EntryRwLockParkingLot> {
127+
let maybe_entry = self.torrents.get(key);
128+
maybe_entry.map(|entry| entry.value().clone())
129+
}
130+
131+
fn get_metrics(&self) -> TorrentsMetrics {
132+
let mut metrics = TorrentsMetrics::default();
133+
134+
for entry in &self.torrents {
135+
let stats = entry.value().read().get_swarm_metadata();
136+
metrics.complete += u64::from(stats.complete);
137+
metrics.downloaded += u64::from(stats.downloaded);
138+
metrics.incomplete += u64::from(stats.incomplete);
139+
metrics.torrents += 1;
140+
}
141+
142+
metrics
143+
}
144+
145+
fn get_paginated(&self, pagination: Option<&Pagination>) -> Vec<(InfoHash, EntryRwLockParkingLot)> {
146+
match pagination {
147+
Some(pagination) => self
148+
.torrents
149+
.iter()
150+
.skip(pagination.offset as usize)
151+
.take(pagination.limit as usize)
152+
.map(|entry| (*entry.key(), entry.value().clone()))
153+
.collect(),
154+
None => self
155+
.torrents
156+
.iter()
157+
.map(|entry| (*entry.key(), entry.value().clone()))
158+
.collect(),
159+
}
160+
}
161+
162+
fn import_persistent(&self, persistent_torrents: &PersistentTorrents) {
163+
for (info_hash, completed) in persistent_torrents {
164+
if self.torrents.contains_key(info_hash) {
165+
continue;
166+
}
167+
168+
let entry = EntryRwLockParkingLot::new(
169+
EntrySingle {
170+
swarm: PeerList::default(),
171+
downloaded: *completed,
172+
}
173+
.into(),
174+
);
175+
176+
// Since SkipMap is lock-free the torrent could have been inserted
177+
// after checking if it exists.
178+
self.torrents.get_or_insert(*info_hash, entry);
179+
}
180+
}
181+
182+
fn remove(&self, key: &InfoHash) -> Option<EntryRwLockParkingLot> {
183+
self.torrents.remove(key).map(|entry| entry.value().clone())
184+
}
185+
186+
fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) {
187+
for entry in &self.torrents {
188+
entry.value().remove_inactive_peers(current_cutoff);
189+
}
190+
}
191+
192+
fn remove_peerless_torrents(&self, policy: &TrackerPolicy) {
193+
for entry in &self.torrents {
194+
if entry.value().is_good(policy) {
195+
continue;
196+
}
197+
198+
entry.remove();
199+
}
200+
}
201+
}

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ use torrust_tracker_torrent_repository::repository::{Repository as _, Repository
88
use torrust_tracker_torrent_repository::{
99
EntrySingle, TorrentsDashMapMutexStd, TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio,
1010
TorrentsRwLockTokio, TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio, TorrentsSkipMapMutexStd,
11+
TorrentsSkipMapRwLockParkingLot,
1112
};
1213

1314
#[derive(Debug)]
@@ -19,6 +20,7 @@ pub(crate) enum Repo {
1920
RwLockTokioMutexStd(TorrentsRwLockTokioMutexStd),
2021
RwLockTokioMutexTokio(TorrentsRwLockTokioMutexTokio),
2122
SkipMapMutexStd(TorrentsSkipMapMutexStd),
23+
SkipMapRwLockParkingLot(TorrentsSkipMapRwLockParkingLot),
2224
DashMapMutexStd(TorrentsDashMapMutexStd),
2325
}
2426

@@ -32,6 +34,7 @@ impl Repo {
3234
Repo::RwLockTokioMutexStd(repo) => repo.upsert_peer(info_hash, peer).await,
3335
Repo::RwLockTokioMutexTokio(repo) => repo.upsert_peer(info_hash, peer).await,
3436
Repo::SkipMapMutexStd(repo) => repo.upsert_peer(info_hash, peer),
37+
Repo::SkipMapRwLockParkingLot(repo) => repo.upsert_peer(info_hash, peer),
3538
Repo::DashMapMutexStd(repo) => repo.upsert_peer(info_hash, peer),
3639
}
3740
}
@@ -45,6 +48,7 @@ impl Repo {
4548
Repo::RwLockTokioMutexStd(repo) => repo.get_swarm_metadata(info_hash).await,
4649
Repo::RwLockTokioMutexTokio(repo) => repo.get_swarm_metadata(info_hash).await,
4750
Repo::SkipMapMutexStd(repo) => repo.get_swarm_metadata(info_hash),
51+
Repo::SkipMapRwLockParkingLot(repo) => repo.get_swarm_metadata(info_hash),
4852
Repo::DashMapMutexStd(repo) => repo.get_swarm_metadata(info_hash),
4953
}
5054
}
@@ -58,6 +62,7 @@ impl Repo {
5862
Repo::RwLockTokioMutexStd(repo) => Some(repo.get(key).await?.lock().unwrap().clone()),
5963
Repo::RwLockTokioMutexTokio(repo) => Some(repo.get(key).await?.lock().await.clone()),
6064
Repo::SkipMapMutexStd(repo) => Some(repo.get(key)?.lock().unwrap().clone()),
65+
Repo::SkipMapRwLockParkingLot(repo) => Some(repo.get(key)?.read().clone()),
6166
Repo::DashMapMutexStd(repo) => Some(repo.get(key)?.lock().unwrap().clone()),
6267
}
6368
}
@@ -71,6 +76,7 @@ impl Repo {
7176
Repo::RwLockTokioMutexStd(repo) => repo.get_metrics().await,
7277
Repo::RwLockTokioMutexTokio(repo) => repo.get_metrics().await,
7378
Repo::SkipMapMutexStd(repo) => repo.get_metrics(),
79+
Repo::SkipMapRwLockParkingLot(repo) => repo.get_metrics(),
7480
Repo::DashMapMutexStd(repo) => repo.get_metrics(),
7581
}
7682
}
@@ -111,6 +117,11 @@ impl Repo {
111117
.iter()
112118
.map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone()))
113119
.collect(),
120+
Repo::SkipMapRwLockParkingLot(repo) => repo
121+
.get_paginated(pagination)
122+
.iter()
123+
.map(|(i, t)| (*i, t.read().clone()))
124+
.collect(),
114125
Repo::DashMapMutexStd(repo) => repo
115126
.get_paginated(pagination)
116127
.iter()
@@ -128,6 +139,7 @@ impl Repo {
128139
Repo::RwLockTokioMutexStd(repo) => repo.import_persistent(persistent_torrents).await,
129140
Repo::RwLockTokioMutexTokio(repo) => repo.import_persistent(persistent_torrents).await,
130141
Repo::SkipMapMutexStd(repo) => repo.import_persistent(persistent_torrents),
142+
Repo::SkipMapRwLockParkingLot(repo) => repo.import_persistent(persistent_torrents),
131143
Repo::DashMapMutexStd(repo) => repo.import_persistent(persistent_torrents),
132144
}
133145
}
@@ -141,6 +153,7 @@ impl Repo {
141153
Repo::RwLockTokioMutexStd(repo) => Some(repo.remove(key).await?.lock().unwrap().clone()),
142154
Repo::RwLockTokioMutexTokio(repo) => Some(repo.remove(key).await?.lock().await.clone()),
143155
Repo::SkipMapMutexStd(repo) => Some(repo.remove(key)?.lock().unwrap().clone()),
156+
Repo::SkipMapRwLockParkingLot(repo) => Some(repo.remove(key)?.write().clone()),
144157
Repo::DashMapMutexStd(repo) => Some(repo.remove(key)?.lock().unwrap().clone()),
145158
}
146159
}
@@ -154,6 +167,7 @@ impl Repo {
154167
Repo::RwLockTokioMutexStd(repo) => repo.remove_inactive_peers(current_cutoff).await,
155168
Repo::RwLockTokioMutexTokio(repo) => repo.remove_inactive_peers(current_cutoff).await,
156169
Repo::SkipMapMutexStd(repo) => repo.remove_inactive_peers(current_cutoff),
170+
Repo::SkipMapRwLockParkingLot(repo) => repo.remove_inactive_peers(current_cutoff),
157171
Repo::DashMapMutexStd(repo) => repo.remove_inactive_peers(current_cutoff),
158172
}
159173
}
@@ -167,6 +181,7 @@ impl Repo {
167181
Repo::RwLockTokioMutexStd(repo) => repo.remove_peerless_torrents(policy).await,
168182
Repo::RwLockTokioMutexTokio(repo) => repo.remove_peerless_torrents(policy).await,
169183
Repo::SkipMapMutexStd(repo) => repo.remove_peerless_torrents(policy),
184+
Repo::SkipMapRwLockParkingLot(repo) => repo.remove_peerless_torrents(policy),
170185
Repo::DashMapMutexStd(repo) => repo.remove_peerless_torrents(policy),
171186
}
172187
}
@@ -194,6 +209,9 @@ impl Repo {
194209
Repo::SkipMapMutexStd(repo) => {
195210
repo.torrents.insert(*info_hash, torrent.into());
196211
}
212+
Repo::SkipMapRwLockParkingLot(repo) => {
213+
repo.torrents.insert(*info_hash, torrent.into());
214+
}
197215
Repo::DashMapMutexStd(repo) => {
198216
repo.torrents.insert(*info_hash, torrent.into());
199217
}

0 commit comments

Comments
 (0)