From c380cead577815ce76a636532ecbdebacaf98be6 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Mon, 12 Oct 2026 20:37:54 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20safe=20=E6=A8=A1=E5=BC=8F=E5=B9=B6?= =?UTF-8?q?=E8=A1=8C=20worker=20=E8=BF=9B=E7=A8=8B=E5=8C=96=EF=BC=8CUnix?= =?UTF-8?q?=20socket=20=E5=A4=8D=E7=94=A8=E5=88=86=E7=89=87=E5=B8=A7?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 42 ++-- Cargo.toml | 2 +- crates/oxide_test262/src/config.rs | 41 +++ crates/oxide_test262/src/main.rs | 10 + crates/oxide_test262/src/safe_supervise.rs | 237 ++++++++++++++++++ crates/oxide_test262/src/supervise.rs | 4 +- .../oxide_test262/src/tcp_window_transport.rs | 58 ++++- crates/oxide_test262/src/worker.rs | 47 +++- 8 files changed, 407 insertions(+), 34 deletions(-) create mode 100644 crates/oxide_test262/src/safe_supervise.rs diff --git a/Cargo.lock b/Cargo.lock index 1b25473f..fb919ca8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.17.9" +version = "0.17.10" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.17.9" +version = "0.17.10" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.17.9" +version = "0.17.10" dependencies = [ "ansi_term", "bumpalo", @@ -1228,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.17.9" +version = "0.17.10" dependencies = [ "lru", "oxide_bytecode", @@ -1237,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1258,7 +1258,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1272,7 +1272,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.17.9" +version = "0.17.10" dependencies = [ "num-bigint", "oxc_allocator", @@ -1288,7 +1288,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1300,7 +1300,7 @@ dependencies = [ [[package]] name = "oxide_fold" -version = "0.17.9" +version = "0.17.10" dependencies = [ "num-bigint", "oxide_bytecode", @@ -1312,7 +1312,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1321,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.17.9" +version = "0.17.10" dependencies = [ "bumpalo", "dashmap", @@ -1335,7 +1335,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1347,7 +1347,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.17.9" +version = "0.17.10" dependencies = [ "tracing", "tracing-appender", @@ -1356,11 +1356,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.17.9" +version = "0.17.10" [[package]] name = "oxide_parser" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1375,7 +1375,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1391,7 +1391,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.17.9" +version = "0.17.10" dependencies = [ "num-bigint", "num-traits", @@ -1404,7 +1404,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.17.9" +version = "0.17.10" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1424,7 +1424,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.17.9" +version = "0.17.10" dependencies = [ "bumpalo", "dashmap", @@ -1437,7 +1437,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.17.9" +version = "0.17.10" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index b3f0f519..d4920178 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,7 +25,7 @@ members = [ ] [workspace.package] -version = "0.17.9" +version = "0.17.10" 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 63c30752..2d6dc694 100644 --- a/crates/oxide_test262/src/config.rs +++ b/crates/oxide_test262/src/config.rs @@ -16,6 +16,9 @@ pub(crate) struct RunConfig { /// 远程监督模式:(端点文本, worker 数)。监督者 bind 端点、每窗口 accept 一条 /// worker 连接,经 Tcp 传输执行窗口。 pub(crate) remote_supervise: Option<(String, usize)>, + /// safe 模式:并行 worker 从线程变进程派生,IPC 走 Unix socket 复用分片帧协议。 + /// 缺省 off(unsafe)保现状逐位不变;与 --supervise/--worker/--remote-supervise 互斥。 + pub(crate) safe: bool, /// 关闭 liveness/精确 DCE/RegAlloc 链(开关对比:关优化链 vs 开优化链各跑一遍)。 pub(crate) no_regalloc: bool, /// 逐测试打印 PASS/FAIL/SKIP(开关对比时对比两轮结果集合用)。 @@ -34,6 +37,7 @@ impl RunConfig { supervise: false, worker: false, remote_supervise: None, + safe: false, no_regalloc: false, verbose: false, no_fail_list: false, @@ -94,6 +98,7 @@ impl RunConfig { "--verbose" => config.verbose = true, "--supervise" => config.supervise = true, "--worker" => config.worker = true, + "--safe" => config.safe = true, "--no-fail-list" => config.no_fail_list = true, // --remote-supervise 消费后两个实参:端点文本与 worker 数。 // 端点解析期归一化:缺省补 loopback,非 loopback 拒绝。 @@ -131,6 +136,17 @@ impl RunConfig { if config.supervise && config.remote_supervise.is_some() { return Err(format!("--supervise and --remote-supervise are mutually exclusive\n\n{}", Self::usage())); } + // safe 模式是本地远程监督的 Unix socket 变体(自动派生 worker 进程), + // 与已各自进程化的 --supervise / --remote-supervise 及执行半区 --worker 互斥。 + if config.safe && config.supervise { + return Err(format!("--safe and --supervise are mutually exclusive\n\n{}", Self::usage())); + } + if config.safe && config.worker { + return Err(format!("--safe and --worker are mutually exclusive\n\n{}", Self::usage())); + } + if config.safe && config.remote_supervise.is_some() { + return Err(format!("--safe and --remote-supervise are mutually exclusive\n\n{}", Self::usage())); + } if let Some(root) = positional.first() { config.test262_root = Some(PathBuf::from(root)); @@ -167,6 +183,11 @@ impl RunConfig { \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\ + --safe Run the parallel workers as spawned processes (crash isolation) with IPC over a\n\ + \x20 local Unix socket reusing the shard frame protocol. A worker process is respawned\n\ + \x20 after a timeout/crash and resumes past the culprit. Off by default (unsafe thread\n\ + \x20 path is unchanged). Mutually exclusive with --supervise, --worker and\n\ + \x20 --remote-supervise.\n\ \n\ env tunables (all optional):\n\ \x20 OXIDE_LOG per-subsystem log levels (default off)\n\ @@ -297,6 +318,26 @@ mod tests { assert!(err.contains("mutually exclusive"), "应提示互斥:{err}"); } + /// --safe 旗标置位;与 --supervise / --worker / --remote-supervise 两两互斥; + /// safe 不触发 supervised()(独立执行路径,非监督模式)。 + #[test] + fn parse_safe_flag_and_exclusive() { + let cfg = RunConfig::parse(&args(&["--safe"])).expect("safe 旗标应解析"); + assert!(cfg.safe, "safe 旗标应置位"); + assert!(!cfg.supervise, "safe 不应置位 supervise"); + assert!(!cfg.supervised(false), "safe 不应触发 supervised"); + + let err = RunConfig::parse(&args(&["--safe", "--supervise"])).expect_err("safe+supervise 应互斥"); + assert!(err.contains("mutually exclusive"), "应提示互斥:{err}"); + + let err = RunConfig::parse(&args(&["--safe", "--worker"])).expect_err("safe+worker 应互斥"); + assert!(err.contains("mutually exclusive"), "应提示互斥:{err}"); + + let err = RunConfig::parse(&args(&["--safe", "--remote-supervise", "127.0.0.1:8080", "4"])) + .expect_err("safe+remote 应互斥"); + assert!(err.contains("mutually exclusive"), "应提示互斥:{err}"); + } + /// remote_supervise 置位也触发 supervised()(非子进程);子进程不重入。 #[test] fn supervised_predicate_with_remote_supervise() { diff --git a/crates/oxide_test262/src/main.rs b/crates/oxide_test262/src/main.rs index f77c7903..89b0f95f 100644 --- a/crates/oxide_test262/src/main.rs +++ b/crates/oxide_test262/src/main.rs @@ -13,6 +13,7 @@ mod judge; mod meta; mod report; mod runner; +mod safe_supervise; mod shard_protocol; mod stats; mod supervise; @@ -29,6 +30,7 @@ use judge::{categorize_fail, TestOutcome, TestResult}; use oxide_log::{Level, LogConfig, Output, SUBSYSTEM_COUNT}; use report::{append_fail_log, first_line, format_fail_categories, format_fail_groupings, format_fail_list}; use runner::{build_runner_engine, process_path, CURRENT_TEST_PATH}; +use safe_supervise::run_safe; use stats::RunStats; use supervise::run_supervised; @@ -315,6 +317,14 @@ fn run_tests() -> bool { return run_supervised(&args, skip_until, end_index, config.no_skip, &paths, config.remote_supervise.clone()); } + // safe 模式:并行 worker 变派生进程(崩溃隔离),IPC 走 Unix socket 复用分片帧 + // 协议。缺省 off(unsafe 线程路径逐位不变)。 + if config.safe { + test262_info!("safe mode enabled"); + eprintln!("safe mode enabled: worker processes with Unix socket IPC (crash isolation)"); + return run_safe(skip_until, end_index, config.no_skip, config.no_regalloc, &paths, workers); + } + if let Some(chunk_size) = chunk_size { if !is_chunk_child && filter.is_none() { test262_info!("chunked mode enabled: {} test(s) per child", chunk_size); diff --git a/crates/oxide_test262/src/safe_supervise.rs b/crates/oxide_test262/src/safe_supervise.rs new file mode 100644 index 00000000..e69e22b5 --- /dev/null +++ b/crates/oxide_test262/src/safe_supervise.rs @@ -0,0 +1,237 @@ +//! safe 模式:本地远程监督的 Unix socket 变体加自动派生 worker 进程。 +//! +//! 并行 worker 从线程变派生进程(崩溃隔离),IPC 走本地 Unix socket,复用分片帧 +//! 协议与跨窗口保活传输。监督者先 bind socket(临时目录短路径,防 `Sun_pathname` +//! 上限),再派生 N 个 worker 进程;每个监督线程 accept 一条连接,经 +//! `supervise_window` 跑动态窗口(共享原子游标)。worker 超时/崩溃后重新派生并 +//! 自肇事者后续跑。缺省 off(unsafe)保现状逐位不变。 +//! +//! 连接生命周期(跨窗口保活):每个监督线程一个传输实例,跨窗口复用——连接跨窗口 +//! 保活,worker 常驻摊薄进程冷启动;worker 死亡(超时/崩溃)后 `assign` 重新派生 +//! 进程并 accept 新连接。 + +use crate::print_fail_categories; +use crate::report::{format_fail_groupings, format_fail_list}; +use crate::stats::RunStats; +use crate::supervise::{generate_token, supervise_window}; +use crate::tcp_window_transport::TcpWindowTransport; +use oxide_net::Listener; +use std::os::unix::net::UnixListener; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +/// 派生一个 worker 进程:注入 `--worker` 与透传旗标(`--no-skip` / +/// `--no-regalloc`),端点与 token 经环境变量带外注入(不经网络分发)。 +/// +/// # 边界与前提 +/// - stdout/stderr 置空:safe 模式无 sidecar 日志,现场经 scene 帧回传。 +fn spawn_worker_process( + exe: &Path, socket_path: &Path, token: &str, no_skip: bool, no_regalloc: bool, +) -> std::io::Result { + let mut cmd = Command::new(exe); + cmd.arg("--worker"); + if no_skip { + cmd.arg("--no-skip"); + } + if no_regalloc { + cmd.arg("--no-regalloc"); + } + cmd.env("OXIDE_TEST262_WORKER_ENDPOINT", socket_path) + .env("OXIDE_TEST262_WORKER_TOKEN", token) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() +} + +/// 编排一次 safe 模式全量运行:bind Unix socket、派生 worker、监督线程并行 +/// accept 加跑窗口、归约统计并打印。 +pub(crate) fn run_safe( + skip_until: usize, end_index: usize, no_skip: bool, no_regalloc: bool, paths: &[PathBuf], workers: usize, +) -> bool { + let exe = match std::env::current_exe() { + Ok(path) => path, + Err(err) => { + eprintln!("failed to resolve current executable for safe mode: {err}"); + return false; + } + }; + + // 先 bind Unix socket(临时目录短路径,防 `Sun_pathname` 108 字节上限), + // 再派生 worker(worker 侧另有连接重试兜底 bind 完成前的竞态窗口)。 + let socket_path = std::env::temp_dir().join(format!("oxide_t262_{}.sock", std::process::id())); + let _ = std::fs::remove_file(&socket_path); + let unix_listener = match UnixListener::bind(&socket_path) { + Ok(l) => l, + Err(err) => { + eprintln!("failed to bind Unix socket {}: {err}", socket_path.display()); + return false; + } + }; + let listener = Arc::new(Listener::Unix(unix_listener)); + + // 认证 token:启动时生成,经环境变量注入 worker 进程(带外,不经网络分发)。 + let token = generate_token(); + + let window = std::env::var("OXIDE_TEST262_WINDOW") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|&n| n > 0) + .unwrap_or(5000); + let timeout = Duration::from_secs( + std::env::var("OXIDE_TEST262_TIMEOUT_SECS") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|&n| n > 0) + .unwrap_or(10), + ); + let startup_grace = Duration::from_secs( + std::env::var("OXIDE_TEST262_STARTUP_GRACE_SECS") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|&n| n > 0) + .unwrap_or(60), + ); + + let mut windows: Vec<(usize, usize, usize)> = Vec::new(); + let mut start = skip_until; + let mut id = 0usize; + while start < end_index { + let end = (start + window).min(end_index); + windows.push((id, start, end)); + start = end; + id += 1; + } + + let supervisors = workers.min(windows.len().max(1)); + + // 路径清单经 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 next = AtomicUsize::new(0); + let next = &next; + let windows = &windows; + + eprintln!( + "safe mode: {} window(s) of up to {window} test(s), {supervisors} worker process(es), per-test timeout {}s", + windows.len(), + timeout.as_secs() + ); + + // socket 文件收尾删除用克隆(原值移入 scope 闭包)。 + let socket_for_cleanup = socket_path.clone(); + + let partials: Vec = std::thread::scope(|scope| { + let handles: Vec<_> = (0..supervisors) + .map(|_| { + scope.spawn({ + // 每监督线程克隆一份所需值(原值被 scope 闭包持有,多闭包不得 + // 重复 move)。 + let paths_clone = (*paths_vec).clone(); + let socket_clone = socket_path.clone(); + let token_clone = token.clone(); + let exe_clone = exe.clone(); + let listener_clone = Arc::clone(&listener); + move || { + let mut stats = RunStats::default(); + // 每监督线程一个传输实例,跨窗口复用(跨窗口保活): + // 连接跨窗口存活,worker 超时/崩溃后重新派生并 accept。 + let mut transport = TcpWindowTransport::with_worker_spawn( + listener_clone, + paths_clone, + timeout_secs, + token_clone.clone(), + Box::new(move || { + spawn_worker_process(&exe_clone, &socket_clone, &token_clone, no_skip, no_regalloc) + }), + ); + 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 transport, + ); + stats.merge(window_stats); + } + stats + } + }) + }) + .collect(); + handles + .into_iter() + .map(|h| h.join().expect("supervisor thread panicked")) + .collect() + }); + + // 收尾删除 socket 文件,防临时目录残留。 + let _ = std::fs::remove_file(&socket_for_cleanup); + + let mut stats = RunStats::default(); + for partial_stats in partials { + stats.merge(partial_stats); + } + + let total = stats.pass + stats.fail + stats.skip; + println!(); + println!("═══════════════════════════════════════"); + println!(" test262 safe aggregate"); + println!("═══════════════════════════════════════"); + println!(" total : {total}"); + println!(" pass : {}", stats.pass); + println!(" fail : {}", stats.fail); + println!( + " skip : {} (timeouts/crashes here by default; --no-skip counts them as fail)", + stats.skip + ); + print_fail_categories(&stats, paths); + let fail_list = format_fail_list(&stats, paths); + if !fail_list.is_empty() { + print!("{fail_list}"); + } + let groupings = format_fail_groupings(&stats, paths); + if !groupings.is_empty() { + print!("{groupings}"); + } + let anomalies = stats.spawn_errors + stats.wait_errors + stats.hb_write_errors; + if anomalies > 0 { + println!(" --- safe anomalies ---"); + if stats.spawn_errors > 0 { + println!(" spawn errors : {}", stats.spawn_errors); + } + if stats.wait_errors > 0 { + println!(" wait errors : {}", stats.wait_errors); + } + if stats.hb_write_errors > 0 { + println!(" hb write errors: {}", stats.hb_write_errors); + } + } + if !stats.timeout_crashes.is_empty() { + println!(" --- TIMEOUT/CRASH list ({}) ---", stats.timeout_crashes.len()); + for (idx, elapsed_ms, scene) in &stats.timeout_crashes { + let path = paths + .get(*idx) + .map(|p| p.display().to_string()) + .unwrap_or_else(|| format!("#{idx}")); + match scene { + Some(s) => println!(" {idx} {elapsed_ms}ms {path} [{s}]"), + None => println!(" {idx} {elapsed_ms}ms {path}"), + } + } + } + println!("═══════════════════════════════════════"); + + stats.fail == 0 +} diff --git a/crates/oxide_test262/src/supervise.rs b/crates/oxide_test262/src/supervise.rs index d00cf466..d83b31b0 100644 --- a/crates/oxide_test262/src/supervise.rs +++ b/crates/oxide_test262/src/supervise.rs @@ -24,7 +24,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; /// # 边界与前提 /// - 非 Linux 宿主(`/dev/urandom` 缺位)或读取失败时回落时间加进程号混合熵 /// (本宿主恒为 Linux,回落臂仅保编译)。 -fn generate_token() -> String { +pub(crate) fn generate_token() -> String { let mut bytes = [0u8; 32]; let ok = std::fs::File::open("/dev/urandom") .and_then(|mut file| file.read_exact(&mut bytes)) @@ -90,7 +90,7 @@ fn record_timeout_or_crash(stats: &mut RunStats, no_skip: bool, index: usize, el /// - `paths` 是父进程的全量路径清单(每行一路径,与父扫描同序),仅供肇事者 /// 路径描述。 #[expect(clippy::too_many_arguments)] -fn supervise_window( +pub(crate) fn supervise_window( no_skip: bool, wstart: usize, wend: usize, timeout: Duration, startup_grace: Duration, paths: &[PathBuf], window_id: usize, transport: &mut dyn WindowTransport, ) -> RunStats { diff --git a/crates/oxide_test262/src/tcp_window_transport.rs b/crates/oxide_test262/src/tcp_window_transport.rs index 5cf63609..887bf2e9 100644 --- a/crates/oxide_test262/src/tcp_window_transport.rs +++ b/crates/oxide_test262/src/tcp_window_transport.rs @@ -53,6 +53,11 @@ pub(crate) struct TcpWindowTransport { paths: Vec, /// 单测试超时,随 assign 帧携带供 worker 侧看门狗使用。 timeout_secs: u64, + /// worker 进程派生钩子(safe 模式):连接丢失需重建时派生一个 worker 进程。 + /// 远程模式为 `None`(worker 外部自启,本地不派生)。 + spawn_worker: Option std::io::Result + Send>>, + /// 当前 worker 进程句柄(safe 模式;kill 或派生失败时为 `None`)。 + child: Option, } impl TcpWindowTransport { @@ -71,6 +76,20 @@ impl TcpWindowTransport { timeout_secs, token, auth_read_timeout: AUTH_READ_TIMEOUT, + spawn_worker: None, + child: None, + } + } + + /// 构造带 worker 进程派生钩子的传输(safe 模式):连接丢失需重建时经钩子 + /// 派生一个 worker 进程再 accept。远程模式用 `new`(不派生)。 + pub(crate) fn with_worker_spawn( + listener: Arc, paths: Vec, timeout_secs: u64, token: String, + spawn_worker: Box std::io::Result + Send>, + ) -> Self { + Self { + spawn_worker: Some(spawn_worker), + ..Self::new(listener, paths, timeout_secs, token) } } @@ -80,8 +99,23 @@ impl TcpWindowTransport { self.reader = None; } + /// 回收 worker 进程(safe 模式):非阻塞 `try_wait` 防僵尸残留。 + /// + /// 连接死亡(EOF / 真实 io 错误 / 协议违例)后 worker 进程已退,须 reap; + /// 远程模式无进程句柄,no-op。 + fn reap_child(&mut self) { + if let Some(child) = self.child.as_mut() { + let _ = child.try_wait(); + } + } + /// Done 帧处理:把 `last_heartbeat` 的 phase 改 DONE 并保留计数(每窗口只 - /// 合并一次的计数契约),随后断连判死亡。 + /// 合并一次的计数契约)。 + /// + /// 跨窗口保活:DONE 后不断连——worker 侧一条连接跨多窗口常驻,监督者侧同 + /// 口径保活,下一窗口 `assign` 复用同一连接发新指派帧。传输实例跨窗口复 + /// 用时(safe 模式)连接随实例存活;实例被 drop 时(远程模式每窗口新建实例) + /// 连接随句柄释放,行为不变。 fn mark_done(&mut self) { match self.last_heartbeat.as_mut() { Some(hb) => hb.phase = "DONE".to_string(), @@ -98,7 +132,6 @@ impl TcpWindowTransport { }); } } - self.drop_conn(); } } @@ -121,6 +154,12 @@ impl WindowTransport for TcpWindowTransport { fn assign(&mut self, wstart: usize, wend: usize) -> Result<(), SpawnError> { // 无连接时阻塞 `accept` 取一条 worker 连接(io 错误包成 SpawnError)。 if self.writer.is_none() { + // safe 模式:先派生 worker 进程再 accept(避免 worker 连接时 socket + // 未 bind 的竞态;worker 侧另有连接重试兜底)。派生失败按 spawn + // 错误上报,监督策略自肇事者后续跑。 + if let Some(spawn) = self.spawn_worker.as_mut() { + self.child = Some(spawn().map_err(SpawnError)?); + } let base = self.listener.accept().map_err(SpawnError)?; // 认证门:读超时内读首帧(半开连接不永久占线程),必须是 Auth 事件 // 帧且 token 异或判等匹配,才写回 AuthOk;不匹配或首帧非 Auth 直接 @@ -207,6 +246,12 @@ impl WindowTransport for TcpWindowTransport { let _ = writer.write_all(frame.as_bytes()); } self.drop_conn(); + // safe 模式:杀掉 worker 进程并回收(远程模式无进程句柄,no-op)。 + if let Some(child) = self.child.as_mut() { + let _ = child.kill(); + let _ = child.wait(); + } + self.child = None; } fn poll(&mut self) -> PollResult { @@ -258,17 +303,20 @@ impl WindowTransport for TcpWindowTransport { // Auth 只应作首帧且在认证门消费;此处到达是协议违例, // 判断连(同畸形帧同构恢复)。 self.drop_conn(); + self.reap_child(); return PollResult::Exited(1); } Err(_) => { // 畸形帧:对端状态不定,判断连(同 EOF 同构恢复)。 self.drop_conn(); + self.reap_child(); return PollResult::Exited(1); } }, Ok(None) => { // EOF:对端关闭,判死亡。 self.drop_conn(); + self.reap_child(); return PollResult::Exited(1); } Err(e) if e.kind() == ErrorKind::WouldBlock => { @@ -278,6 +326,7 @@ impl WindowTransport for TcpWindowTransport { Err(_) => { // 真实 io 错误:判断连。 self.drop_conn(); + self.reap_child(); return PollResult::Exited(1); } } @@ -296,8 +345,9 @@ impl WindowTransport for TcpWindowTransport { } fn archive_window(&mut self) { - // 远程无文件归档:失败记录已并入 stats,无数据丢失;drop 连接即收尾。 - self.drop_conn(); + // 跨窗口保活:不断连——传输实例跨窗口复用时(safe 模式)连接随实例存活, + // 下一窗口 `assign` 复用同一连接。失败记录已并入 stats,无数据丢失;实例 + // 被 drop 时(远程模式每窗口新建实例)连接随句柄释放,行为不变。 } } diff --git a/crates/oxide_test262/src/worker.rs b/crates/oxide_test262/src/worker.rs index 16b54864..bd2a316d 100644 --- a/crates/oxide_test262/src/worker.rs +++ b/crates/oxide_test262/src/worker.rs @@ -13,9 +13,11 @@ use std::io::BufReader; use std::net::TcpStream; +use std::os::unix::net::UnixStream; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{mpsc, Arc, RwLock}; +use std::time::{Duration, Instant}; use crate::config::RunConfig; use crate::harness::{HarnessModuleCache, HarnessPrefixCache, HarnessSources, HARNESS}; @@ -65,14 +67,14 @@ pub(crate) fn run_worker(config: &RunConfig) -> bool { } }; - let stream = match TcpStream::connect(&endpoint) { - Ok(s) => s, - Err(e) => { - eprintln!("worker: failed to connect to {endpoint}: {e}"); + let conn = match connect_with_retry(&endpoint) { + Some(c) => c, + None => { + eprintln!("worker: failed to connect to {endpoint} after retries"); return false; } }; - let mut conn = Conn::Tcp(stream); + let mut conn = conn; // 认证门:读超时 30 秒(半开连接不永久占进程),发 Auth 帧后读回首帧, // 必须是 AuthOk,否则报错退出。握手必须在看门狗线程启动前完成:看门狗 @@ -166,7 +168,6 @@ pub(crate) fn run_worker(config: &RunConfig) -> bool { .filter(|&n| n > 0) .map(|n| n.min(512)) .unwrap_or(256); - let mut stats = RunStats::default(); let mut tests_since_kernel_reset = 0usize; for msg in msg_rx { @@ -178,6 +179,10 @@ pub(crate) fn run_worker(config: &RunConfig) -> bool { return true; } }; + // 每窗口一份统计:DONE 心跳的 pass/fail/skip 是窗口级计数(监督者每窗口 + // 只合并一次最终心跳的契约),跨窗口累计会使后窗口心跳含前窗口计数而双计。 + // 失败记录帧逐测试随发(带全局下标),不依赖本统计跨窗口保留。 + let mut stats = RunStats::default(); for (offset, path_str) in window.paths.iter().enumerate() { let i = window.wstart + offset; // 检查 kill 标志(看门狗读 kill 帧后置位)。 @@ -230,6 +235,36 @@ pub(crate) fn run_worker(config: &RunConfig) -> bool { true } +/// 按端点形态连接监督者:含 `:` 走 TCP(host:port),否则走 Unix socket 路径 +/// (safe 模式临时目录短路径,防 `Sun_pathname` 上限)。 +/// +/// # 步骤 +/// 1. 按端点形态选传输臂尝试连接。 +/// 2. 失败(含 socket 文件尚未 bind 的竞态)睡 100 ms 重试,至 30 秒上限。 +/// +/// # 边界与前提 +/// - safe 模式监督者先 bind 再派生 worker,正常路径首连即成;重试只兜底 +/// 进程启动早于 bind 完成的竞态窗口。 +/// +/// # 副作用 +/// - 无(只建连接,不发帧)。 +fn connect_with_retry(endpoint: &str) -> Option { + let deadline = Instant::now() + Duration::from_secs(30); + loop { + if endpoint.contains(':') { + if let Ok(s) = TcpStream::connect(endpoint) { + return Some(Conn::Tcp(s)); + } + } else if let Ok(s) = UnixStream::connect(endpoint) { + return Some(Conn::Unix(s)); + } + if Instant::now() >= deadline { + return None; + } + std::thread::sleep(Duration::from_millis(100)); + } +} + /// 发送事件帧(write_all 即 flush)。 fn send_event_frame(conn: &mut Conn, event: &ShardEvent) -> std::io::Result<()> { let frame = encode_event(event);