feat: 新增通用消息队列组件（泛型 mpsc 薄封装）

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",

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"

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 各链注册面）。

crates/oxide_vm/src/message_queue.rs

@@ -0,0 +1,267 @@
+//! 通用消息队列：`std::sync::mpsc` 的薄封装，提供命名发送端/接收端类型与统一入口。
+//!
+//! 关键约定：
+//! - `Sender<T>` 可克隆、可跨线程共享（要求 `T: Send`）；`Receiver<T>` 单消费者、非 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<T>() -> (Sender<T>, Receiver<T>) {
+    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<T> {
+    inner: mpsc::Sender<T>,
+    live: Arc<AtomicUsize>,
+}
+
+impl<T> Sender<T> {
+    /// 发送一个值进队列。
+    ///
+    /// # 边界与前提
+    /// - 接收端已全部 drop 时返回 `SendError`，值归还调用方。
+    ///
+    /// # 注意事项
+    /// - `Sender<T>: Send` 要求 `T: Send`，值随发送跨线程转移。
+    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
+        self.inner.send(value).map_err(|e| SendError(e.0))
+    }
+}
+
+// std 的 `mpsc::Sender` 对任意 `T` 都可克隆，手动实现避免 derive 引入多余的 `T: Clone` 约束。
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        self.live.fetch_add(1, Ordering::Relaxed);
+        Self {
+            inner: self.inner.clone(),
+            live: self.live.clone(),
+        }
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        self.live.fetch_sub(1, Ordering::Release);
+    }
+}
+
+/// 接收端：消息队列的单消费者端，非 Send。
+///
+/// 接收端只在持有线程内使用；阻塞 `recv` 与非阻塞 `try_recv` 均可用。
+pub struct Receiver<T> {
+    inner: mpsc::Receiver<T>,
+    live: Arc<AtomicUsize>,
+}
+
+impl<T> Receiver<T> {
+    /// 阻塞接收：等待直到有值到达或队列断开。
+    ///
+    /// # 边界与前提
+    /// - 队列已断开（全部发送端 drop 且无值可取）时返回 `RecvError`。
+    ///
+    /// # 注意事项
+    /// - 阻塞调用，调用方须保证所在线程可被阻塞。
+    pub fn recv(&self) -> Result<T, RecvError> {
+        self.inner.recv().map_err(|_| RecvError)
+    }
+
+    /// 非阻塞接收：立即返回。
+    ///
+    /// # 边界与前提
+    /// - 队列为空时返回 `TryRecvError::Empty`（发送端仍存活，其后可有值）。
+    /// - 队列已断开时返回 `TryRecvError::Disconnected`（其后无值）。
+    pub fn try_recv(&self) -> Result<T, TryRecvError> {
+        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<T>(pub T);
+
+/// 接收失败：队列已断开（全部发送端已 drop）且无值可取。
+pub struct RecvError;
+
+/// 非阻塞接收失败：队列为空或队列已断开。
+pub enum TryRecvError {
+    /// 队列为空（发送端仍存活，其后可有值）。
+    Empty,
+    /// 队列已断开（全部发送端已 drop，其后无值）。
+    Disconnected,
+}
+
+impl<T: std::fmt::Debug> std::fmt::Debug for SendError<T> {
+    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<T> std::fmt::Display for SendError<T> {
+    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<T> std::error::Error for SendError<T> 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<i32> = (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::<i32>();
+        assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty)));
+    }
+
+    #[test]
+    fn try_recv_disconnected() {
+        let (tx, rx) = channel::<i32>();
+        drop(tx);
+        assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected)));
+    }
+
+    #[test]
+    fn recv_disconnected() {
+        let (tx, rx) = channel::<i32>();
+        drop(tx);
+        assert!(rx.recv().is_err());
+    }
+
+    #[test]
+    fn is_disconnected_flag() {
+        let (tx, rx) = channel::<i32>();
+        assert!(!rx.is_disconnected());
+        drop(tx);
+        assert!(rx.is_disconnected());
+    }
+
+    #[test]
+    fn is_disconnected_with_pending_values() {
+        // 断开语义不取决于队列是否还有待取值：发送端 drop 即断开。
+        let (tx, rx) = channel::<i32>();
+        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::<i32>();
+        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::<String>();
+        let handle = std::thread::spawn(move || {
+            tx.send("from worker".to_string()).unwrap();
+        });
+        handle.join().unwrap();
+        assert_eq!(rx.recv().unwrap(), "from worker");
+    }
+}