Stats on steals

crates/librqbit/src/torrent_state/live/mod.rs

@@ -1055,10 +1055,8 @@ impl PeerHandler {
             }
         };
 
-        // Send cancellations to old peer.
+        // Send cancellations to old peer and bump counters.
-        {
+        self.state.peers.on_steal(from_peer, self.addr, stolen_idx);
-            self.state.peers.send_cancellations(from_peer, stolen_idx);
-        }
 
         Some(stolen_idx)
     }

crates/librqbit/src/torrent_state/live/peer/stats/atomic.rs

@@ -20,6 +20,8 @@ pub(crate) struct PeerCountersAtomic {
     pub downloaded_and_checked_pieces: AtomicU32,
     pub downloaded_and_checked_bytes: AtomicU64,
     pub total_piece_download_ms: AtomicU64,
+    pub times_stolen_from_me: AtomicU32,
+    pub times_i_stole: AtomicU32,
 }
 
 impl PeerCountersAtomic {

crates/librqbit/src/torrent_state/live/peer/stats/snapshot.rs

@@ -15,6 +15,8 @@ pub struct PeerCounters {
     pub fetched_chunks: u32,
     pub downloaded_and_checked_pieces: u32,
     pub total_piece_download_ms: u64,
+    pub times_stolen_from_me: u32,
+    pub times_i_stole: u32,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -39,6 +41,8 @@ impl From<&super::atomic::PeerCountersAtomic> for PeerCounters {
                 .downloaded_and_checked_pieces
                 .load(Ordering::Relaxed),
             total_piece_download_ms: counters.total_piece_download_ms.load(Ordering::Relaxed),
+            times_i_stole: counters.times_i_stole.load(Ordering::Relaxed),
+            times_stolen_from_me: counters.times_stolen_from_me.load(Ordering::Relaxed),
         }
     }
 }

crates/librqbit/src/torrent_state/live/peers/mod.rs

@@ -113,7 +113,20 @@ impl PeerStates {
         Some(prev)
     }
 
-    pub(crate) fn send_cancellations(&self, from_peer: SocketAddr, stolen_idx: ValidPieceIndex) {
+    pub(crate) fn on_steal(
+        &self,
+        from_peer: SocketAddr,
+        to_peer: SocketAddr,
+        stolen_idx: ValidPieceIndex,
+    ) {
+        self.with_peer(to_peer, |p| {
+            atomic_inc(&p.stats.counters.times_i_stole);
+        });
+        self.with_peer(from_peer, |p| {
+            atomic_inc(&p.stats.counters.times_stolen_from_me);
+        });
+        self.stats.inc_steals();
+
         self.with_live_mut(from_peer, "send_cancellations", |live| {
             let to_remove = live
                 .inflight_requests

crates/librqbit/src/torrent_state/live/peers/stats/atomic.rs

@@ -15,6 +15,7 @@ pub(crate) struct AggregatePeerStatsAtomic {
     pub seen: AtomicU32,
     pub dead: AtomicU32,
     pub not_needed: AtomicU32,
+    pub steals: AtomicU32,
 }
 
 impl AggregatePeerStatsAtomic {
@@ -40,4 +41,8 @@ impl AggregatePeerStatsAtomic {
         self.dec(old);
         self.inc(new);
     }
+
+    pub fn inc_steals(&self) {
+        atomic_inc(&self.steals);
+    }
 }

crates/librqbit/src/torrent_state/live/peers/stats/snapshot.rs

@@ -12,6 +12,7 @@ pub struct AggregatePeerStats {
     pub seen: usize,
     pub dead: usize,
     pub not_needed: usize,
+    pub steals: usize,
 }
 
 impl<'a> From<&'a AggregatePeerStatsAtomic> for AggregatePeerStats {
@@ -24,6 +25,7 @@ impl<'a> From<&'a AggregatePeerStatsAtomic> for AggregatePeerStats {
             seen: s.seen.load(ordering) as usize,
             dead: s.dead.load(ordering) as usize,
             not_needed: s.not_needed.load(ordering) as usize,
+            steals: s.steals.load(ordering) as usize,
         }
     }
 }