From 2c8815f17834c0c05a032454f1e7f1984887c968 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Sun, 11 Oct 2026 17:07:23 +0000 Subject: [PATCH] =?UTF-8?q?refactor:=20=E7=9B=91=E7=9D=A3=E7=AA=97?= =?UTF-8?q?=E5=8F=A3=E6=96=87=E4=BB=B6=E5=8D=8F=E8=AE=AE=E6=8A=BD=E8=B1=A1?= =?UTF-8?q?=E4=B8=BA=E5=85=AD=E6=93=8D=E4=BD=9C=E4=BC=A0=E8=BE=93=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3?= 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/main.rs | 1 + crates/oxide_test262/src/supervise.rs | 293 +++-------------- crates/oxide_test262/src/window_transport.rs | 326 +++++++++++++++++++ 5 files changed, 402 insertions(+), 260 deletions(-) create mode 100644 crates/oxide_test262/src/window_transport.rs diff --git a/Cargo.lock b/Cargo.lock index 0d5841c3..209d7636 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.15.15" +version = "0.15.16" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.15.15" +version = "0.15.16" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.15.15" +version = "0.15.16" dependencies = [ "ansi_term", "bumpalo", @@ -1228,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.15.15" +version = "0.15.16" dependencies = [ "lru", "oxide_bytecode", @@ -1237,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1257,7 +1257,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1271,7 +1271,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.15.15" +version = "0.15.16" dependencies = [ "num-bigint", "oxc_allocator", @@ -1286,7 +1286,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1298,7 +1298,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1307,7 +1307,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.15.15" +version = "0.15.16" dependencies = [ "bumpalo", "dashmap", @@ -1321,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1333,7 +1333,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.15.15" +version = "0.15.16" dependencies = [ "tracing", "tracing-appender", @@ -1342,11 +1342,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.15.15" +version = "0.15.16" [[package]] name = "oxide_parser" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1377,7 +1377,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.15.15" +version = "0.15.16" dependencies = [ "num-bigint", "num-traits", @@ -1390,7 +1390,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.15.15" +version = "0.15.16" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1408,7 +1408,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.15.15" +version = "0.15.16" dependencies = [ "bumpalo", "dashmap", @@ -1421,7 +1421,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.15.15" +version = "0.15.16" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 7fe01bd3..96211a21 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.15.15" +version = "0.15.16" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_test262/src/main.rs b/crates/oxide_test262/src/main.rs index 66a5c96d..1d0af95e 100644 --- a/crates/oxide_test262/src/main.rs +++ b/crates/oxide_test262/src/main.rs @@ -17,6 +17,7 @@ mod runner; mod stats; mod supervise; mod test262_log; +mod window_transport; use chunked::run_chunked; use config::RunConfig; use discovery::discover_tests; diff --git a/crates/oxide_test262/src/supervise.rs b/crates/oxide_test262/src/supervise.rs index 54752b84..cbc4d3de 100644 --- a/crates/oxide_test262/src/supervise.rs +++ b/crates/oxide_test262/src/supervise.rs @@ -1,67 +1,22 @@ //! 监督模式子进程监督与聚合：把测试区间切窗口，监督线程并行派生/轮询/杀死单 worker //! 子进程；心跳并入累计统计，旁路失败行按已消费偏移并入；超时/崩溃杀肇事者自其后续跑（默认计 skip，--no-skip 计 fail）。 +//! +//! 文件协议（心跳/旁路失败行/last-pc 现场/sidecar 日志/子进程派生与杀死）收口在 +//! window_transport 的 WindowTransport 传输接口，本模块只保留监督策略：超时判定、 +//! 重启循环、记账。 -use crate::heartbeat::{merge_heartbeat, read_heartbeat}; +use crate::heartbeat::merge_heartbeat; use crate::print_fail_categories; -use crate::report::{format_fail_groupings, format_fail_list, parse_fail_log}; +use crate::report::{format_fail_groupings, format_fail_list}; use crate::stats::RunStats; -use std::path::{Path, PathBuf}; -use std::process::Command; +use crate::window_transport::{LocalFileWindowTransport, PollResult, WindowTransport}; +use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant}; -/// 把 supervise 旁路失败行文件并入统计：逐行重建 FailRecord（类别经本表 id -/// 重映射 + push_fail_record 的 cap 逻辑）。 -/// -/// # 边界与前提 -/// - 失败计数（fail_categories）已由对应心跳合并覆盖，此处只补 fail_records， -/// 不得再累计计数，避免双计。worker 循环先写心跳后追加旁路，故"无心跳退出" -/// 分支旁路必为空（防御性 no-op）。 -/// - 旁路文件跨子进程重启追加保留，本函数只解析自上次消费以来的新增段 -/// （`consumed` 由调用方按窗口持有）；文件缩至已消费偏移之下（删除/重建） -/// 回绕到文件头重读，避免丢记录。 -fn merge_fail_log(stats: &mut RunStats, hb_path: &Path, consumed: &mut usize) { - let content = match std::fs::read_to_string(hb_path.with_extension("fails")) { - Ok(c) => c, - Err(_) => return, - }; - if content.len() < *consumed { - *consumed = 0; - } - let fresh = &content[*consumed..]; - *consumed = content.len(); - for (index, cat, subkey, message) in parse_fail_log(fresh) { - stats.push_fail_record(index, cat, subkey, message); - } -} - -/// 读子进程被杀前留下的 last-pc 现场行：取文件末行，须含 pc/op/frames 三字段 -/// 才接受（缺文件 / 空文件 / 残行均返回 None）。 -fn read_pc_scene(path: &Path) -> Option { - let content = std::fs::read_to_string(path).ok()?; - let last = content.lines().last()?.trim(); - if last.starts_with("pc=") && last.contains(" op=") && last.contains(" frames=") { - Some(last.to_string()) - } else { - None - } -} - -/// 读 sidecar 日志的末 `n` 行：文件缺失/空返回 None，有内容时返回整文件或 -/// 末 N 行（超时/崩溃后供现场重建）。 -fn tail_lines(path: &Path, n: usize) -> Option { - let content = std::fs::read_to_string(path).ok()?; - let lines: Vec<&str> = content.lines().collect(); - let start = lines.len().saturating_sub(n); - if start == lines.len() { - return None; - } - Some(lines[start..].join("\n")) -} - /// 超时/崩溃后把 sidecar 日志的末 ~20 行打到父进程 stderr，供现场重建。 -fn print_log_tail(window_id: usize, path: &Path) { - if let Some(tail) = tail_lines(path, 20) { +fn print_log_tail(window_id: usize, transport: &dyn WindowTransport) { + if let Some(tail) = transport.log_tail() { eprintln!(" [log tail] window {window_id}:\n{tail}"); } } @@ -93,23 +48,18 @@ fn record_timeout_or_crash(stats: &mut RunStats, no_skip: bool, index: usize, el /// 读取旁路失败行并入 fail_records。超时/崩溃默认计为 skip，`--no-skip` 下计为失败。 /// /// # 注意事项 -/// - `paths_file` 是父进程写出的全量路径清单文件（每行一路径，与父扫描同序）， -/// 经 `OXIDE_TEST262_PATHS_FILE` 环境变量传给子进程，子进程读文件代替全量重扫 -/// 目录；文件在窗口内子进程多次重启间只读共享，不得按 spawn 删除。 +/// - `transport` 是窗口级传输实例（跨子进程重启存活），六操作与文件生命周期 +/// 都经它调用；路径清单文件由传输持有并经 `OXIDE_TEST262_PATHS_FILE` 传给 +/// 子进程，在窗口内子进程多次重启间只读共享，不得按 spawn 删除。 +/// - `paths` 是父进程的全量路径清单（每行一路径，与父扫描同序），仅供肇事者 +/// 路径描述。 #[expect(clippy::too_many_arguments)] fn supervise_window( - exe: &Path, args: &[String], no_skip: bool, wstart: usize, wend: usize, timeout: Duration, startup_grace: Duration, - paths: &[PathBuf], paths_file: &Path, window_id: usize, + no_skip: bool, wstart: usize, wend: usize, timeout: Duration, startup_grace: Duration, paths: &[PathBuf], + window_id: usize, transport: &mut dyn WindowTransport, ) -> RunStats { let mut stats = RunStats::default(); let mut cur = wstart; - // 旁路失败行已消费字节偏移：跨重启只并入新增段（见 merge_fail_log）。 - let mut fails_consumed: usize = 0; - let hb_path = std::env::temp_dir().join(format!("oxide_t262_hb_{}_{}.txt", std::process::id(), window_id)); - // sidecar 日志：子进程 stdout/stderr 双写此文件（与 .fails 同目录同生命周期）。 - let log_path = hb_path.with_extension("log"); - // last-pc 现场文件：子进程运行期定频追加写，监督者杀子进程后读回末行。 - let pc_path = std::env::temp_dir().join(format!("oxide_t262_pc_{}_{}.txt", std::process::id(), window_id)); let describe = |idx: usize| -> String { paths @@ -121,76 +71,43 @@ fn supervise_window( while cur < wend { // 心跳文件每子进程重开；旁路失败行与 sidecar 日志跨重启保留（追加而非 // 清空）：超时/崩溃重启若删旁路会丢掉此前子进程的全部失败记录与输出， - // 收尾差分只剩末段。merge_fail_log 按已消费字节偏移只并入新增段（每行恰 + // 收尾差分只剩末段。merge_fail 按已消费字节偏移只并入新增段（每行恰 // 并入一次）；重跑测试至多重录一条（心跳先写、旁路后追加的写序）， // 新行是真实新事件而非重复并入。 - let _ = std::fs::remove_file(&hb_path); - let _ = std::fs::remove_file(format!("{}.tmp", hb_path.display())); - let _ = std::fs::remove_file(&pc_path); - let max_tests = wend - cur; + transport.reset_round(); - // sidecar 日志经手：子进程 stdout/stderr 双写此文件（直接文件写，无 - // pipe 缓冲死锁面）；打开失败回退丢弃，不中断运行。 - let (stdout, stderr) = match std::fs::OpenOptions::new().create(true).append(true).open(&log_path) { - Ok(file) => { - // 文件经手克隆：底层 fd 复制，stdout/stderr 各持一份。 - let err_file = file.try_clone().expect("sidecar 日志经手克隆失败"); - let out = std::process::Stdio::from(file); - let err = std::process::Stdio::from(err_file); - (out, err) - } - Err(_) => (std::process::Stdio::null(), std::process::Stdio::null()), - }; - - let mut child = match Command::new(exe) - .args(args.iter().skip(1)) - .env("OXIDE_SKIP_UNTIL", cur.to_string()) - .env("OXIDE_MAX_TESTS", max_tests.to_string()) - .env("OXIDE_TEST262_WORKERS", "1") - .env("OXIDE_TEST262_HEARTBEAT", &hb_path) - .env("OXIDE_TEST262_PC_WATCH", &pc_path) - .env("OXIDE_TEST262_PATHS_FILE", paths_file) - .env("OXIDE_TEST262_CHILD_CHUNK", "1") - .env("OXIDE_TEST262_ALLOW_FAIL_EXIT", "1") - .env_remove("OXIDE_TEST262_CHUNK_SIZE") - .stdout(stdout) - .stderr(stderr) - .spawn() - { - Ok(child) => child, - Err(err) => { - eprintln!(" window {window_id}: failed to spawn child at index {cur}: {err}"); - stats.spawn_errors += 1; - record_timeout_or_crash(&mut stats, no_skip, cur, 0, None); - cur += 1; - continue; - } - }; - - let spawn_time = Instant::now(); let mut last_index: Option = None; let mut last_change = Instant::now(); + if let Err(err) = transport.assign(cur, wend) { + eprintln!(" window {window_id}: failed to spawn child at index {cur}: {err}"); + stats.spawn_errors += 1; + record_timeout_or_crash(&mut stats, no_skip, cur, 0, None); + cur += 1; + continue; + } + let spawn_time = Instant::now(); + loop { - match child.try_wait() { - Ok(Some(status)) => { - match read_heartbeat(&hb_path) { + match transport.poll() { + PollResult::Exited(status) => { + match transport.heartbeat() { Some(hb) if hb.phase == "DONE" => { merge_heartbeat(&mut stats, &hb); - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); + transport.merge_fail(&mut stats); cur = wend; } Some(hb) => { merge_heartbeat(&mut stats, &hb); let culprit = hb.index + 1; - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - let scene = read_pc_scene(&pc_path); + transport.merge_fail(&mut stats); + let scene = transport.scene(); eprintln!( " [warn] window {window_id}: child exited ({status}) mid-test #{}: {}", culprit, describe(culprit) ); - print_log_tail(window_id, &log_path); + print_log_tail(window_id, transport); record_timeout_or_crash(&mut stats, no_skip, culprit, 0, scene.as_deref()); cur = culprit; } @@ -198,31 +115,30 @@ fn supervise_window( eprintln!( " [warn] window {window_id}: child exited ({status}) with no heartbeat at index {cur}; skipping one" ); - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - let scene = read_pc_scene(&pc_path); - print_log_tail(window_id, &log_path); + transport.merge_fail(&mut stats); + let scene = transport.scene(); + print_log_tail(window_id, transport); record_timeout_or_crash(&mut stats, no_skip, cur, 0, scene.as_deref()); cur += 1; } } break; } - Ok(None) => {} - Err(err) => { + PollResult::Alive => {} + PollResult::WaitError(err) => { eprintln!(" window {window_id}: try_wait error: {err}"); stats.wait_errors += 1; - let _ = child.kill(); - let _ = child.wait(); - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - let scene = read_pc_scene(&pc_path); - print_log_tail(window_id, &log_path); + transport.kill(); + transport.merge_fail(&mut stats); + let scene = transport.scene(); + print_log_tail(window_id, transport); record_timeout_or_crash(&mut stats, no_skip, cur, 0, scene.as_deref()); cur += 1; break; } } - if let Some(hb) = read_heartbeat(&hb_path) { + if let Some(hb) = transport.heartbeat() { if Some(hb.index) != last_index { last_index = Some(hb.index); last_change = Instant::now(); @@ -237,22 +153,21 @@ fn supervise_window( if elapsed > deadline { // 先杀再读：心跳、旁路失败行与 last-pc 现场都须在子进程死亡后取最终状态。 - let _ = child.kill(); - let _ = child.wait(); - let hb = read_heartbeat(&hb_path); + transport.kill(); + let hb = transport.heartbeat(); let culprit = hb.as_ref().map(|h| h.index + 1).unwrap_or(cur); if let Some(h) = &hb { merge_heartbeat(&mut stats, h); } - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - let scene = read_pc_scene(&pc_path); + transport.merge_fail(&mut stats); + let scene = transport.scene(); eprintln!( " [timeout] window {window_id}: TIMEOUT ({}s) on test #{culprit}: {}{}", deadline.as_secs(), describe(culprit), scene.as_deref().map(|s| format!(" [{s}]")).unwrap_or_default() ); - print_log_tail(window_id, &log_path); + print_log_tail(window_id, transport); record_timeout_or_crash(&mut stats, no_skip, culprit, elapsed.as_millis() as u64, scene.as_deref()); cur = culprit + 1; break; @@ -262,14 +177,7 @@ fn supervise_window( } } - let _ = std::fs::remove_file(&hb_path); - let _ = std::fs::remove_file(format!("{}.tmp", hb_path.display())); - let _ = std::fs::remove_file(&pc_path); - // 旁路失败行归档而非删除：统计已由心跳合并入父进程，此文件是收尾逐文件 - // 差分的终态依据（删除即永久丢记录）。 - let _ = std::fs::rename(hb_path.with_extension("fails"), hb_path.with_extension("fails.done")); - // sidecar 日志同式归档：整窗子进程输出是现场重建的终态依据。 - let _ = std::fs::rename(&log_path, log_path.with_extension("log.done")); + transport.archive_window(); stats } @@ -357,17 +265,17 @@ pub(crate) fn run_supervised( break; } let (window_id, wstart, wend) = windows[wi]; + // 每窗口一个传输实例（跨该窗口内子进程重启存活）， + // 已消费偏移等窗口级状态随之保留。 let window_stats = supervise_window( - exe, - args, no_skip, wstart, wend, timeout, startup_grace, paths, - paths_file, window_id, + &mut LocalFileWindowTransport::new(exe, args, paths_file, window_id), ); stats.merge(window_stats); } @@ -444,7 +352,6 @@ pub(crate) fn run_supervised( #[cfg(test)] mod tests { use super::*; - use crate::report::append_fail_log; /// 超时/崩溃记账：计数（no_skip 转 fail）与 timeout_crashes 入列（scene 原样 /// 存串，缺现场为 None）。 @@ -462,96 +369,4 @@ mod tests { vec![(3, 2500, Some("pc=1 op=ADD flat_id=0 frames=1".to_string())), (4, 0, None)] ); } - - /// read_pc_scene 防御解析：缺文件 / 空文件 / 残行（缺 frames 字段）均 None， - /// 末行完整时接受（忽略中间行）。 - #[test] - fn read_pc_scene_defensive_parse() { - let dir = std::env::temp_dir(); - let path = dir.join(format!("oxide_t262_pc_test_{}.txt", std::process::id())); - assert_eq!(read_pc_scene(&path), None, "缺文件返回 None"); - std::fs::write(&path, "").expect("写失败"); - assert_eq!(read_pc_scene(&path), None, "空文件返回 None"); - std::fs::write(&path, "pc=1 op=ADD flat_id=0 frames=1\npc=2 op=ADD flat_id=0 frames=1\n").expect("写失败"); - assert_eq!(read_pc_scene(&path), Some("pc=2 op=ADD flat_id=0 frames=1".to_string())); - // 残行：末行缺 frames 字段，返回 None。 - std::fs::write(&path, "pc=1 op=ADD flat_id=0 frames=1\npc=2 op=AD").expect("写失败"); - assert_eq!(read_pc_scene(&path), None, "残行返回 None"); - let _ = std::fs::remove_file(&path); - } - - /// tail_lines 防御解析：缺文件 / 空文件返回 None，不足 N 行返回整文件， - /// 超出 N 行只取末 N 行。 - #[test] - fn tail_lines_defensive_parse() { - let dir = std::env::temp_dir(); - let path = dir.join(format!("oxide_t262_log_test_{}.log", std::process::id())); - assert_eq!(tail_lines(&path, 20), None, "缺文件返回 None"); - std::fs::write(&path, "").expect("写失败"); - assert_eq!(tail_lines(&path, 20), None, "空文件返回 None"); - std::fs::write(&path, "a\nb\n").expect("写失败"); - assert_eq!(tail_lines(&path, 20), Some("a\nb".to_string()), "不足 N 行返回整文件"); - std::fs::write(&path, "1\n2\n3\n4\n").expect("写失败"); - assert_eq!(tail_lines(&path, 2), Some("3\n4".to_string()), "超出 N 行取末 N 行"); - let _ = std::fs::remove_file(&path); - } - - /// merge_fail_log 重建 fail_records：类别经 id 表重映射、消息 cap； - /// fail_categories 保持空（计数不双计契约）。 - #[test] - fn merge_fail_log_reconstructs_fail_records() { - let dir = std::env::temp_dir(); - let hb_path = dir.join(format!("oxide_t262_hb_test_mrg_{}.txt", std::process::id())); - let sidecar = hb_path.with_extension("fails"); - append_fail_log(&sidecar, 2, "vm: not callable", "", "x is not callable").expect("追加失败"); - append_fail_log(&sidecar, 7, "compile: unsupported", "", "y").expect("追加失败"); - let mut stats = RunStats::default(); - let mut fails_consumed = 0usize; - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - assert_eq!(stats.fail_records.len(), 2); - // 无新增内容的二次合并不产生重复记录（已消费偏移）。 - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - assert_eq!(stats.fail_records.len(), 2); - assert_eq!(stats.fail_records[0].index, 2); - assert_eq!(stats.categories[stats.fail_records[0].category_id as usize], "vm: not callable"); - assert_eq!(stats.fail_records[0].message, "x is not callable"); - assert_eq!(stats.fail_records[1].index, 7); - assert_eq!(stats.categories[stats.fail_records[1].category_id as usize], "compile: unsupported"); - assert!(stats.fail_categories.is_empty(), "计数不双计：fail_categories 必须保持空"); - let _ = std::fs::remove_file(&sidecar); - let _ = std::fs::remove_file(&hb_path); - } - - /// 跨重启合并序列：旁路文件两次"子进程死亡"间追加保留，第二次合并只并入 - /// 新增段（不重复并入前段行）；重跑测试的新行是真实新事件；文件重建得比 - /// 已消费偏移更短时回绕重读。 - #[test] - fn merge_fail_log_restart_sequence_no_reread_duplicates() { - let dir = std::env::temp_dir(); - let hb_path = dir.join(format!("oxide_t262_hb_test_rst_{}.txt", std::process::id())); - let sidecar = hb_path.with_extension("fails"); - // 段 1：首个子进程死亡前的失败行。 - append_fail_log(&sidecar, 10, "vm: not callable", "", "a").expect("追加失败"); - append_fail_log(&sidecar, 11, "vm: not callable", "", "b").expect("追加失败"); - let mut stats = RunStats::default(); - let mut fails_consumed = 0usize; - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - assert_eq!(stats.fail_records.len(), 2); - // 段 2：第二个子进程追加（重跑的测试 11 至多重录一条 + 新测试 12）。 - append_fail_log(&sidecar, 11, "vm: not callable", "", "b again").expect("追加失败"); - append_fail_log(&sidecar, 12, "compile: unsupported", "", "c").expect("追加失败"); - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - assert_eq!(stats.fail_records.len(), 4, "只应并入新增段: {:?}", stats.fail_records); - assert_eq!(stats.fail_records.iter().map(|r| r.index).collect::>(), vec![10, 11, 11, 12]); - // 段 3：无新增内容的合并幂等。 - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - assert_eq!(stats.fail_records.len(), 4); - // 文件重建得比已消费偏移短：回绕到文件头重读。 - std::fs::write(&sidecar, "20\tvm: not callable\t\td\n").expect("重写失败"); - merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); - assert_eq!(stats.fail_records.len(), 5); - assert_eq!(stats.fail_records[4].index, 20); - let _ = std::fs::remove_file(&sidecar); - let _ = std::fs::remove_file(&hb_path); - } } diff --git a/crates/oxide_test262/src/window_transport.rs b/crates/oxide_test262/src/window_transport.rs new file mode 100644 index 00000000..6e35c173 --- /dev/null +++ b/crates/oxide_test262/src/window_transport.rs @@ -0,0 +1,326 @@ +//! 监督窗口传输抽象：把 supervise 子进程窗口与父进程之间的文件协议（心跳文件、 +//! 旁路失败行、last-pc 现场、sidecar 日志、子进程派生与杀死）收口成六操作传输 +//! 接口；LocalFile 实现即现状文件协议逐位不变，远程模式将来加 Tcp 实现。 +//! +//! 传输只负责「怎么把窗口跑起来、怎么读回状态」；超时判定、重启循环、记账等 +//! 监督策略留在 supervise 侧。传输实例是窗口级状态：跨子进程重启存活， +//! `assign` 只重置子进程句柄，不重置已消费偏移。 + +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}; + +/// 派生窗口子进程失败：spawn 的 io 错误。 +pub(crate) struct SpawnError(std::io::Error); + +impl std::fmt::Display for SpawnError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} + +/// 子进程等待三态：死亡（带退出状态）/ 存活 / 等待错误。 +/// 三臂在监督策略里处理不同（死亡走心跳裁决、存活继续轮询、错误走 kill 加 +/// 记账），不得压扁。 +pub(crate) enum PollResult { + Exited(ExitStatus), + Alive, + WaitError(std::io::Error), +} + +/// 监督窗口传输：六操作（assign/heartbeat/merge_fail/scene/kill/poll）加 +/// 生命周期与日志辅助。LocalFile 臂是现状文件协议，远程臂将来是窗口指派/ +/// 心跳/失败记录/现场/kill/DONE 帧。 +pub(crate) trait WindowTransport { + /// 派生一个跑 `[wstart, wend)` 的子进程（LocalFile：九项环境变量加 + /// sidecar 双写，打开失败回退丢弃不中断）。 + fn assign(&mut self, wstart: usize, wend: usize) -> Result<(), SpawnError>; + /// 读最新心跳（缺失/畸形返回 None，调用方下一拍重试）。 + fn heartbeat(&self) -> Option; + /// 把旁路失败行新增段并入统计（已消费偏移跨重启保留，防双计）。 + fn merge_fail(&mut self, stats: &mut RunStats); + /// 读子进程被杀前留下的 last-pc 现场行（缺文件/残行返回 None）。 + fn scene(&self) -> Option; + /// 杀死子进程并等待回收。 + fn kill(&mut self); + /// 非阻塞查询子进程状态（三态）。 + fn poll(&mut self) -> PollResult; + /// 读 sidecar 日志末 ~20 行（超时/崩溃后供现场重建）。 + fn log_tail(&self) -> Option; + /// 每轮重启前清理：删心跳文件、其 .tmp 与 last-pc 现场文件；旁路失败行 + /// 与 sidecar 日志跨重启追加保留。 + fn reset_round(&mut self); + /// 窗口收尾：删心跳与 .tmp 与 last-pc 文件，旁路失败行与 sidecar 日志 + /// 归档为 .done 而非删除（收尾逐文件差分的终态依据）。 + fn archive_window(&mut self); +} + +/// 本地文件传输：现状文件协议实现（心跳文件 + 旁路失败行 + last-pc 现场 + +/// sidecar 日志 + 子进程句柄）。 +pub(crate) struct LocalFileWindowTransport { + exe: PathBuf, + args: Vec, + paths_file: PathBuf, + hb_path: PathBuf, + log_path: PathBuf, + pc_path: PathBuf, + /// 旁路失败行已消费字节偏移：窗口级状态，跨子进程重启保留（防双计）。 + fails_consumed: usize, + child: Option, +} + +impl LocalFileWindowTransport { + /// 构造传输：按 pid 加 window_id 命名规则算好三个文件路径（与现状一致）。 + pub(crate) fn new(exe: &Path, args: &[String], paths_file: &Path, window_id: usize) -> Self { + let hb_path = std::env::temp_dir().join(format!("oxide_t262_hb_{}_{}.txt", std::process::id(), window_id)); + let log_path = hb_path.with_extension("log"); + let pc_path = std::env::temp_dir().join(format!("oxide_t262_pc_{}_{}.txt", std::process::id(), window_id)); + Self { + exe: exe.to_path_buf(), + args: args.to_vec(), + paths_file: paths_file.to_path_buf(), + hb_path, + log_path, + pc_path, + fails_consumed: 0, + child: None, + } + } +} + +impl WindowTransport for LocalFileWindowTransport { + fn assign(&mut self, wstart: usize, wend: usize) -> Result<(), SpawnError> { + let max_tests = wend - wstart; + // sidecar 日志经手：子进程 stdout/stderr 双写此文件（直接文件写，无 + // pipe 缓冲死锁面）；打开失败回退丢弃，不中断运行。 + let (stdout, stderr) = match std::fs::OpenOptions::new().create(true).append(true).open(&self.log_path) { + Ok(file) => { + // 文件经手克隆：底层 fd 复制，stdout/stderr 各持一份。 + let err_file = file.try_clone().expect("sidecar 日志经手克隆失败"); + let out = std::process::Stdio::from(file); + let err = std::process::Stdio::from(err_file); + (out, err) + } + Err(_) => (std::process::Stdio::null(), std::process::Stdio::null()), + }; + + let child = Command::new(&self.exe) + .args(self.args.iter().skip(1)) + .env("OXIDE_SKIP_UNTIL", wstart.to_string()) + .env("OXIDE_MAX_TESTS", max_tests.to_string()) + .env("OXIDE_TEST262_WORKERS", "1") + .env("OXIDE_TEST262_HEARTBEAT", &self.hb_path) + .env("OXIDE_TEST262_PC_WATCH", &self.pc_path) + .env("OXIDE_TEST262_PATHS_FILE", &self.paths_file) + .env("OXIDE_TEST262_CHILD_CHUNK", "1") + .env("OXIDE_TEST262_ALLOW_FAIL_EXIT", "1") + .env_remove("OXIDE_TEST262_CHUNK_SIZE") + .stdout(stdout) + .stderr(stderr) + .spawn() + .map_err(SpawnError)?; + self.child = Some(child); + Ok(()) + } + + fn heartbeat(&self) -> Option { + read_heartbeat(&self.hb_path) + } + + fn merge_fail(&mut self, stats: &mut RunStats) { + merge_fail_log(stats, &self.hb_path, &mut self.fails_consumed); + } + + fn scene(&self) -> Option { + read_pc_scene(&self.pc_path) + } + + fn kill(&mut self) { + if let Some(child) = self.child.as_mut() { + let _ = child.kill(); + let _ = child.wait(); + } + self.child = None; + } + + fn poll(&mut self) -> PollResult { + let Some(child) = self.child.as_mut() else { + return PollResult::Alive; + }; + match child.try_wait() { + Ok(Some(status)) => PollResult::Exited(status), + Ok(None) => PollResult::Alive, + Err(err) => PollResult::WaitError(err), + } + } + + fn log_tail(&self) -> Option { + tail_lines(&self.log_path, 20) + } + + fn reset_round(&mut self) { + let _ = std::fs::remove_file(&self.hb_path); + let _ = std::fs::remove_file(format!("{}.tmp", self.hb_path.display())); + let _ = std::fs::remove_file(&self.pc_path); + } + + fn archive_window(&mut self) { + let _ = std::fs::remove_file(&self.hb_path); + let _ = std::fs::remove_file(format!("{}.tmp", self.hb_path.display())); + let _ = std::fs::remove_file(&self.pc_path); + // 旁路失败行归档而非删除：统计已由心跳合并入父进程，此文件是收尾逐文件 + // 差分的终态依据（删除即永久丢记录）。 + let _ = std::fs::rename(self.hb_path.with_extension("fails"), self.hb_path.with_extension("fails.done")); + // sidecar 日志同式归档：整窗子进程输出是现场重建的终态依据。 + let _ = std::fs::rename(&self.log_path, self.log_path.with_extension("log.done")); + } +} + +/// 把 supervise 旁路失败行文件并入统计：逐行重建 FailRecord（类别经本表 id +/// 重映射 + push_fail_record 的 cap 逻辑）。 +/// +/// # 边界与前提 +/// - 失败计数（fail_categories）已由对应心跳合并覆盖，此处只补 fail_records， +/// 不得再累计计数，避免双计。worker 循环先写心跳后追加旁路，故"无心跳退出" +/// 分支旁路必为空（防御性 no-op）。 +/// - 旁路文件跨子进程重启追加保留，本函数只解析自上次消费以来的新增段 +/// （`consumed` 由传输实例按窗口持有）；文件缩至已消费偏移之下（删除/重建） +/// 回绕到文件头重读，避免丢记录。 +fn merge_fail_log(stats: &mut RunStats, hb_path: &Path, consumed: &mut usize) { + let content = match std::fs::read_to_string(hb_path.with_extension("fails")) { + Ok(c) => c, + Err(_) => return, + }; + if content.len() < *consumed { + *consumed = 0; + } + let fresh = &content[*consumed..]; + *consumed = content.len(); + for (index, cat, subkey, message) in parse_fail_log(fresh) { + stats.push_fail_record(index, cat, subkey, message); + } +} + +/// 读子进程被杀前留下的 last-pc 现场行：取文件末行，须含 pc/op/frames 三字段 +/// 才接受（缺文件 / 空文件 / 残行均返回 None）。 +fn read_pc_scene(path: &Path) -> Option { + let content = std::fs::read_to_string(path).ok()?; + let last = content.lines().last()?.trim(); + if last.starts_with("pc=") && last.contains(" op=") && last.contains(" frames=") { + Some(last.to_string()) + } else { + None + } +} + +/// 读 sidecar 日志的末 `n` 行：文件缺失/空返回 None，有内容时返回整文件或 +/// 末 N 行（超时/崩溃后供现场重建）。 +fn tail_lines(path: &Path, n: usize) -> Option { + let content = std::fs::read_to_string(path).ok()?; + let lines: Vec<&str> = content.lines().collect(); + let start = lines.len().saturating_sub(n); + if start == lines.len() { + return None; + } + Some(lines[start..].join("\n")) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::report::append_fail_log; + + /// read_pc_scene 防御解析：缺文件 / 空文件 / 残行（缺 frames 字段）均 None， + /// 末行完整时接受（忽略中间行）。 + #[test] + fn read_pc_scene_defensive_parse() { + let dir = std::env::temp_dir(); + let path = dir.join(format!("oxide_t262_pc_test_{}.txt", std::process::id())); + assert_eq!(read_pc_scene(&path), None, "缺文件返回 None"); + std::fs::write(&path, "").expect("写失败"); + assert_eq!(read_pc_scene(&path), None, "空文件返回 None"); + std::fs::write(&path, "pc=1 op=ADD flat_id=0 frames=1\npc=2 op=ADD flat_id=0 frames=1\n").expect("写失败"); + assert_eq!(read_pc_scene(&path), Some("pc=2 op=ADD flat_id=0 frames=1".to_string())); + // 残行：末行缺 frames 字段，返回 None。 + std::fs::write(&path, "pc=1 op=ADD flat_id=0 frames=1\npc=2 op=AD").expect("写失败"); + assert_eq!(read_pc_scene(&path), None, "残行返回 None"); + let _ = std::fs::remove_file(&path); + } + + /// tail_lines 防御解析：缺文件 / 空文件返回 None，不足 N 行返回整文件， + /// 超出 N 行只取末 N 行。 + #[test] + fn tail_lines_defensive_parse() { + let dir = std::env::temp_dir(); + let path = dir.join(format!("oxide_t262_log_test_{}.log", std::process::id())); + assert_eq!(tail_lines(&path, 20), None, "缺文件返回 None"); + std::fs::write(&path, "").expect("写失败"); + assert_eq!(tail_lines(&path, 20), None, "空文件返回 None"); + std::fs::write(&path, "a\nb\n").expect("写失败"); + assert_eq!(tail_lines(&path, 20), Some("a\nb".to_string()), "不足 N 行返回整文件"); + std::fs::write(&path, "1\n2\n3\n4\n").expect("写失败"); + assert_eq!(tail_lines(&path, 2), Some("3\n4".to_string()), "超出 N 行取末 N 行"); + let _ = std::fs::remove_file(&path); + } + + /// merge_fail_log 重建 fail_records：类别经 id 表重映射、消息 cap； + /// fail_categories 保持空（计数不双计契约）。 + #[test] + fn merge_fail_log_reconstructs_fail_records() { + let dir = std::env::temp_dir(); + let hb_path = dir.join(format!("oxide_t262_hb_test_mrg_{}.txt", std::process::id())); + let sidecar = hb_path.with_extension("fails"); + append_fail_log(&sidecar, 2, "vm: not callable", "", "x is not callable").expect("追加失败"); + append_fail_log(&sidecar, 7, "compile: unsupported", "", "y").expect("追加失败"); + let mut stats = RunStats::default(); + let mut fails_consumed = 0usize; + merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); + assert_eq!(stats.fail_records.len(), 2); + // 无新增内容的二次合并不产生重复记录（已消费偏移）。 + merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); + assert_eq!(stats.fail_records.len(), 2); + assert_eq!(stats.fail_records[0].index, 2); + assert_eq!(stats.categories[stats.fail_records[0].category_id as usize], "vm: not callable"); + assert_eq!(stats.fail_records[0].message, "x is not callable"); + assert_eq!(stats.fail_records[1].index, 7); + assert_eq!(stats.categories[stats.fail_records[1].category_id as usize], "compile: unsupported"); + assert!(stats.fail_categories.is_empty(), "计数不双计：fail_categories 必须保持空"); + let _ = std::fs::remove_file(&sidecar); + let _ = std::fs::remove_file(&hb_path); + } + + /// 跨重启合并序列：旁路文件两次"子进程死亡"间追加保留，第二次合并只并入 + /// 新增段（不重复并入前段行）；重跑测试的新行是真实新事件；文件重建得比 + /// 已消费偏移更短时回绕重读。 + #[test] + fn merge_fail_log_restart_sequence_no_reread_duplicates() { + let dir = std::env::temp_dir(); + let hb_path = dir.join(format!("oxide_t262_hb_test_rst_{}.txt", std::process::id())); + let sidecar = hb_path.with_extension("fails"); + // 段 1：首个子进程死亡前的失败行。 + append_fail_log(&sidecar, 10, "vm: not callable", "", "a").expect("追加失败"); + append_fail_log(&sidecar, 11, "vm: not callable", "", "b").expect("追加失败"); + let mut stats = RunStats::default(); + let mut fails_consumed = 0usize; + merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); + assert_eq!(stats.fail_records.len(), 2); + // 段 2：第二个子进程追加（重跑的测试 11 至多重录一条 + 新测试 12）。 + append_fail_log(&sidecar, 11, "vm: not callable", "", "b again").expect("追加失败"); + append_fail_log(&sidecar, 12, "compile: unsupported", "", "c").expect("追加失败"); + merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); + assert_eq!(stats.fail_records.len(), 4, "只应并入新增段: {:?}", stats.fail_records); + assert_eq!(stats.fail_records.iter().map(|r| r.index).collect::>(), vec![10, 11, 11, 12]); + // 段 3：无新增内容的合并幂等。 + merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); + assert_eq!(stats.fail_records.len(), 4); + // 文件重建得比已消费偏移短：回绕到文件头重读。 + std::fs::write(&sidecar, "20\tvm: not callable\t\td\n").expect("重写失败"); + merge_fail_log(&mut stats, &hb_path, &mut fails_consumed); + assert_eq!(stats.fail_records.len(), 5); + assert_eq!(stats.fail_records[4].index, 20); + let _ = std::fs::remove_file(&sidecar); + let _ = std::fs::remove_file(&hb_path); + } +}