//! 进程级共享引擎:**一个数据根一个 `DownloadEngine`**,跨请求复用。 //! //! # 为什么要有这一层(M165) //! //! 在此之前,宿主侧每处理一次「请求」(一次检索 / 一次详情 / 一次装文件 / 一次拉清单)都会 //! 现建一个引擎(`DownloadEngine::new`)。建的时候什么都会重新来一遍,于是下面这些**跨请求 //! 本该积累的状态每次都清零**: //! //! | 引擎里的东西 | 每建一次就丢什么 | //! |--------------|------------------| //! | `reqwest::Client`(`http::build_client`) | 整条连接池:每个请求重新 TCP + TLS 握手 | //! | `Semaphore`(`max_concurrency`) | 限流变成「每请求 64」,N 个并发请求 = 64×N 条在飞 | //! | `HealthRegistry`(`open_in`) | 从盘上重读 `health.json`;`last_save_ms` 被重置成「现在」⇒ 收尾那次 `flush()` 落在 5s 节流窗口里直接返回,**这个短命引擎记下的失败数一并丢掉**(只有「进/出冷却」会立刻落盘) | //! | `ContentCache`(`open`) | 从盘上重读并解析整份 `index.json`;`persist()` 用**本请求这一份内存索引**覆写全文件 ⇒ 两个并发请求各自覆写,后写的把先写的登记条目冲掉 | //! | `MirrorTable::builtin()` | 重新解析一遍内置 `mirrors.json` | //! | 地区档案 + `Jitter` | 重读 `region.json`;抖动序列重新播种 | //! //! 失败分类、镜像回退、`Retry-After` 语义一概不动——复用只改「谁来建、建几次」,不改 //! [`DownloadEngine`] 内部的任何判定。 //! //! # 边界(谁持有 / 活多久 / 并发多少) //! //! * **谁持有**:本模块的进程级登记处(`OnceLock>`),键是**数据根的缓存目录** //! (`/cache`)。同一数据根的多次请求拿到同一个 `Arc`; //! 不同数据根各有一份,缓存/健康度不会串味(有测试钉住这一条)。 //! * **活多久**:跟随进程。登记处**不做淘汰**——淘汰会让后续请求建出第二个引擎,同一个根上 //! 同时存在两条连接池,等于把要修的问题换个地方复现。要让地区判定等「一次性派生值」重算, //! 用 [`forget`] 显式失效(设置页「重新检测地区」就这么做)。 //! * **`forget` 到什么范围**:只让登记处不再持有那一份。**已经拿到 `Arc` 的人继续用旧的**: //! 短命请求自然结束(它们的窗口只有一次下载),但**长命持有者必须自己重建**——presenter 的 //! 自更新子系统(`host::update::Factory`)从 `CoreHost::build` 起就把引擎存在字段里, //! 只 forget 的话它会一直用旧实例(含旧地区档案)直到进程结束。所以「重新检测地区」那条 //! 设置动作除了 [`forget`] 还调 `CoreHost::refresh_update_factory()` 把新引擎接过去 //! (M165 复审 P2-1:这是本模块文档原先说漏的一半)。 //! * **并发多少**:`EngineConfig::default().max_concurrency`(64)现在是**进程级**上限, //! 不再是「每请求 64」。单请求独占时与从前一致;多请求并发时它们分摊这 64 张许可。 //! * **镜像切换 / 失效**:一律走引擎原有的候选规划 + 健康度冷却,本模块不参与判定;唯一可见的 //! 差别是健康度记忆现在真的跨请求累积(这本就是它存在的理由)。 //! //! # 与 [`super::blocking::block_on`] 的绑定关系 //! //! `reqwest` 的连接池绑在**驱动连接的运行时**上:引擎复用了而每次 `block_on` 都新起一个运行时, //! 池里的空闲连接就属于已经销毁的运行时,复用会退化成「拿到的都是死连接」。所以引擎复用必须 //! 配运行时常驻(`blocking` 模块那头一并改成进程内唯一运行时)。两者是同一件事的两半。 use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, OnceLock}; use super::cache::{now_ms, ContentCache}; use super::config::EngineConfig; use super::engine::DownloadEngine; use super::health::HealthRegistry; use super::mirrors::MirrorTable; use super::region::{decide_from, load_cached, store_cached}; use super::region_system::detect_signals; /// 登记处:缓存根 → 引擎。 fn registry() -> &'static Mutex>> { static ENGINES: OnceLock>>> = OnceLock::new(); ENGINES.get_or_init(|| Mutex::new(HashMap::new())) } /// 「这个根建过几次引擎」的账本(守卫/诊断用,**不**随 [`forget`] 清零)。 /// /// 与 [`registry`] **分两把锁**:记账发生在 `DownloadEngine::new` 里(见 [`record_build`]), /// 而 [`engine_for`] 是**持着**登记处那把锁去调构造的——合成一把锁就是自己等自己。 fn ledger() -> &'static Mutex> { static BUILDS: OnceLock>> = OnceLock::new(); BUILDS.get_or_init(|| Mutex::new(HashMap::new())) } /// 缓存根的键:去掉尾部分隔符(`D:/x/cache` 与 `D:/x/cache/` 视作同一根)。 fn key_of(cache_root: &Path) -> PathBuf { let text = cache_root.to_string_lossy(); let trimmed = text.trim_end_matches(['/', '\\']); if trimmed.is_empty() { cache_root.to_path_buf() } else { PathBuf::from(trimmed) } } /// 取该数据根上的共享引擎(第一次调用时构造,之后一律复用)。 /// /// 构造过程只做本地 I/O(读 `region.json` / `health.json` / `index.json`,解析内置 /// `mirrors.json`),**不联网**;因此持锁构造不会把网络等待带进临界区。 #[must_use] pub fn engine_for(cache_root: impl AsRef) -> Arc { let key = key_of(cache_root.as_ref()); let mut engines = registry() .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); if let Some(engine) = engines.get(&key) { return Arc::clone(engine); } let engine = Arc::new(build(&key)); engines.insert(key, Arc::clone(&engine)); engine } /// 让该数据根的共享引擎失效(下次 [`engine_for`] 重新构造)。 /// /// **只让登记处不再持有那一份**:已经拿到 `Arc` 的人继续用旧的。短命请求自然结束;长命持有者 /// (presenter 的自更新 `Factory`)必须自己重建,否则它会一直用旧实例——本仓在「重新检测地区」 /// 的设置动作里调 `CoreHost::refresh_update_factory()` 补上这一步(见模块文档「`forget` 到什么范围」)。 /// 返回是否真淘汰了一份。 pub fn forget(cache_root: impl AsRef) -> bool { registry() .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) .remove(&key_of(cache_root.as_ref())) .is_some() } /// 记一笔「这个缓存根上又造了一个引擎」。 /// /// **唯一调用点是 [`DownloadEngine::new`]**(不是 [`engine_for`]):记账卡在构造这个卡口上, /// 于是 [`constructions`] 统计的是**进程里真实发生过几次构造**,与「谁造的」无关。 /// 若只在 `engine_for` 里记账,任何绕过登记处、自己 `DownloadEngine::new` 的调用点都统计不到 /// (M165 复审备注 2 的诉求)。 pub(crate) fn record_build(cache_root: &Path) { *ledger() .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) .entry(key_of(cache_root)) .or_default() += 1; } /// 该数据根在本进程里**实际构造过几次引擎**(不论谁构造的)。 /// /// 这是给守卫与诊断用的:跨请求复用成立时,同一个根无论被请求多少次,这个数都停在 `1` /// (`forget` 之后每重建一次 +1)。只断言「结果一样」测不出复用——引擎建几次都能下同一个文件。 #[must_use] pub fn constructions(cache_root: impl AsRef) -> usize { ledger() .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) .get(&key_of(cache_root.as_ref())) .copied() .unwrap_or(0) } /// 真实构造(**唯一**一处 `DownloadEngine::new` 的宿主侧装配)。 fn build(cache_root: &Path) -> DownloadEngine { let now = now_ms(); let region_path = cache_root.join("region.json"); let region = load_cached(®ion_path, now).unwrap_or_else(|| { let profile = decide_from(&detect_signals(), now); let _ = store_cached(®ion_path, &profile); profile }); DownloadEngine::new( EngineConfig::default(), Arc::new(MirrorTable::builtin()), Arc::new(region), Arc::new(HealthRegistry::open_in(cache_root, now)), Arc::new(ContentCache::open(cache_root)), ) } #[cfg(test)] mod tests { use super::*; /// 每个用例一个独立数据根(登记处是进程级的,键撞了会互相干扰)。 fn root(tag: &str) -> PathBuf { static SEQ: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::SeqCst); let dir = std::env::temp_dir().join(format!("ull-shared-{tag}-{}-{seq}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir_all(&dir).expect("建缓存根"); dir } /// 守卫:同一个根上「多次请求」只构造一个引擎(**断言构造次数,不是只看结果**)。 /// /// 抓得住的变异:把 [`engine_for`] 里的 `engines.get` 命中分支删掉(每次现建), /// 本用例立刻红在 `constructions == 1` 与 `Arc::ptr_eq` 两处。 #[test] fn one_engine_per_cache_root() { let dir = root("one"); let first = engine_for(&dir); let second = engine_for(&dir); assert_eq!(constructions(&dir), 1, "同一根只该构造一次引擎"); assert!(Arc::ptr_eq(&first, &second), "两次请求必须是同一个引擎"); // 尾部分隔符不该被当成另一个根。 let with_sep = PathBuf::from(format!("{}/", dir.display())); assert!( Arc::ptr_eq(&first, &engine_for(&with_sep)), "尾部分隔符不该造成第二个引擎" ); assert_eq!(constructions(&dir), 1); std::fs::remove_dir_all(&dir).ok(); } /// 两个数据根**不共享**引擎:缓存 / 健康度串味比多建一个引擎严重得多。 #[test] fn different_cache_roots_do_not_share() { let a = root("a"); let b = root("b"); let engine_a = engine_for(&a); let engine_b = engine_for(&b); assert!(!Arc::ptr_eq(&engine_a, &engine_b)); assert_eq!(constructions(&a), 1); assert_eq!(constructions(&b), 1); assert_eq!(engine_a.cache().root(), a.as_path()); assert_eq!(engine_b.cache().root(), b.as_path()); std::fs::remove_dir_all(&a).ok(); std::fs::remove_dir_all(&b).ok(); } /// 失效后重建(设置页「重新检测地区」依赖这条语义)。 #[test] fn forget_rebuilds_once() { let dir = root("forget"); let before = engine_for(&dir); assert!(forget(&dir), "已存在的引擎应被淘汰"); assert!(!forget(&dir), "淘汰后不该再有可淘汰的"); let after = engine_for(&dir); assert!(!Arc::ptr_eq(&before, &after)); assert_eq!(constructions(&dir), 2, "失效一次 = 多构造一次"); std::fs::remove_dir_all(&dir).ok(); } /// 离线回环 HTTP 服务:记下「接受了几条连接 / 处理了几条请求」。 /// /// 只回固定 JSON、只讲 HTTP/1.1 keep-alive,够用来判「两次下载有没有复用同一条连接」。 fn loopback_server( expected_requests: usize, ) -> (String, std::sync::mpsc::Receiver<(usize, usize)>) { use std::io::{Read as _, Write as _}; use std::net::TcpListener; let listener = TcpListener::bind("127.0.0.1:0").expect("绑定回环端口"); let addr = listener.local_addr().expect("取端口"); let (tx, rx) = std::sync::mpsc::channel(); std::thread::spawn(move || { let mut conns = 0usize; let mut requests = 0usize; while requests < expected_requests { let Ok((mut socket, _)) = listener.accept() else { break; }; conns += 1; let _ = socket.set_read_timeout(Some(std::time::Duration::from_secs(5))); let mut seen = Vec::new(); loop { let mut chunk = [0u8; 1024]; match socket.read(&mut chunk) { Ok(0) | Err(_) => break, Ok(read) => seen.extend_from_slice(&chunk[..read]), } if !seen.windows(4).any(|window| window == b"\r\n\r\n") { continue; } requests += 1; let body = br#"{"ok":true}"#; let head = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n", body.len() ); if socket.write_all(head.as_bytes()).is_err() || socket.write_all(body).is_err() { break; } seen.clear(); if requests >= expected_requests { let _ = tx.send((conns, requests)); return; } } } let _ = tx.send((conns, requests)); }); (format!("http://{addr}/meta.json"), rx) } /// **复用真的落到连接上**:同一个引擎跨两次 `block_on` 取同一个 host,只应该拨号一次。 /// /// 这条钉住的是「引擎复用」与「运行时常驻」必须成对出现的那半个承诺:只复用引擎、 /// 每次 `block_on` 新起运行时的话,连接池里的空闲连接属于已销毁的运行时, /// 第二次请求会重新 TCP 握手——`conns` 就是 2 而不是 1。 /// /// 离线可跑:服务端是 127.0.0.1 的回环监听器,不碰任何外网站点。 #[test] fn pooled_connection_survives_across_block_on_calls() { use super::super::blocking::block_on; use super::super::types::{DownloadRequest, ResourceKind}; let dir = root("pool"); let (url, counts) = loopback_server(2); let engine = engine_for(&dir); for seq in 0..2 { let request = DownloadRequest::new( ResourceKind::Other, url.clone(), dir.join(format!("meta-{seq}.json")), ); let bytes = block_on(engine.download_bytes(request)).expect("回环请求应成功"); assert_eq!(bytes, br#"{"ok":true}"#.to_vec()); } let (conns, requests) = counts .recv_timeout(std::time::Duration::from_secs(10)) .expect("服务端应报告计数"); assert_eq!(requests, 2, "两次下载 = 两条请求"); assert_eq!(conns, 1, "两次下载应复用同一条连接"); std::fs::remove_dir_all(&dir).ok(); } }