LFeng/Thing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

性能优化

Browse Source
This commit is contained in:
zhongluofeng
2026-08-06 10:33:16 +08:00
parent c7578a2e6b
commit e66c53e66d
105 changed files with 7273 additions and 5002 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/download_engine/engine.rs
+152 -88
View File
@@ -1,6 +1,6 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
@@ -10,9 +10,10 @@ use super::http_dl::{HttpDownloader, split_segments};
use super::rate_limit::RateLimiter;
use super::storage::{EngineState, Storage};
use super::task::{DownloadTask, DownloaderSettings, ProbeResult, Segment, TaskStatus};
use specta::Type;
/// 重复类型
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[derive(Debug, Clone, PartialEq, serde::Serialize, Type)]
#[serde(rename_all = "camelCase")]
pub enum DuplicateKind {
/// 无重复
@@ -26,7 +27,7 @@ pub enum DuplicateKind {
}
/// 已存在的任务信息（用于前端展示）
#[derive(Debug, Clone, serde::Serialize)]
#[derive(Debug, Clone, serde::Serialize, Type)]
#[serde(rename_all = "camelCase")]
pub struct ExistingTaskInfo {
pub id: String,
@@ -35,7 +36,7 @@ pub struct ExistingTaskInfo {
}
/// check_url 命令返回的结果
#[derive(Debug, Clone, serde::Serialize)]
#[derive(Debug, Clone, serde::Serialize, Type)]
#[serde(rename_all = "camelCase")]
pub struct CheckUrlResult {
/// 探测是否成功
@@ -77,6 +78,9 @@ pub struct CompletePayload {
/// 活跃下载句柄
struct TaskHandle {
/// 代际号：同一任务每次 start_download 递增。
/// 旧代际任务完成时不删除新任务句柄、不覆盖新任务状态（防 pause→resume 竞态）
gen: u64,
cancel: Arc<AtomicBool>,
/// 每个分段的已下载字节（与 segments 一一对应）
progress: Vec<Arc<std::sync::atomic::AtomicU64>>,
@@ -107,6 +111,8 @@ struct EngineInner {
app_handle: AppHandle,
/// 引擎是否已启动
started: AtomicBool,
/// 任务代际计数器（每次 start_download 递增，分配给新句柄）
next_gen: AtomicU64,
/// 持久化节流：上次保存时间
last_save: Mutex<Instant>,
}
@@ -149,6 +155,7 @@ impl DownloadEngine {
http: HttpDownloader::new(),
app_handle,
started: AtomicBool::new(false),
next_gen: AtomicU64::new(0),
last_save: Mutex::new(Instant::now()),
}),
};
@@ -182,7 +189,7 @@ impl DownloadEngine {
headers: &HashMap<String, String>,
) -> (Result<ProbeResult, String>, DuplicateKind, Option<ExistingTaskInfo>) {
let probe = self.inner.http.probe(url, headers).await;
let settings = self.inner.settings.lock().unwrap().clone();
let settings = self.inner.settings.lock().unwrap_or_else(|e| e.into_inner()).clone();
let task_dir = dir.map(|d| d.to_string()).unwrap_or_else(|| settings.download_dir.clone());
let filename = probe.as_ref().ok()
@@ -199,7 +206,7 @@ impl DownloadEngine {
let mut existing: Option<ExistingTaskInfo> = None;
{
let tasks = self.inner.tasks.lock().unwrap();
let tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
for t in tasks.values() {
// URL 完全相同
if t.url == url {
@@ -280,7 +287,7 @@ impl DownloadEngine {
// 探测资源信息
let probe = self.inner.http.probe(&url, &headers).await;
let settings = self.inner.settings.lock().unwrap().clone();
let settings = self.inner.settings.lock().unwrap_or_else(|e| e.into_inner()).clone();
let task_dir = dir.unwrap_or_else(|| settings.download_dir.clone());
// 确定文件名
@@ -347,14 +354,14 @@ impl DownloadEngine {
};
{
let mut tasks = self.inner.tasks.lock().unwrap();
let mut tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
tasks.insert(id.clone(), task);
}
self.persist_now();
// 如果探测成功，尝试调度
let should_schedule = {
let tasks = self.inner.tasks.lock().unwrap();
let tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
tasks.get(&id).map(|t| t.status == TaskStatus::Queued).unwrap_or(false)
};
if should_schedule {
@@ -362,7 +369,7 @@ impl DownloadEngine {
}
// 通知前端有新任务加入（扩展通过 HTTP API 添加时，前端需要刷新）
let _ = self.inner.app_handle.emit("download-added", serde_json::json!({ "id": id }));
let _ = self.inner.app_handle.emit(crate::constants::events::DOWNLOAD_ADDED, serde_json::json!({ "id": id }));
Ok(id)
}
@@ -371,7 +378,7 @@ impl DownloadEngine {
pub fn pause_task(&self, id: &str) -> Result<(), String> {
// 1. 设置取消标志 + 读取进度（锁 handles）
let progress_values: Vec<u64> = {
let handles = self.inner.handles.lock().unwrap();
let handles = self.inner.handles.lock().unwrap_or_else(|e| e.into_inner());
if let Some(handle) = handles.get(id) {
handle.cancel.store(true, Ordering::SeqCst);
handle.progress.iter().map(|p| p.load(Ordering::Relaxed)).collect()
@@ -382,7 +389,7 @@ impl DownloadEngine {
// 2. 更新任务状态 + 同步进度（锁 tasks，不嵌套锁 handles）
{
let mut tasks = self.inner.tasks.lock().unwrap();
let mut tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
if let Some(task) = tasks.get_mut(id) {
if task.status == TaskStatus::Active || task.status == TaskStatus::Queued {
task.status = TaskStatus::Paused;
@@ -407,7 +414,7 @@ impl DownloadEngine {
/// 恢复任务
pub fn resume_task(&self, id: &str) -> Result<(), String> {
{
let mut tasks = self.inner.tasks.lock().unwrap();
let mut tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
if let Some(task) = tasks.get_mut(id) {
if task.status != TaskStatus::Paused && task.status != TaskStatus::Error {
return Err("任务不在可恢复状态".to_string());
@@ -434,7 +441,7 @@ impl DownloadEngine {
pub fn remove_task(&self, id: &str, delete_files: bool) -> Result<(), String> {
// 1. 设置取消标志 + abort join handle（锁 handles）
{
let mut handles = self.inner.handles.lock().unwrap();
let mut handles = self.inner.handles.lock().unwrap_or_else(|e| e.into_inner());
if let Some(handle) = handles.remove(id) {
handle.cancel.store(true, Ordering::SeqCst);
if let Ok(mut join) = handle.join.lock() {
@@ -447,7 +454,7 @@ impl DownloadEngine {
// 2. 从任务列表移除（锁 tasks）
let task = {
let mut tasks = self.inner.tasks.lock().unwrap();
let mut tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
tasks.remove(id)
};
@@ -466,12 +473,12 @@ impl DownloadEngine {
/// 获取所有任务
pub fn get_tasks(&self) -> Vec<DownloadTask> {
self.inner.tasks.lock().unwrap().values().cloned().collect()
self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner()).values().cloned().collect()
}
/// 获取设置
pub fn get_settings(&self) -> DownloaderSettings {
self.inner.settings.lock().unwrap().clone()
self.inner.settings.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// 保存设置
@@ -485,7 +492,7 @@ impl DownloadEngine {
self.inner.global_limiter.set_limit(new_limit);
{
let mut s = self.inner.settings.lock().unwrap();
let mut s = self.inner.settings.lock().unwrap_or_else(|e| e.into_inner());
*s = settings;
}
self.persist_now();
@@ -500,7 +507,7 @@ impl DownloadEngine {
pub fn cleanup_on_exit(&self) {
// 取消所有活跃下载
{
let handles = self.inner.handles.lock().unwrap();
let handles = self.inner.handles.lock().unwrap_or_else(|e| e.into_inner());
for handle in handles.values() {
handle.cancel.store(true, Ordering::SeqCst);
}
@@ -508,7 +515,7 @@ impl DownloadEngine {
// 将 Active 任务标记为 Paused
{
let mut tasks = self.inner.tasks.lock().unwrap();
let mut tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
for task in tasks.values_mut() {
if task.status == TaskStatus::Active {
task.status = TaskStatus::Paused;
@@ -528,10 +535,10 @@ impl DownloadEngine {
/// 调度：如果活跃任务数 < max_concurrent，启动排队任务
fn schedule(&self) {
let max_concurrent = self.inner.settings.lock().unwrap().max_concurrent as usize;
let max_concurrent = self.inner.settings.lock().unwrap_or_else(|e| e.into_inner()).max_concurrent as usize;
let (active_count, queued_ids) = {
let tasks = self.inner.tasks.lock().unwrap();
let tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
let active = tasks.values().filter(|t| t.status == TaskStatus::Active).count();
let mut queued: Vec<_> = tasks
.values()
@@ -554,7 +561,7 @@ impl DownloadEngine {
/// 启动单个下载任务
fn start_download(&self, id: String) {
let task = {
let mut tasks = self.inner.tasks.lock().unwrap();
let mut tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
match tasks.get_mut(&id) {
Some(task) if task.status == TaskStatus::Queued => {
task.status = TaskStatus::Active;
@@ -565,6 +572,9 @@ impl DownloadEngine {
}
};
// 分配代际号（同一任务每次重启递增）
let gen = self.inner.next_gen.fetch_add(1, Ordering::SeqCst) + 1;
// 创建取消标志和进度计数器
let cancel = Arc::new(AtomicBool::new(false));
let progress: Vec<Arc<std::sync::atomic::AtomicU64>> = task
@@ -573,13 +583,22 @@ impl DownloadEngine {
.map(|s| Arc::new(std::sync::atomic::AtomicU64::new(s.completed)))
.collect();
// 存储句柄
let handle = TaskHandle {
cancel: cancel.clone(),
progress: progress.iter().map(|p| p.clone()).collect(),
join: Mutex::new(None),
};
self.inner.handles.lock().unwrap().insert(id.clone(), handle);
// 存储句柄：先取消旧代际（若存在），确保旧任务尽快退出，避免新旧并发写同一临时文件
{
let mut handles = self.inner.handles.lock().unwrap_or_else(|e| e.into_inner());
if let Some(old) = handles.get(&id) {
old.cancel.store(true, Ordering::SeqCst);
}
handles.insert(
id.clone(),
TaskHandle {
gen,
cancel: cancel.clone(),
progress: progress.iter().map(|p| p.clone()).collect(),
join: Mutex::new(None),
},
);
}
// 生成下载 future
let engine = self.clone();
@@ -596,69 +615,94 @@ impl DownloadEngine {
let http = self.inner.http.clone();
let join = tauri::async_runtime::spawn(async move {
let my_gen = gen;
let result = http
.download(&url, &headers, &segments, &temp_file_path, cancel_clone, &progress_clone, limiter)
.await;
// 下载结束，更新任务状态
let final_status = match &result {
Ok(()) => TaskStatus::Complete,
Err(e) if e == "已取消" => TaskStatus::Paused,
Err(_) => TaskStatus::Error,
// 代际守卫：仅最新代际的任务能更新状态 / 移除句柄 / 发完成事件。
// pause→resume 后旧代际任务才退出，此时句柄已被新代际替换，
// 若仍按旧逻辑执行会覆盖新任务状态并误删新句柄（pause/remove 失效）
let is_current = {
let handles = engine.inner.handles.lock().unwrap_or_else(|e| e.into_inner());
handles.get(&id_clone).map(|h| h.gen == my_gen).unwrap_or(false)
};
// 下载成功后，将临时文件重命名为最终文件名
if final_status == TaskStatus::Complete {
let _ = tokio::fs::rename(&temp_file_path, &final_file_path).await;
}
if is_current {
// 下载结束，更新任务状态
let mut final_status = match &result {
Ok(()) => TaskStatus::Complete,
Err(e) if e == "已取消" => TaskStatus::Paused,
Err(_) => TaskStatus::Error,
};
let mut final_error: Option<String> = match &result {
Err(e) if e != "已取消" => Some(e.clone()),
_ => None,
};
// 同步最终进度到任务
{
let mut tasks = engine.inner.tasks.lock().unwrap();
if let Some(task) = tasks.get_mut(&id_clone) {
for (i, prog) in progress_clone.iter().enumerate() {
if let Some(seg) = task.segments.get_mut(i) {
seg.completed = prog.load(Ordering::Relaxed);
}
}
task.recalc_completed();
task.speed = 0;
task.status = final_status.clone();
if let Err(e) = &result {
if e != "已取消" {
task.error = Some(e.clone());
}
}
if final_status == TaskStatus::Complete {
task.completed_size = task.total_size.max(task.completed_size);
// 下载成功后，将临时文件重命名为最终文件名
if final_status == TaskStatus::Complete {
if let Err(e) = tokio::fs::rename(&temp_file_path, &final_file_path).await {
// 重命名失败（如目标被占用/路径不可写）→ 置 Error，
// 避免"标记完成但文件缺失"的状态不一致
final_status = TaskStatus::Error;
final_error = Some(format!("移动文件到最终路径失败: {}", e));
}
}
// 同步最终进度到任务
{
let mut tasks = engine.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
if let Some(task) = tasks.get_mut(&id_clone) {
for (i, prog) in progress_clone.iter().enumerate() {
if let Some(seg) = task.segments.get_mut(i) {
seg.completed = prog.load(Ordering::Relaxed);
}
}
task.recalc_completed();
task.speed = 0;
task.status = final_status.clone();
if let Some(e) = final_error {
task.error = Some(e);
}
if final_status == TaskStatus::Complete {
task.completed_size = task.total_size.max(task.completed_size);
}
}
}
// 从活跃句柄中移除（仅移除自己代际的句柄）
{
let mut handles = engine.inner.handles.lock().unwrap_or_else(|e| e.into_inner());
if let Some(h) = handles.get(&id_clone) {
if h.gen == my_gen {
handles.remove(&id_clone);
}
}
}
// 发送完成事件
let task = engine.inner.tasks.lock().unwrap_or_else(|e| e.into_inner()).get(&id_clone).cloned();
if let Some(task) = task {
let _ = engine.inner.app_handle.emit(
"download-complete",
CompletePayload {
id: id_clone.clone(),
filename: task.filename.clone(),
status: task.status.clone(),
error: task.error.clone(),
},
);
}
// 持久化 + 调度下一个
engine.persist_now();
engine.schedule();
}
// 从活跃句柄中移除
engine.inner.handles.lock().unwrap().remove(&id_clone);
// 发送完成事件
let task = engine.inner.tasks.lock().unwrap().get(&id_clone).cloned();
if let Some(task) = task {
let _ = engine.inner.app_handle.emit(
"download-complete",
CompletePayload {
id: id_clone.clone(),
filename: task.filename.clone(),
status: task.status.clone(),
error: task.error.clone(),
},
);
}
// 持久化 + 调度下一个
engine.persist_now();
engine.schedule();
});
// 存储 JoinHandle
if let Some(h) = self.inner.handles.lock().unwrap().get_mut(&id) {
if let Some(h) = self.inner.handles.lock().unwrap_or_else(|e| e.into_inner()).get_mut(&id) {
if let Ok(mut join_guard) = h.join.lock() {
*join_guard = Some(join);
}
@@ -670,6 +714,7 @@ impl DownloadEngine {
let progress_monitor = progress;
let cancel_monitor = cancel;
tauri::async_runtime::spawn(async move {
let my_gen = gen;
// 初始化为当前已下载量，避免恢复下载时首次计算速度异常
let mut last_completed: u64 = progress_monitor
.iter()
@@ -681,9 +726,17 @@ impl DownloadEngine {
loop {
interval.tick().await;
// 代际守卫：pause→resume 后旧监控立即退出，避免用过期进度覆盖新任务
let is_current = {
let handles = engine.inner.handles.lock().unwrap_or_else(|e| e.into_inner());
handles.get(&id_monitor).map(|h| h.gen == my_gen).unwrap_or(false)
};
if !is_current {
break;
}
// 如果任务已不在活跃状态，停止监控
let is_active = {
let tasks = engine.inner.tasks.lock().unwrap();
let tasks = engine.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
tasks.get(&id_monitor).map(|t| t.status == TaskStatus::Active).unwrap_or(false)
};
if !is_active || cancel_monitor.load(Ordering::SeqCst) {
@@ -709,7 +762,7 @@ impl DownloadEngine {
// 更新任务状态 + 发送进度事件
let total_size = {
let mut tasks = engine.inner.tasks.lock().unwrap();
let mut tasks = engine.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
if let Some(task) = tasks.get_mut(&id_monitor) {
task.completed_size = completed;
task.speed = speed;
@@ -743,7 +796,7 @@ impl DownloadEngine {
/// 节流持久化（至少间隔 3 秒）
fn persist_throttled(&self) {
let should_save = {
let last = self.inner.last_save.lock().unwrap();
let last = self.inner.last_save.lock().unwrap_or_else(|e| e.into_inner());
last.elapsed() >= Duration::from_secs(3)
};
if should_save {
@@ -751,19 +804,30 @@ impl DownloadEngine {
}
}
/// 立即持久化
/// 立即持久化。
/// 保存失败（磁盘不可写/rename 失败）时把进行中任务标记为 Error，
/// 防止用户误以为任务已持久化而关闭应用导致数据丢失。
fn persist_now(&self) {
let tasks: Vec<DownloadTask> = {
let tasks = self.inner.tasks.lock().unwrap();
let tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
tasks.values().cloned().collect()
};
let settings = self.inner.settings.lock().unwrap().clone();
let settings = self.inner.settings.lock().unwrap_or_else(|e| e.into_inner()).clone();
let state = EngineState {
tasks,
settings,
next_id: 0, // storage.save 会从 id_counter 读取
};
self.inner.storage.save(state);
*self.inner.last_save.lock().unwrap() = Instant::now();
if let Err(e) = self.inner.storage.save(state) {
crate::logger::log_error("download", &format!("状态持久化失败，进行中任务标记为 Error: {}", e));
let mut tasks = self.inner.tasks.lock().unwrap_or_else(|e| e.into_inner());
for task in tasks.values_mut() {
if task.status == TaskStatus::Active {
task.status = TaskStatus::Error;
}
}
return; // 不更新 last_save，下次定时器会重试
}
*self.inner.last_save.lock().unwrap_or_else(|e| e.into_inner()) = Instant::now();
}
}