leyoda/rqbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

When stealing pieces, only steal from slower peers

Browse source
This commit is contained in:
Igor Katson 2023-12-25 09:08:16 -05:00
parent 1e525eb91e
commit 5d4ffae6de
No known key found for this signature in database
GPG key ID: B4EC22B66D61A3F5
2 changed files with 13 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

27
crates/librqbit/src/torrent_state/live/mod.rs
View file

@ -1030,18 +1030,15 @@ impl PeerHandler {
.map(|r| r.flatten()) .map(|r| r.flatten())
} }
/// Try to steal a piece from a slower peer. Threshold is
/// "how many times is my average download speed faster to be able to steal".
///
/// If this returns, an existing in-flight piece was marked to be ours.
fn try_steal_old_slow_piece(&self, threshold: f64) -> Option<ValidPieceIndex> { fn try_steal_old_slow_piece(&self, threshold: f64) -> Option<ValidPieceIndex> {
let total = self let my_avg_time = match self.counters.average_piece_download_time() {
.state Some(t) => t,
.stats None => return None,
.downloaded_and_checked_pieces };
.load(Ordering::Acquire);
// heuristic for not enough precision in average time
if total < 20 {
return None;
}
let avg_time = self.state.stats.average_piece_download_time()?;
let mut g = self.state.lock_write("try_steal_old_slow_piece"); let mut g = self.state.lock_write("try_steal_old_slow_piece");
let (idx, elapsed, piece_req) = g let (idx, elapsed, piece_req) = g
@ -1053,10 +1050,10 @@ impl PeerHandler {
.max_by_key(|(_, e, _)| *e)?; .max_by_key(|(_, e, _)| *e)?;
// heuristic for "too slow peer" // heuristic for "too slow peer"
if elapsed.as_secs_f64() > avg_time.as_secs_f64() * threshold { if elapsed.as_secs_f64() > my_avg_time.as_secs_f64() * threshold {
debug!( debug!(
"will steal piece {} from {}: elapsed time {:?}, avg piece time: {:?}", "will steal piece {} from {}: elapsed time {:?}, my avg piece time: {:?}",
idx, piece_req.peer, elapsed, avg_time idx, piece_req.peer, elapsed, my_avg_time
); );
piece_req.peer = self.addr; piece_req.peer = self.addr;
piece_req.started = Instant::now(); piece_req.started = Instant::now();
@ -1469,7 +1466,7 @@ impl PeerHandler {
.fetch_add(piece_len, Ordering::Relaxed); .fetch_add(piece_len, Ordering::Relaxed);
self.state.stats.total_piece_download_ms.fetch_add( self.state.stats.total_piece_download_ms.fetch_add(
full_piece_download_time.as_millis() as u64, full_piece_download_time.as_millis() as u64,
Ordering::Release, Ordering::Relaxed,
); );
// Per-peer piece counters. // Per-peer piece counters.

16
crates/librqbit/src/torrent_state/live/stats/atomic.rs
View file

@ -1,7 +1,4 @@
use std::{ use std::sync::atomic::AtomicU64;
sync::atomic::{AtomicU64, Ordering},
time::Duration,
};
#[derive(Default, Debug)] #[derive(Default, Debug)]
pub struct AtomicStats { pub struct AtomicStats {
@ -12,14 +9,3 @@ pub struct AtomicStats {
pub fetched_bytes: AtomicU64, pub fetched_bytes: AtomicU64,
pub total_piece_download_ms: AtomicU64, pub total_piece_download_ms: AtomicU64,
} }
impl AtomicStats {
pub fn average_piece_download_time(&self) -> Option<Duration> {
let d = self.downloaded_and_checked_pieces.load(Ordering::Acquire);
let t = self.total_piece_download_ms.load(Ordering::Acquire);
if d == 0 {
return None;
}
Some(Duration::from_secs_f64(t as f64 / d as f64 / 1000f64))
}
}