diff --git a/Cargo.lock b/Cargo.lock index 24070ed9..a2e1886a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.8.92" +version = "0.8.93" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.8.92" +version = "0.8.93" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.8.92" +version = "0.8.93" dependencies = [ "ansi_term", "bumpalo", @@ -1226,7 +1226,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.8.92" +version = "0.8.93" dependencies = [ "lru", "oxide_bytecode", @@ -1235,7 +1235,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1255,7 +1255,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1269,7 +1269,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.8.92" +version = "0.8.93" dependencies = [ "num-bigint", "oxc_allocator", @@ -1284,7 +1284,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1296,7 +1296,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1305,7 +1305,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.8.92" +version = "0.8.93" dependencies = [ "bumpalo", "dashmap", @@ -1319,7 +1319,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1331,7 +1331,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.8.92" +version = "0.8.93" dependencies = [ "tracing", "tracing-appender", @@ -1340,7 +1340,7 @@ dependencies = [ [[package]] name = "oxide_parser" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1371,7 +1371,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.8.92" +version = "0.8.93" dependencies = [ "num-bigint", "num-traits", @@ -1384,7 +1384,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.8.92" +version = "0.8.93" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1402,7 +1402,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.8.92" +version = "0.8.93" dependencies = [ "bumpalo", "dashmap", @@ -1415,7 +1415,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.8.92" +version = "0.8.93" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 8217a407..1a44997c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,7 @@ members = [ ] [workspace.package] -version = "0.8.92" +version = "0.8.93" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_cli/src/main.rs b/crates/oxide_cli/src/main.rs index 9ea160b5..22d00561 100644 --- a/crates/oxide_cli/src/main.rs +++ b/crates/oxide_cli/src/main.rs @@ -645,6 +645,9 @@ fn eval(code: &str, engine: &Engine, trace: bool, profile: bool, print_result: b let kernel = engine.kernel(); format_result(guard.vm(), kernel.perm_interner().as_ref(), kernel.shape_forge().as_ref(), result); } + // 主线程事件循环:消费 worker 消息、清理已断开 worker,直至无活跃 + // worker(脚本建了 Worker 不 terminate 时恒转,符合页面寿命语义)。 + run_worker_event_loop(guard.vm_mut()); ExitCode::SUCCESS } Err(err) => { @@ -655,6 +658,27 @@ fn eval(code: &str, engine: &Engine, trace: bool, profile: bool, print_result: b } } +/// 主线程事件循环:消费 worker 消息并清理已断开 worker,循环直至无活跃 worker。 +/// +/// # 步骤 +/// 1. 循环条件:仍有活跃 worker。 +/// 2. `deliver_worker_messages` 交付 worker 消息到 `onmessage`。 +/// 3. `cleanup_disconnected_workers` 清理已退出 worker(通道断开即 join 加移除)。 +/// 4. 本轮未交付消息时 1ms 轮询(避免忙等)。 +/// +/// # 边界与前提 +/// - 终止条件为无活跃 worker;用户脚本建 Worker 不 terminate 时恒转(有意, +/// 页面寿命 = worker 寿命)。 +fn run_worker_event_loop(vm: &mut Vm) { + while !vm.active_workers().is_empty() { + let delivered = vm.deliver_worker_messages(); + vm.cleanup_disconnected_workers(); + if !delivered { + std::thread::sleep(Duration::from_millis(1)); + } + } +} + fn format_result(vm: &oxide_vm::vm::Vm, string_forge: &PermInterner, shape_forge: &ShapeForge, val: JsValue) { println!("{}", format_js_value(vm, string_forge, shape_forge, val)); } diff --git a/crates/oxide_vm/src/bindings/bind_worker.rs b/crates/oxide_vm/src/bindings/bind_worker.rs index 107144c9..fdf60cc1 100644 --- a/crates/oxide_vm/src/bindings/bind_worker.rs +++ b/crates/oxide_vm/src/bindings/bind_worker.rs @@ -156,6 +156,18 @@ pub fn bind_worker(core: &Arc, session: &KernelSession, global: &mut ], ); } + // self 全局属性:值为 global 对象本身(worker realm 经 self === global 可达, + // self.onmessage / self.postMessage 由此成立)。幂等守卫。 + let si_self = sf.intern("self").0; + if sh.lookup_position(global.shape_id(), si_self).is_none() { + let global_ptr = global as *mut JsObject; + let shape = sh.make_shape(global.shape_id(), si_self); + global.set_shape_id(shape); + global.ensure_hash_props().push(JsValue::from_js_object(global_ptr)); + let pos = global.hash_props_vec().map_or(0, |v| v.len() as u32).saturating_sub(1); + global.set_data_meta(pos, PropAttributes::new(true, false, true)); + global.bump_generation(); + } // name / location getter(幂等守卫)。 // let si_name_global = sf.intern("name").0; // if sh.lookup_position(global.shape_id(), si_name_global).is_none() { diff --git a/crates/oxide_vm/src/vm/gc_hooks.rs b/crates/oxide_vm/src/vm/gc_hooks.rs index 91e08fd2..7978ca2b 100644 --- a/crates/oxide_vm/src/vm/gc_hooks.rs +++ b/crates/oxide_vm/src/vm/gc_hooks.rs @@ -8,7 +8,7 @@ use super::frames::Completion; use super::Vm; use crate::session_gc::SessionGc; -/// GC 根组来源:`for_each_value` 枚举的 24 个根组,变体顺序与根清单 +/// GC 根组来源:`for_each_value` 枚举的 25 个根组,变体顺序与根清单 /// 的枚举顺序一致。 /// /// 变体下标是逐组计数数组(`SessionGc::root_counts`)的下标;`COUNT` 与 @@ -39,19 +39,20 @@ pub(crate) enum RootGroup { AtomicsWaiters, ForInIters, Global, + WorkerObjects, } impl RootGroup { /// 根组总数(与变体数同源,绑定逐组计数数组的长度)。 - pub const COUNT: usize = 24; + pub const COUNT: usize = 25; } impl Vm { /// GC 根收集的统一遍历(对象与字符串都产出)。 /// 覆盖执行核心的全部 JsValue 持有点:regs/帧/各栈段/cell/在途异常与完成/ - /// 挂起信号/迭代器/微任务/global。 + /// 挂起信号/迭代器/微任务/global/Worker 对象注册表。 /// - /// 闭包接收根组来源(`RootGroup`)与根值:组是枚举顺序中的 24 根组之一, + /// 闭包接收根组来源(`RootGroup`)与根值:组是枚举顺序中的 25 根组之一, /// 调用方按组计数(如 `SessionGc` 的逐组计数)。 pub(crate) fn for_each_value(&self, mut f: impl FnMut(RootGroup, JsValue)) { for value in &self.regs { @@ -157,6 +158,11 @@ impl Vm { RootGroup::Global, JsValue::from_js_object(self.realm.session.borrow().global_object().as_ptr() as *mut JsObject), ); + // Worker 对象注册表:Worker 对象是 GC 根(注册表保活至 terminate / + // full_reset 清表),漏根 → sweep 释放 → 事件循环反查悬垂。 + for &worker_obj in self.worker_objects.values() { + f(RootGroup::WorkerObjects, worker_obj); + } } /// GC 根统一枚举入口:遍历的字段清单与 `for_each_value` 相同。 diff --git a/crates/oxide_vm/src/vm/mod.rs b/crates/oxide_vm/src/vm/mod.rs index ba016a2d..8b5eae17 100644 --- a/crates/oxide_vm/src/vm/mod.rs +++ b/crates/oxide_vm/src/vm/mod.rs @@ -390,6 +390,11 @@ pub struct Vm { pub(crate) worker_registry: std::collections::HashMap, /// 下一个 worker 编号(单调递增,首个为 0)。 pub(crate) worker_next_id: u64, + /// Worker 对象注册表:键 = worker 编号,值 = Worker 对象(主 realm session + /// 对象)。主线程事件循环据编号反查 Worker 对象读 `onmessage`。Worker 对象 + /// 是 GC 根(`for_each_value` 遍历),注册表保活至 `worker_terminate` / + /// `full_reset` 清表,无悬垂风险。 + pub(crate) worker_objects: std::collections::HashMap, } impl Drop for Vm { @@ -438,6 +443,10 @@ impl Drop for Vm { // poll_worker_messages 访问、不跨线程移动;MessageValue 是 Send 中间表示 // (无 realm 局部指针)。跨线程移动 Vm 时 worker_registry 整体迁移, // 其中无指向 session 堆的裸指针,不引入别名。 +// +// 6. worker_objects 字段(HashMap)持本 Vm session 堆的 Worker +// 对象指针,归不变量 1 覆盖:对象由本 Vm 独占分配、地址稳定、随 Vm 整体 +// 迁移,跨线程移动不悬垂。 unsafe impl Send for Vm {} impl Vm { diff --git a/crates/oxide_vm/src/vm_support.rs b/crates/oxide_vm/src/vm_support.rs index 110c5b2d..b1fede22 100644 --- a/crates/oxide_vm/src/vm_support.rs +++ b/crates/oxide_vm/src/vm_support.rs @@ -193,6 +193,7 @@ impl Vm { max_steps_override: None, worker_registry: std::collections::HashMap::new(), worker_next_id: 0, + worker_objects: std::collections::HashMap::new(), }; vm.init_generator_intrinsics(); vm.init_promise_intrinsics(); @@ -352,6 +353,7 @@ impl Vm { max_steps_override: None, worker_registry: std::collections::HashMap::new(), worker_next_id: 0, + worker_objects: std::collections::HashMap::new(), }; vm.init_generator_intrinsics(); vm.init_promise_intrinsics(); @@ -392,6 +394,9 @@ impl Vm { /// /// 用于在多次 JS 执行之间达到完全隔离:session 内未被污染的 builtin 保留原指针。 pub fn full_reset(&mut self) { + // 池回收路径:若有活跃 worker,先终止并 join 全部(防线程泄漏与注册表 + // 残留:worker 线程各持自有 Vm,不 join 则线程孤儿、registry 条目残留)。 + self.shutdown_workers(); // 防御兜底:属性值写原语已推进 generation,常规覆盖写由快照对比发现; // 若未来出现绕过属性写原语的裸属性区改写,session 对象会被 session GC 回收, // 保留 global 将持悬垂指针,故带 session 对象时强制 bump 保证 global 重建。 @@ -473,6 +478,9 @@ impl Vm { // BroadcastChannel 注册表持 session 对象裸指针:teardown 释放全部 // session 对象前须清表,否则残留悬垂指针。 self.realm.gc.borrow_mut().broadcast_channels.clear(); + // Worker 对象注册表持 session 对象指针:teardown 释放全部 session 对象 + // 前须清表,否则残留悬垂指针。 + self.worker_objects.clear(); self.realm.teardown_session_heap_data(); self.realm.gc.borrow_mut().session_bytes_allocated = 0; self.realm.gc.borrow_mut().session_bytes_peak = 0; diff --git a/crates/oxide_vm/src/worker/bindings.rs b/crates/oxide_vm/src/worker/bindings.rs index ba5a9de2..b60ebfe9 100644 --- a/crates/oxide_vm/src/worker/bindings.rs +++ b/crates/oxide_vm/src/worker/bindings.rs @@ -72,7 +72,8 @@ pub(crate) fn worker_close_requested() -> bool { /// - Worker 对象经 `alloc_object` 入对象表即成 GC 根;`workerId` 是数值(无 GC 边)。 /// /// # 副作用 -/// - 派生一个 OS 线程(`spawn_worker`);登记一个 `WorkerHandle`。 +/// - 派生一个 OS 线程(`spawn_worker`);登记一个 `WorkerHandle` 与一条 +/// Worker 对象注册表条目(GC 根)。 pub(crate) fn worker_constructor(vm: &mut Vm, args: &[u8]) -> NativeResult { // 构造形态校验(HTML 规范只有 [[Construct]])。 if !vm.constructing_native { @@ -132,6 +133,9 @@ pub(crate) fn worker_constructor(vm: &mut Vm, args: &[u8]) -> NativeResult { let si_worker_id = vm.perm_intern("workerId"); let _ = vm.define_data_property(worker_obj, si_worker_id, JsValue::float(id as f64), PropAttributes::new(true, false, true)); + // 登记 Worker 对象进注册表(GC 根,主线程事件循环据编号反查 onmessage)。 + vm.worker_objects.insert(id, JsValue::from_js_object(ptr)); + NativeResult::Ok(JsValue::from_js_object(ptr)) } @@ -293,7 +297,6 @@ pub(crate) fn self_location(vm: &mut Vm, _args: &[u8]) -> NativeResult { /// /// # 副作用 /// - 新建一个 MessageEvent 对象(经 `alloc_object` 入对象表)。 -#[expect(dead_code)] // 供 935.5 主线程事件循环消费(onmessage 交付),935.4 未接线 pub(crate) fn message_event_constructor(vm: &mut Vm, args: &[u8]) -> NativeResult { let data_val = if args.len() > 1 { vm.reg(args[1]) } else { JsValue::undefined() }; let type_val = if args.len() > 2 { vm.reg(args[2]) } else { JsValue::undefined() }; diff --git a/crates/oxide_vm/src/worker/mod.rs b/crates/oxide_vm/src/worker/mod.rs index c6edaef0..06daef0e 100644 --- a/crates/oxide_vm/src/worker/mod.rs +++ b/crates/oxide_vm/src/worker/mod.rs @@ -13,15 +13,15 @@ pub mod bindings; -use std::collections::HashSet; use std::sync::Arc; use std::thread::JoinHandle; use std::time::Duration; -use oxide_builtins::message_value::{detach_message, rehydrate_message, MessageValue}; +use oxide_builtins::message_value::{rehydrate_message, MessageValue}; use oxide_kernel::kernel::KernelCore; use oxide_kernel::message_queue::{channel, Receiver, Sender, Timeout}; -use oxide_runtime_api::CompilerService; +use oxide_runtime_api::{CompilerService, NativeResult, VmHost}; +use oxide_types::object::JsObject; use oxide_types::value::JsValue; use crate::vm::Vm; @@ -108,13 +108,9 @@ fn worker_event_loop( } match rx.recv_timeout(Duration::from_millis(100)) { Ok(WorkerMail::Message(value)) => { - // rehydrate → execute_task → drain_microtasks,再回显处理值。 - let value = rehydrate_message(&mut vm, &value); - let _ = vm.execute_task(|_vm| Ok(value)); - vm.drain_microtasks(); - if let Ok(mv) = detach_message(&mut vm, value, &HashSet::new()) { - let _ = out_tx.send(mv); - } + // rehydrate → 建 MessageEvent → 交付 self.onmessage → drain 微任务 + // (onmessage 语义替换早期回显语义)。 + vm.deliver_self_message(value); } Ok(WorkerMail::Error(err)) => { // 主线程上报的错误记入 last_uncaught_value。 @@ -196,13 +192,14 @@ impl Vm { /// 1. 发 `WorkerMail::Terminate`(worker 收到即退出事件循环)。 /// 2. 从 `worker_registry` 移除(取走 `WorkerHandle`)。 /// 3. join 线程(须在句柄 drop 前完成,防线程泄漏)。 + /// 4. 从 `worker_objects` 移除 Worker 对象注册表条目(GC 根解除)。 /// /// # 边界与前提 /// - `id` 不存在时返回 `Err`。 /// - 线程 panic 时返回 `Err`。 /// /// # 副作用 - /// - 移除一个 `WorkerHandle`;join 一个 OS 线程。 + /// - 移除一个 `WorkerHandle` 与一条 Worker 对象注册表条目;join 一个 OS 线程。 pub fn worker_terminate(&mut self, id: u64) -> Result<(), String> { // 发关停信号(worker 收到即退出事件循环)。 if let Some(worker) = self.worker_registry.get_mut(&id) { @@ -211,6 +208,8 @@ impl Vm { // 从注册表移除并 join 线程(句柄 drop 前必须 join)。 let worker = self.worker_registry.remove(&id).ok_or_else(|| format!("worker {id} 不存在"))?; worker.handle.join().map_err(|_| format!("worker {id} 线程异常退出"))?; + // 解除 Worker 对象注册表条目(GC 根解除,对象可被回收)。 + self.worker_objects.remove(&id); Ok(()) } @@ -266,6 +265,160 @@ impl Vm { let _ = self.worker_terminate(id); } } + + /// 按 worker 编号反查 Worker 对象(注册表查找,GC 根)。 + /// + /// # 返回值 + /// 注册表中该编号的 Worker 对象值;编号不存在时 `None`。 + pub(crate) fn worker_object(&self, id: u64) -> Option { + self.worker_objects.get(&id).copied() + } + + /// 主线程事件循环消息交付:排空 worker → 主线程通道、rehydrate 进主 realm, + /// 交付到 Worker 对象的 `onmessage`。 + /// + /// # 步骤 + /// 1. 遍历活跃 worker 编号。 + /// 2. 对每个 worker `poll_worker_messages` 排空通道并 rehydrate 进主 realm。 + /// 3. 反查 Worker 对象、读 `onmessage` 属性。 + /// 4. `onmessage` 可调用时,逐条消息建 MessageEvent 经 `execute_task` + + /// `call_function_sync` 触发,后 `drain_microtasks`。 + /// + /// # 返回值 + /// 本轮是否交付了消息(false 时调用方 1ms 轮询,避免忙等)。 + /// + /// # 边界与前提 + /// - `onmessage` 缺失或非可调用时静默跳过(消息已消费,符合浏览器 + /// "无 handler 即丢弃"语义)。 + /// - 交付前二次 `worker_object` 校验存活(GC 防护):worker 在事件循环中 + /// 被终止(注册表条目移除)即停止交付。 + /// + /// # 副作用 + /// - 消费 worker → 主线程通道消息;触发 `onmessage` 回调与微任务 drain。 + pub fn deliver_worker_messages(&mut self) -> bool { + let mut any = false; + for id in self.active_workers() { + let messages = self.poll_worker_messages(id); + if messages.is_empty() { + continue; + } + any = true; + // Worker 对象注册表反查(GC 根,注册即存活)。 + let Some(worker_obj) = self.worker_object(id) else { + continue; + }; + // 读 onmessage 属性(undefined 或非可调用即静默跳过)。 + let si_onmessage = self.perm_intern("onmessage"); + let handler = match self.ordinary_get(unsafe { &*worker_obj.as_js_object_ptr() }, si_onmessage, worker_obj) + { + Ok(h) => h, + Err(_) => continue, + }; + if !is_callable(handler) { + continue; + } + for data in messages { + // 交付前二次校验存活(GC 防护):worker 在事件循环中被终止 + // (注册表条目移除)即停止交付。 + if self.worker_object(id).is_none() { + break; + } + let event = self.build_message_event(data, "message"); + let _ = self.execute_task(|vm| vm.call_function_sync(handler, worker_obj, &[event])); + self.drain_microtasks(); + } + } + any + } + + /// 建 MessageEvent 对象(`data` + `type` 属性),委托 935.4 构造辅助。 + /// + /// # 步骤 + /// 1. `type` 串物化为 session 串。 + /// 2. `data` 与 `type` 写入寄存器 1 / 2,委托 `message_event_constructor` + /// (按寄存器下标读取)建对象。 + /// + /// # 返回值 + /// 新建的 MessageEvent 对象值(构造失败时返回错误对象值)。 + /// + /// # 副作用 + /// - 新建一个 MessageEvent 对象(经 `alloc_object` 入对象表)。 + fn build_message_event(&mut self, data: JsValue, type_str: &str) -> JsValue { + let type_val = self.new_string(type_str); + self.set_reg(1, data); + self.set_reg(2, type_val); + match bindings::message_event_constructor(self, &[0, 1, 2]) { + NativeResult::Ok(v) => v, + NativeResult::Err(e) => e, + // 构造辅助只建对象,不产生尾调用(防御臂,不可达)。 + NativeResult::TailCall { .. } => JsValue::undefined(), + } + } + + /// 清理已断开 worker:通道断开即 worker 线程已退出,join 并移除。 + /// + /// # 步骤 + /// 1. 遍历活跃 worker 编号。 + /// 2. 对每个 worker 查 `rx_out.is_disconnected()`(worker 线程退出后其 + /// 发送端 drop,接收端转断开)。 + /// 3. 断开者 `worker_terminate`(发 `Terminate` + join + 移除注册表条目)。 + /// + /// # 副作用 + /// - 移除已断开 worker 的注册表条目;join 其 OS 线程。 + pub fn cleanup_disconnected_workers(&mut self) { + for id in self.active_workers() { + let disconnected = self.worker_registry.get(&id).is_some_and(|w| w.rx_out.is_disconnected()); + if disconnected { + let _ = self.worker_terminate(id); + } + } + } + + /// worker 侧消息交付:rehydrate 进 worker realm、建 MessageEvent、交付到 + /// `self.onmessage`(worker global 的 `onmessage` 属性)。 + /// + /// # 步骤 + /// 1. rehydrate 消息值进 worker realm。 + /// 2. 建 MessageEvent(data + type "message")。 + /// 3. 读 worker global 的 `onmessage` 属性。 + /// 4. 可调用时经 `execute_task` + `call_function_sync` 触发,后 `drain_microtasks`。 + /// + /// # 边界与前提 + /// - `onmessage` 缺失或非可调用时静默跳过(消息已消费,符合浏览器 + /// "无 handler 即丢弃"语义)。 + /// + /// # 副作用 + /// - 触发 `self.onmessage` 回调与微任务 drain。 + pub(crate) fn deliver_self_message(&mut self, value: MessageValue) { + let data = rehydrate_message(self, &value); + let event = self.build_message_event(data, "message"); + // 取 worker global 对象指针(Ref 守卫在语句块内消费,不跨 &mut 借用长存)。 + let global_ptr = { + let session = self.realm.session.borrow(); + session.global_object().as_ptr() as *mut JsObject + }; + // SAFETY: global_ptr 是当前 session 的 global 对象,存活。 + let global_obj = unsafe { &*global_ptr }; + let si_onmessage = self.perm_intern("onmessage"); + let handler = match self.ordinary_get(global_obj, si_onmessage, JsValue::from_js_object(global_ptr)) { + Ok(h) => h, + Err(_) => return, + }; + if !is_callable(handler) { + return; + } + let _ = self.execute_task(|vm| vm.call_function_sync(handler, JsValue::from_js_object(global_ptr), &[event])); + self.drain_microtasks(); + } +} + +/// 判定值是否可调用(对象且函数对象,header bit 31)。 +/// +/// # 边界与前提 +/// - 非对象值恒不可调用。 +/// - 对象经 `as_js_object_ptr` 解引用读函数位(unsafe,惯例形态)。 +fn is_callable(val: JsValue) -> bool { + val.is_object() && unsafe { &*val.as_js_object_ptr() }.is_function() } #[cfg(test)] @@ -273,8 +426,10 @@ mod tests { use super::*; use std::time::Instant; + use oxide_compiler::compiler::Compiler; use oxide_compiler::DefaultCompilerService; use oxide_kernel::kernel::KernelConfig; + use oxide_parser::Allocator; /// 建一个带真实编译服务的 Vm(worker 脚本编译需要)。 fn vm_with_compiler() -> Vm { @@ -296,12 +451,15 @@ mod tests { } } - /// 端到端:spawn → post → 回显 → terminate,验证 worker 基础设施闭环。 + /// 端到端:spawn → post → onmessage 回显 → terminate,验证 worker 基础设施闭环。 #[test] fn worker_round_trip() { let mut vm = vm_with_compiler(); - let id = vm.spawn_worker("1 + 1").expect("worker 应派生成功"); + // worker 脚本设 onmessage 回显(onmessage 语义替换早期回显语义)。 + let id = vm + .spawn_worker("self.onmessage = function(e) { self.postMessage(e.data); };") + .expect("worker 应派生成功"); assert_eq!(vm.active_workers(), vec![id], "应有唯一活跃 worker"); vm.worker_post_message(id, MessageValue::Number(42.0)).expect("投递应成功"); @@ -332,8 +490,13 @@ mod tests { fn multiple_workers_isolated() { let mut vm = vm_with_compiler(); - let id_a = vm.spawn_worker("1").expect("worker A 应派生成功"); - let id_b = vm.spawn_worker("2").expect("worker B 应派生成功"); + // 两 worker 各设 onmessage 回显(onmessage 语义)。 + let id_a = vm + .spawn_worker("self.onmessage = function(e) { self.postMessage(e.data); };") + .expect("worker A 应派生成功"); + let id_b = vm + .spawn_worker("self.onmessage = function(e) { self.postMessage(e.data); };") + .expect("worker B 应派生成功"); assert_eq!(vm.active_workers(), vec![id_a, id_b], "应有两个活跃 worker"); vm.worker_post_message(id_a, MessageValue::Number(1.0)).expect("投递 A 应成功"); @@ -347,4 +510,145 @@ mod tests { vm.shutdown_workers(); assert!(vm.active_workers().is_empty(), "shutdown 后无活跃 worker"); } + + /// 编译主脚本源码为 `CompiledModule`(测试驱动主 Vm 执行)。 + fn compile_script(source: &str) -> oxide_bytecode::module::CompiledModule { + let allocator = Allocator::default(); + let program = oxide_parser::parse(&allocator, source).expect("parse 应成功"); + Compiler::new().compile(&program).expect("compile 应成功") + } + + /// 写 worker 脚本到临时文件(进程 id 加纳秒唯一名防并发单测冲突),返回路径。 + fn write_worker_script(content: &str) -> std::path::PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("系统时间应有效") + .as_nanos(); + let name = format!("oxide_worker_{}_{}.js", std::process::id(), nanos); + let path = std::env::temp_dir().join(name); + std::fs::write(&path, content).expect("应能写 worker 脚本临时文件"); + path + } + + /// 读全局属性(缺失返 None)。 + fn global_value_opt(vm: &Vm, name: &str) -> Option { + let session = vm.realm.session.borrow(); + let global = session.global_object(); + let si = vm.kernel_core().perm_interner().intern(name).0; + vm.resolve_property(global, si) + } + + /// 数值值转 f64(int 与 double 两表示)。 + fn as_number(v: &JsValue) -> f64 { + if v.is_int() { + v.as_int() as f64 + } else { + v.as_double() + } + } + + /// 端到端:主 → worker → main 双向消息(onmessage 交付、e.data 值)。 + /// + /// 主脚本建 Worker、设 onmessage、投递 21;worker 脚本收到翻倍回发 42; + /// 事件循环交付到主 onmessage 置 `globalThis.received`。 + #[test] + fn worker_e2e_onmessage_round_trip() { + let mut vm = vm_with_compiler(); + + // worker 脚本:收到消息翻倍后回发。 + let worker_path = write_worker_script("self.onmessage = function(e) { self.postMessage(e.data * 2); };"); + // 主脚本:建 Worker、设 onmessage、投递 21。 + let main_script = format!( + "var w = new Worker('{}'); w.onmessage = function(e) {{ globalThis.received = e.data; }}; w.postMessage(21);", + worker_path.display() + ); + vm.run(&Arc::new(compile_script(&main_script))).expect("主脚本应运行"); + + // 事件循环轮询至 globalThis.received === 42(带截止,防 flaky)。 + let deadline = Instant::now() + Duration::from_secs(5); + loop { + vm.deliver_worker_messages(); + vm.cleanup_disconnected_workers(); + if let Some(v) = global_value_opt(&vm, "received") { + if as_number(&v) == 42.0 { + break; + } + } + if Instant::now() >= deadline { + panic!("5 秒内未收到回显消息 42"); + } + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!( + as_number(&global_value_opt(&vm, "received").expect("received 应被设置")), + 42.0, + "回显值应为 42" + ); + + vm.shutdown_workers(); + let _ = std::fs::remove_file(&worker_path); + } + + /// 端到端:terminate 后注册表清空、二次 postMessage 返 Err。 + #[test] + fn worker_e2e_terminate() { + let mut vm = vm_with_compiler(); + + let id = vm.spawn_worker("1").expect("worker 应派生成功"); + assert_eq!(vm.active_workers(), vec![id], "应有唯一活跃 worker"); + + vm.worker_terminate(id).expect("终止应成功"); + assert!(vm.active_workers().is_empty(), "终止后无活跃 worker"); + + let result = vm.worker_post_message(id, MessageValue::Number(1.0)); + assert!(result.is_err(), "二次 postMessage 应返 Err"); + } + + /// 端到端:realm 回收计数配对(spawn +1、terminate 回基线)。 + #[test] + fn worker_e2e_realm_reclaimed() { + let mut vm = vm_with_compiler(); + // 克隆 Arc 解耦借用(kernel_core 返回 &Arc,长存会阻塞后续 &mut 借用)。 + let core = Arc::clone(vm.kernel_core()); + let baseline = core.active_vms(); + + let id = vm.spawn_worker("1").expect("worker 应派生成功"); + // worker 线程异步建 Vm:轮询至 realm 计数 +1(带截止,防 flaky)。 + let deadline = Instant::now() + Duration::from_secs(5); + while core.active_vms() < baseline + 1 { + if Instant::now() >= deadline { + panic!("5 秒内 worker realm 未创建"); + } + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(core.active_vms(), baseline + 1, "spawn 后应 +1 realm"); + + vm.worker_terminate(id).expect("终止应成功"); + assert_eq!(core.active_vms(), baseline, "terminate 后应回基线"); + } + + /// 端到端:postMessage 不可克隆值抛 DataCloneError、事件循环不挂死。 + #[test] + fn worker_e2e_data_clone_error() { + let mut vm = vm_with_compiler(); + + let worker_path = write_worker_script("1"); + // 主脚本:建 Worker、postMessage 函数(不可克隆),捕获 DataCloneError。 + let main_script = format!( + "var w = new Worker('{}'); try {{ w.postMessage(function(){{}}); globalThis.cloneError = false; }} catch (e) {{ globalThis.cloneError = true; }}", + worker_path.display() + ); + vm.run(&Arc::new(compile_script(&main_script))).expect("主脚本应运行"); + + assert_eq!( + global_value_opt(&vm, "cloneError"), + Some(JsValue::bool(true)), + "postMessage 函数应抛 DataCloneError" + ); + // worker 仍存活(postMessage 失败不影响 worker)。 + assert!(!vm.active_workers().is_empty(), "worker 应仍存活"); + + vm.shutdown_workers(); + let _ = std::fs::remove_file(&worker_path); + } }