From 6422c142ede56ca393d89c3b2f62a82434dab149 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Tue, 6 Oct 2026 15:21:38 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20server=20=E5=9B=BA=E5=AE=9A=20worker=20?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E6=A8=A1=E5=9E=8B=EF=BC=8Cmpsc=20=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=E4=B8=8E=E6=8E=A7=E5=88=B6=E6=89=A7=E8=A1=8C=E5=88=86?= =?UTF-8?q?=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.toml | 2 +- crates/oxide_cli/src/server/mod.rs | 13 +- crates/oxide_cli/src/server/server.rs | 303 +++++++++++++++++++++---- crates/oxide_cli/src/server/workers.rs | 214 +++++++++++++++++ crates/oxide_vm/src/vm_pool.rs | 61 ++++- 5 files changed, 543 insertions(+), 50 deletions(-) create mode 100644 crates/oxide_cli/src/server/workers.rs diff --git a/Cargo.toml b/Cargo.toml index 6c7d8d6f..1d9fa9ad 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ members = [ ] [workspace.package] -version = "0.8.28" +version = "0.8.29" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_cli/src/server/mod.rs b/crates/oxide_cli/src/server/mod.rs index 0d670257..0d3dd4b4 100644 --- a/crates/oxide_cli/src/server/mod.rs +++ b/crates/oxide_cli/src/server/mod.rs @@ -1,18 +1,21 @@ -//! 持久 server 子系统：协议数据面、身份注册、执行路径与 server 进程主体。 +//! 持久 server 子系统：协议数据面、身份注册、执行路径、worker 池与 server 进程主体。 //! //! 持久 server 是常驻进程，复用预热池、近零 spawn 成本； //! 通信走 Unix socket + NDJSON（换行分隔 JSON）。 -//! 模块分四层：协议数据面（帧格式与 serde 结构体）、身份注册（sidecar 文件 +//! 模块分五层：协议数据面（帧格式与 serde 结构体）、身份注册（sidecar 文件 //! 与 liveness 检查原语）、执行请求路径（parse → compile → spawn → run → -//! format → drop 六段纯函数）、server 进程主体（accept 循环、控制请求直接 -//! 处理、优雅退出）。并发模型在后续任务。 +//! format → drop 六段纯函数）、worker 池（固定 N 常驻 worker、各持自有 VM +//! 池、mpsc 轮询路由）、server 进程主体（accept 循环、控制请求直接处理、 +//! 执行请求路由、优雅退出）。 /// 执行请求路径：parse → compile → spawn → run → format → drop 六段纯函数。 pub mod eval; /// 协议数据面：NDJSON 帧格式与 serde 结构体。 pub mod protocol; -/// server 进程主体：accept 循环、控制请求分派与优雅退出。 +/// server 进程主体：accept 循环、控制请求分派、执行请求路由与优雅退出。 #[allow(clippy::module_inception)] pub mod server; /// 身份注册：sidecar 文件、liveness 检查与陈旧清理。 pub mod sidecar; +/// worker 池：固定 N 常驻 worker、各持自有 VM 池、mpsc 轮询路由。 +pub mod workers; diff --git a/crates/oxide_cli/src/server/server.rs b/crates/oxide_cli/src/server/server.rs index 92accf2c..6fd6479a 100644 --- a/crates/oxide_cli/src/server/server.rs +++ b/crates/oxide_cli/src/server/server.rs @@ -1,16 +1,17 @@ -//! server 进程主体：accept 循环、控制请求直接处理与优雅退出。 +//! server 进程主体：accept 循环、控制请求直接处理、执行请求路由与优雅退出。 //! //! 关键约定： //! - 启动顺序契约（与身份注册一致）：先认领 sidecar，再绑定 socket； //! 反向会让客户端在 sidecar 缺位时误判无 server 并启动第二实例。 //! - accept 循环为非阻塞 accept 加 10 毫秒轮询加原子关闭标志：唤醒路径 -//! 确定、无锁、无唤醒丢失；每连接一个标准库线程是临时模型，并发任务 -//! 整体替换为固定 worker 加消息队列路由，`handle_connection` 签名保持稳定。 -//! - 五类控制请求全部直接应答，不 spawn 虚拟机；执行请求暂以协议错误帧 -//! 答复（执行路径由后续任务补齐）。 -//! - 优雅退出顺序：关闭请求 → 在途归零 → drop 池 → drop 内核 → -//! 删 sidecar 与 socket 文件。 -//! - 池本体不是 Send/Sync（Vm 不跨线程），上下文只持池的原子状态句柄。 +//! 确定、无锁、无唤醒丢失。 +//! - 并发模型：主线程只跑 accept 循环；每连接一个标准库线程只做 I/O +//! （帧读写）；五类控制请求由连接线程直接应答、不占 worker；执行请求 +//! 经 mpsc 路由到固定 N 个常驻 worker 执行（各持自己线程上的 VM 池）。 +//! - 优雅退出顺序：关闭请求 → 在途归零 → 清空发送端 → join worker → +//! drop 内核 → 删 sidecar 与 socket 文件。 +//! - 池本体不是 Send/Sync（Vm 不跨线程），上下文只持池的原子状态句柄； +//! 状态响应读全体 worker 池的共享聚合计数器。 use std::fs; use std::io; @@ -22,11 +23,11 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use oxide_kernel::kernel::{KernelConfig, KernelCore}; -use oxide_vm::vm_pool::VmPool; +use oxide_vm::vm_pool::PoolCounters; -use super::eval; use super::protocol::{self, FrameReader, ProtocolError, ServerRequest, ServerResponse}; use super::sidecar::{self, ClaimResult}; +use super::workers::{self, WorkerRouter}; /// server 启动与运行错误：绑定失败、认领被拒、接受连接失败、读取身份失败。 #[derive(Debug, Clone, PartialEq, Eq)] @@ -39,6 +40,8 @@ pub enum ServerError { AcceptFailed(String), /// 读取身份失败（sidecar 写后读回解析失败）。 IdentityReadFailed(String), + /// worker 线程异常退出（panic 载荷文本）。 + WorkerPanic(String), } impl std::fmt::Display for ServerError { @@ -48,6 +51,7 @@ impl std::fmt::Display for ServerError { ServerError::BindFailed(msg) => write!(f, "socket 绑定失败：{msg}"), ServerError::AcceptFailed(msg) => write!(f, "接受连接失败：{msg}"), ServerError::IdentityReadFailed(msg) => write!(f, "读取身份失败：{msg}"), + ServerError::WorkerPanic(msg) => write!(f, "worker 线程异常退出：{msg}"), } } } @@ -61,38 +65,52 @@ pub struct ServerConfig { pub socket_path: PathBuf, /// sidecar 文件路径。 pub sidecar_path: PathBuf, + /// 常驻 worker 线程数；缺省取宿主核数，下限钳制为 1。 + pub worker_count: usize, } impl ServerConfig { - /// 生产默认配置：socket 与 sidecar 取 well-known 位置。 + /// 生产默认配置：socket 与 sidecar 取 well-known 位置，worker 数取宿主核数。 pub fn well_known() -> Self { ServerConfig { socket_path: sidecar::well_known_socket_path(), sidecar_path: sidecar::well_known_sidecar_path(), + worker_count: default_worker_count(), } } } -/// server 运行上下文：共享内核、启动时刻、关闭标志、在途计数。 +/// 缺省 worker 数：宿主核数，下限钳制为 1。 +fn default_worker_count() -> usize { + std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1) + .max(1) +} + +/// server 运行上下文：共享内核、启动时刻、关闭标志、在途计数、池聚合计数器。 /// /// 内核跨线程共享（`Arc` 是 Send/Sync）；池本体不进上下文 -/// （Vm 不跨线程），每连接线程自建池（见 `handle_connection`）。 +/// （Vm 不跨线程，池归各 worker 线程所有），状态响应读共享聚合计数器。 struct ServerContext { kernel: Arc, started_at: Instant, shutdown: Arc, in_flight: Arc, + pool_counters: Arc, socket_path: String, } -/// server 进程主体：认领 sidecar → 绑定 socket → 建内核与池 → accept 循环 → -/// 在途归零 → 按序 drop 池与内核 → 删 sidecar 与 socket 文件。 +/// server 进程主体：认领 sidecar → 绑定 socket → 建内核 → 派生 worker → +/// accept 循环 → 在途归零 → 清空发送端 → join worker → drop 内核 → +/// 删 sidecar 与 socket 文件。 /// /// # 步骤 /// 1. 认领 sidecar（启动顺序契约：先写 sidecar 再 bind socket）。 /// 2. 绑定监听器；失败且路径存在时删 stale socket 文件重试一次。 -/// 3. 建内核（标准配置）与池（按 min_pool_size 预热）。 -/// 4. 进入 accept 循环；关闭标志置位后在途归零、按序释放、删文件。 +/// 3. 建内核（标准配置）与共享聚合计数器。 +/// 4. 派生 N 个常驻 worker（各在自己线程上建自有池）。 +/// 5. 进入 accept 循环；关闭标志置位后按关闭序列退出。 /// /// # 边界与前提 /// - 认领被拒（AlreadyRunning / RefusedAmbiguous）立即返回错误，不双注册。 @@ -100,11 +118,13 @@ struct ServerContext { /// /// # 副作用 /// - 创建 sidecar 与 socket 文件；正常退出时删除两者。 -/// - 每连接派生一个标准库线程（临时模型）。 +/// - 派生 N 个常驻 worker 线程与每连接一个 I/O 线程。 /// /// # 注意事项 /// - 在调用线程上运行（后续 CLI 骨架任务在主线程分派到它）。 /// - 关闭标志由关闭请求置位；信号处理由后续任务在同一标志上接线。 +/// - 关闭序列固定：清空发送端 → join 全部 worker → drop 内核（内核 drop +/// 的调试断言要求存活 VM 数为零，worker 池随线程退出先释放，前提满足）。 pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { // 认领 sidecar：已有存活 server 或存活证据不足即返回错误。 match sidecar::claim_sidecar(&config.sidecar_path, &config.socket_path) { @@ -115,24 +135,45 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { // 绑定监听器：stale socket 文件删一次重试，失败回滚 sidecar。 let listener = bind_listener(config)?; - // 建内核（标准配置）；池每连接线程自建（Vm 不跨线程，池本体不共享）。 + // 建内核（标准配置）与共享聚合计数器（全体 worker 池的增减累加到同一聚合）。 let kernel = KernelCore::new(KernelConfig::standard()); + let counters = PoolCounters::shared(); + + // 派生 N 个常驻 worker：各持自己线程上建的池（Vm 不跨线程）。 + let (router, worker_handles) = workers::spawn_workers( + Arc::clone(&kernel), + Arc::clone(&counters), + kernel.config.min_pool_size, + kernel.config.max_pool_size, + config.worker_count, + ); + let router = Arc::new(router); let ctx = Arc::new(ServerContext { kernel: Arc::clone(&kernel), started_at: Instant::now(), shutdown: Arc::new(AtomicBool::new(false)), in_flight: Arc::new(AtomicUsize::new(0)), + pool_counters: counters, socket_path: config.socket_path.to_string_lossy().into_owned(), }); // accept 循环：非阻塞 accept 加 10 毫秒轮询加原子关闭标志。 - let accept_result = accept_loop(&listener, &ctx); + let accept_result = accept_loop(&listener, &ctx, &router); - // 在途归零：轮询至零，5 秒安全超时后记警告继续。 + // 在途归零：轮询至零，5 秒安全超时后记警告继续（余留连接线程持内核 + // 与路由器的共享句柄，不悬垂；其后续投递经清空的路由立即失败）。 wait_in_flight_drained(&ctx.in_flight, Duration::from_secs(5)); - // 按序释放：上下文（持内核句柄）→ 内核。 + // 关闭序列：清空发送端 → worker 的 recv 断开退出 → 逐一 join。 + router.close(); + for handle in worker_handles { + if let Err(payload) = handle.join() { + return Err(ServerError::WorkerPanic(panic_payload_str(&payload))); + } + } + + // 按序释放：上下文（持内核句柄）→ 内核（worker 池已随线程退出释放）。 drop(ctx); drop(kernel); @@ -143,6 +184,18 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { accept_result } +/// 从 `catch_unwind` 的 panic payload 提取可读文本：`&str` 与 `String` 两种 +/// 常见形态，其余返回占位文本。 +fn panic_payload_str(payload: &Box) -> String { + if let Some(s) = payload.downcast_ref::<&str>() { + return (*s).to_string(); + } + if let Some(s) = payload.downcast_ref::() { + return s.clone(); + } + "non-string payload".into() +} + /// 绑定监听器：失败且路径存在时删 stale socket 文件重试一次。 /// /// # 边界与前提 @@ -179,8 +232,12 @@ fn bind_listener(config: &ServerConfig) -> Result { /// 其余错误以 `AcceptFailed` 返回。 /// /// # 副作用 -/// - 每连接派生一个标准库线程，线程结束时减一在途计数。 -fn accept_loop(listener: &UnixListener, ctx: &Arc) -> Result<(), ServerError> { +/// - 每连接派生一个标准库线程（只做 I/O），线程结束时减一在途计数。 +fn accept_loop( + listener: &UnixListener, + ctx: &Arc, + router: &Arc, +) -> Result<(), ServerError> { listener .set_nonblocking(true) .map_err(|e| ServerError::AcceptFailed(e.to_string()))?; @@ -195,8 +252,9 @@ fn accept_loop(listener: &UnixListener, ctx: &Arc) -> Result<(), Ok((stream, _addr)) => { ctx.in_flight.fetch_add(1, Ordering::SeqCst); let ctx = Arc::clone(ctx); + let router = Arc::clone(router); std::thread::spawn(move || { - handle_connection(stream, &ctx); + handle_connection(stream, &ctx, &router); ctx.in_flight.fetch_sub(1, Ordering::SeqCst); }); } @@ -213,15 +271,17 @@ fn accept_loop(listener: &UnixListener, ctx: &Arc) -> Result<(), /// # 步骤 /// 1. 克隆写端，原流交给帧读取器，设 30 秒读超时。 /// 2. 循环读帧：EOF、读超时、非法 UTF-8 退出；帧长超限按协议约定关连接。 -/// 3. 畸形帧以 `Error` 帧答复后继续；正常帧分派并写回。 +/// 3. 畸形帧以 `Error` 帧答复后继续；执行请求路由到 worker 并阻塞等回复； +/// 控制请求直接应答并写回。 /// 4. 关闭确认帧写回即退出循环。 /// /// # 边界与前提 /// - 半开连接（对端静默不关）由 30 秒读超时有界化。 +/// - 本线程只做 I/O 与分派，不建池（池归 worker 线程所有）。 /// /// # 副作用 /// - 关闭请求会置位关闭标志。 -fn handle_connection(stream: UnixStream, ctx: &ServerContext) { +fn handle_connection(stream: UnixStream, ctx: &ServerContext, router: &WorkerRouter) { // 克隆写端：原流交给帧读取器，写端独立用于响应写回。 let mut writer = match stream.try_clone() { Ok(writer) => writer, @@ -231,15 +291,14 @@ fn handle_connection(stream: UnixStream, ctx: &ServerContext) { // 30 秒读超时：半开连接不永久占住线程。 let _ = stream.set_read_timeout(Some(Duration::from_secs(30))); - // 每连接自建池（Vm 不跨线程，池是每线程资产），按内核配置预热。 - let pool = VmPool::new(Arc::clone(&ctx.kernel), ctx.kernel.config.min_pool_size, ctx.kernel.config.max_pool_size); - let mut reader = FrameReader::new(io::BufReader::new(stream)); // 读帧：EOF、读超时、非法 UTF-8 均退出。 while let Ok(Some(frame)) = reader.read_frame() { - // 分派：帧长超限按协议约定关连接；畸形帧以 Error 帧答复后继续。 + // 分派：帧长超限按协议约定关连接；畸形帧以 Error 帧答复后继续； + // 执行请求走 worker 路由，控制请求直接应答。 let response = match protocol::parse_request(&frame) { - Ok(request) => dispatch_control(&request, ctx, &pool), + Ok(ServerRequest::Eval { code, max_steps }) => dispatch_eval(&code, max_steps, router), + Ok(request) => dispatch_control(&request, ctx), Err(ProtocolError::FrameTooLarge) => break, Err(error) => error.to_error_frame(), }; @@ -257,17 +316,36 @@ fn handle_connection(stream: UnixStream, ctx: &ServerContext) { } } -/// 控制请求分派：五类控制请求直接应答；执行请求走 `handle_eval`。 +/// 执行请求分派：路由到 worker 执行，阻塞在一次性回复通道上取回响应帧。 /// -/// 池是每连接资产（`handle_connection` 创建），Status 响应读本连接池状态。 -fn dispatch_control(request: &ServerRequest, ctx: &ServerContext, pool: &Arc) -> ServerResponse { +/// # 步骤 +/// 1. 建一次性回复通道，发送端随任务进 worker。 +/// 2. 轮询投递到 worker；发送端已清空（关闭序列进行中）时立即以错误帧答复。 +/// 3. 阻塞 recv 至回复；worker 已退出（通道断开）时以错误帧答复，不挂起。 +fn dispatch_eval(code: &str, max_steps: Option, router: &WorkerRouter) -> ServerResponse { + let (reply_tx, reply_rx) = std::sync::mpsc::channel(); + let task = workers::WorkerTask::Eval { + code: code.to_string(), + max_steps, + reply: reply_tx, + }; + match router.route(task) { + Ok(()) => reply_rx.recv().unwrap_or_else(|_| ServerResponse::eval_err("worker 已退出，执行请求未处理")), + Err(_) => ServerResponse::eval_err("server 正在关闭，执行请求未受理"), + } +} + +/// 控制请求分派：五类控制请求直接应答，不占 worker。 +/// +/// Status 响应读全体 worker 池的共享聚合计数器。 +fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerResponse { match request { ServerRequest::Version => ServerResponse::Version { version: env!("CARGO_PKG_VERSION").to_string(), }, ServerRequest::Status => ServerResponse::Status { - pool_available: pool.available_count(), - pool_total: pool.total_count(), + pool_available: ctx.pool_counters.available(), + pool_total: ctx.pool_counters.total(), uptime_ms: ctx.started_at.elapsed().as_millis() as u64, }, ServerRequest::Health => ServerResponse::Health { healthy: true }, @@ -281,7 +359,9 @@ fn dispatch_control(request: &ServerRequest, ctx: &ServerContext, pool: &Arc eval::handle_eval(code, *max_steps, &ctx.kernel, pool), + // 防御性兜底：正常路径 eval 在调用本函数前已被拦截走 worker 路由； + // 若到达此臂说明上游分流缺失，以错误帧答复而非静默执行。 + ServerRequest::Eval { .. } => ServerResponse::eval_err("执行请求应经 worker 路由，不应到达控制分派"), } } @@ -303,12 +383,14 @@ mod tests { use std::thread::JoinHandle; /// 唯一临时路径（进程号加测试名）：单进程内唯一，进程间以进程号隔离。 + /// worker 数取小值 2：测试在宿主核数上跑，大值徒增预热成本。 fn unique_paths(test_name: &str) -> ServerConfig { let dir = std::env::temp_dir().join(format!("oxide_server_test_{}_{}", std::process::id(), test_name)); fs::create_dir_all(&dir).expect("测试目录创建应成功"); ServerConfig { socket_path: dir.join("server.sock"), sidecar_path: dir.join("sidecar.json"), + worker_count: 2, } } @@ -361,13 +443,26 @@ mod tests { stop_server(&mut stream, handle); } - /// 状态请求得 Status 帧：空闲数不超过总数、总数不小于 1（标准配置预热 1 个）、运行时长合理。 + /// 状态请求得 Status 帧：空闲数不超过总数、总数不小于 1（标准配置每 worker 预热 1 个）、运行时长合理。 #[test] fn status_response() { let config = unique_paths("status"); let handle = start_server(config.clone()); let mut stream = connect(&config); - let response = send_and_recv(&mut stream, &ServerRequest::Status); + + // worker 池异步预热：轮询至聚合总数不小于 1（预热完成）。 + let deadline = Instant::now() + Duration::from_secs(5); + let response = loop { + let response = send_and_recv(&mut stream, &ServerRequest::Status); + if let ServerResponse::Status { pool_total, .. } = &response { + if *pool_total >= 1 { + break response; + } + } + assert!(Instant::now() < deadline, "worker 池预热应完成"); + std::thread::sleep(Duration::from_millis(10)); + }; + match response { ServerResponse::Status { pool_available, @@ -491,4 +586,132 @@ mod tests { stop_server(&mut stream, handle); } + + /// 端到端并发：两条连接各发 3 个 eval，6 个响应帧全部正确且逐连接顺序与请求一致。 + #[test] + fn concurrent_evals_across_connections() { + let config = unique_paths("concurrent_evals"); + let handle = start_server(config.clone()); + let mut stream_a = connect(&config); + let mut stream_b = connect(&config); + + // 两条连接各发 3 个 eval，响应逐帧按请求顺序送达。 + for code in ["1 + 1", "2 * 3", "10 - 4"] { + let response = send_and_recv(&mut stream_a, &ServerRequest::Eval { code: code.into(), max_steps: None }); + assert!( + matches!(response, ServerResponse::EvalResult { ref value, .. } if value.is_some()), + "连接 A 应得完成值：{response:?}" + ); + } + for code in ["21 / 7", "2 + 30", "100 % 9"] { + let response = send_and_recv(&mut stream_b, &ServerRequest::Eval { code: code.into(), max_steps: None }); + assert!( + matches!(response, ServerResponse::EvalResult { ref value, .. } if value.is_some()), + "连接 B 应得完成值：{response:?}" + ); + } + + // 完成值逐条核对（顺序与请求一致）。 + let expected_a = ["2", "6", "6"]; + let expected_b = ["3", "32", "1"]; + for (code, expected) in [ + ("1 + 1", expected_a[0]), + ("2 * 3", expected_a[1]), + ("10 - 4", expected_a[2]), + ] { + let response = send_and_recv(&mut stream_a, &ServerRequest::Eval { code: code.into(), max_steps: None }); + match response { + ServerResponse::EvalResult { ref value, .. } => assert_eq!(value.as_deref(), Some(expected), "{code} 应得 {expected}"), + other => panic!("应得 EvalResult 帧：{other:?}"), + } + } + for (code, expected) in [ + ("21 / 7", expected_b[0]), + ("2 + 30", expected_b[1]), + ("100 % 9", expected_b[2]), + ] { + let response = send_and_recv(&mut stream_b, &ServerRequest::Eval { code: code.into(), max_steps: None }); + match response { + ServerResponse::EvalResult { ref value, .. } => assert_eq!(value.as_deref(), Some(expected), "{code} 应得 {expected}"), + other => panic!("应得 EvalResult 帧：{other:?}"), + } + } + + drop(stream_a); + stop_server(&mut stream_b, handle); + } + + /// 控制请求不占 worker：一条连接发长 eval 占住 worker 期间， + /// 另一连接发 Status 立即返回（不受 eval 阻塞）。 + #[test] + fn control_request_not_blocked_by_eval() { + let config = unique_paths("control_not_blocked"); + let handle = start_server(config.clone()); + let mut stream_a = connect(&config); + let mut stream_b = connect(&config); + + // 连接 A 发长 eval（步数上限使执行有界，占住 worker 数百毫秒）。 + stream_a + .write_all( + protocol::encode_request(&ServerRequest::Eval { + code: "for(;;){}".into(), + max_steps: Some(20_000_000), + }) + .as_bytes(), + ) + .expect("写长 eval 应成功"); + + // 连接 B 发 Status：控制请求直接应答，须在有界时间内返回。 + let started = Instant::now(); + let response = send_and_recv(&mut stream_b, &ServerRequest::Status); + assert!(started.elapsed() < Duration::from_secs(3), "Status 不应被 eval 阻塞"); + assert!(matches!(response, ServerResponse::Status { .. }), "应得 Status 帧：{response:?}"); + + // 连接 A 的长 eval 最终得步数超限错误帧。 + let mut reader = FrameReader::new(io::BufReader::new(&mut stream_a)); + let frame = reader.read_frame().expect("读响应应成功").expect("应读到响应帧"); + let response = protocol::parse_response(&frame).expect("响应帧解析应成功"); + match response { + ServerResponse::EvalResult { ref error, .. } => { + let err = error.as_deref().expect("长 eval 应得步数超限错误"); + assert!(err.contains("step limit"), "错误应含 step limit：{err}"); + } + other => panic!("应得 EvalResult 帧：{other:?}"), + } + + drop(stream_a); + stop_server(&mut stream_b, handle); + } + + /// 关闭序列：在有在途 eval 时发关闭请求，server 线程正常退出， + /// sidecar 与 socket 文件均被删除。 + #[test] + fn shutdown_with_in_flight_eval() { + let config = unique_paths("shutdown_inflight"); + let handle = start_server(config.clone()); + let mut stream_a = connect(&config); + let mut stream_b = connect(&config); + + // 连接 A 发长 eval（在途）；连接 B 发关闭请求。 + stream_a + .write_all( + protocol::encode_request(&ServerRequest::Eval { + code: "for(;;){}".into(), + max_steps: Some(20_000_000), + }) + .as_bytes(), + ) + .expect("写长 eval 应成功"); + let response = send_and_recv(&mut stream_b, &ServerRequest::Shutdown); + assert!(matches!(response, ServerResponse::Shutdown), "应得关闭确认帧：{response:?}"); + + // 关闭连接 A：在途连接线程写回响应后退出，在途归零。 + drop(stream_a); + drop(stream_b); + + let result = handle.join().expect("server 线程应正常退出"); + assert!(result.is_ok(), "server 应正常退出：{result:?}"); + assert!(!config.sidecar_path.exists(), "sidecar 文件应被删除"); + assert!(!config.socket_path.exists(), "socket 文件应被删除"); + } } diff --git a/crates/oxide_cli/src/server/workers.rs b/crates/oxide_cli/src/server/workers.rs new file mode 100644 index 00000000..f3d397ec --- /dev/null +++ b/crates/oxide_cli/src/server/workers.rs @@ -0,0 +1,214 @@ +//! worker 池：固定 N 个常驻 worker 线程，各持自己线程上建的 VM 池， +//! 执行请求经 mpsc 路由到 worker 执行。 +//! +//! 关键约定： +//! - `Vm` 不是 Send，池在 worker 线程闭包内创建、永不跨线程移动； +//! `Arc` 是 Send/Sync，由全体 worker 共享（字节码缓存跨 worker 命中）。 +//! - 标准库 mpsc 的接收端不是 Send，不能跨线程共享给多个 worker， +//! 因此每 worker 一条独立通道，发送端收进 `WorkerRouter` 按轮询投递。 +//! - 关闭序列固定：`close` 清空发送端 → worker 的 recv 断开退出 → +//! 主线程逐一 join → 再 drop 内核。 + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{mpsc, Arc, Mutex}; +use std::thread::JoinHandle; + +use oxide_kernel::kernel::KernelCore; +use oxide_vm::vm_pool::{PoolCounters, VmPool}; + +use super::eval; +use super::protocol::ServerResponse; + +/// worker 任务：执行请求加一次性回复通道。 +#[derive(Debug)] +pub enum WorkerTask { + /// 执行请求：源码、可选最大指令数、一次性回复通道（worker 执行完送回响应帧）。 + Eval { + code: String, + max_steps: Option, + reply: mpsc::Sender, + }, +} + +/// worker 路由器：N 条独立任务通道的发送端加轮询计数。 +/// +/// 发送端收进互斥锁：连接线程可能存活超过主线程的关闭序列（在途排水超时后 +/// 仍有余留连接），`close` 清空发送端后它们的投递立即失败，不挂起、不悬垂。 +pub struct WorkerRouter { + senders: Mutex>>, + next: AtomicUsize, +} + +impl WorkerRouter { + fn from_senders(senders: Vec>) -> Self { + WorkerRouter { + senders: Mutex::new(senders), + next: AtomicUsize::new(0), + } + } + + /// 轮询投递任务到 worker。 + /// + /// # 步骤 + /// 1. 计数加一取模 N 选发送端。 + /// 2. 投递；发送端已清空（关闭序列进行中）或接收端已断开时归还任务。 + /// + /// # 边界与前提 + /// - 返回 `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) { + Ok(()) => Ok(()), + Err(sent) => Err(sent.0), + } + } + + /// 关闭序列第一步：清空全部发送端，worker 的 recv 断开退出。 + pub fn close(&self) { + self.senders.lock().unwrap().clear(); + } +} + +/// 派生 N 个常驻 worker 线程：各线程在自己线程上建自有 VM 池（共享聚合计数器） +/// 并进入任务循环。 +/// +/// # 步骤 +/// 1. 建 N 对通道，接收端各移入一个 worker 线程。 +/// 2. worker 线程内按内核配置的池规模建池（共享聚合计数器）。 +/// 3. 各 worker 进入任务循环：recv → 执行 → 经回复通道送回响应帧。 +/// +/// # 边界与前提 +/// - `worker_count` 下限钳制为 1。 +/// +/// # 副作用 +/// - 派生 N 个常驻线程；发送端全部清空后线程退出。 +/// +/// # 注意事项 +/// - 返回的句柄必须在关闭序列中逐一 join（drop 发送端 → join → drop 内核）。 +pub fn spawn_workers( + kernel: Arc, + counters: Arc, + min_pool_size: usize, + max_pool_size: Option, + worker_count: usize, +) -> (WorkerRouter, Vec>) { + let n = worker_count.max(1); + let mut senders = Vec::with_capacity(n); + let mut handles = Vec::with_capacity(n); + for _ in 0..n { + let (tx, rx) = mpsc::channel(); + let kernel = Arc::clone(&kernel); + let counters = Arc::clone(&counters); + handles.push(std::thread::spawn(move || { + // 池在本线程上创建，永不跨线程移动（Vm 不是 Send）。 + let pool = VmPool::with_counters(Arc::clone(&kernel), min_pool_size, max_pool_size, counters); + worker_loop(rx, kernel, pool); + })); + senders.push(tx); + } + (WorkerRouter::from_senders(senders), handles) +} + +/// worker 任务循环：recv 任务 → 执行 → 经回复通道送回响应帧；发送端断开即退出。 +fn worker_loop(receiver: mpsc::Receiver, kernel: Arc, pool: Arc) { + for task in receiver { + match task { + WorkerTask::Eval { + code, + max_steps, + reply, + } => { + let response = eval::handle_eval(&code, max_steps, &kernel, &pool); + let _ = reply.send(response); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use oxide_kernel::kernel::KernelConfig; + + /// 建一个哑任务（回复通道只挂不读）。 + fn dummy_task() -> WorkerTask { + let (reply, _rx) = mpsc::channel(); + WorkerTask::Eval { + code: "1".into(), + max_steps: None, + reply, + } + } + + /// 抽干接收端，返回收到的任务数。 + fn drain_count(receiver: &mpsc::Receiver) -> usize { + let mut n = 0; + while receiver.try_recv().is_ok() { + n += 1; + } + n + } + + /// 轮询分发：N 等于 2，投 4 个任务，两个 worker 各收到 2 个。 + #[test] + fn round_robin_distribution() { + let (tx1, rx1) = mpsc::channel(); + let (tx2, rx2) = mpsc::channel(); + let router = WorkerRouter::from_senders(vec![tx1, tx2]); + for _ in 0..4 { + router.route(dummy_task()).expect("投递应成功"); + } + assert_eq!(drain_count(&rx1), 2, "第一个 worker 应收到 2 个任务"); + assert_eq!(drain_count(&rx2), 2, "第二个 worker 应收到 2 个任务"); + } + + /// 关闭序列：清空发送端后 worker 线程退出，join 在有界时间内返回。 + #[test] + fn workers_exit_when_senders_cleared() { + let kernel = KernelCore::new(KernelConfig::minimal()); + let counters = PoolCounters::shared(); + let (router, handles) = spawn_workers(kernel, counters, 1, Some(2), 2); + router.close(); + for handle in handles { + handle.join().expect("worker 线程应正常退出"); + } + } + + /// 发送端清空后投递归还任务：调用方不挂起，可立即以错误帧应答。 + #[test] + fn route_after_close_returns_task() { + let (tx, _rx) = mpsc::channel(); + let router = WorkerRouter::from_senders(vec![tx]); + router.close(); + assert!(router.route(dummy_task()).is_err(), "清空后投递应归还任务"); + } + + /// 端到端：真实 worker 执行 eval 并经回复通道送回响应帧。 + #[test] + fn worker_executes_eval_and_replies() { + let kernel = KernelCore::new(KernelConfig::minimal()); + let counters = PoolCounters::shared(); + let (router, handles) = spawn_workers(kernel, counters, 1, Some(2), 1); + let (reply_tx, reply_rx) = mpsc::channel(); + router + .route(WorkerTask::Eval { + code: "1 + 1".into(), + max_steps: None, + reply: reply_tx, + }) + .expect("投递应成功"); + let response = reply_rx.recv().expect("回复通道应收到响应"); + assert!( + matches!(response, ServerResponse::EvalResult { ref value, .. } if value.as_deref() == Some("2")), + "应得完成值为 2 的响应：{response:?}" + ); + router.close(); + for handle in handles { + handle.join().expect("worker 线程应正常退出"); + } + } +} diff --git a/crates/oxide_vm/src/vm_pool.rs b/crates/oxide_vm/src/vm_pool.rs index 3e24e702..8744cf1c 100644 --- a/crates/oxide_vm/src/vm_pool.rs +++ b/crates/oxide_vm/src/vm_pool.rs @@ -23,7 +23,10 @@ pub struct PoolCounters { } impl PoolCounters { - fn new() -> Arc { + /// 创建新的聚合计数器（零初始化），供多个池共享。 + /// + /// 共享后各池的原子增减自然累加到同一聚合，读取方拿到的是全体池的合计。 + pub fn shared() -> Arc { Arc::new(PoolCounters { available: AtomicUsize::new(0), total: AtomicUsize::new(0), @@ -66,9 +69,32 @@ pub struct VmGuard { impl VmPool { /// 创建 VM 池并同步预热 `min_size` 个 Vm（数量受 `max_size` 钳制）， /// 首次 `spawn` 直接命中池。`max_size` 为池上限，`None` 表示不限。 + /// + /// 计数器为本池私有（不与其他池共享）。 pub fn new(kernel_core: Arc, min_size: usize, max_size: Option) -> Arc { + Self::with_counters(kernel_core, min_size, max_size, PoolCounters::shared()) + } + + /// 创建共享外部聚合计数器的 VM 池：多个池的原子增减累加到同一聚合， + /// 读取方（如 server 的状态响应）拿到的是全体池的合计。 + /// + /// # 步骤 + /// 1. 以给定计数器建池结构。 + /// 2. 同步预热 `min_size` 个 Vm（受 `max_size` 钳制），计数器以累加方式同步 + /// （共享聚合上不得覆盖写，各池各自累加）。 + /// + /// # 边界与前提 + /// - 计数器通常经 `PoolCounters::shared` 跨线程共享；池本体不跨线程移动。 + /// + /// # 副作用 + /// - 同步创建 `min_size` 个 Vm。 + pub fn with_counters( + kernel_core: Arc, + min_size: usize, + max_size: Option, + counters: Arc, + ) -> Arc { let warm = min_size.min(max_size.unwrap_or(min_size)); - let counters = PoolCounters::new(); let pool = Arc::new(Self { kernel_core: Arc::clone(&kernel_core), inner: Mutex::new(VmPoolInner { @@ -85,8 +111,8 @@ impl VmPool { inner.total_count += 1; } drop(inner); - counters.available.store(warm, Ordering::Relaxed); - counters.total.store(warm, Ordering::Relaxed); + counters.available.fetch_add(warm, Ordering::Relaxed); + counters.total.fetch_add(warm, Ordering::Relaxed); pool } @@ -276,4 +302,31 @@ mod tests { assert_eq!(inner.available.len(), 1); assert_eq!(inner.total_count, 1); } + + #[test] + fn test_with_counters_shared_aggregate() { + let kernel = test_kernel(); + let counters = PoolCounters::shared(); + let pool_a = VmPool::with_counters(Arc::clone(&kernel), 1, Some(4), Arc::clone(&counters)); + let pool_b = VmPool::with_counters(Arc::clone(&kernel), 2, Some(4), Arc::clone(&counters)); + + // 预热后聚合等于两池之和：总数 1 + 2 = 3，全部空闲。 + assert_eq!(counters.total(), 3, "聚合总数应为两池预热之和"); + assert_eq!(counters.available(), 3, "聚合空闲数应为两池预热之和"); + + // 各池借出再归还：聚合空闲数与总数等于两池之和。 + let g1 = pool_a.spawn(); + let g2 = pool_b.spawn(); + let g3 = pool_b.spawn(); + assert_eq!(counters.available(), 0, "三笔借出后聚合空闲数应为零"); + drop(g1); + drop(g2); + drop(g3); + assert_eq!(counters.available(), 3, "全部归还后聚合空闲数应复原"); + assert_eq!(counters.total(), 3, "聚合总数应不变"); + + // 各池的读取面读同一聚合，两池读数一致。 + assert_eq!(pool_a.available_count(), counters.available()); + assert_eq!(pool_b.available_count(), counters.available()); + } }