feat: server 连接上限与 worker 队列有界背压

Cargo.lock

@@ -1160,7 +1160,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_builtins"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "chrono",
  "fast-float",
@@ -1185,7 +1185,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_bytecode"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "num-bigint",
  "oxide_log",
@@ -1193,7 +1193,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_cfg"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_emit",
@@ -1204,7 +1204,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_cli"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "ansi_term",
  "bumpalo",
@@ -1228,7 +1228,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_code_cache"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "lru",
  "oxide_bytecode",
@@ -1237,7 +1237,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_compiler"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1257,7 +1257,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_dce"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1271,7 +1271,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_emit"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "num-bigint",
  "oxc_allocator",
@@ -1286,7 +1286,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_engine"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_compiler",
@@ -1298,7 +1298,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_ir"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_log",
@@ -1307,7 +1307,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_kernel"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "bumpalo",
  "dashmap",
@@ -1321,7 +1321,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_liveness"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1333,7 +1333,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_log"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "tracing",
  "tracing-appender",
@@ -1342,11 +1342,11 @@ dependencies = [
 
 [[package]]
 name = "oxide_net"
-version = "0.15.12"
+version = "0.15.13"
 
 [[package]]
 name = "oxide_parser"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxc_allocator",
  "oxc_ast",
@@ -1361,7 +1361,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_regalloc"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1377,7 +1377,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_runtime_api"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "num-bigint",
  "num-traits",
@@ -1390,7 +1390,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_test262"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "oxide_bytecode",
  "oxide_compiler",
@@ -1408,7 +1408,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_types"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "bumpalo",
  "dashmap",
@@ -1421,7 +1421,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_vm"
-version = "0.15.12"
+version = "0.15.13"
 dependencies = [
  "bumpalo",
  "chrono",

Cargo.toml

@@ -24,7 +24,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.15.12"
+version = "0.15.13"
 edition = "2021"
 license = "MIT"
 rust-version = "1.80"

crates/oxide_cli/src/server/server.rs

@@ -10,9 +10,9 @@
 //! - accept 循环为非阻塞 accept 加 10 毫秒轮询加原子关闭标志：唤醒路径
 //!   确定、无锁、无唤醒丢失。
 //! - 并发模型：主线程只跑 accept 循环；每连接一个标准库线程只做 I/O
-//!   （帧读写）；六类控制请求由连接线程直接应答、不占 worker；执行请求与
+//!   （帧读写）；accept 循环带并发连接上限（超限连接立即关闭）；六类控制
-//!   forge 查询经 mpsc 路由到固定 N 个常驻 worker 执行（各持自己线程上的
+//!   请求由连接线程直接应答、不占 worker；执行请求与 forge 查询经 mpsc
-//!   VM 池）。
+//!   路由到固定 N 个常驻 worker 执行（各持自己线程上的 VM 池）。
 //! - 优雅退出顺序：关闭请求、SIGINT/SIGTERM 信号或 yield 请求置位关闭标志
 //!   → 在途归零 → 清空发送端 → join worker → drop 内核 → 先删 socket 文件
 //!   （Unix 臂）后删 sidecar（新 server 以 sidecar 消失为接管信号）。
@@ -113,6 +113,10 @@ fn default_worker_count() -> usize {
 /// --rm 模式缺省空闲超时：30 秒。
 const RM_DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
 
+/// 并发连接上限：accept 循环容忍的在途连接线程数上限；超限连接立即关闭
+/// （客户端表现为连接建立随即被关，不发协议帧）。
+const MAX_CONCURRENT_CONNECTIONS: usize = 64;
+
 /// --rm 独立模式运行配置：监听端点、worker 数、空闲超时、共享 token。
 ///
 /// 不含 sidecar 路径字段：「不注册 sidecar」的不变量在类型上成立，
@@ -708,8 +712,13 @@ fn bind_listener(config: &ServerConfig) -> Result<Listener, ServerError> {
 ///
 /// # 步骤
 /// 1. 检查关闭标志（原子布尔，序一致排序）。
-/// 2. accept：成功则派生连接线程（在途计数加一）；WouldBlock 睡 10 毫秒再查；
+/// 2. accept：成功则先查并发连接上限，未超限派生连接线程（在途计数加一）、
-///    其余错误以 `AcceptFailed` 返回。
+///    超限立即关闭该流；WouldBlock 睡 10 毫秒再查；其余错误以 `AcceptFailed`
+///    返回。
+///
+/// # 边界与前提
+/// - 上限检查与派生同在单线程 accept 循环内，无竞态；在途计数是既有的
+///   连接线程生命周期计数，直接复用。
 ///
 /// # 副作用
 /// - 每连接派生一个标准库线程（只做 I/O），线程结束时减一在途计数。
@@ -726,6 +735,12 @@ fn accept_loop(listener: &Listener, ctx: &Arc<ServerContext>, router: &Arc<Worke
 
         match listener.accept() {
             Ok(stream) => {
+                // 并发连接上限：在途计数达上限时立即关闭该流（客户端表现为
+                // 连接建立随即被关，不发协议帧），不派生线程、不加计数。
+                if ctx.in_flight.load(Ordering::SeqCst) >= MAX_CONCURRENT_CONNECTIONS {
+                    drop(stream);
+                    continue;
+                }
                 ctx.in_flight.fetch_add(1, Ordering::SeqCst);
                 let ctx = Arc::clone(ctx);
                 let router = Arc::clone(router);
@@ -1366,6 +1381,60 @@ mod tests {
         stop_server(&mut stream_b, handle);
     }
 
+    /// 认证探针：发 Auth 帧并检查是否读到 AuthOk 帧；连接被关时返回假
+    /// （不 panic），供连接上限类测试探活。
+    fn auth_probe(stream: &mut Conn, token: &str) -> bool {
+        if stream
+            .write_all(protocol::encode_request(&ServerRequest::Auth { token: token.to_string() }).as_bytes())
+            .is_err()
+        {
+            return false;
+        }
+        let mut reader = FrameReader::new(io::BufReader::new(stream));
+        match reader.read_frame() {
+            Ok(Some(frame)) => matches!(protocol::parse_response(&frame).ok(), Some(ServerResponse::AuthOk)),
+            _ => false,
+        }
+    }
+
+    /// 连接上限：开满上限条连接后，下一条连接被立即关闭（读首帧得 EOF）；
+    /// 释放一条连接后，新连接恢复正常受理。
+    #[test]
+    fn accept_loop_enforces_concurrent_limit() {
+        let config = unique_paths("conn_limit");
+        let handle = start_server(config.clone());
+        let token = read_token(&config);
+
+        // 开满上限：每条连接的连接线程存活，在途计数达上限。
+        let mut streams: Vec<Conn> = (0..MAX_CONCURRENT_CONNECTIONS).map(|_| connect(&config)).collect();
+
+        // 第 N+1 条连接：server 立即关闭，读首帧得 EOF。
+        let mut extra = connect(&config);
+        assert!(!auth_probe(&mut extra, &token), "超限连接应被立即关闭（EOF）");
+
+        // 释放一条连接后，新连接恢复正常受理（轮询容忍连接线程退出延迟）。
+        drop(streams.pop().expect("应有一条连接可释放"));
+        let deadline = Instant::now() + Duration::from_secs(5);
+        let mut admitted = false;
+        while !admitted {
+            assert!(Instant::now() < deadline, "上限释放后新连接应被受理");
+            let mut probe = connect(&config);
+            admitted = auth_probe(&mut probe, &token);
+            if !admitted {
+                std::thread::sleep(Duration::from_millis(10));
+            }
+        }
+
+        // 清理：关闭全部连接，再以一条已认证连接发关闭请求停 server。
+        let mut stop_stream = streams.pop().expect("应有一条连接用于停 server");
+        authenticate(&mut stop_stream, &token);
+        for stream in streams {
+            drop(stream);
+        }
+        drop(extra);
+        stop_server(&mut stop_stream, handle);
+    }
+
     /// 控制请求不占 worker：一条连接发长 eval 占住 worker 期间，
     /// 另一连接发 Status 立即返回（不受 eval 阻塞）。
     #[test]

crates/oxide_cli/src/server/workers.rs

@@ -6,6 +6,8 @@
 //!   `Arc<KernelCore>` 是 Send/Sync，由全体 worker 共享（字节码缓存跨 worker 命中）。
 //! - 标准库 mpsc 的接收端不是 Send，不能跨线程共享给多个 worker，
 //!   因此每 worker 一条独立通道，发送端收进 `WorkerRouter` 按轮询投递。
+//! - 每条任务通道有界（每 worker 容量 8）：队列满时投递阻塞形成背压，
+//!   待执行任务内存有界。
 //! - 关闭序列固定：`close` 清空发送端 → worker 的 recv 断开退出 →
 //!   主线程逐一 join → 再 drop 内核。
 
@@ -21,6 +23,10 @@ use super::eval;
 use super::forge;
 use super::protocol::{ForgeTarget, ServerResponse};
 
+/// 任务队列容量（每 worker 一条，非全局共享）：队列满时投递阻塞形成背压，
+/// 待执行任务内存有界（每任务至多一帧长上限的源码字符串）。
+const QUEUE_CAPACITY: usize = 8;
+
 /// worker 任务：执行请求加一次性回复通道。
 #[derive(Debug)]
 pub enum WorkerTask {
@@ -40,17 +46,18 @@ pub enum WorkerTask {
     },
 }
 
-/// worker 路由器：N 条独立任务通道的发送端加轮询计数。
+/// worker 路由器：N 条独立有界任务通道的发送端加轮询计数。
 ///
 /// 发送端收进互斥锁：连接线程可能存活超过主线程的关闭序列（在途排水超时后
 /// 仍有余留连接），`close` 清空发送端后它们的投递立即失败，不挂起、不悬垂。
+/// 锁只保护发送端的索引与克隆，投递（可能因队列满而阻塞）在锁外进行。
 pub struct WorkerRouter {
-    senders: Mutex<Vec<mpsc::Sender<WorkerTask>>>,
+    senders: Mutex<Vec<mpsc::SyncSender<WorkerTask>>>,
     next: AtomicUsize,
 }
 
 impl WorkerRouter {
-    fn from_senders(senders: Vec<mpsc::Sender<WorkerTask>>) -> Self {
+    fn from_senders(senders: Vec<mpsc::SyncSender<WorkerTask>>) -> Self {
         WorkerRouter {
             senders: Mutex::new(senders),
             next: AtomicUsize::new(0),
@@ -60,18 +67,26 @@ impl WorkerRouter {
     /// 轮询投递任务到 worker。
     ///
     /// # 步骤
-    /// 1. 计数加一取模 N 选发送端。
+    /// 1. 锁内计数加一取模 N 克隆发送端，随即释放锁。
-    /// 2. 投递；发送端已清空（关闭序列进行中）或接收端已断开时归还任务。
+    /// 2. 锁外投递；队列满时阻塞等待空位（背压）。
+    /// 3. 发送端已清空（关闭序列进行中）或接收端已断开时归还任务。
     ///
     /// # 边界与前提
     /// - 返回 `Err` 时任务未被任何 worker 接收，调用方须以错误帧应答，不得阻塞等待。
+    /// - 队列满时调用线程阻塞至空位腾出；阻塞期间不持有发送端锁，
+    ///   其余连接线程的投递不被串行卡死。
     pub fn route(&self, task: WorkerTask) -> Result<(), WorkerTask> {
+        // 锁内克隆发送端、锁外投递：队列满时 send 阻塞，若持锁阻塞会把
+        // 全体连接线程的投递串行卡死。
+        let sender = {
             let senders = self.senders.lock().unwrap();
             if senders.is_empty() {
                 return Err(task);
             }
             let idx = self.next.fetch_add(1, Ordering::Relaxed) % senders.len();
-        match senders[idx].send(task) {
+            senders[idx].clone()
+        };
+        match sender.send(task) {
             Ok(()) => Ok(()),
             Err(sent) => Err(sent.0),
         }
@@ -87,7 +102,8 @@ impl WorkerRouter {
 /// 并进入任务循环。
 ///
 /// # 步骤
-/// 1. 建 N 对通道，接收端各移入一个 worker 线程。
+/// 1. 建 N 对有界通道（每 worker 容量 `QUEUE_CAPACITY`），接收端各移入一个
+///    worker 线程。
 /// 2. worker 线程内按内核配置的池规模建池（共享聚合计数器）。
 /// 3. 各 worker 进入任务循环：recv → 执行 → 经回复通道送回响应帧。
 ///
@@ -107,7 +123,8 @@ pub fn spawn_workers(
     let mut senders = Vec::with_capacity(n);
     let mut handles = Vec::with_capacity(n);
     for _ in 0..n {
-        let (tx, rx) = mpsc::channel();
+        // 有界通道：队列满时投递阻塞（背压），待执行任务内存有界。
+        let (tx, rx) = mpsc::sync_channel(QUEUE_CAPACITY);
         let kernel = Arc::clone(&kernel);
         let counters = Arc::clone(&counters);
         handles.push(std::thread::spawn(move || {
@@ -153,6 +170,7 @@ fn worker_loop(receiver: mpsc::Receiver<WorkerTask>, kernel: Arc<KernelCore>, po
 mod tests {
     use super::*;
     use oxide_kernel::kernel::KernelConfig;
+    use std::time::Duration;
 
     /// 建一个哑任务（回复通道只挂不读）。
     fn dummy_task() -> WorkerTask {
@@ -176,8 +194,8 @@ mod tests {
     /// 轮询分发：N 等于 2，投 4 个任务，两个 worker 各收到 2 个。
     #[test]
     fn round_robin_distribution() {
-        let (tx1, rx1) = mpsc::channel();
+        let (tx1, rx1) = mpsc::sync_channel(QUEUE_CAPACITY);
-        let (tx2, rx2) = mpsc::channel();
+        let (tx2, rx2) = mpsc::sync_channel(QUEUE_CAPACITY);
         let router = WorkerRouter::from_senders(vec![tx1, tx2]);
         for _ in 0..4 {
             router.route(dummy_task()).expect("投递应成功");
@@ -186,6 +204,30 @@ mod tests {
         assert_eq!(drain_count(&rx2), 2, "第二个 worker 应收到 2 个任务");
     }
 
+    /// 队列满时投递阻塞、队列腾出后投递成功：有界队列的背压语义。
+    #[test]
+    fn route_blocks_when_queue_full() {
+        let (tx, rx) = mpsc::sync_channel(QUEUE_CAPACITY);
+        let router = Arc::new(WorkerRouter::from_senders(vec![tx]));
+
+        // 填满队列至容量。
+        for _ in 0..QUEUE_CAPACITY {
+            router.route(dummy_task()).expect("队列未满时投递应成功");
+        }
+
+        // 队列已满：投递阻塞。在独立线程中经共享路由器投递，主线程验证其
+        // 未在有界时间内完成（防挂死的超时断言）。
+        let router_for_send = Arc::clone(&router);
+        let send_handle = std::thread::spawn(move || router_for_send.route(dummy_task()).is_ok());
+        std::thread::sleep(Duration::from_millis(200));
+        assert!(!send_handle.is_finished(), "队列满时投递应阻塞");
+
+        // 腾出一个空位：阻塞的投递应完成。
+        let _ = rx.try_recv();
+        let delivered = send_handle.join().expect("投递线程应正常退出");
+        assert!(delivered, "腾出空位后投递应成功");
+    }
+
     /// 关闭序列：清空发送端后 worker 线程退出，join 在有界时间内返回。
     #[test]
     fn workers_exit_when_senders_cleared() {
@@ -201,7 +243,7 @@ mod tests {
     /// 发送端清空后投递归还任务：调用方不挂起，可立即以错误帧应答。
     #[test]
     fn route_after_close_returns_task() {
-        let (tx, _rx) = mpsc::channel();
+        let (tx, _rx) = mpsc::sync_channel(QUEUE_CAPACITY);
         let router = WorkerRouter::from_senders(vec![tx]);
         router.close();
         assert!(router.route(dummy_task()).is_err(), "清空后投递应归还任务");