From 45acd5a2265cb8472239554a520b7e923be81118 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Sun, 11 Oct 2026 16:36:49 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20server=20=E8=BF=9E=E6=8E=A5=E4=B8=8A?= =?UTF-8?q?=E9=99=90=E4=B8=8E=20worker=20=E9=98=9F=E5=88=97=E6=9C=89?= =?UTF-8?q?=E7=95=8C=E8=83=8C=E5=8E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 40 ++++++------- Cargo.toml | 2 +- crates/oxide_cli/src/server/server.rs | 79 ++++++++++++++++++++++++-- crates/oxide_cli/src/server/workers.rs | 74 ++++++++++++++++++------ 4 files changed, 153 insertions(+), 42 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a7325531..118ced0c 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index 069874ce..8f837258 100644 --- a/Cargo.toml +++ b/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" diff --git a/crates/oxide_cli/src/server/server.rs b/crates/oxide_cli/src/server/server.rs index 8b5a051d..c40a4b19 100644 --- a/crates/oxide_cli/src/server/server.rs +++ b/crates/oxide_cli/src/server/server.rs @@ -10,9 +10,9 @@ //! - accept 循环为非阻塞 accept 加 10 毫秒轮询加原子关闭标志：唤醒路径 //! 确定、无锁、无唤醒丢失。 //! - 并发模型：主线程只跑 accept 循环；每连接一个标准库线程只做 I/O -//! （帧读写）；六类控制请求由连接线程直接应答、不占 worker；执行请求与 -//! forge 查询经 mpsc 路由到固定 N 个常驻 worker 执行（各持自己线程上的 -//! VM 池）。 +//! （帧读写）；accept 循环带并发连接上限（超限连接立即关闭）；六类控制 +//! 请求由连接线程直接应答、不占 worker；执行请求与 forge 查询经 mpsc +//! 路由到固定 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 { /// /// # 步骤 /// 1. 检查关闭标志（原子布尔，序一致排序）。 -/// 2. accept：成功则派生连接线程（在途计数加一）；WouldBlock 睡 10 毫秒再查； -/// 其余错误以 `AcceptFailed` 返回。 +/// 2. accept：成功则先查并发连接上限，未超限派生连接线程（在途计数加一）、 +/// 超限立即关闭该流；WouldBlock 睡 10 毫秒再查；其余错误以 `AcceptFailed` +/// 返回。 +/// +/// # 边界与前提 +/// - 上限检查与派生同在单线程 accept 循环内，无竞态；在途计数是既有的 +/// 连接线程生命周期计数，直接复用。 /// /// # 副作用 /// - 每连接派生一个标准库线程（只做 I/O），线程结束时减一在途计数。 @@ -726,6 +735,12 @@ fn accept_loop(listener: &Listener, ctx: &Arc, router: &Arc { + // 并发连接上限：在途计数达上限时立即关闭该流（客户端表现为 + // 连接建立随即被关，不发协议帧），不派生线程、不加计数。 + 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 = (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] diff --git a/crates/oxide_cli/src/server/workers.rs b/crates/oxide_cli/src/server/workers.rs index 4bbad0ac..d2a1e82e 100644 --- a/crates/oxide_cli/src/server/workers.rs +++ b/crates/oxide_cli/src/server/workers.rs @@ -6,6 +6,8 @@ //! `Arc` 是 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>>, + senders: Mutex>>, next: AtomicUsize, } impl WorkerRouter { - fn from_senders(senders: Vec>) -> Self { + fn from_senders(senders: Vec>) -> Self { WorkerRouter { senders: Mutex::new(senders), next: AtomicUsize::new(0), @@ -60,18 +67,26 @@ impl WorkerRouter { /// 轮询投递任务到 worker。 /// /// # 步骤 - /// 1. 计数加一取模 N 选发送端。 - /// 2. 投递；发送端已清空（关闭序列进行中）或接收端已断开时归还任务。 + /// 1. 锁内计数加一取模 N 克隆发送端，随即释放锁。 + /// 2. 锁外投递；队列满时阻塞等待空位（背压）。 + /// 3. 发送端已清空（关闭序列进行中）或接收端已断开时归还任务。 /// /// # 边界与前提 /// - 返回 `Err` 时任务未被任何 worker 接收，调用方须以错误帧应答，不得阻塞等待。 + /// - 队列满时调用线程阻塞至空位腾出；阻塞期间不持有发送端锁， + /// 其余连接线程的投递不被串行卡死。 pub fn route(&self, task: WorkerTask) -> Result<(), WorkerTask> { - 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) { + // 锁内克隆发送端、锁外投递：队列满时 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(); + 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, kernel: Arc, 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 (tx2, rx2) = mpsc::channel(); + let (tx1, rx1) = mpsc::sync_channel(QUEUE_CAPACITY); + 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(), "清空后投递应归还任务");