From 842062fe44f6f8d7842b954afa7fcbafa755077c Mon Sep 17 00:00:00 2001 From: Huang Xin Date: Mon, 28 Jul 2025 00:56:41 +0800 Subject: [PATCH] perf: multi-part download with range access (#1690) --- Cargo.lock | 1 + apps/readest-app/src-tauri/Cargo.toml | 1 + .../src-tauri/src/transfer_file.rs | 151 ++++++++++++++---- .../src/app/library/hooks/useBooksSync.ts | 1 + 4 files changed, 123 insertions(+), 31 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9e84758a..c5638804 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,6 +14,7 @@ version = "0.2.2" dependencies = [ "block", "cocoa", + "futures", "futures-util", "log", "objc", diff --git a/apps/readest-app/src-tauri/Cargo.toml b/apps/readest-app/src-tauri/Cargo.toml index 781aeb71..fbc3a90b 100644 --- a/apps/readest-app/src-tauri/Cargo.toml +++ b/apps/readest-app/src-tauri/Cargo.toml @@ -29,6 +29,7 @@ thiserror = "2" tokio = { version = "1", features = ["fs"] } tokio-util = { version = "0.7", features = ["codec"] } futures-util = "0.3" +futures = "0.3.31" read-progress-stream = "1.0.0" reqwest = { version = "0.12", default-features = false, features = [ "json", diff --git a/apps/readest-app/src-tauri/src/transfer_file.rs b/apps/readest-app/src-tauri/src/transfer_file.rs index a58d098f..f3d3234f 100644 --- a/apps/readest-app/src-tauri/src/transfer_file.rs +++ b/apps/readest-app/src-tauri/src/transfer_file.rs @@ -17,8 +17,8 @@ use tokio_util::codec::{BytesCodec, FramedRead}; use read_progress_stream::ReadProgressStream; -use std::collections::HashMap; use std::time::Instant; +use std::{collections::HashMap, sync::Arc}; type Result = std::result::Result; @@ -109,41 +109,130 @@ pub async fn download_file( body: Option, on_progress: Channel, ) -> Result<()> { + use futures::stream::{self, StreamExt}; + use std::cmp::min; + use tokio::io::AsyncSeekExt; + + const PART_SIZE: u64 = 1024 * 1024 * 1; + let client = reqwest::Client::new(); - let mut request = if let Some(body) = body { - client.post(url).body(body) - } else { - client.get(url) - }; - // Loop trought the headers keys and values - // and add them to the request object. - for (key, value) in headers { - request = request.header(&key, value); + + // Check if server supports range requests + let range_resp = client.get(url).header("Range", "bytes=0-0").send().await?; + let accept_ranges = range_resp + .headers() + .get("accept-ranges") + .map(|v| v.to_str().unwrap_or("")) + .unwrap_or("") + .to_ascii_lowercase() + == "bytes"; + let total = range_resp + .headers() + .get("content-range") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.split('/').nth(1)) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + + if !accept_ranges || total == 0 { + // Fallback to original single-threaded logic + let mut request = if let Some(body) = body { + client.post(url).body(body) + } else { + client.get(url) + }; + + for (key, value) in headers.iter() { + request = request.header(key, value); + } + + let response = request.send().await?; + if !response.status().is_success() { + return Err(Error::HttpErrorCode( + response.status().as_u16(), + response.text().await.unwrap_or_default(), + )); + } + + let total = response.content_length().unwrap_or(0); + let mut file = BufWriter::new(File::create(file_path).await?); + let mut stream = response.bytes_stream(); + + let mut stats = TransferStats::default(); + while let Some(chunk) = stream.try_next().await? { + file.write_all(&chunk).await?; + stats.record_chunk_transfer(chunk.len()); + let _ = on_progress.send(ProgressPayload { + progress: stats.total_transferred, + total, + transfer_speed: stats.transfer_speed, + }); + } + file.flush().await?; + return Ok(()); } - let response = request.send().await?; - if !response.status().is_success() { - return Err(Error::HttpErrorCode( - response.status().as_u16(), - response.text().await.unwrap_or_default(), - )); - } - let total = response.content_length().unwrap_or(0); + // Multi-part download with range access + let part_count = (total + PART_SIZE - 1) / PART_SIZE; + let file = File::create(file_path).await?; + file.set_len(total).await?; - let mut file = BufWriter::new(File::create(file_path).await?); - let mut stream = response.bytes_stream(); + let file = Arc::new(tokio::sync::Mutex::new(file)); + let progress = Arc::new(tokio::sync::Mutex::new(TransferStats::default())); - let mut stats = TransferStats::default(); - while let Some(chunk) = stream.try_next().await? { - file.write_all(&chunk).await?; - stats.record_chunk_transfer(chunk.len()); - let _ = on_progress.send(ProgressPayload { - progress: stats.total_transferred, - total, - transfer_speed: stats.transfer_speed, - }); - } - file.flush().await?; + stream::iter(0..part_count) + .for_each_concurrent(8, |i| { + let client = client.clone(); + let file = Arc::clone(&file); + let progress = Arc::clone(&progress); + let headers = headers.clone(); + let url = url.to_string(); + let on_progress = on_progress.clone(); + + async move { + let start = i * PART_SIZE; + let end = min(start + PART_SIZE - 1, total - 1); + let range_header = format!("bytes={}-{}", start, end); + + let mut req = client.get(&url).header("Range", range_header); + for (key, value) in headers { + req = req.header(key, value); + } + + let resp = match req.send().await { + Ok(r) => r, + Err(_) => return, + }; + + if !resp.status().is_success() + && resp.status() != reqwest::StatusCode::PARTIAL_CONTENT + { + return; + } + + let bytes = match resp.bytes().await { + Ok(b) => b, + Err(_) => return, + }; + + { + let mut f = file.lock().await; + f.seek(std::io::SeekFrom::Start(start)).await.unwrap(); + f.write_all(&bytes).await.unwrap(); + } + + { + let mut stat = progress.lock().await; + stat.record_chunk_transfer(bytes.len()); + let _ = on_progress.send(ProgressPayload { + progress: stat.total_transferred, + total, + transfer_speed: stat.transfer_speed, + }); + } + } + }) + .await; Ok(()) } diff --git a/apps/readest-app/src/app/library/hooks/useBooksSync.ts b/apps/readest-app/src/app/library/hooks/useBooksSync.ts index d16a64c8..cedab7c1 100644 --- a/apps/readest-app/src/app/library/hooks/useBooksSync.ts +++ b/apps/readest-app/src/app/library/hooks/useBooksSync.ts @@ -92,6 +92,7 @@ export const useBooksSync = ({ onSyncStart, onSyncEnd }: UseBooksSyncProps) => { console.error('Failed to download book:', newBook); } finally { newBook.coverImageUrl = await appService?.generateCoverImageUrl(newBook); + newBook.coverImageUrl += '?timestamp=' + Date.now(); updatedLibrary.push(newBook); setLibrary(updatedLibrary); }