diff --git a/Cargo.lock b/Cargo.lock index f5144ff3..2dc7652b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.8.90" +version = "0.8.91" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.8.90" +version = "0.8.91" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.8.90" +version = "0.8.91" dependencies = [ "ansi_term", "bumpalo", @@ -1226,7 +1226,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.8.90" +version = "0.8.91" dependencies = [ "lru", "oxide_bytecode", @@ -1235,7 +1235,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1255,7 +1255,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1269,7 +1269,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.8.90" +version = "0.8.91" dependencies = [ "num-bigint", "oxc_allocator", @@ -1284,7 +1284,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1296,7 +1296,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1305,7 +1305,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.8.90" +version = "0.8.91" dependencies = [ "bumpalo", "dashmap", @@ -1319,7 +1319,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1331,7 +1331,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.8.90" +version = "0.8.91" dependencies = [ "tracing", "tracing-appender", @@ -1340,7 +1340,7 @@ dependencies = [ [[package]] name = "oxide_parser" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1371,7 +1371,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.8.90" +version = "0.8.91" dependencies = [ "num-bigint", "num-traits", @@ -1384,7 +1384,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.8.90" +version = "0.8.91" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1402,7 +1402,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.8.90" +version = "0.8.91" dependencies = [ "bumpalo", "dashmap", @@ -1415,7 +1415,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.8.90" +version = "0.8.91" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index f60538c5..7d0022ef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,7 @@ members = [ ] [workspace.package] -version = "0.8.90" +version = "0.8.91" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_kernel/src/message_queue.rs b/crates/oxide_kernel/src/message_queue.rs index d40dfad0..0b7623ab 100644 --- a/crates/oxide_kernel/src/message_queue.rs +++ b/crates/oxide_kernel/src/message_queue.rs @@ -93,6 +93,20 @@ impl Receiver { }) } + /// 带超时接收：在 `duration` 内返回一个值，否则返回超时错误。 + /// + /// # 边界与前提 + /// - `duration` 内无值到达且队列未断开时返回 `Err(Timeout)`。 + /// - 队列已断开（全部发送端 drop）且无值可取时同样返回 `Err(Timeout)` + /// （与 std `recv_timeout` 语义一致，不区分超时与断开；断开判定经 + /// [`Receiver::is_disconnected`] 单独查询）。 + /// + /// # 注意事项 + /// - 透传 std `mpsc::Receiver::recv_timeout` 语义。 + pub fn recv_timeout(&self, duration: std::time::Duration) -> Result { + self.inner.recv_timeout(duration).map_err(|_| Timeout) + } + /// 队列是否已断开（全部发送端已 drop）。 /// /// 与 std 断开语义一致：不取决于队列是否还有待取值。断开是单向状态： @@ -116,6 +130,9 @@ pub enum TryRecvError { Disconnected, } +/// 带超时接收失败：指定时间内无值到达且队列未断开（或已断开且无值可取）。 +pub struct Timeout; + impl std::fmt::Debug for SendError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_tuple("SendError").field(&self.0).finish() @@ -137,6 +154,12 @@ impl std::fmt::Debug for TryRecvError { } } +impl std::fmt::Debug for Timeout { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Timeout") + } +} + impl std::fmt::Display for SendError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "发送失败：接收端已 drop") @@ -158,9 +181,16 @@ impl std::fmt::Display for TryRecvError { } } +impl std::fmt::Display for Timeout { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "接收超时") + } +} + impl std::error::Error for SendError where T: std::fmt::Debug {} impl std::error::Error for RecvError {} impl std::error::Error for TryRecvError {} +impl std::error::Error for Timeout {} #[cfg(test)] mod tests { @@ -255,6 +285,27 @@ mod tests { assert_eq!(err.0, 7); } + #[test] + fn recv_timeout_returns_value_in_time() { + let (tx, rx) = channel::(); + tx.send(42).unwrap(); + assert_eq!(rx.recv_timeout(std::time::Duration::from_millis(50)).unwrap(), 42); + } + + #[test] + fn recv_timeout_empty_returns_timeout() { + let (_tx, rx) = channel::(); + assert!(rx.recv_timeout(std::time::Duration::from_millis(50)).is_err()); + } + + #[test] + fn recv_timeout_disconnected_returns_timeout() { + let (tx, rx) = channel::(); + drop(tx); + assert!(rx.recv_timeout(std::time::Duration::from_millis(50)).is_err()); + assert!(rx.is_disconnected()); + } + #[test] fn sender_is_send_across_threads() { let (tx, rx) = channel::(); diff --git a/crates/oxide_vm/src/lib.rs b/crates/oxide_vm/src/lib.rs index 14618913..71edca6b 100644 --- a/crates/oxide_vm/src/lib.rs +++ b/crates/oxide_vm/src/lib.rs @@ -45,5 +45,7 @@ mod vm_props; mod vm_runtime; mod vm_state; mod vm_support; +/// worker 线程基础设施：WorkerMail 消息通道、WorkerHandle 句柄与 worker 事件循环。 +pub mod worker; /// JS 值类型再导出（来自 `oxide_types`），作为本 crate 公共 API 的一部分。 pub use oxide_types::value::JsValue; diff --git a/crates/oxide_vm/src/vm/mod.rs b/crates/oxide_vm/src/vm/mod.rs index e4204288..ba016a2d 100644 --- a/crates/oxide_vm/src/vm/mod.rs +++ b/crates/oxide_vm/src/vm/mod.rs @@ -384,10 +384,19 @@ pub struct Vm { /// 池回收的 `full_reset` 路径清位；`run()` 入口的 `clear_execution_state` /// 刻意不清（清在那会把执行路径刚设的覆盖抹掉）。 pub(crate) max_steps_override: Option, + /// worker 注册表：键 = worker 编号，值 = `WorkerHandle`（主线程 → worker + /// 通道 + worker → 主线程通道 + OS 线程句柄）。worker 对象被 GC 不终止 + /// worker 线程，条目孤儿至 `shutdown_workers()`（`Drop for Vm` 调用）。 + pub(crate) worker_registry: std::collections::HashMap, + /// 下一个 worker 编号（单调递增，首个为 0）。 + pub(crate) worker_next_id: u64, } impl Drop for Vm { fn drop(&mut self) { + // 先终止全部 worker 并 join（防线程泄漏）：worker 线程各持自有 Vm， + // 其收尾独立于本 Vm，join 后线程不再引用本 Vm 的 kernel_core。 + self.shutdown_workers(); // 边界守卫计数：与构造器登记恰好配对（Rust 所有权保证恰好一次）。 self.kernel_core.note_vm_ended(); // 活跃 for-in / for-of 迭代器是 per-VM 执行态：收尾路径逐条释放， @@ -422,6 +431,13 @@ impl Drop for Vm { // realm 字段为 Arc：Realm 的 session 堆不跨线程共享，Arc 引用计数为 // 原子操作，跨线程移动 Arc 安全；Realm 只经 Vm（单线程 &mut）访问， // 其非 Sync 性对 Send 无碍。 +// +// 5. worker_registry 字段（HashMap）不持 session 堆指针： +// WorkerHandle 的 tx（Sender）与 handle（JoinHandle<()>）均 Send， +// rx_out（Receiver）虽非 Send 但只在主线程（Vm 属主线程）经 +// poll_worker_messages 访问、不跨线程移动；MessageValue 是 Send 中间表示 +// （无 realm 局部指针）。跨线程移动 Vm 时 worker_registry 整体迁移， +// 其中无指向 session 堆的裸指针，不引入别名。 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 37c7144c..110c5b2d 100644 --- a/crates/oxide_vm/src/vm_support.rs +++ b/crates/oxide_vm/src/vm_support.rs @@ -191,6 +191,8 @@ impl Vm { trace_instructions: false, pc_watch: None, max_steps_override: None, + worker_registry: std::collections::HashMap::new(), + worker_next_id: 0, }; vm.init_generator_intrinsics(); vm.init_promise_intrinsics(); @@ -348,6 +350,8 @@ impl Vm { trace_instructions: false, pc_watch: None, max_steps_override: None, + worker_registry: std::collections::HashMap::new(), + worker_next_id: 0, }; vm.init_generator_intrinsics(); vm.init_promise_intrinsics(); diff --git a/crates/oxide_vm/src/worker/mod.rs b/crates/oxide_vm/src/worker/mod.rs new file mode 100644 index 00000000..bb3b1ce1 --- /dev/null +++ b/crates/oxide_vm/src/worker/mod.rs @@ -0,0 +1,364 @@ +//! worker 线程基础设施：WorkerMail 消息通道、WorkerHandle 句柄与 worker 事件循环。 +//! +//! 关键约定： +//! - 每 worker 是独立 OS 线程，线程内经 `Vm::with_kernel_core` 建专属 Vm（新 realm）， +//! 线程退出即 drop（`Drop for Vm` → `Drop for Realm`）。 +//! - `Arc` 跨线程共享（Send/Sync），字节码缓存跨 worker 命中。 +//! - 跨线程消息经 `MessageValue`（Send 中间表示，detach/rehydrate），`JsValue` 不 +//! 跨线程（session 堆指针是 realm 局部地址，跨线程无效）。 +//! - 主线程 → worker 通道载 `WorkerMail`（数据 / 错误 / 关停三面）；worker → 主线程 +//! 通道载 `MessageValue`，主线程轮询时 rehydrate 进主 realm。 +//! - worker 脚本编译失败不 panic：经 worker → 主线程通道上报错误串，worker 继续 +//! 事件循环（可被干净终止）。 + +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_kernel::kernel::KernelCore; +use oxide_kernel::message_queue::{channel, Receiver, Sender, Timeout}; +use oxide_runtime_api::CompilerService; +use oxide_types::value::JsValue; + +use crate::vm::Vm; +use crate::vm_debug; +use crate::vm_warn; + +/// worker 邮件（主线程 → worker 线程）：数据 / 错误 / 关停三面。 +/// +/// `Send`（`MessageValue` 与 `String` 均 Send），可经 `Sender` 跨线程投递。 +pub enum WorkerMail { + /// 数据面：一条结构化克隆消息值，worker rehydrate 后处理。 + Message(MessageValue), + /// 错误面：主线程向 worker 上报错误（worker 记入 `last_uncaught_value`）。 + Error(String), + /// 关停信号：worker 退出事件循环。 + Terminate, +} + +/// worker 句柄：主线程对单个 worker 的持有。 +/// +/// `id` 是单调递增的 worker 编号（主线程分配）；`tx` 是主线程 → worker 通道 +/// （发 `WorkerMail`）；`rx_out` 是 worker → 主线程消息通道（收 `MessageValue`， +/// 轮询时 rehydrate 进主 realm）；`handle` 是 OS 线程句柄（终止时 join）。 +pub struct WorkerHandle { + /// worker 编号（主线程分配，单调递增）。 + pub id: u64, + /// 主线程 → worker 通道（发 `WorkerMail`）。 + pub tx: Sender, + /// worker → 主线程消息通道（收 `MessageValue`）。 + pub(crate) rx_out: Receiver, + /// OS 线程句柄（终止时 join）。 + pub(crate) handle: JoinHandle<()>, +} + +/// worker 事件循环：在 worker 线程内运行。 +/// +/// # 步骤 +/// 1. 经 `Vm::with_kernel_core` 建专属 Vm（新 realm）并注入编译服务。 +/// 2. 编译 worker 脚本（worker 的程序）；失败则经 worker → 主线程通道上报错误串， +/// 不 panic，worker 继续事件循环（可被干净终止）。 +/// 3. 编译成功则运行脚本（worker 的程序）。 +/// 4. 循环 `recv_timeout(100ms)`：处理 `Message` / `Error` / `Terminate` / 超时。 +/// +/// # 边界与前提 +/// - `script` 是 worker 的程序源码（普通脚本），编译与运行均在 worker 线程内完成。 +/// - 超时后查 `rx.is_disconnected()`：主线程全部发送端 drop 即断开，worker 退出。 +/// +/// # 副作用 +/// - 创建并 drop 一个 Vm（退出时 drop 触发 `Drop for Vm` → `Drop for Realm`）。 +/// - 每条 `Message` 经 rehydrate → execute_task → drain_microtasks 处理后，把处理 +/// 值 detach 回 worker → 主线程通道（935.3 的回显语义，935.4 的 onmessage 替换之）。 +fn worker_event_loop( + core: Arc, + compiler: Arc, + script: String, + rx: Receiver, + out_tx: Sender, +) { + let mut vm = Vm::with_kernel_core(core); + vm.set_compiler_service(Arc::clone(&compiler)); + + // 编译 worker 脚本（worker 的程序）。失败不 panic：上报错误串后继续循环。 + let script_module = match compiler.compile_script(&script) { + Ok(module) => Some(module), + Err(err) => { + vm_warn!("worker: script compile failed: {err}"); + let units: Box<[u16]> = err.encode_utf16().collect::>().into(); + let _ = out_tx.send(MessageValue::String(units)); + None + } + }; + + // 运行脚本（worker 的程序）。运行期错误记入 last_uncaught_value，不中断循环。 + if let Some(module) = script_module { + if let Err(err) = vm.run(&Arc::new(module)) { + vm.last_uncaught_value = Some(vm.new_string(&err)); + } + } + + // 事件循环：recv_timeout 驱动，超时查断开。 + 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); + } + } + Ok(WorkerMail::Error(err)) => { + // 主线程上报的错误记入 last_uncaught_value。 + vm.last_uncaught_value = Some(vm.new_string(&err)); + } + Ok(WorkerMail::Terminate) => break, + Err(Timeout) => { + // 超时：主线程全部发送端 drop 即断开，worker 退出。 + if rx.is_disconnected() { + break; + } + } + } + } + + // 退出：drop(vm) 触发 Drop for Vm → Drop for Realm（per-realm 收尾）。 + drop(vm); + vm_debug!("worker: event loop exited"); +} + +impl Vm { + /// 派生一个 worker：建 `WorkerHandle` 并 spawn 一个 OS 线程运行事件循环。 + /// + /// # 步骤 + /// 1. 分配 worker 编号（`worker_next_id` 自增）。 + /// 2. 建主线程 → worker 与 worker → 主线程两条通道。 + /// 3. spawn 线程，线程内建专属 Vm 并进入事件循环。 + /// 4. 把 `WorkerHandle` 登记进 `worker_registry`。 + /// + /// # 边界与前提 + /// - `script` 是 worker 的程序源码（普通脚本）；编译在 worker 线程内完成， + /// 编译失败经 worker → 主线程通道上报，不 panic。 + /// - realm 计数受 512 上界约束（每 worker 一个 realm），worker 数须远低于 512。 + /// + /// # 返回值 + /// 新 worker 的编号。 + /// + /// # 副作用 + /// - 派生一个 OS 线程；登记一个 `WorkerHandle`。 + pub fn spawn_worker(&mut self, script: &str) -> Result { + let id = self.worker_next_id; + self.worker_next_id += 1; + + let (tx, rx) = channel::(); + let (out_tx, out_rx) = channel::(); + + let core = Arc::clone(&self.kernel_core); + let compiler = Arc::clone(&self.compiler); + let script_owned = script.to_string(); + + let handle = std::thread::spawn(move || { + worker_event_loop(core, compiler, script_owned, rx, out_tx); + }); + + let worker = WorkerHandle { + id, + tx, + rx_out: out_rx, + handle, + }; + self.worker_registry.insert(id, worker); + Ok(id) + } + + /// 向 worker 投递一条消息（经主线程 → worker 通道的 `WorkerMail::Message`）。 + /// + /// # 边界与前提 + /// - `id` 不存在时返回 `Err`。 + /// - worker 通道已断开（worker 线程退出）时返回 `Err`，消息归还调用方。 + /// + /// # 返回值 + /// 投递成功 `Ok(())`，失败 `Err`（含错误描述）。 + pub fn worker_post_message(&mut self, id: u64, msg: MessageValue) -> Result<(), String> { + let worker = self + .worker_registry + .get_mut(&id) + .ok_or_else(|| format!("worker {id} 不存在"))?; + worker + .tx + .send(WorkerMail::Message(msg)) + .map_err(|_| format!("worker {id} 通道已断开")) + } + + /// 终止 worker：发 `Terminate` 并 join 线程。 + /// + /// # 步骤 + /// 1. 发 `WorkerMail::Terminate`（worker 收到即退出事件循环）。 + /// 2. 从 `worker_registry` 移除（取走 `WorkerHandle`）。 + /// 3. join 线程（须在句柄 drop 前完成，防线程泄漏）。 + /// + /// # 边界与前提 + /// - `id` 不存在时返回 `Err`。 + /// - 线程 panic 时返回 `Err`。 + /// + /// # 副作用 + /// - 移除一个 `WorkerHandle`；join 一个 OS 线程。 + pub fn worker_terminate(&mut self, id: u64) -> Result<(), String> { + // 发关停信号（worker 收到即退出事件循环）。 + if let Some(worker) = self.worker_registry.get_mut(&id) { + let _ = worker.tx.send(WorkerMail::Terminate); + } + // 从注册表移除并 join 线程（句柄 drop 前必须 join）。 + let worker = self + .worker_registry + .remove(&id) + .ok_or_else(|| format!("worker {id} 不存在"))?; + worker + .handle + .join() + .map_err(|_| format!("worker {id} 线程异常退出"))?; + Ok(()) + } + + /// 轮询 worker 消息：排空 worker → 主线程通道，rehydrate 进主 realm。 + /// + /// # 步骤 + /// 1. 取 `WorkerHandle`（`id` 不存在时返回空列表）。 + /// 2. `try_recv` 循环排空 `rx_out`，每条 rehydrate 进主 realm。 + /// + /// # 返回值 + /// 本批排空的消息值列表（无消息时为空列表）。 + /// + /// # 边界与前提 + /// - 非阻塞（`try_recv`），不等待新消息。 + pub fn poll_worker_messages(&mut self, id: u64) -> Vec { + // 先排空 worker → 主线程通道到本地列表（注册表借用与 rehydrate 借用 + // 不重叠，避免对 self 的双重可变借用）。 + let drained: Vec = { + let Some(worker) = self.worker_registry.get_mut(&id) else { + return Vec::new(); + }; + let mut drained = Vec::new(); + while let Ok(mv) = worker.rx_out.try_recv() { + drained.push(mv); + } + drained + }; + let mut out = Vec::new(); + for mv in &drained { + out.push(rehydrate_message(self, mv)); + } + out + } + + /// 列出活跃 worker 编号（升序）。 + pub fn active_workers(&self) -> Vec { + let mut ids: Vec = self.worker_registry.keys().copied().collect(); + ids.sort_unstable(); + ids + } + + /// 终止全部 worker 并 join（`Drop for Vm` 调用，防线程泄漏）。 + /// + /// # 步骤 + /// 1. 收集全部活跃 worker 编号。 + /// 2. 逐一 `worker_terminate`（发 `Terminate` + join）。 + /// + /// # 副作用 + /// - 清空 `worker_registry`；join 全部 worker 线程。 + pub fn shutdown_workers(&mut self) { + let ids = self.active_workers(); + for id in ids { + let _ = self.worker_terminate(id); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Instant; + + use oxide_compiler::DefaultCompilerService; + use oxide_kernel::kernel::KernelConfig; + + /// 建一个带真实编译服务的 Vm（worker 脚本编译需要）。 + fn vm_with_compiler() -> Vm { + let core = KernelCore::new(KernelConfig::minimal()); + let mut vm = Vm::with_kernel_core(core); + vm.set_compiler_service(Arc::new(DefaultCompilerService)); + vm + } + + /// 轮询至 worker 回显一条消息（带截止，防 flaky）。 + fn poll_until_message(vm: &mut Vm, id: u64) -> Vec { + let deadline = Instant::now() + Duration::from_secs(2); + loop { + let messages = vm.poll_worker_messages(id); + if !messages.is_empty() || Instant::now() >= deadline { + return messages; + } + std::thread::sleep(Duration::from_millis(10)); + } + } + + /// 端到端：spawn → post → 回显 → terminate，验证 worker 基础设施闭环。 + #[test] + fn worker_round_trip() { + let mut vm = vm_with_compiler(); + + let id = vm.spawn_worker("1 + 1").expect("worker 应派生成功"); + assert_eq!(vm.active_workers(), vec![id], "应有唯一活跃 worker"); + + vm.worker_post_message(id, MessageValue::Number(42.0)) + .expect("投递应成功"); + let messages = poll_until_message(&mut vm, id); + assert_eq!(messages.len(), 1, "应回显一条消息"); + // 整数值经 number_to_js 归为 Int 表示。 + assert_eq!(messages[0], JsValue::int(42), "回显值应为 42"); + + vm.worker_terminate(id).expect("终止应成功"); + assert!(vm.active_workers().is_empty(), "终止后无活跃 worker"); + } + + /// 编译失败不 panic：worker 上报错误串，主线程可经轮询取回。 + #[test] + fn worker_script_compile_failure_reports_error() { + let mut vm = vm_with_compiler(); + + let id = vm + .spawn_worker("function { 语法错误") + .expect("worker 应派生成功"); + let messages = poll_until_message(&mut vm, id); + assert!(!messages.is_empty(), "编译失败应上报错误串"); + assert!(messages[0].is_string(), "错误串应为字符串值"); + + vm.worker_terminate(id).expect("终止应成功"); + } + + /// 多 worker 隔离：各 worker 独立 realm，消息不串扰。 + #[test] + 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 应派生成功"); + assert_eq!(vm.active_workers(), vec![id_a, id_b], "应有两个活跃 worker"); + + vm.worker_post_message(id_a, MessageValue::Number(1.0)) + .expect("投递 A 应成功"); + vm.worker_post_message(id_b, MessageValue::Number(2.0)) + .expect("投递 B 应成功"); + + let messages_a = poll_until_message(&mut vm, id_a); + assert_eq!(messages_a[0], JsValue::int(1), "A 应回显 1"); + let messages_b = poll_until_message(&mut vm, id_b); + assert_eq!(messages_b[0], JsValue::int(2), "B 应回显 2"); + + vm.shutdown_workers(); + assert!(vm.active_workers().is_empty(), "shutdown 后无活跃 worker"); + } +}