diff --git a/Cargo.lock b/Cargo.lock index 2170ef57..6707b90f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.8.63" +version = "0.8.64" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.8.63" +version = "0.8.64" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.8.63" +version = "0.8.64" dependencies = [ "ansi_term", "bumpalo", @@ -1225,7 +1225,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.8.63" +version = "0.8.64" dependencies = [ "lru", "oxide_bytecode", @@ -1234,7 +1234,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1253,7 +1253,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1267,7 +1267,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.8.63" +version = "0.8.64" dependencies = [ "num-bigint", "oxc_allocator", @@ -1282,7 +1282,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1291,7 +1291,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.8.63" +version = "0.8.64" dependencies = [ "bumpalo", "dashmap", @@ -1306,7 +1306,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1318,7 +1318,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.8.63" +version = "0.8.64" dependencies = [ "tracing", "tracing-appender", @@ -1327,7 +1327,7 @@ dependencies = [ [[package]] name = "oxide_parser" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1342,7 +1342,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1358,7 +1358,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.8.63" +version = "0.8.64" dependencies = [ "num-bigint", "num-traits", @@ -1370,7 +1370,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.8.63" +version = "0.8.64" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1387,7 +1387,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.8.63" +version = "0.8.64" dependencies = [ "bumpalo", "hashbrown 0.14.5", @@ -1399,7 +1399,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.8.63" +version = "0.8.64" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 5db51fd0..2a5ccd08 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ members = [ ] [workspace.package] -version = "0.8.63" +version = "0.8.64" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_vm/src/lib.rs b/crates/oxide_vm/src/lib.rs index cf2daacc..c7998733 100644 --- a/crates/oxide_vm/src/lib.rs +++ b/crates/oxide_vm/src/lib.rs @@ -17,6 +17,8 @@ pub mod bindings; mod dispatch; mod generator; mod ic_helper; +/// 通用消息队列:`std::sync::mpsc` 的薄封装,提供命名发送端/接收端类型与统一入口。 +pub mod message_queue; /// native 函数签名类型([`native::NativeFn`]),builtin 绑定与 VM 调用约定依赖它。 pub mod native; /// native 载荷家族单点分类与每家族边函数引用(家族表,GC 各链注册面)。 diff --git a/crates/oxide_vm/src/message_queue.rs b/crates/oxide_vm/src/message_queue.rs new file mode 100644 index 00000000..d40dfad0 --- /dev/null +++ b/crates/oxide_vm/src/message_queue.rs @@ -0,0 +1,267 @@ +//! 通用消息队列:`std::sync::mpsc` 的薄封装,提供命名发送端/接收端类型与统一入口。 +//! +//! 关键约定: +//! - `Sender` 可克隆、可跨线程共享(要求 `T: Send`);`Receiver` 单消费者、非 Send。 +//! - FIFO 序:`send` 的顺序与 `recv` 的顺序一致。 +//! - 断开检测:全部发送端(含克隆)drop 后,`recv` / `try_recv` 返回断开, +//! `is_disconnected` 返回 true(与 std 语义一致,不取决于队列是否还有待取值)。 +//! - 语义与 `std::sync::mpsc` 完全一致;本模块只提供命名类型与统一入口,零新依赖。 + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{mpsc, Arc}; + +/// 创建一对消息队列发送端与接收端。 +/// +/// 发送端可克隆、可跨线程共享;接收端单消费者。全部发送端 drop 后队列断开, +/// 接收端此后的所有接收都返回断开。 +pub fn channel() -> (Sender, Receiver) { + let (tx, rx) = mpsc::channel(); + // 存活发送端计数:std 的 `mpsc::Receiver::is_disconnected` 尚不稳定, + // 用自维护计数复现 std 断开语义(全部发送端 drop 即断开)。 + let live = Arc::new(AtomicUsize::new(1)); + (Sender { inner: tx, live: live.clone() }, Receiver { inner: rx, live }) +} + +/// 发送端:消息队列的可克隆、可跨线程共享端。 +/// +/// 多个 `Sender` 克隆(含移入其他线程的)写入同一队列;全部发送端 drop 后 +/// 队列断开。 +pub struct Sender { + inner: mpsc::Sender, + live: Arc, +} + +impl Sender { + /// 发送一个值进队列。 + /// + /// # 边界与前提 + /// - 接收端已全部 drop 时返回 `SendError`,值归还调用方。 + /// + /// # 注意事项 + /// - `Sender: Send` 要求 `T: Send`,值随发送跨线程转移。 + pub fn send(&self, value: T) -> Result<(), SendError> { + self.inner.send(value).map_err(|e| SendError(e.0)) + } +} + +// std 的 `mpsc::Sender` 对任意 `T` 都可克隆,手动实现避免 derive 引入多余的 `T: Clone` 约束。 +impl Clone for Sender { + fn clone(&self) -> Self { + self.live.fetch_add(1, Ordering::Relaxed); + Self { + inner: self.inner.clone(), + live: self.live.clone(), + } + } +} + +impl Drop for Sender { + fn drop(&mut self) { + self.live.fetch_sub(1, Ordering::Release); + } +} + +/// 接收端:消息队列的单消费者端,非 Send。 +/// +/// 接收端只在持有线程内使用;阻塞 `recv` 与非阻塞 `try_recv` 均可用。 +pub struct Receiver { + inner: mpsc::Receiver, + live: Arc, +} + +impl Receiver { + /// 阻塞接收:等待直到有值到达或队列断开。 + /// + /// # 边界与前提 + /// - 队列已断开(全部发送端 drop 且无值可取)时返回 `RecvError`。 + /// + /// # 注意事项 + /// - 阻塞调用,调用方须保证所在线程可被阻塞。 + pub fn recv(&self) -> Result { + self.inner.recv().map_err(|_| RecvError) + } + + /// 非阻塞接收:立即返回。 + /// + /// # 边界与前提 + /// - 队列为空时返回 `TryRecvError::Empty`(发送端仍存活,其后可有值)。 + /// - 队列已断开时返回 `TryRecvError::Disconnected`(其后无值)。 + pub fn try_recv(&self) -> Result { + self.inner.try_recv().map_err(|e| match e { + mpsc::TryRecvError::Empty => TryRecvError::Empty, + mpsc::TryRecvError::Disconnected => TryRecvError::Disconnected, + }) + } + + /// 队列是否已断开(全部发送端已 drop)。 + /// + /// 与 std 断开语义一致:不取决于队列是否还有待取值。断开是单向状态: + /// 一旦为 true 不再回到 false。 + pub fn is_disconnected(&self) -> bool { + self.live.load(Ordering::Acquire) == 0 + } +} + +/// 发送失败:接收端已 drop,值归还调用方。 +pub struct SendError(pub T); + +/// 接收失败:队列已断开(全部发送端已 drop)且无值可取。 +pub struct RecvError; + +/// 非阻塞接收失败:队列为空或队列已断开。 +pub enum TryRecvError { + /// 队列为空(发送端仍存活,其后可有值)。 + Empty, + /// 队列已断开(全部发送端已 drop,其后无值)。 + Disconnected, +} + +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() + } +} + +impl std::fmt::Debug for RecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "RecvError") + } +} + +impl std::fmt::Debug for TryRecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TryRecvError::Empty => write!(f, "TryRecvError::Empty"), + TryRecvError::Disconnected => write!(f, "TryRecvError::Disconnected"), + } + } +} + +impl std::fmt::Display for SendError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "发送失败:接收端已 drop") + } +} + +impl std::fmt::Display for RecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "接收失败:消息队列已断开") + } +} + +impl std::fmt::Display for TryRecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TryRecvError::Empty => write!(f, "队列为空"), + TryRecvError::Disconnected => 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 {} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn send_recv_roundtrip() { + let (tx, rx) = channel(); + tx.send(42).unwrap(); + assert_eq!(rx.recv().unwrap(), 42); + } + + #[test] + fn fifo_order() { + let (tx, rx) = channel(); + for i in 0..3 { + tx.send(i).unwrap(); + } + for i in 0..3 { + assert_eq!(rx.recv().unwrap(), i); + } + } + + #[test] + fn multi_producer() { + let (tx, rx) = channel(); + let mut handles = Vec::new(); + for t in 0..4 { + let tx = tx.clone(); + handles.push(std::thread::spawn(move || { + for i in 0..25 { + tx.send(t * 100 + i).unwrap(); + } + })); + } + drop(tx); + let mut received = Vec::new(); + for _ in 0..100 { + received.push(rx.recv().unwrap()); + } + received.sort_unstable(); + let expected: Vec = (0..4).flat_map(|t| (0..25).map(move |i| t * 100 + i)).collect(); + assert_eq!(received, expected); + for h in handles { + h.join().unwrap(); + } + } + + #[test] + fn try_recv_empty() { + let (_tx, rx) = channel::(); + assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); + } + + #[test] + fn try_recv_disconnected() { + let (tx, rx) = channel::(); + drop(tx); + assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected))); + } + + #[test] + fn recv_disconnected() { + let (tx, rx) = channel::(); + drop(tx); + assert!(rx.recv().is_err()); + } + + #[test] + fn is_disconnected_flag() { + let (tx, rx) = channel::(); + assert!(!rx.is_disconnected()); + drop(tx); + assert!(rx.is_disconnected()); + } + + #[test] + fn is_disconnected_with_pending_values() { + // 断开语义不取决于队列是否还有待取值:发送端 drop 即断开。 + let (tx, rx) = channel::(); + tx.send(1).unwrap(); + drop(tx); + assert!(rx.is_disconnected()); + assert_eq!(rx.recv().unwrap(), 1); + } + + #[test] + fn send_error_returns_value() { + let (tx, rx) = channel::(); + drop(rx); + let err = tx.send(7).unwrap_err(); + assert_eq!(err.0, 7); + } + + #[test] + fn sender_is_send_across_threads() { + let (tx, rx) = channel::(); + let handle = std::thread::spawn(move || { + tx.send("from worker".to_string()).unwrap(); + }); + handle.join().unwrap(); + assert_eq!(rx.recv().unwrap(), "from worker"); + } +}