diff --git a/Cargo.lock b/Cargo.lock index 42d3a731..ba34b865 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.15.17" +version = "0.15.18" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.15.17" +version = "0.15.18" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.15.17" +version = "0.15.18" dependencies = [ "ansi_term", "bumpalo", @@ -1228,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.15.17" +version = "0.15.18" dependencies = [ "lru", "oxide_bytecode", @@ -1237,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1257,7 +1257,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1271,7 +1271,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.15.17" +version = "0.15.18" dependencies = [ "num-bigint", "oxc_allocator", @@ -1286,7 +1286,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1298,7 +1298,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1307,7 +1307,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.15.17" +version = "0.15.18" dependencies = [ "bumpalo", "dashmap", @@ -1321,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1333,7 +1333,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.15.17" +version = "0.15.18" dependencies = [ "tracing", "tracing-appender", @@ -1342,11 +1342,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.15.17" +version = "0.15.18" [[package]] name = "oxide_parser" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1377,7 +1377,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.15.17" +version = "0.15.18" dependencies = [ "num-bigint", "num-traits", @@ -1390,7 +1390,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.15.17" +version = "0.15.18" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1398,6 +1398,7 @@ dependencies = [ "oxide_engine", "oxide_kernel", "oxide_log", + "oxide_net", "oxide_parser", "oxide_types", "oxide_vm", @@ -1409,7 +1410,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.15.17" +version = "0.15.18" dependencies = [ "bumpalo", "dashmap", @@ -1422,7 +1423,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.15.17" +version = "0.15.18" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 24c0899b..4bfc6033 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.15.17" +version = "0.15.18" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_test262/Cargo.toml b/crates/oxide_test262/Cargo.toml index 8ceaafbb..37df0780 100644 --- a/crates/oxide_test262/Cargo.toml +++ b/crates/oxide_test262/Cargo.toml @@ -20,6 +20,7 @@ oxide_kernel = { path = "../oxide_kernel" } oxide_engine = { path = "../oxide_engine" } oxide_types = { path = "../oxide_types" } oxide_log = { path = "../oxide_log" } +oxide_net = { path = "../oxide_net" } serde = { version = "1", features = ["derive"] } serde_json.workspace = true serde_yaml = "0.9" diff --git a/crates/oxide_test262/src/config.rs b/crates/oxide_test262/src/config.rs index 6d4e5f30..5ff63819 100644 --- a/crates/oxide_test262/src/config.rs +++ b/crates/oxide_test262/src/config.rs @@ -1,5 +1,5 @@ //! test262 runner 运行配置：`RunConfig` 结构、CLI 解析（`parse`）与用法输出（`usage`）。 -//! 7 字段与 3 关联函数放宽为 `pub(crate)`；`usage` 全文即 `--help` 用户可见输出，逐字保留。 +//! 8 字段与 3 关联函数放宽为 `pub(crate)`；`usage` 全文即 `--help` 用户可见输出，逐字保留。 use std::path::PathBuf; @@ -10,6 +10,8 @@ pub(crate) struct RunConfig { pub(crate) filter: Option, pub(crate) no_skip: bool, pub(crate) supervise: bool, + /// 远程分片 worker 模式：连监督者端点收窗口指派帧，in-process 执行并发帧。 + pub(crate) worker: bool, /// 关闭 liveness/精确 DCE/RegAlloc 链（开关对比：关优化链 vs 开优化链各跑一遍）。 pub(crate) no_regalloc: bool, /// 逐测试打印 PASS/FAIL/SKIP（开关对比时对比两轮结果集合用）。 @@ -26,6 +28,7 @@ impl RunConfig { filter: None, no_skip: false, supervise: false, + worker: false, no_regalloc: false, verbose: false, no_fail_list: false, @@ -49,6 +52,7 @@ impl RunConfig { "--no-regalloc" => config.no_regalloc = true, "--verbose" => config.verbose = true, "--supervise" => config.supervise = true, + "--worker" => config.worker = true, "--no-fail-list" => config.no_fail_list = true, "--help" | "-h" => return Err(Self::usage()), _ if arg.starts_with("--") => return Err(format!("unknown option: {arg}\n\n{}", Self::usage())), @@ -56,6 +60,12 @@ impl RunConfig { } } + // worker 与 supervise 互斥：前者是远程分片的执行半区（连监督者端点）， + // 后者是本地子进程监督，两者不共存。 + if config.worker && config.supervise { + return Err(format!("--worker and --supervise are mutually exclusive\n\n{}", Self::usage())); + } + if let Some(root) = positional.first() { config.test262_root = Some(PathBuf::from(root)); } @@ -71,7 +81,7 @@ impl RunConfig { /// 打印用法说明。 pub(crate) fn usage() -> String { - "usage: test262-runner [--no-skip] [--no-regalloc] [--verbose] [--supervise] [--version] [test262-root] [path-filter]\n\ + "usage: test262-runner [--no-skip] [--no-regalloc] [--verbose] [--supervise] [--worker] [--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\ @@ -81,6 +91,9 @@ impl RunConfig { --supervise Run the suite as single-worker child-process windows with a hard per-test timeout and\n\ \x20 automatic resume past any hanging/crashing test. A hang or crash is reported by path.\n\ \x20 Combines with [path-filter]: the filter is applied per-test inside each child window.\n\ + --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\ \n\ env tunables (all optional):\n\ \x20 OXIDE_LOG per-subsystem log levels (default off)\n\ @@ -95,6 +108,7 @@ impl RunConfig { \x20 OXIDE_TEST262_CHUNK_SIZE tests per child in chunked mode\n\ \x20 OXIDE_TEST262_CHILD_CHUNK marks a spawned child as chunk worker (any value)\n\ \x20 OXIDE_TEST262_ALLOW_FAIL_EXIT child may exit 1 on fail (any value)\n\ + \x20 OXIDE_TEST262_WORKER_ENDPOINT supervisor endpoint for --worker mode (host:port)\n\ \x20 OXIDE_TEST262_KERNEL_BATCH kernel rebuilds every N tests (default 256, clamped to the 512 realm bound)\n\ \x20 OXIDE_TEST262_LOG_LEVEL runner log level (default info)\n\ \x20 OXIDE_SKIP_UNTIL skip test indexes below N (chunking/supervise resume)\n\ @@ -137,6 +151,17 @@ mod tests { assert_eq!(cfg.filter.as_deref(), Some("language")); } + /// --worker 旗标置位；与 --supervise 同时给返回互斥错误。 + #[test] + fn parse_worker_flag_and_supervise_exclusive() { + let cfg = RunConfig::parse(&args(&["--worker"])).expect("worker flag should parse"); + assert!(cfg.worker, "worker 旗标应置位"); + assert!(!cfg.supervise, "supervise 旗标不应置位"); + + let err = RunConfig::parse(&args(&["--worker", "--supervise"])).expect_err("worker+supervise must be exclusive"); + assert!(err.contains("mutually exclusive"), "应提示互斥：{err}"); + } + /// 未知选项返回含 "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 d81aa02d..335bee61 100644 --- a/crates/oxide_test262/src/main.rs +++ b/crates/oxide_test262/src/main.rs @@ -1,7 +1,6 @@ #![allow(clippy::arc_with_non_send_sync)] use std::path::PathBuf; -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, RwLock}; use std::time::Duration; @@ -19,12 +18,13 @@ mod stats; mod supervise; mod test262_log; mod window_transport; +mod worker; use chunked::run_chunked; use config::RunConfig; use discovery::discover_tests; use harness::{HarnessModuleCache, HarnessPrefixCache, HarnessSources, HARNESS}; use heartbeat::write_heartbeat; -use judge::{categorize_fail, TestOutcome}; +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}; @@ -71,6 +71,95 @@ fn print_fail_categories(stats: &RunStats, paths: &[PathBuf]) { } } +/// 单 worker 窗口执行参数：路径清单、过滤器、各开关、kernel 批次与共享缓存。 +/// `harness_cache` 与 `harness_module_cache` 在并行模式跨 worker 共享， +/// worker 模式单 worker 独享。 +struct WorkerParams { + paths: Vec, + filter: Option, + no_skip: bool, + no_regalloc: bool, + verbose: bool, + log_running_tests: bool, + kernel_batch: usize, + harness_cache: Arc>, + harness_module_cache: Arc>, +} + +/// 在窗口 `[start, end)` 上顺序执行测试，返回窗口聚合统计。 +/// +/// 窗口区间是测试下标范围；引擎参数经 `WorkerParams` 传入；心跳回调在每个测试 +/// 完成后（`stats.record` 之后）调用，返回需累加到 `hb_write_errors` 的写失败数。 +/// kernel 重建与批内 sweep 时机与并行模式逐位一致（每 `kernel_batch` 测试重建， +/// 批内每 500 测试 sweep）。 +fn run_worker_window( + start: usize, + end: usize, + params: &WorkerParams, + on_heartbeat: &mut impl FnMut(usize, &TestResult, &RunStats) -> usize, +) -> RunStats { + let mut engine = build_runner_engine(); + let harness_sources = HARNESS.get_or_init(HarnessSources::new); + let mut stats = RunStats::default(); + let mut tests_since_kernel_reset = 0usize; + + for i in start..end { + let path_str = params.paths[i].display().to_string(); + // 始终记录当前测试路径，使 panic hook 能标识 Rust panic。 + // 诊断需要最后一行 stderr 的 OS 级崩溃时设置 OXIDE_TEST262_RUNNING_LOG=1。 + CURRENT_TEST_PATH.with(|p| *p.borrow_mut() = path_str.clone()); + if params.log_running_tests { + test262_debug!("running: {}", path_str); + eprintln!(" [worker] running: {path_str}"); + } + + let result = process_path( + ¶ms.paths[i], + ¶ms.filter, + params.no_skip, + params.no_regalloc, + &engine, + harness_sources, + ¶ms.harness_cache, + ¶ms.harness_module_cache, + ); + stats.record(i, &result); + // COMPLETED 心跳：先写心跳再追加旁路失败行（崩溃恢复 + // 无漏项、无双计的写序契约）。 + stats.hb_write_errors += on_heartbeat(i, &result, &stats); + if params.verbose { + match &result.outcome { + TestOutcome::Pass(_) => println!("PASS {}", params.paths[i].display()), + TestOutcome::Fail(msg) => { + let (cat, _) = categorize_fail(msg); + println!("FAIL {} [{}] {}", params.paths[i].display(), cat, first_line(msg)); + } + TestOutcome::Skip(_) => println!("SKIP {}", params.paths[i].display()), + } + } + tests_since_kernel_reset += 1; + + if tests_since_kernel_reset >= params.kernel_batch { + // 重建即全新 forge（结构性必清）：旧核连同其表整体丢弃， + // 先 sweep 是对 doomed 核白做一次 O(50k) 清理。 + // 重建边界契约：旧核上无存活 VM（VM 每测试局部，此处作用域外）。 + debug_assert!(engine.kernel().active_vms() == 0, "kernel rebuild requires no live VMs"); + engine = build_runner_engine(); + tests_since_kernel_reset = 0; + } else if tests_since_kernel_reset % 500 == 0 { + engine.kernel().sweep_runner_forges(); // 批内 50k 兜底（数据依赖） + } + let done = i - start + 1; + let total = end - start; + if done % 500 == 0 || done == total { + test262_info!("progress: {}/{} ({}%)", done, total, done * 100 / total); + eprintln!(" progress: {done}/{total} ({}%)", done * 100 / total); + } + } + + stats +} + /// 测试主流程：解析配置 → 发现测试 → 按 supervise/chunked/并行 三种模式执行 → /// 汇总并打印统计；任何失败使返回值为 false（进程退出码 1）。 fn run_tests() -> bool { @@ -111,6 +200,14 @@ fn run_tests() -> bool { levels: [log_level; SUBSYSTEM_COUNT], }); + // worker 模式：远程分片的执行半区，连监督者端点收窗口指派帧，in-process + // 执行并发帧。不依赖本地 test262 根目录与路径发现（路径随指派帧携带）。 + if config.worker { + test262_info!("worker mode enabled"); + eprintln!("worker mode enabled: remote shard worker, in-process window execution"); + return worker::run_worker(&config); + } + let test262_root = if let Some(root) = config.test262_root.clone() { root } else { @@ -231,65 +328,41 @@ fn run_tests() -> bool { eprintln!("running on {workers} worker thread(s)"); test262_info!("kernel reset batch: {} test(s)", kernel_batch); eprintln!("kernel reset batch: {kernel_batch} test(s)"); - let cursor = AtomicUsize::new(skip_until); - let progress = AtomicUsize::new(skip_until); - let filter = &filter; - let no_skip = config.no_skip; - let no_regalloc = config.no_regalloc; - let verbose = config.verbose; - let paths_ref = &paths; + + let params = WorkerParams { + paths: paths.clone(), + filter: filter.clone(), + no_skip: config.no_skip, + no_regalloc: config.no_regalloc, + verbose: config.verbose, + log_running_tests, + kernel_batch, + harness_cache: Arc::new(RwLock::new(HarnessPrefixCache::new())), + harness_module_cache: Arc::new(RwLock::new(HarnessModuleCache::new())), + }; let heartbeat_ref = &heartbeat_path; - let harness_cache = Arc::new(RwLock::new(HarnessPrefixCache::new())); - let harness_module_cache = Arc::new(RwLock::new(HarnessModuleCache::new())); + + // 静态窗口划分：把 [skip_until, end_index) 切成 workers 个连续窗口， + // 每 worker 顺序处理一个窗口（动态负载均衡改静态窗口，测试结果不变）。 + let total_window = end_index - skip_until; + let base = total_window / workers; + let rem = total_window % workers; // 保持内存平稳：worker 只返回聚合统计。保留数万个 `TestResult` 会使 // `--no-skip` 运行在套件末尾附近累积 path/error 字符串直至进程 OOM。 let partials: Vec = std::thread::scope(|scope| { let handles: Vec<_> = (0..workers) - .map(|_| { - let cursor = &cursor; - let progress = &progress; - let harness_cache = Arc::clone(&harness_cache); - let harness_module_cache = Arc::clone(&harness_module_cache); + .map(|w| { + let start = skip_until + w * base + w.min(rem); + let end = start + base + if w < rem { 1 } else { 0 }; + let params_ref = ¶ms; // 与 main() 的 16MB 栈一致：VM 在深层嵌套测试程序上递归， // 默认 worker 栈会溢出。 std::thread::Builder::new() .stack_size(16 * 1024 * 1024) .spawn_scoped(scope, move || { - let mut engine = build_runner_engine(); - let harness_sources = HARNESS.get_or_init(HarnessSources::new); - let mut stats = RunStats::default(); - let mut tests_since_kernel_reset = 0usize; - - let tid = std::thread::current().id(); - loop { - let i = cursor.fetch_add(1, Ordering::Relaxed); - if i >= end_index { - break; - } - let path_str = paths_ref[i].display().to_string(); - // 始终记录当前测试路径，使 panic hook 能标识 Rust panic。 - // 诊断需要最后一行 stderr 的 OS 级崩溃时设置 - // OXIDE_TEST262_RUNNING_LOG=1。 - CURRENT_TEST_PATH.with(|p| *p.borrow_mut() = path_str.clone()); - if log_running_tests { - test262_debug!("running: {}", path_str); - eprintln!(" [{tid:?}] running: {path_str}"); - } - - let result = process_path( - &paths_ref[i], - filter, - no_skip, - no_regalloc, - &engine, - harness_sources, - &harness_cache, - &harness_module_cache, - ); - stats.record(i, &result); - // COMPLETED 心跳：先写心跳再追加旁路失败行（崩溃恢复 - // 无漏项、无双计的写序契约）。 + let mut on_heartbeat = |i: usize, result: &TestResult, stats: &RunStats| -> usize { + let mut errors = 0; if let Some(hb) = heartbeat_ref { if let Err(e) = write_heartbeat( hb, @@ -301,49 +374,21 @@ fn run_tests() -> bool { &stats.fail_categories, stats.hb_write_errors, ) { - stats.hb_write_errors += 1; + errors += 1; eprintln!(" [warn] heartbeat write failed at #{i}: {e}"); } if let TestOutcome::Fail(msg) = &result.outcome { // 旁路行携带真 subkey，父进程并入 fail_records 后直接生效 subkey 分组。 let (cat, subkey) = categorize_fail(msg); - if let Err(e) = append_fail_log(&hb.with_extension("fails"), i, &cat, &subkey, msg) - { - stats.hb_write_errors += 1; + if let Err(e) = append_fail_log(&hb.with_extension("fails"), i, &cat, &subkey, msg) { + errors += 1; eprintln!(" [warn] fail log append failed at #{i}: {e}"); } } } - if verbose { - match &result.outcome { - TestOutcome::Pass(_) => println!("PASS {}", paths_ref[i].display()), - TestOutcome::Fail(msg) => { - let (cat, _) = categorize_fail(msg); - println!("FAIL {} [{}] {}", paths_ref[i].display(), cat, first_line(msg)); - } - TestOutcome::Skip(_) => println!("SKIP {}", paths_ref[i].display()), - } - } - tests_since_kernel_reset += 1; - - let done = progress.fetch_add(1, Ordering::Relaxed) + 1; - if tests_since_kernel_reset >= kernel_batch { - // 重建即全新 forge（结构性必清）：旧核连同其表整体丢弃， - // 先 sweep 是对 doomed 核白做一次 O(50k) 清理。 - // 重建边界契约：旧核上无存活 VM（VM 每测试局部，此处作用域外）。 - debug_assert!(engine.kernel().active_vms() == 0, "kernel rebuild requires no live VMs"); - engine = build_runner_engine(); - tests_since_kernel_reset = 0; - } else if tests_since_kernel_reset % 500 == 0 { - engine.kernel().sweep_runner_forges(); // 批内 50k 兜底（数据依赖） - } - if done % 500 == 0 || done == total { - test262_info!("progress: {}/{} ({}%)", done, total, done * 100 / total); - eprintln!(" progress: {done}/{total} ({}%)", done * 100 / total); - } - } - - stats + errors + }; + run_worker_window(start, end, params_ref, &mut on_heartbeat) }) .expect("failed to spawn test262 worker thread") }) diff --git a/crates/oxide_test262/src/worker.rs b/crates/oxide_test262/src/worker.rs new file mode 100644 index 00000000..409959cc --- /dev/null +++ b/crates/oxide_test262/src/worker.rs @@ -0,0 +1,229 @@ +//! Worker 模式：远程分片的执行半区。连监督者端点收窗口指派帧，in-process +//! 执行测试，发心跳/失败/现场/DONE 帧。 +//! +//! 看门狗线程独立读控制帧（kill → 置标志），主线程在测试间隙检查标志后从 +//! 主线程上下文发现场帧再退出。`KernelCore` 是 `!Send`，看门狗不得触引擎； +//! 现场帧内容来自主线程上下文（last-pc 采样行）。断连按死亡处理（恢复由 +//! 监督者侧同构完成）。 + +use std::net::TcpStream; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{mpsc, Arc, RwLock}; + +use crate::config::RunConfig; +use crate::harness::{HarnessModuleCache, HarnessPrefixCache, HarnessSources, HARNESS}; +use crate::judge::{categorize_fail, TestOutcome}; +use crate::runner::{build_runner_engine, process_path, CURRENT_TEST_PATH}; +use crate::shard_protocol::{encode_event, parse_command, FrameReader, ShardCommand, ShardEvent}; +use crate::stats::RunStats; +use oxide_net::Conn; + +/// 窗口指派帧载荷：窗口起始下标与路径清单（`wend` 与 `timeout_secs` 由 +/// 监督者侧消费，worker 侧按路径清单长度执行，超时由监督者发 kill 帧兜底）。 +struct AssignWindow { + wstart: usize, + paths: Vec, +} + +/// 看门狗到主线程的通道消息：窗口指派或 kill 信号。 +/// kill 信号使主线程在测试间隙（或窗口间）发现场帧并退出， +/// 不依赖 `kill_flag` 的轮询（主线程阻塞在通道时也能收到）。 +enum ChannelMsg { + Assign(AssignWindow), + Kill, +} + +/// Worker 模式入口：连监督者端点，收窗口指派帧，in-process 执行，发帧。 +/// +/// 端点经 `OXIDE_TEST262_WORKER_ENDPOINT` 环境变量传入（host:port 形态）。 +/// 看门狗线程独立读控制帧（kill → 置标志），主线程在测试间隙检查标志后 +/// 发现场帧再退出。断连按死亡处理（恢复由监督者侧完成）。 +pub(crate) fn run_worker(config: &RunConfig) -> bool { + let endpoint = match std::env::var("OXIDE_TEST262_WORKER_ENDPOINT") { + Ok(ep) => ep, + Err(_) => { + eprintln!("worker mode requires OXIDE_TEST262_WORKER_ENDPOINT (host:port)"); + return false; + } + }; + + let stream = match TcpStream::connect(&endpoint) { + Ok(s) => s, + Err(e) => { + eprintln!("worker: failed to connect to {endpoint}: {e}"); + return false; + } + }; + let mut conn = Conn::Tcp(stream); + + // 看门狗线程：独立读控制帧。指派帧经通道转主线程（主线程持引擎， + // 看门狗不得触引擎）；kill 帧置标志并经通道发 kill 信号（主线程阻塞 + // 在通道时也能收到，不依赖标志轮询）。 + let kill_flag = Arc::new(AtomicBool::new(false)); + let (msg_tx, msg_rx) = mpsc::channel::(); + let watchdog_kill = Arc::clone(&kill_flag); + let watchdog_tx = msg_tx; + let watchdog_conn = conn.try_clone().expect("failed to clone conn for watchdog"); + let watchdog_buf = std::io::BufReader::new(watchdog_conn); + std::thread::spawn(move || { + let mut reader = FrameReader::new(watchdog_buf); + loop { + match reader.read_frame() { + Ok(Some(frame)) => match parse_command(&frame) { + Ok(ShardCommand::Assign { wstart, wend: _, paths, timeout_secs: _ }) => { + let _ = watchdog_tx.send(ChannelMsg::Assign(AssignWindow { wstart, paths })); + } + Ok(ShardCommand::Kill) => { + watchdog_kill.store(true, Ordering::SeqCst); + let _ = watchdog_tx.send(ChannelMsg::Kill); + break; + } + Err(_) => continue, + }, + Ok(None) => break, + Err(_) => break, + } + } + }); + + // 主线程：收窗口指派帧，in-process 执行，发心跳/失败/现场/DONE 帧。 + let mut engine = build_runner_engine(); + let harness_sources = HARNESS.get_or_init(HarnessSources::new); + let harness_cache = Arc::new(RwLock::new(HarnessPrefixCache::new())); + let harness_module_cache = Arc::new(RwLock::new(HarnessModuleCache::new())); + let kernel_batch = std::env::var("OXIDE_TEST262_KERNEL_BATCH") + .ok() + .and_then(|s| s.parse::().ok()) + .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 { + let window = match msg { + ChannelMsg::Assign(window) => window, + ChannelMsg::Kill => { + // 窗口间收到 kill：发现场帧并退出。 + send_scene_frame(&mut conn); + return true; + } + }; + for (offset, path_str) in window.paths.iter().enumerate() { + let i = window.wstart + offset; + // 检查 kill 标志（看门狗读 kill 帧后置位）。 + if kill_flag.load(Ordering::SeqCst) { + send_scene_frame(&mut conn); + return true; + } + + let path = PathBuf::from(path_str); + let path_str = path.display().to_string(); + CURRENT_TEST_PATH.with(|p| *p.borrow_mut() = path_str.clone()); + + let result = process_path( + &path, + &None, + config.no_skip, + config.no_regalloc, + &engine, + harness_sources, + &harness_cache, + &harness_module_cache, + ); + stats.record(i, &result); + + // 发心跳帧（COMPLETED 阶段，计数覆盖 ≤ i 的全部测试）。 + stats.hb_write_errors += send_heartbeat_frame(&mut conn, i, &stats); + + // 发失败记录帧（一测试失败一帧）。 + if let TestOutcome::Fail(msg) = &result.outcome { + let (cat, subkey) = categorize_fail(msg); + stats.hb_write_errors += send_fail_frame(&mut conn, i, &cat, &subkey, msg); + } + + tests_since_kernel_reset += 1; + if tests_since_kernel_reset >= kernel_batch { + debug_assert!(engine.kernel().active_vms() == 0, "kernel rebuild requires no live VMs"); + engine = build_runner_engine(); + tests_since_kernel_reset = 0; + } else if tests_since_kernel_reset % 500 == 0 { + engine.kernel().sweep_runner_forges(); + } + } + + // 发 DONE 帧（退出前 flush，监督者不致误判死亡重跑窗口）。 + if let Err(e) = send_event_frame(&mut conn, &ShardEvent::Done) { + eprintln!("worker: failed to send DONE frame: {e}"); + } + } + + true +} + +/// 发送事件帧（write_all 即 flush）。 +fn send_event_frame(conn: &mut Conn, event: &ShardEvent) -> std::io::Result<()> { + let frame = encode_event(event); + conn.write_all(frame.as_bytes()) +} + +/// 发心跳帧；返回写失败数（0 或 1）。 +fn send_heartbeat_frame(conn: &mut Conn, index: usize, stats: &RunStats) -> usize { + let event = ShardEvent::Heartbeat { + phase: "COMPLETED".into(), + index, + pass: stats.pass, + fail: stats.fail, + skip: stats.skip, + categories: stats.fail_categories.clone(), + hb_write_errors: stats.hb_write_errors, + }; + match send_event_frame(conn, &event) { + Ok(()) => 0, + Err(e) => { + eprintln!("worker: failed to send heartbeat frame at #{index}: {e}"); + 1 + } + } +} + +/// 发失败记录帧；返回写失败数（0 或 1）。 +fn send_fail_frame(conn: &mut Conn, index: usize, category: &str, subkey: &str, message: &str) -> usize { + let event = ShardEvent::Fail { + index, + category: category.to_string(), + subkey: subkey.to_string(), + message: message.to_string(), + }; + match send_event_frame(conn, &event) { + Ok(()) => 0, + Err(e) => { + eprintln!("worker: failed to send fail frame at #{index}: {e}"); + 1 + } + } +} + +/// 发现场帧：读 last-pc 现场文件末行，作为现场帧发出。 +/// +/// 须主线程上下文调用（引擎 `!Send`，看门狗不得触引擎）。现场文件缺或 +/// 末行不合法时静默返回（监督者侧按无现场处理）。 +fn send_scene_frame(conn: &mut Conn) { + let pc_path = match std::env::var_os("OXIDE_TEST262_PC_WATCH") { + Some(p) => PathBuf::from(p), + None => return, + }; + let content = match std::fs::read_to_string(&pc_path) { + Ok(c) => c, + Err(_) => return, + }; + let last = content.lines().last().unwrap_or("").trim(); + if last.is_empty() { + return; + } + let event = ShardEvent::Scene { line: last.to_string(), log_tail: None }; + if let Err(e) = send_event_frame(conn, &event) { + eprintln!("worker: failed to send scene frame: {e}"); + } +}