fix(lib): report a download's rate from the bytes actually read
Progress assumed every read had filled the buffer, so a peer trickling small chunks was reported at up to several times its real rate. The rate now comes from the bytes received since the last report.
This commit is contained in:
@@ -16,7 +16,7 @@ use crate::utils::path::{PART_SUFFIX, expand_tilde};
|
|||||||
const READ_BUFFER_SIZE: usize = 8192;
|
const READ_BUFFER_SIZE: usize = 8192;
|
||||||
const CANCEL_POLL: Duration = Duration::from_secs(1);
|
const CANCEL_POLL: Duration = Duration::from_secs(1);
|
||||||
const STALL_DEADLINE: Duration = Duration::from_secs(30);
|
const STALL_DEADLINE: Duration = Duration::from_secs(30);
|
||||||
const PROGRESS_UPDATE_CHUNKS: usize = 15; // ~120KB (15 * 8192 bytes)
|
const PROGRESS_UPDATE_CHUNKS: usize = 15;
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum DownloadError {
|
pub enum DownloadError {
|
||||||
@@ -353,6 +353,7 @@ impl DownloadPeer {
|
|||||||
) -> Result<(Download, String), DownloadError> {
|
) -> Result<(Download, String), DownloadError> {
|
||||||
let mut read_buffer = [0u8; READ_BUFFER_SIZE];
|
let mut read_buffer = [0u8; READ_BUFFER_SIZE];
|
||||||
let mut chunk_counter = 0usize;
|
let mut chunk_counter = 0usize;
|
||||||
|
let mut bytes_since_last_update = 0usize;
|
||||||
let mut last_update_time = Instant::now();
|
let mut last_update_time = Instant::now();
|
||||||
let mut last_data = Instant::now();
|
let mut last_data = Instant::now();
|
||||||
stream
|
stream
|
||||||
@@ -414,11 +415,10 @@ impl DownloadPeer {
|
|||||||
|
|
||||||
part.write(data)?;
|
part.write(data)?;
|
||||||
chunk_counter += 1;
|
chunk_counter += 1;
|
||||||
|
bytes_since_last_update += bytes_read;
|
||||||
|
|
||||||
if chunk_counter.is_multiple_of(PROGRESS_UPDATE_CHUNKS) {
|
if chunk_counter.is_multiple_of(PROGRESS_UPDATE_CHUNKS) {
|
||||||
let elapsed = last_update_time.elapsed().as_secs_f64();
|
let elapsed = last_update_time.elapsed().as_secs_f64();
|
||||||
let bytes_since_last_update =
|
|
||||||
PROGRESS_UPDATE_CHUNKS * READ_BUFFER_SIZE;
|
|
||||||
let speed = if elapsed > 0.0 {
|
let speed = if elapsed > 0.0 {
|
||||||
bytes_since_last_update as f64 / elapsed
|
bytes_since_last_update as f64 / elapsed
|
||||||
} else {
|
} else {
|
||||||
@@ -430,6 +430,7 @@ impl DownloadPeer {
|
|||||||
part.written,
|
part.written,
|
||||||
speed,
|
speed,
|
||||||
);
|
);
|
||||||
|
bytes_since_last_update = 0;
|
||||||
last_update_time = Instant::now();
|
last_update_time = Instant::now();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -3296,3 +3296,74 @@ fn a_third_party_client_reads_our_user_info() {
|
|||||||
|
|
||||||
let _ = std::fs::remove_dir_all(share_dir);
|
let _ = std::fs::remove_dir_all(share_dir);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Progress assumed every read filled the buffer, so a peer that trickles
|
||||||
|
// small chunks was reported at several times its real rate.
|
||||||
|
#[test]
|
||||||
|
fn download_progress_reports_the_rate_actually_received() {
|
||||||
|
let server = server_or_skip!();
|
||||||
|
|
||||||
|
let listen_port = free_port().expect("free listen port");
|
||||||
|
let mut client = Client::with_settings(server.listening_settings(
|
||||||
|
"e2e_rate_dl",
|
||||||
|
"pw",
|
||||||
|
listen_port,
|
||||||
|
));
|
||||||
|
client.connect().expect("connect");
|
||||||
|
assert!(client.login().expect("login"));
|
||||||
|
|
||||||
|
let filename = "trickle.mp3";
|
||||||
|
let content: Vec<u8> = (0..400_000u32).map(|i| (i % 251) as u8).collect();
|
||||||
|
let download_dir = unique_download_dir();
|
||||||
|
let (ready_tx, ready_rx) = std::sync::mpsc::channel();
|
||||||
|
let cfg = MockUpload {
|
||||||
|
listen_addr: format!("127.0.0.1:{listen_port}"),
|
||||||
|
peer_username: "e2e_rate_up".to_string(),
|
||||||
|
filename: filename.to_string(),
|
||||||
|
content: content.clone(),
|
||||||
|
token: 424_260_u32,
|
||||||
|
ready: ready_tx,
|
||||||
|
token_delay: Duration::ZERO,
|
||||||
|
};
|
||||||
|
let uploader =
|
||||||
|
std::thread::spawn(move || run_mock_slow_uploader(&cfg, None));
|
||||||
|
ready_rx
|
||||||
|
.recv_timeout(Duration::from_secs(5))
|
||||||
|
.expect("mock uploader P connection");
|
||||||
|
std::thread::sleep(Duration::from_millis(1500));
|
||||||
|
|
||||||
|
let (_download, status_rx) = client
|
||||||
|
.download(
|
||||||
|
filename.to_string(),
|
||||||
|
"e2e_rate_up".to_string(),
|
||||||
|
content.len() as u64,
|
||||||
|
download_dir.display().to_string(),
|
||||||
|
)
|
||||||
|
.expect("start download");
|
||||||
|
|
||||||
|
let mut fastest = 0.0f64;
|
||||||
|
let mut completed = false;
|
||||||
|
let deadline = Instant::now() + Duration::from_secs(30);
|
||||||
|
while Instant::now() < deadline && !completed {
|
||||||
|
match status_rx.recv_timeout(Duration::from_millis(500)) {
|
||||||
|
Ok(DownloadStatus::InProgress {
|
||||||
|
speed_bytes_per_sec,
|
||||||
|
..
|
||||||
|
}) => fastest = fastest.max(speed_bytes_per_sec),
|
||||||
|
Ok(DownloadStatus::Completed) => completed = true,
|
||||||
|
Ok(DownloadStatus::Failed(_) | DownloadStatus::TimedOut) => break,
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert!(completed, "the trickled download should complete");
|
||||||
|
assert!(fastest > 0.0, "progress should report a rate");
|
||||||
|
// The mock sends 4 KiB every 20 ms.
|
||||||
|
let sent_rate = 4096.0 / 0.020;
|
||||||
|
assert!(
|
||||||
|
fastest < 1.5 * sent_rate,
|
||||||
|
"reported {fastest} B/s for a peer sending {sent_rate} B/s"
|
||||||
|
);
|
||||||
|
|
||||||
|
let _ = uploader.join();
|
||||||
|
let _ = std::fs::remove_dir_all(&download_dir);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user