refactor: performance optimization in tracker announce

The tracker announce method uses two locks:

1. To write the new peer in the torrent entry peer list (write lock).
2. To read the peers list from the torrent entry (read lock).

I was wondering if the repo would be faster removing the second lock
since we can get the peer list in the first lock.

This are the benchmarking results:

BEFORE:

tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, Entry>>
add_one_torrent: Avg/AdjAvg: (57ns, 59ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (15.769541ms, 16.69948ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (20.6541ms, 20.010245ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (18.326633ms, 18.326633ms)

std::sync::RwLock<std::collections::BTreeMap<InfoHash, Entry>>
add_one_torrent: Avg/AdjAvg: (38ns, 39ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (7.154301ms, 7.296907ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (8.370377ms, 8.370377ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (6.902327ms, 6.902327ms)

std::sync::RwLock<std::collections::BTreeMap<InfoHash, Arc<std::sync::Mutex<Entry>>>>
add_one_torrent: Avg/AdjAvg: (44ns, 40ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (6.023789ms, 6.023789ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (12.306719ms, 12.306719ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (8.429203ms, 8.429203ms)

tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, Arc<std::sync::Mutex<Entry>>>>
add_one_torrent: Avg/AdjAvg: (71ns, 69ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (7.022376ms, 7.022376ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (28.125712ms, 27.811162ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (8.908545ms, 9.036288ms)

tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, Arc<tokio::sync::Mutex<Entry>>>>
add_one_torrent: Avg/AdjAvg: (85ns, 81ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (15.285006ms, 15.555171ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (29.794849ms, 30.254531ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (9.302051ms, 9.302051ms)

AFTER:

tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, Entry>>
add_one_torrent: Avg/AdjAvg: (64ns, 64ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (15.652934ms, 15.54617ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (23.834278ms, 22.935421ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (17.973855ms, 18.28944ms)

std::sync::RwLock<std::collections::BTreeMap<InfoHash, Entry>>
add_one_torrent: Avg/AdjAvg: (43ns, 39ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (6.362147ms, 6.362147ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (8.869102ms, 8.869102ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (6.589264ms, 6.510441ms)

std::sync::RwLock<std::collections::BTreeMap<InfoHash, Arc<std::sync::Mutex<Entry>>>>
add_one_torrent: Avg/AdjAvg: (48ns, 49ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (6.194211ms, 6.194211ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (12.50352ms, 12.50352ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (8.411279ms, 8.411279ms)

tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, Arc<std::sync::Mutex<Entry>>>>
add_one_torrent: Avg/AdjAvg: (79ns, 79ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (6.966462ms, 6.966462ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (27.584361ms, 27.411202ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (6.965519ms, 7.474346ms)

tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, Arc<tokio::sync::Mutex<Entry>>>>
add_one_torrent: Avg/AdjAvg: (97ns, 97ns)
update_one_torrent_in_parallel: Avg/AdjAvg: (16.046499ms, 16.430313ms)
add_multiple_torrents_in_parallel: Avg/AdjAvg: (28.42327ms, 27.876115ms)
update_multiple_torrents_in_parallel: Avg/AdjAvg: (6.576256ms, 6.326548ms)

Only in a couple of cases is a litlte bit faster. But I think the code
is simpler.

Author: josecelano

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

@@ -19,9 +19,7 @@ pub async fn async_add_one_torrent<T: TRepositoryAsync + Send + Sync + 'static>(
 
         let start_time = std::time::Instant::now();
 
-        torrent_repository
-            .update_torrent_with_peer_and_get_stats(&info_hash, &DEFAULT_PEER)
-            .await;
+        torrent_repository.announce(&info_hash, &DEFAULT_PEER).await;
 
         let result = start_time.elapsed();
 
@@ -45,19 +43,15 @@ pub async fn async_update_one_torrent_in_parallel<T: TRepositoryAsync + Send + S
         let handles = FuturesUnordered::new();
 
         // Add the torrent/peer to the torrent repository
-        torrent_repository
-            .update_torrent_with_peer_and_get_stats(info_hash, &DEFAULT_PEER)
-            .await;
+        torrent_repository.announce(info_hash, &DEFAULT_PEER).await;
 
         let start_time = std::time::Instant::now();
 
         for _ in 0..10_000 {
             let torrent_repository_clone = torrent_repository.clone();
 
             let handle = runtime.spawn(async move {
-                torrent_repository_clone
-                    .update_torrent_with_peer_and_get_stats(info_hash, &DEFAULT_PEER)
-                    .await;
+                torrent_repository_clone.announce(info_hash, &DEFAULT_PEER).await;
 
                 if let Some(sleep_time) = args.sleep {
                     let start_time = std::time::Instant::now();
@@ -99,9 +93,7 @@ pub async fn async_add_multiple_torrents_in_parallel<T: TRepositoryAsync + Send
             let torrent_repository_clone = torrent_repository.clone();
 
             let handle = runtime.spawn(async move {
-                torrent_repository_clone
-                    .update_torrent_with_peer_and_get_stats(&info_hash, &DEFAULT_PEER)
-                    .await;
+                torrent_repository_clone.announce(&info_hash, &DEFAULT_PEER).await;
 
                 if let Some(sleep_time) = args.sleep {
                     let start_time = std::time::Instant::now();
@@ -139,9 +131,7 @@ pub async fn async_update_multiple_torrents_in_parallel<T: TRepositoryAsync + Se
 
         // Add the torrents/peers to the torrent repository
         for info_hash in &info_hashes {
-            torrent_repository
-                .update_torrent_with_peer_and_get_stats(info_hash, &DEFAULT_PEER)
-                .await;
+            torrent_repository.announce(info_hash, &DEFAULT_PEER).await;
         }
 
         let start_time = std::time::Instant::now();
@@ -150,9 +140,7 @@ pub async fn async_update_multiple_torrents_in_parallel<T: TRepositoryAsync + Se
             let torrent_repository_clone = torrent_repository.clone();
 
             let handle = runtime.spawn(async move {
-                torrent_repository_clone
-                    .update_torrent_with_peer_and_get_stats(&info_hash, &DEFAULT_PEER)
-                    .await;
+                torrent_repository_clone.announce(&info_hash, &DEFAULT_PEER).await;
 
                 if let Some(sleep_time) = args.sleep {
                     let start_time = std::time::Instant::now();

src/core/mod.rs

@@ -651,10 +651,7 @@ impl Tracker {
         peer.change_ip(&assign_ip_address_to_peer(remote_client_ip, self.external_ip));
         debug!("After: {peer:?}");
 
-        // we should update the torrent and get the stats before we get the peer list.
-        let stats = self.update_torrent_with_peer_and_get_stats(info_hash, peer).await;
-
-        let peers = self.get_torrent_peers_for_peer(info_hash, peer).await;
+        let (stats, peers) = self.inner_announce(info_hash, peer).await;
 
         AnnounceData {
             peers,
@@ -722,19 +719,6 @@ impl Tracker {
         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) {
-            None => vec![],
-            Some(entry) => entry
-                .get_peers_for_peer(peer, TORRENT_PEERS_LIMIT)
-                .into_iter()
-                .copied()
-                .collect(),
-        }
-    }
-
     /// # Context: Tracker
     ///
     /// Get all torrent peers for a given torrent
@@ -752,11 +736,8 @@ impl Tracker {
     /// needed for a `announce` request response.
     ///
     /// # Context: Tracker
-    pub async fn update_torrent_with_peer_and_get_stats(&self, info_hash: &InfoHash, peer: &peer::Peer) -> torrent::SwarmStats {
-        // 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;
+    pub async fn inner_announce(&self, info_hash: &InfoHash, peer: &peer::Peer) -> (torrent::SwarmStats, Vec<Peer>) {
+        let (stats, stats_updated, peers) = self.torrents.announce(info_hash, peer).await;
 
         if self.policy.persistent_torrent_completed_stat && stats_updated {
             let completed = stats.downloaded;
@@ -765,7 +746,7 @@ impl Tracker {
             drop(self.database.save_persistent_torrent(&info_hash, completed).await);
         }
 
-        stats
+        (stats, peers)
     }
 
     /// It calculates and returns the general `Tracker`
@@ -1228,7 +1209,7 @@ mod tests {
             let info_hash = sample_info_hash();
             let peer = sample_peer();
 
-            tracker.update_torrent_with_peer_and_get_stats(&info_hash, &peer).await;
+            tracker.inner_announce(&info_hash, &peer).await;
 
             let peers = tracker.get_torrent_peers(&info_hash).await;
 
@@ -1242,9 +1223,7 @@ mod tests {
             let info_hash = sample_info_hash();
             let peer = sample_peer();
 
-            tracker.update_torrent_with_peer_and_get_stats(&info_hash, &peer).await;
-
-            let peers = tracker.get_torrent_peers_for_peer(&info_hash, &peer).await;
+            let (_stats, peers) = tracker.inner_announce(&info_hash, &peer).await;
 
             assert_eq!(peers, vec![]);
         }
@@ -1253,9 +1232,7 @@ mod tests {
         async fn it_should_return_the_torrent_metrics() {
             let tracker = public_tracker();
 
-            tracker
-                .update_torrent_with_peer_and_get_stats(&sample_info_hash(), &leecher())
-                .await;
+            tracker.inner_announce(&sample_info_hash(), &leecher()).await;
 
             let torrent_metrics = tracker.get_torrents_metrics().await;
 
@@ -1764,11 +1741,11 @@ mod tests {
                 let mut peer = sample_peer();
 
                 peer.event = AnnounceEvent::Started;
-                let swarm_stats = tracker.update_torrent_with_peer_and_get_stats(&info_hash, &peer).await;
+                let (swarm_stats, _peers) = tracker.inner_announce(&info_hash, &peer).await;
                 assert_eq!(swarm_stats.downloaded, 0);
 
                 peer.event = AnnounceEvent::Completed;
-                let swarm_stats = tracker.update_torrent_with_peer_and_get_stats(&info_hash, &peer).await;
+                let (swarm_stats, _peers) = tracker.inner_announce(&info_hash, &peer).await;
                 assert_eq!(swarm_stats.downloaded, 1);
 
                 // Remove the newly updated torrent from memory

src/core/services/torrent.rs

@@ -213,9 +213,7 @@ mod tests {
 
             let hash = "9e0217d0fa71c87332cd8bf9dbeabcb2c2cf3c4d".to_owned();
             let info_hash = InfoHash::from_str(&hash).unwrap();
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash, &sample_peer())
-                .await;
+            tracker.inner_announce(&info_hash, &sample_peer()).await;
 
             let torrent_info = get_torrent_info(tracker.clone(), &info_hash).await.unwrap();
 
@@ -265,9 +263,7 @@ mod tests {
             let hash = "9e0217d0fa71c87332cd8bf9dbeabcb2c2cf3c4d".to_owned();
             let info_hash = InfoHash::from_str(&hash).unwrap();
 
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash, &sample_peer())
-                .await;
+            tracker.inner_announce(&info_hash, &sample_peer()).await;
 
             let torrents = get_torrents_page(tracker.clone(), &Pagination::default()).await;
 
@@ -291,12 +287,8 @@ mod tests {
             let hash2 = "03840548643af2a7b63a9f5cbca348bc7150ca3a".to_owned();
             let info_hash2 = InfoHash::from_str(&hash2).unwrap();
 
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash1, &sample_peer())
-                .await;
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash2, &sample_peer())
-                .await;
+            tracker.inner_announce(&info_hash1, &sample_peer()).await;
+            tracker.inner_announce(&info_hash2, &sample_peer()).await;
 
             let offset = 0;
             let limit = 1;
@@ -315,12 +307,8 @@ mod tests {
             let hash2 = "03840548643af2a7b63a9f5cbca348bc7150ca3a".to_owned();
             let info_hash2 = InfoHash::from_str(&hash2).unwrap();
 
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash1, &sample_peer())
-                .await;
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash2, &sample_peer())
-                .await;
+            tracker.inner_announce(&info_hash1, &sample_peer()).await;
+            tracker.inner_announce(&info_hash2, &sample_peer()).await;
 
             let offset = 1;
             let limit = 4000;
@@ -345,15 +333,11 @@ mod tests {
 
             let hash1 = "9e0217d0fa71c87332cd8bf9dbeabcb2c2cf3c4d".to_owned();
             let info_hash1 = InfoHash::from_str(&hash1).unwrap();
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash1, &sample_peer())
-                .await;
+            tracker.inner_announce(&info_hash1, &sample_peer()).await;
 
             let hash2 = "03840548643af2a7b63a9f5cbca348bc7150ca3a".to_owned();
             let info_hash2 = InfoHash::from_str(&hash2).unwrap();
-            tracker
-                .update_torrent_with_peer_and_get_stats(&info_hash2, &sample_peer())
-                .await;
+            tracker.inner_announce(&info_hash2, &sample_peer()).await;
 
             let torrents = get_torrents_page(tracker.clone(), &Pagination::default()).await;