From 8426d001334915594b873af490fea4f2381abd34 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Sun, 11 Oct 2026 17:56:53 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=9B=91=E7=9D=A3=E8=80=85=E8=BF=9C?= =?UTF-8?q?=E7=A8=8B=E6=A8=A1=E5=BC=8F=20TCP=20=E4=BC=A0=E8=BE=93=E4=B8=8E?= =?UTF-8?q?=20--remote-supervise=20=E6=97=97=E6=A0=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 40 +- Cargo.toml | 2 +- crates/oxide_test262/src/config.rs | 89 +++- crates/oxide_test262/src/main.rs | 3 +- crates/oxide_test262/src/supervise.rs | 195 +++++--- .../oxide_test262/src/tcp_window_transport.rs | 425 ++++++++++++++++++ crates/oxide_test262/src/window_transport.rs | 15 +- 7 files changed, 677 insertions(+), 92 deletions(-) create mode 100644 crates/oxide_test262/src/tcp_window_transport.rs diff --git a/Cargo.lock b/Cargo.lock index ba34b865..d758de61 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.15.18" +version = "0.15.19" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.15.18" +version = "0.15.19" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.15.18" +version = "0.15.19" dependencies = [ "ansi_term", "bumpalo", @@ -1228,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.15.18" +version = "0.15.19" dependencies = [ "lru", "oxide_bytecode", @@ -1237,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1257,7 +1257,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1271,7 +1271,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.15.18" +version = "0.15.19" dependencies = [ "num-bigint", "oxc_allocator", @@ -1286,7 +1286,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1298,7 +1298,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1307,7 +1307,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.15.18" +version = "0.15.19" dependencies = [ "bumpalo", "dashmap", @@ -1321,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1333,7 +1333,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.15.18" +version = "0.15.19" dependencies = [ "tracing", "tracing-appender", @@ -1342,11 +1342,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.15.18" +version = "0.15.19" [[package]] name = "oxide_parser" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1377,7 +1377,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.15.18" +version = "0.15.19" dependencies = [ "num-bigint", "num-traits", @@ -1390,7 +1390,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.15.18" +version = "0.15.19" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1410,7 +1410,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.15.18" +version = "0.15.19" dependencies = [ "bumpalo", "dashmap", @@ -1423,7 +1423,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.15.18" +version = "0.15.19" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 4bfc6033..79f7595c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.15.18" +version = "0.15.19" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_test262/src/config.rs b/crates/oxide_test262/src/config.rs index 7ac9490b..971918b9 100644 --- a/crates/oxide_test262/src/config.rs +++ b/crates/oxide_test262/src/config.rs @@ -12,6 +12,9 @@ pub(crate) struct RunConfig { pub(crate) supervise: bool, /// 远程分片 worker 模式：连监督者端点收窗口指派帧，in-process 执行并发帧。 pub(crate) worker: bool, + /// 远程监督模式：（端点文本, worker 数）。监督者 bind 端点、每窗口 accept 一条 + /// worker 连接，经 Tcp 传输执行窗口。 + pub(crate) remote_supervise: Option<(String, usize)>, /// 关闭 liveness/精确 DCE/RegAlloc 链（开关对比：关优化链 vs 开优化链各跑一遍）。 pub(crate) no_regalloc: bool, /// 逐测试打印 PASS/FAIL/SKIP（开关对比时对比两轮结果集合用）。 @@ -29,24 +32,28 @@ impl RunConfig { no_skip: false, supervise: false, worker: false, + remote_supervise: None, no_regalloc: false, verbose: false, no_fail_list: false, } } - /// 监督模式判定：`--supervise` 且非监督派生的子进程（子进程不重入监督）。 - /// 路径 filter 不参与判定——filter 在子进程内逐测试应用，不构成监督屏障。 + /// 监督模式判定：`--supervise` 或 `--remote-supervise` 且非监督派生的子进程 + /// （子进程不重入监督）。路径 filter 不参与判定——filter 在子进程内逐测试 + /// 应用，不构成监督屏障。 pub(crate) fn supervised(&self, is_chunk_child: bool) -> bool { - self.supervise && !is_chunk_child + (self.supervise || self.remote_supervise.is_some()) && !is_chunk_child } - /// 解析命令行参数为运行配置；未知选项或参数过多返回错误。 + /// 解析命令行参数为运行配置；未知选项、参数过多或 `--remote-supervise` + /// 缺实参 / worker 数非法返回错误。 pub(crate) fn parse(args: &[String]) -> Result { let mut config = Self::new(); let mut positional = Vec::new(); - - for arg in args.iter().skip(1) { + let mut i = 1; + while i < args.len() { + let arg = &args[i]; match arg.as_str() { "--no-skip" => config.no_skip = true, "--no-regalloc" => config.no_regalloc = true, @@ -54,10 +61,25 @@ impl RunConfig { "--supervise" => config.supervise = true, "--worker" => config.worker = true, "--no-fail-list" => config.no_fail_list = true, + // --remote-supervise 消费后两个实参：端点文本与 worker 数。 + "--remote-supervise" => { + let endpoint = args + .get(i + 1) + .ok_or_else(|| format!("--remote-supervise requires an endpoint\n\n{}", Self::usage()))?; + let workers_str = args + .get(i + 2) + .ok_or_else(|| format!("--remote-supervise requires a worker count\n\n{}", Self::usage()))?; + let workers = workers_str + .parse::() + .map_err(|_| format!("invalid worker count: {workers_str}\n\n{}", Self::usage()))?; + config.remote_supervise = Some((endpoint.clone(), workers)); + i += 2; + } "--help" | "-h" => return Err(Self::usage()), _ if arg.starts_with("--") => return Err(format!("unknown option: {arg}\n\n{}", Self::usage())), _ => positional.push(arg.clone()), } + i += 1; } // worker 与 supervise 互斥：前者是远程分片的执行半区（连监督者端点）， @@ -65,6 +87,13 @@ impl RunConfig { if config.worker && config.supervise { return Err(format!("--worker and --supervise are mutually exclusive\n\n{}", Self::usage())); } + // remote-supervise 是远程监督半区，与 worker（执行半区）及本地 supervise 互斥。 + if config.worker && config.remote_supervise.is_some() { + return Err(format!("--worker and --remote-supervise are mutually exclusive\n\n{}", Self::usage())); + } + if config.supervise && config.remote_supervise.is_some() { + return Err(format!("--supervise and --remote-supervise are mutually exclusive\n\n{}", Self::usage())); + } if let Some(root) = positional.first() { config.test262_root = Some(PathBuf::from(root)); @@ -81,7 +110,7 @@ impl RunConfig { /// 打印用法说明。 pub(crate) fn usage() -> String { - "usage: test262-runner [--no-skip] [--no-regalloc] [--verbose] [--supervise] [--worker] [--version] [test262-root] [path-filter]\n\ + "usage: test262-runner [--no-skip] [--no-regalloc] [--verbose] [--supervise] [--worker] [--remote-supervise ] [--version] [test262-root] [path-filter]\n\ \n\ --no-skip Run capability-excluded tests and count unsupported compile/runtime results as failures.\n\ --no-regalloc Disable the liveness/precise-DCE/RegAlloc compiler chain (vregs stay as physical numbers).\n\ @@ -94,6 +123,11 @@ impl RunConfig { --worker Run as a remote shard worker: connect to the supervisor endpoint and execute assigned\n\ \x20 test windows in-process, sending heartbeat/fail/scene/DONE frames over the connection.\n\ \x20 Mutually exclusive with --supervise.\n\ + --remote-supervise Run as a remote supervisor: bind (host:port) and execute test\n\ + \x20 windows on remote workers, sending assign frames and receiving\n\ + \x20 heartbeat/fail/scene/DONE frames over the accepted connections. A worker\n\ + \x20 disconnect is recovered as a child-process death. Mutually exclusive with\n\ + \x20 --supervise and --worker.\n\ \n\ env tunables (all optional):\n\ \x20 OXIDE_LOG per-subsystem log levels (default off)\n\ @@ -163,6 +197,47 @@ mod tests { assert!(err.contains("mutually exclusive"), "应提示互斥：{err}"); } + /// --remote-supervise 正常解析端点与 worker 数；缺实参、worker 数非法返回错误。 + #[test] + fn parse_remote_supervise_flag() { + let cfg = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "4"])) + .expect("remote-supervise 应解析成功"); + assert_eq!(cfg.remote_supervise, Some(("127.0.0.1:8080".to_string(), 4))); + assert!(!cfg.supervise, "remote-supervise 不应置位 supervise"); + assert!(!cfg.worker, "remote-supervise 不应置位 worker"); + + let err = RunConfig::parse(&args(&["--remote-supervise"])).expect_err("缺端点应报错"); + assert!(err.contains("--remote-supervise requires an endpoint"), "应提示缺端点：{err}"); + + let err = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080"])).expect_err("缺 worker 数应报错"); + assert!(err.contains("--remote-supervise requires a worker count"), "应提示缺 worker 数：{err}"); + + let err = + RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "abc"])).expect_err("worker 数非法应报错"); + assert!(err.contains("invalid worker count"), "应提示 worker 数非法：{err}"); + } + + /// --remote-supervise 与 --worker、--supervise 互斥。 + #[test] + fn parse_remote_supervise_exclusive() { + let err = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "4", "--worker"])) + .expect_err("worker+remote-supervise 应互斥"); + assert!(err.contains("mutually exclusive"), "应提示互斥：{err}"); + + let err = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "4", "--supervise"])) + .expect_err("supervise+remote-supervise 应互斥"); + assert!(err.contains("mutually exclusive"), "应提示互斥：{err}"); + } + + /// remote_supervise 置位也触发 supervised()（非子进程）；子进程不重入。 + #[test] + fn supervised_predicate_with_remote_supervise() { + let mut cfg = RunConfig::new(); + cfg.remote_supervise = Some(("127.0.0.1:8080".to_string(), 4)); + assert!(cfg.supervised(false), "remote_supervise 应触发 supervised"); + assert!(!cfg.supervised(true), "子进程不应重入 supervised"); + } + /// 未知选项返回含 "unknown option" 的错误；--help 的错误文案与 usage 逐字相等。 #[test] fn parse_unknown_option_and_help_err_text() { diff --git a/crates/oxide_test262/src/main.rs b/crates/oxide_test262/src/main.rs index a638de0a..f77c7903 100644 --- a/crates/oxide_test262/src/main.rs +++ b/crates/oxide_test262/src/main.rs @@ -16,6 +16,7 @@ mod runner; mod shard_protocol; mod stats; mod supervise; +mod tcp_window_transport; mod test262_log; mod window_transport; mod worker; @@ -311,7 +312,7 @@ fn run_tests() -> bool { if config.supervised(is_chunk_child) { test262_info!("supervised mode enabled"); eprintln!("supervised mode enabled: child-window execution with per-test timeout + auto-resume"); - return run_supervised(&args, skip_until, end_index, config.no_skip, &paths); + return run_supervised(&args, skip_until, end_index, config.no_skip, &paths, config.remote_supervise.clone()); } if let Some(chunk_size) = chunk_size { diff --git a/crates/oxide_test262/src/supervise.rs b/crates/oxide_test262/src/supervise.rs index cbc4d3de..bb355177 100644 --- a/crates/oxide_test262/src/supervise.rs +++ b/crates/oxide_test262/src/supervise.rs @@ -9,9 +9,12 @@ use crate::heartbeat::merge_heartbeat; use crate::print_fail_categories; use crate::report::{format_fail_groupings, format_fail_list}; use crate::stats::RunStats; +use crate::tcp_window_transport::TcpWindowTransport; use crate::window_transport::{LocalFileWindowTransport, PollResult, WindowTransport}; +use oxide_net::Listener; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; use std::time::{Duration, Instant}; /// 超时/崩溃后把 sidecar 日志的末 ~20 行打到父进程 stderr，供现场重建。 @@ -186,6 +189,7 @@ fn supervise_window( /// 从不持有 `KernelCore`，因此按引用共享 `paths`/`args` 是安全的。 pub(crate) fn run_supervised( args: &[String], skip_until: usize, end_index: usize, no_skip: bool, paths: &[PathBuf], + remote_supervise: Option<(String, usize)>, ) -> bool { let exe = match std::env::current_exe() { Ok(path) => path, @@ -195,14 +199,19 @@ pub(crate) fn run_supervised( } }; - // 路径清单文件：派生窗口前写一次（每行一路径，与父扫描同序），跨窗口只读 - // 共享；子进程读文件代替全量重扫目录。运行结束删除，防临时目录残留。 - let paths_file = std::env::temp_dir().join(format!("oxide_t262_paths_{}.txt", std::process::id())); - let list = paths.iter().map(|p| p.to_string_lossy()).collect::>().join("\n"); - if let Err(err) = std::fs::write(&paths_file, list) { - eprintln!("failed to write paths file {}: {err}", paths_file.display()); - return false; - } + // 路径清单文件：仅本地模式需要（子进程读文件代替全量重扫目录）；远程模式 + // 路径随 assign 帧携带，不写文件。运行结束删除，防临时目录残留。 + let paths_file: Option = if remote_supervise.is_none() { + let pf = std::env::temp_dir().join(format!("oxide_t262_paths_{}.txt", std::process::id())); + let list = paths.iter().map(|p| p.to_string_lossy()).collect::>().join("\n"); + if let Err(err) = std::fs::write(&pf, list) { + eprintln!("failed to write paths file {}: {err}", pf.display()); + return false; + } + Some(pf) + } else { + None + }; let window = std::env::var("OXIDE_TEST262_WINDOW") .ok() @@ -234,68 +243,140 @@ pub(crate) fn run_supervised( id += 1; } - // 缺省 16：全量复测验证过的安全并发；未经复测不得抬高。 - let supervisors = std::env::var("OXIDE_TEST262_SUPERVISORS") - .ok() - .and_then(|s| s.parse::().ok()) - .filter(|&n| n > 0) - .unwrap_or(16) - .min(windows.len().max(1)); - - eprintln!( - "supervised mode: {} window(s) of up to {window} test(s), {supervisors} supervisor(s), per-test timeout {}s", - windows.len(), - timeout.as_secs() - ); + // 缺省 16：全量复测验证过的安全并发；未经复测不得抬高。远程模式用旗标 worker 数。 + let (supervisors, endpoint) = match &remote_supervise { + Some((endpoint, workers)) => ((*workers).min(windows.len().max(1)), Some(endpoint.clone())), + None => ( + std::env::var("OXIDE_TEST262_SUPERVISORS") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|&n| n > 0) + .unwrap_or(16) + .min(windows.len().max(1)), + None, + ), + }; let next = AtomicUsize::new(0); let next = &next; let windows = &windows; let exe = &exe; - let paths_file = &paths_file; - let partials: Vec = std::thread::scope(|scope| { - let handles: Vec<_> = (0..supervisors) - .map(|_| { - scope.spawn(move || { - let mut stats = RunStats::default(); - loop { - let wi = next.fetch_add(1, Ordering::Relaxed); - if wi >= windows.len() { - break; - } - let (window_id, wstart, wend) = windows[wi]; - // 每窗口一个传输实例（跨该窗口内子进程重启存活）， - // 已消费偏移等窗口级状态随之保留。 - let window_stats = supervise_window( - no_skip, - wstart, - wend, - timeout, - startup_grace, - paths, - window_id, - &mut LocalFileWindowTransport::new(exe, args, paths_file, window_id), - ); - stats.merge(window_stats); - } - stats - }) + let partials: Vec = match &endpoint { + Some(endpoint) => { + // 远程分支：bind 监听端点，每窗口 accept 一条 worker 连接，worker 数控制 + // 并发线程数。路径随 assign 帧携带，不写 paths_file。 + let listener = match std::net::TcpListener::bind(endpoint) { + Ok(l) => Arc::new(Listener::Tcp(l)), + Err(err) => { + eprintln!("failed to bind remote supervisor endpoint {endpoint}: {err}"); + return false; + } + }; + let bound = listener.local_addr().unwrap_or_else(|_| endpoint.clone()); + eprintln!( + "remote supervised mode: {} window(s) of up to {window} test(s), {supervisors} worker(s), endpoint {bound}, per-test timeout {}s", + windows.len(), + timeout.as_secs() + ); + // 路径清单经 Arc 跨监督线程共享（每窗口克隆一份随 assign 帧携带）。 + let paths_vec: Arc> = + Arc::new(paths.iter().map(|p| p.to_string_lossy().into_owned()).collect()); + let timeout_secs = timeout.as_secs(); + let listener_ref = &listener; + let paths_vec_ref = &paths_vec; + std::thread::scope(|scope| { + let handles: Vec<_> = (0..supervisors) + .map(|_| { + scope.spawn(move || { + let mut stats = RunStats::default(); + loop { + let wi = next.fetch_add(1, Ordering::Relaxed); + if wi >= windows.len() { + break; + } + let (window_id, wstart, wend) = windows[wi]; + // 每窗口一个传输实例（跨该窗口内 worker 重启存活）。 + let window_stats = supervise_window( + no_skip, + wstart, + wend, + timeout, + startup_grace, + paths, + window_id, + &mut TcpWindowTransport::new( + Arc::clone(listener_ref), + (**paths_vec_ref).clone(), + timeout_secs, + ), + ); + stats.merge(window_stats); + } + stats + }) + }) + .collect(); + handles + .into_iter() + .map(|h| h.join().expect("supervisor thread panicked")) + .collect() }) - .collect(); - handles - .into_iter() - .map(|h| h.join().expect("supervisor thread panicked")) - .collect() - }); + } + None => { + // 本地分支：现状不变（写 paths_file、LocalFileWindowTransport、缺省 16）。 + let paths_file = paths_file.as_ref().expect("本地模式已写 paths_file"); + eprintln!( + "supervised mode: {} window(s) of up to {window} test(s), {supervisors} supervisor(s), per-test timeout {}s", + windows.len(), + timeout.as_secs() + ); + std::thread::scope(|scope| { + let handles: Vec<_> = (0..supervisors) + .map(|_| { + scope.spawn(move || { + let mut stats = RunStats::default(); + loop { + let wi = next.fetch_add(1, Ordering::Relaxed); + if wi >= windows.len() { + break; + } + let (window_id, wstart, wend) = windows[wi]; + // 每窗口一个传输实例（跨该窗口内子进程重启存活）， + // 已消费偏移等窗口级状态随之保留。 + let window_stats = supervise_window( + no_skip, + wstart, + wend, + timeout, + startup_grace, + paths, + window_id, + &mut LocalFileWindowTransport::new(exe, args, paths_file, window_id), + ); + stats.merge(window_stats); + } + stats + }) + }) + .collect(); + handles + .into_iter() + .map(|h| h.join().expect("supervisor thread panicked")) + .collect() + }) + } + }; let mut stats = RunStats::default(); for partial_stats in partials { stats.merge(partial_stats); } - // 全部窗口结束、子进程已读完清单，删除路径清单文件防临时目录残留。 - let _ = std::fs::remove_file(paths_file); + // 全部窗口结束、子进程已读完清单，删除路径清单文件防临时目录残留（远程模式无文件）。 + if let Some(pf) = &paths_file { + let _ = std::fs::remove_file(pf); + } let total = stats.pass + stats.fail + stats.skip; println!(); diff --git a/crates/oxide_test262/src/tcp_window_transport.rs b/crates/oxide_test262/src/tcp_window_transport.rs new file mode 100644 index 00000000..33faf99a --- /dev/null +++ b/crates/oxide_test262/src/tcp_window_transport.rs @@ -0,0 +1,425 @@ +//! 远程监督窗口传输：`WindowTransport` 的 Tcp 臂。监督者 `accept` 一条 worker +//! 连接，经该连接发 assign 帧、收事件帧；断连（EOF / 真实 io 错误 / 畸形帧） +//! 映射为 `PollResult::Exited`（合成退出码），走与本地子进程死亡完全同构的恢复 +//! 路径。 +//! +//! 写读侧分离：`writer` 是 `try_clone` 出的写侧句柄，`reader` 包住底层连接读侧 +//! （含余量缓冲）。`kill` 后须 drop 两侧句柄真正断连，否则连接半开。 +//! +//! 计数合并契约：监督者每窗口只合并一次最终心跳（DONE，带全窗累计计数）； +//! 中间 COMPLETED 心跳只用于超时跟踪（last_index），不合并计数。`Done` 帧把 +//! `last_heartbeat` 的 phase 改 DONE 并保留计数；worker 发 Done 前无 Heartbeat +//! 时合成零计数 DONE。 +//! +//! 非阻塞读陷阱：`set_read_timeout(0)` 后 `WouldBlock` 必须判「无数据、仍存活」 +//! 而非断连，否则正常窗口被误杀重启；仅真实 io 错误与 EOF 判断连。残行留在 +//! `FrameReader` 余量缓冲，跨 poll 不丢。 + +use std::collections::HashMap; +use std::io::BufReader; +use std::io::ErrorKind; +use std::sync::Arc; + +use crate::heartbeat::Heartbeat; +use crate::shard_protocol::{encode_command, parse_event, FrameReader, ShardCommand, ShardEvent}; +use crate::stats::RunStats; +use crate::window_transport::{PollResult, SpawnError, WindowTransport}; +use oxide_net::{Conn, Listener}; + +/// 远程监督窗口传输：`WindowTransport` 的 Tcp 臂。`listener` 供跨子进程重启 +/// 重新 `accept`；`writer` / `reader` 是当前 worker 连接的写读两侧句柄。 +pub(crate) struct TcpWindowTransport { + listener: Arc, + /// 写侧句柄（`try_clone` 出，与读侧共享底层套接字）。 + writer: Option, + /// 读侧（含余量缓冲，跨 poll 保留残行）。 + reader: Option>>, + /// 最近一次心跳帧（中间 COMPLETED 只跟踪不合并；Done 帧改 phase 为 DONE）。 + last_heartbeat: Option, + /// 待并入的失败记录帧（一测试失败一帧，按序消费，无偏移态，天然无双计）。 + pending_fails: Vec<(usize, String, String, String)>, + /// 最近一次现场帧（last-pc 现场行加可选日志尾）。 + last_scene: Option<(String, Option)>, + /// 全量路径文本（assign 时切窗口片随帧携带，worker 读不到监督者本机文件）。 + paths: Vec, + /// 单测试超时，随 assign 帧携带供 worker 侧看门狗使用。 + timeout_secs: u64, +} + +impl TcpWindowTransport { + /// 构造传输：`listener` 是监督者绑定端点后的监听句柄（跨重启 `accept` 用）， + /// `paths` 是全量路径文本，`timeout_secs` 随 assign 帧携带。 + pub(crate) fn new(listener: Arc, paths: Vec, timeout_secs: u64) -> Self { + Self { + listener, + writer: None, + reader: None, + last_heartbeat: None, + pending_fails: Vec::new(), + last_scene: None, + paths, + timeout_secs, + } + } + + /// 写读两侧句柄都 drop，真正断连（防连接半开）。 + fn drop_conn(&mut self) { + self.writer = None; + self.reader = None; + } + + /// Done 帧处理：把 `last_heartbeat` 的 phase 改 DONE 并保留计数（每窗口只 + /// 合并一次的计数契约），随后断连判死亡。 + fn mark_done(&mut self) { + match self.last_heartbeat.as_mut() { + Some(hb) => hb.phase = "DONE".to_string(), + None => { + // worker 发 Done 前无 Heartbeat：合成零计数 DONE。 + self.last_heartbeat = Some(Heartbeat { + phase: "DONE".to_string(), + index: 0, + pass: 0, + fail: 0, + skip: 0, + categories: HashMap::new(), + hb_write_errors: 0, + }); + } + } + self.drop_conn(); + } +} + +impl WindowTransport for TcpWindowTransport { + fn assign(&mut self, wstart: usize, wend: usize) -> Result<(), SpawnError> { + // 无连接时阻塞 `accept` 取一条 worker 连接（io 错误包成 SpawnError）。 + if self.writer.is_none() { + let base = self.listener.accept().map_err(SpawnError)?; + // 读侧置非阻塞：std 拒绝 0 时长读超时，改用 `set_nonblocking` 使无数据 + // 读返回 `WouldBlock`（判「无数据、仍存活」，不误杀正常窗口）。 + match &base { + Conn::Tcp(s) => s.set_nonblocking(true).map_err(SpawnError)?, + Conn::Unix(s) => s.set_nonblocking(true).map_err(SpawnError)?, + } + let writer = base.try_clone().map_err(SpawnError)?; + let reader = FrameReader::new(BufReader::new(base)); + self.writer = Some(writer); + self.reader = Some(reader); + } + + // 发 assign 帧：窗口路径片随帧携带（worker 读不到监督者本机临时文件）。 + let window_paths = self.paths[wstart..wend].to_vec(); + let cmd = ShardCommand::Assign { + wstart, + wend, + paths: window_paths, + timeout_secs: self.timeout_secs, + }; + let frame = encode_command(&cmd); + let writer = self.writer.as_mut().expect("assign 刚建立写侧句柄"); + writer.write_all(frame.as_bytes()).map_err(SpawnError)?; + Ok(()) + } + + fn heartbeat(&self) -> Option { + // `Heartbeat` 未派生 `Clone`（心跳模块零改动），逐字段手工克隆。 + self.last_heartbeat.as_ref().map(|hb| Heartbeat { + phase: hb.phase.clone(), + index: hb.index, + pass: hb.pass, + fail: hb.fail, + skip: hb.skip, + categories: hb.categories.clone(), + hb_write_errors: hb.hb_write_errors, + }) + } + + fn merge_fail(&mut self, stats: &mut RunStats) { + // 排空待并入失败帧逐条重建 FailRecord（帧按序消费，无偏移态，无双计）。 + for (index, category, subkey, message) in self.pending_fails.drain(..) { + stats.push_fail_record(index, category, subkey, message); + } + } + + fn scene(&self) -> Option { + self.last_scene.as_ref().map(|(line, _)| line.clone()) + } + + fn kill(&mut self) { + // 发 kill 帧后 drop 写读两侧句柄真正断连（否则连接半开）。 + if let Some(writer) = self.writer.as_mut() { + let frame = encode_command(&ShardCommand::Kill); + let _ = writer.write_all(frame.as_bytes()); + } + self.drop_conn(); + } + + fn poll(&mut self) -> PollResult { + // 先取帧（读侧借用在此结束，帧是拥有值），再改状态，避免借用冲突。 + let frame_result = match self.reader.as_mut() { + Some(reader) => reader.read_frame(), + None => return PollResult::Alive, + }; + match frame_result { + Ok(Some(frame)) => match parse_event(&frame) { + Ok(ShardEvent::Heartbeat { + phase, + index, + pass, + fail, + skip, + categories, + hb_write_errors, + }) => { + // 中间心跳只跟踪不合并计数（合并契约：每窗口只合并一次 DONE）。 + self.last_heartbeat = Some(Heartbeat { + phase, + index, + pass, + fail, + skip, + categories, + hb_write_errors, + }); + } + Ok(ShardEvent::Fail { + index, + category, + subkey, + message, + }) => { + self.pending_fails.push((index, category, subkey, message)); + } + Ok(ShardEvent::Scene { line, log_tail }) => { + self.last_scene = Some((line, log_tail)); + } + Ok(ShardEvent::Done) => { + // 窗口结束：改末次心跳 phase 为 DONE 保留计数，断连判死亡 + // （与本地子进程退出同构）。 + self.mark_done(); + return PollResult::Exited(1); + } + Err(_) => { + // 畸形帧：对端状态不定，判断连（同 EOF 同构恢复）。 + self.drop_conn(); + return PollResult::Exited(1); + } + }, + Ok(None) => { + // EOF：对端关闭，判死亡。 + self.drop_conn(); + return PollResult::Exited(1); + } + Err(e) if e.kind() == ErrorKind::WouldBlock => { + // 非阻塞读无数据：仍存活，下一拍重试（不得误判断连）。 + return PollResult::Alive; + } + Err(_) => { + // 真实 io 错误：判断连。 + self.drop_conn(); + return PollResult::Exited(1); + } + } + PollResult::Alive + } + + fn log_tail(&self) -> Option { + self.last_scene.as_ref().and_then(|(_, log_tail)| log_tail.clone()) + } + + fn reset_round(&mut self) { + // 清事件缓冲（心跳 / 待并入失败帧 / 现场归零），跨重启不保留。 + self.last_heartbeat = None; + self.pending_fails.clear(); + self.last_scene = None; + } + + fn archive_window(&mut self) { + // 远程无文件归档：失败记录已并入 stats，无数据丢失；drop 连接即收尾。 + self.drop_conn(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use std::net::TcpStream; + use std::time::{Duration, Instant}; + + use crate::shard_protocol::encode_event; + + /// TcpWindowTransport 回环：bind 动态端口，客户端连接后发 Heartbeat / Fail / + /// Scene 帧，断言 heartbeat / scene / log_tail / merge_fail 读回正确； + /// 客户端断连后 poll 返回 Exited（合成码）。 + #[test] + fn tcp_transport_loopback() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind 失败"); + let addr = listener.local_addr().expect("local_addr 失败"); + let listener = Arc::new(Listener::Tcp(listener)); + + let mut transport = + TcpWindowTransport::new(Arc::clone(&listener), vec!["a/b.js".to_string(), "c/d.js".to_string()], 10); + + // assign：accept 客户端连接并发 assign 帧。 + let mut client = TcpStream::connect(addr).expect("客户端连接失败"); + transport.assign(0, 2).expect("assign 应成功"); + + // 发 Heartbeat / Fail / Scene 帧（按序，一测试失败一帧）。 + client + .write_all( + encode_event(&ShardEvent::Heartbeat { + phase: "COMPLETED".into(), + index: 0, + pass: 1, + fail: 0, + skip: 0, + categories: HashMap::new(), + hb_write_errors: 0, + }) + .as_bytes(), + ) + .expect("写心跳帧失败"); + client + .write_all( + encode_event(&ShardEvent::Fail { + index: 0, + category: "vm: not callable".into(), + subkey: "x".into(), + message: "x is not callable".into(), + }) + .as_bytes(), + ) + .expect("写失败帧失败"); + client + .write_all( + encode_event(&ShardEvent::Scene { + line: "pc=1 op=ADD frames=1".into(), + log_tail: Some("log line".into()), + }) + .as_bytes(), + ) + .expect("写现场帧失败"); + + // 读帧：poll 逐帧消费，直到 Scene（末帧）读回。 + let deadline = Instant::now() + Duration::from_secs(5); + while transport.scene().is_none() && Instant::now() < deadline { + match transport.poll() { + PollResult::Exited(_) | PollResult::WaitError(_) => break, + PollResult::Alive => {} + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!(transport.scene().is_some(), "现场帧应读回"); + let hb = transport.heartbeat().expect("心跳帧应读回"); + assert_eq!(hb.phase, "COMPLETED"); + assert_eq!(hb.index, 0); + assert_eq!(hb.pass, 1); + assert_eq!(transport.scene(), Some("pc=1 op=ADD frames=1".to_string())); + assert_eq!(transport.log_tail(), Some("log line".to_string())); + + // merge_fail 排空待并入失败帧；二次合并不重复（无偏移态，排空即空）。 + let mut stats = RunStats::default(); + transport.merge_fail(&mut stats); + assert_eq!(stats.fail_records.len(), 1, "应并入一条失败记录"); + assert_eq!(stats.fail_records[0].index, 0); + transport.merge_fail(&mut stats); + assert_eq!(stats.fail_records.len(), 1, "二次合并不应重复"); + + // 客户端断连：poll 返回 Exited（合成码）。 + drop(client); + let mut exited = false; + let deadline = Instant::now() + Duration::from_secs(5); + while !exited && Instant::now() < deadline { + if matches!(transport.poll(), PollResult::Exited(_)) { + exited = true; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!(exited, "断连后 poll 应返回 Exited"); + } + + /// Done 帧：把末次心跳 phase 改 DONE 保留计数，poll 返回 Exited；worker 发 + /// Done 前无 Heartbeat 时合成零计数 DONE。 + #[test] + fn tcp_transport_done_frame_keeps_counts() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind 失败"); + let addr = listener.local_addr().expect("local_addr 失败"); + let listener = Arc::new(Listener::Tcp(listener)); + + let mut transport = TcpWindowTransport::new(Arc::clone(&listener), vec!["a/b.js".to_string()], 10); + let mut client = TcpStream::connect(addr).expect("客户端连接失败"); + transport.assign(0, 1).expect("assign 应成功"); + + // 先发一条带计数的心跳，再发 Done：Done 应把 phase 改 DONE 并保留计数。 + client + .write_all( + encode_event(&ShardEvent::Heartbeat { + phase: "COMPLETED".into(), + index: 0, + pass: 5, + fail: 2, + skip: 1, + categories: HashMap::new(), + hb_write_errors: 0, + }) + .as_bytes(), + ) + .expect("写心跳帧失败"); + let deadline = Instant::now() + Duration::from_secs(5); + while transport.heartbeat().is_none() && Instant::now() < deadline { + let _ = transport.poll(); + std::thread::sleep(Duration::from_millis(10)); + } + client + .write_all(encode_event(&ShardEvent::Done).as_bytes()) + .expect("写 Done 帧失败"); + + // poll 读到 Done 帧应返回 Exited，且心跳 phase 改 DONE、计数保留。 + let mut exited = false; + let deadline = Instant::now() + Duration::from_secs(5); + while !exited && Instant::now() < deadline { + if matches!(transport.poll(), PollResult::Exited(_)) { + exited = true; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!(exited, "Done 帧后 poll 应返回 Exited"); + let hb = transport.heartbeat().expect("Done 后心跳应可读回"); + assert_eq!(hb.phase, "DONE", "Done 帧应把 phase 改 DONE"); + assert_eq!(hb.pass, 5, "Done 帧应保留末次心跳计数"); + assert_eq!(hb.fail, 2); + assert_eq!(hb.skip, 1); + } + + /// Done 帧前无 Heartbeat：合成零计数 DONE（不 panic、计数全零）。 + #[test] + fn tcp_transport_done_without_heartbeat_synthesizes_zero() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind 失败"); + let addr = listener.local_addr().expect("local_addr 失败"); + let listener = Arc::new(Listener::Tcp(listener)); + + let mut transport = TcpWindowTransport::new(Arc::clone(&listener), vec!["a/b.js".to_string()], 10); + let mut client = TcpStream::connect(addr).expect("客户端连接失败"); + transport.assign(0, 1).expect("assign 应成功"); + + client + .write_all(encode_event(&ShardEvent::Done).as_bytes()) + .expect("写 Done 帧失败"); + let mut exited = false; + let deadline = Instant::now() + Duration::from_secs(5); + while !exited && Instant::now() < deadline { + if matches!(transport.poll(), PollResult::Exited(_)) { + exited = true; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!(exited, "Done 帧后 poll 应返回 Exited"); + let hb = transport.heartbeat().expect("无心跳 Done 应合成零计数 DONE"); + assert_eq!(hb.phase, "DONE"); + assert_eq!(hb.pass, 0); + assert_eq!(hb.fail, 0); + assert_eq!(hb.skip, 0); + assert!(hb.categories.is_empty()); + } +} diff --git a/crates/oxide_test262/src/window_transport.rs b/crates/oxide_test262/src/window_transport.rs index 6e35c173..e618913c 100644 --- a/crates/oxide_test262/src/window_transport.rs +++ b/crates/oxide_test262/src/window_transport.rs @@ -10,10 +10,11 @@ use crate::heartbeat::{read_heartbeat, Heartbeat}; use crate::report::parse_fail_log; use crate::stats::RunStats; use std::path::{Path, PathBuf}; -use std::process::{Child, Command, ExitStatus}; +use std::process::{Child, Command}; /// 派生窗口子进程失败：spawn 的 io 错误。 -pub(crate) struct SpawnError(std::io::Error); +#[derive(Debug)] +pub(crate) struct SpawnError(pub(crate) std::io::Error); impl std::fmt::Display for SpawnError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -21,11 +22,12 @@ impl std::fmt::Display for SpawnError { } } -/// 子进程等待三态：死亡（带退出状态）/ 存活 / 等待错误。 +/// 子进程等待三态：死亡（带退出码）/ 存活 / 等待错误。 /// 三臂在监督策略里处理不同（死亡走心跳裁决、存活继续轮询、错误走 kill 加 -/// 记账），不得压扁。 +/// 记账），不得压扁。退出码仅用于打印（`ExitStatus` 无公开构造器，远程断连 +/// 合成码 1 走同一打印路径，不参与逻辑分支）。 pub(crate) enum PollResult { - Exited(ExitStatus), + Exited(usize), Alive, WaitError(std::io::Error), } @@ -150,7 +152,8 @@ impl WindowTransport for LocalFileWindowTransport { return PollResult::Alive; }; match child.try_wait() { - Ok(Some(status)) => PollResult::Exited(status), + // 退出码仅用于打印：正常退出取 code，信号杀死取缺省 1。 + Ok(Some(status)) => PollResult::Exited(status.code().unwrap_or(1) as usize), Ok(None) => PollResult::Alive, Err(err) => PollResult::WaitError(err), }