diff --git a/Cargo.lock b/Cargo.lock index e48f5392..c84bf2dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.15.4" +version = "0.15.5" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.15.4" +version = "0.15.5" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.15.4" +version = "0.15.5" dependencies = [ "ansi_term", "bumpalo", @@ -1217,6 +1217,7 @@ dependencies = [ "oxide_engine", "oxide_kernel", "oxide_log", + "oxide_net", "oxide_parser", "oxide_types", "oxide_vm", @@ -1227,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.15.4" +version = "0.15.5" dependencies = [ "lru", "oxide_bytecode", @@ -1236,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1256,7 +1257,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1270,7 +1271,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.15.4" +version = "0.15.5" dependencies = [ "num-bigint", "oxc_allocator", @@ -1285,7 +1286,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1297,7 +1298,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1306,7 +1307,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.15.4" +version = "0.15.5" dependencies = [ "bumpalo", "dashmap", @@ -1320,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1332,7 +1333,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.15.4" +version = "0.15.5" dependencies = [ "tracing", "tracing-appender", @@ -1341,11 +1342,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.15.4" +version = "0.15.5" [[package]] name = "oxide_parser" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1360,7 +1361,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1376,7 +1377,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.15.4" +version = "0.15.5" dependencies = [ "num-bigint", "num-traits", @@ -1389,7 +1390,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.15.4" +version = "0.15.5" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1407,7 +1408,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.15.4" +version = "0.15.5" dependencies = [ "bumpalo", "dashmap", @@ -1420,7 +1421,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.15.4" +version = "0.15.5" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 92f7d4da..b6122eda 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.15.4" +version = "0.15.5" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_cli/Cargo.toml b/crates/oxide_cli/Cargo.toml index b04b5ec4..43cb89d8 100644 --- a/crates/oxide_cli/Cargo.toml +++ b/crates/oxide_cli/Cargo.toml @@ -19,6 +19,7 @@ oxide_compiler = { path = "../oxide_compiler" } oxide_vm = { path = "../oxide_vm" } oxide_kernel = { path = "../oxide_kernel" } oxide_engine = { path = "../oxide_engine" } +oxide_net = { path = "../oxide_net" } oxide_log = { path = "../oxide_log" } bumpalo.workspace = true clap.workspace = true diff --git a/crates/oxide_cli/src/server/server.rs b/crates/oxide_cli/src/server/server.rs index f863ba15..696a7d5a 100644 --- a/crates/oxide_cli/src/server/server.rs +++ b/crates/oxide_cli/src/server/server.rs @@ -15,13 +15,14 @@ //! VM 池)。 //! - 优雅退出顺序:关闭请求、SIGINT/SIGTERM 信号或 yield 请求置位关闭标志 //! → 在途归零 → 清空发送端 → join worker → drop 内核 → 先删 socket 文件 -//! 后删 sidecar(新 server 以 sidecar 消失为接管信号)。 +//! (Unix 臂)后删 sidecar(新 server 以 sidecar 消失为接管信号)。 //! - 池本体不是 Send/Sync(Vm 不跨线程),上下文只持池的原子状态句柄; //! 状态响应读全体 worker 池的共享聚合计数器。 use std::fs; use std::io; use std::io::Write; +use std::net::TcpListener; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -31,6 +32,7 @@ use std::time::{Duration, Instant}; use oxide_kernel::kernel::{KernelConfig, KernelCore}; use oxide_log::{Level, LogConfig, Output, SUBSYSTEM_COUNT}; +use oxide_net::{Conn, Endpoint, Listener}; use oxide_vm::vm_pool::PoolCounters; use super::liveness; @@ -72,13 +74,13 @@ impl std::fmt::Display for ServerError { } } -/// server 运行配置:socket 与 sidecar 路径可注入。 +/// server 运行配置:监听端点与 sidecar 路径可注入。 /// /// 生产默认取 well-known 位置;测试注入临时路径。 #[derive(Debug, Clone)] pub struct ServerConfig { - /// Unix socket 路径。 - pub socket_path: PathBuf, + /// 监听端点(Unix 套接字路径或 TCP 地址)。 + pub endpoint: Endpoint, /// sidecar 文件路径。 pub sidecar_path: PathBuf, /// 常驻 worker 线程数;缺省取宿主核数,下限钳制为 1。 @@ -88,11 +90,11 @@ pub struct ServerConfig { } impl ServerConfig { - /// 生产默认配置:socket 与 sidecar 取 well-known 位置,worker 数取宿主 + /// 生产默认配置:endpoint 与 sidecar 取 well-known 位置,worker 数取宿主 /// 核数,版本取构建期版本。 pub fn well_known() -> Self { ServerConfig { - socket_path: sidecar::well_known_socket_path(), + endpoint: Endpoint::Unix(sidecar::well_known_socket_path()), sidecar_path: sidecar::well_known_sidecar_path(), worker_count: default_worker_count(), version: env!("CARGO_PKG_VERSION").to_string(), @@ -108,14 +110,14 @@ fn default_worker_count() -> usize { /// --rm 模式缺省空闲超时:30 秒。 const RM_DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(30); -/// --rm 独立模式运行配置:唯一 socket 路径、worker 数、空闲超时。 +/// --rm 独立模式运行配置:监听端点、worker 数、空闲超时。 /// /// 不含 sidecar 路径字段:「不注册 sidecar」的不变量在类型上成立, /// 不需要运行时守卫。 #[derive(Debug, Clone)] pub struct RmServerConfig { - /// Unix socket 路径;缺省为进程唯一路径。 - pub socket_path: PathBuf, + /// 监听端点;缺省为进程唯一 Unix 路径。 + pub endpoint: Endpoint, /// 常驻 worker 线程数;缺省取宿主核数,下限钳制为 1。 pub worker_count: usize, /// 空闲超时:绑定后该时长内无连接、或连接建立后该时长内无数据, @@ -126,7 +128,7 @@ pub struct RmServerConfig { impl Default for RmServerConfig { fn default() -> Self { RmServerConfig { - socket_path: sidecar::rm_socket_path(), + endpoint: Endpoint::Unix(sidecar::rm_socket_path()), worker_count: default_worker_count(), idle_timeout: RM_DEFAULT_IDLE_TIMEOUT, } @@ -146,13 +148,14 @@ struct ServerContext { shutdown: Arc, in_flight: Arc, pool_counters: Arc, + // 监听端点文本形态(Unix 臂为路径、TCP 臂为 host:port),Info 响应用。 socket_path: String, } /// server 进程主体:初始化日志文件输出 → 启动时 liveness 扫描 → 认领 sidecar → -/// 注册信号处理器 → 绑定 socket → 建内核 → 派生 worker → accept 循环 → -/// 在途归零 → 清空发送端 → join worker → drop 内核 → 先删 socket 文件后删 -/// sidecar。 +/// 注册信号处理器 → 绑定监听器 → 建内核 → 派生 worker → accept 循环 → +/// 在途归零 → 清空发送端 → join worker → drop 内核 → 先删 socket 文件 +/// (Unix 臂)后删 sidecar。 /// /// # 步骤 /// 1. 初始化日志文件输出(sidecar 同主名的追加文件、Info 级;init 幂等, @@ -163,8 +166,9 @@ struct ServerContext { /// 3. 认领 sidecar(启动顺序契约:先写 sidecar 再 bind socket);已有同版本 /// 存活 server 时拒绝启动,版本不匹配时转入交接路径(向旧 server 发 /// yield 请求,等旧 server 排空退出后重新认领)。 -/// 4. 注册信号处理器(绑定 socket 之前;失败回滚 sidecar,此时无存活线程)。 -/// 5. 绑定监听器;失败且路径存在时删 stale socket 文件重试一次。 +/// 4. 注册信号处理器(绑定监听器之前;失败回滚 sidecar,此时无存活线程)。 +/// 5. 绑定监听器:Unix 臂失败且路径存在时删 stale socket 文件重试一次, +/// TCP 臂直接绑地址(无文件、无 stale、无重试)。 /// 6. 建内核(标准配置)与共享聚合计数器。 /// 7. 派生 N 个常驻 worker(各在自己线程上建自有池)。 /// 8. 进入 accept 循环;关闭标志置位后按关闭序列退出。 @@ -178,7 +182,8 @@ struct ServerContext { /// - stale socket 文件删除重试后再失败时回滚删除 sidecar 并以错误返回。 /// /// # 副作用 -/// - 创建 sidecar 与 socket 文件;正常退出时删除两者。 +/// - 创建 sidecar 与 socket 文件(Unix 臂);正常退出时删除两者(TCP 臂 +/// 不是文件,不删)。 /// - 创建日志文件(追加写、sidecar 同主名);正常退出时不删除(跨重启的 /// 诊断工件)。 /// - 启动时 liveness 扫描可能向 sidecar 记录的进程发 SIGTERM、删除陈旧的 @@ -203,20 +208,27 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { levels: [Level::Info; SUBSYSTEM_COUNT], }); - // 启动时 liveness 扫描:认领裁决前清掉陈旧残留(僵尸态自动恢复)。 - // 杀进程失败映射既有的 ClaimRefused(RefusedAmbiguous),诊断指向人工清理 - // 命令。 - match liveness::liveness_scan(&config.sidecar_path, &config.socket_path) { - liveness::LivenessOutcome::Noop | liveness::LivenessOutcome::Cleaned => {} - liveness::LivenessOutcome::KillFailed => return Err(ServerError::ClaimRefused(ClaimResult::RefusedAmbiguous)), - } + // liveness 扫描、sidecar 认领与交接协议都在 Unix 文件面上;TCP 臂跳过 + // sidecar 注册,同端口裁决交给绑定失败。 + if let Endpoint::Unix(socket_path) = &config.endpoint { + // 启动时 liveness 扫描:认领裁决前清掉陈旧残留(僵尸态自动恢复)。 + // 杀进程失败映射既有的 ClaimRefused(RefusedAmbiguous),诊断指向人工 + // 清理命令。 + match liveness::liveness_scan(&config.sidecar_path, socket_path) { + liveness::LivenessOutcome::Noop | liveness::LivenessOutcome::Cleaned => {} + liveness::LivenessOutcome::KillFailed => { + return Err(ServerError::ClaimRefused(ClaimResult::RefusedAmbiguous)) + } + } - // 认领 sidecar:同版本拒绝启动,异版本走交接路径,存活证据不足即返回错误。 - match sidecar::claim_sidecar(&config.sidecar_path, &config.socket_path, &config.version) { - ClaimResult::Registered => {} - ClaimResult::AlreadyRunning => return Err(ServerError::ClaimRefused(ClaimResult::AlreadyRunning)), - ClaimResult::VersionMismatch => yield_and_takeover(config)?, - ClaimResult::RefusedAmbiguous => return Err(ServerError::ClaimRefused(ClaimResult::RefusedAmbiguous)), + // 认领 sidecar(启动顺序契约:先写 sidecar 再绑定监听器):同版本 + // 拒绝启动,异版本走交接路径,存活证据不足即返回错误。 + match sidecar::claim_sidecar(&config.sidecar_path, socket_path, &config.version) { + ClaimResult::Registered => {} + ClaimResult::AlreadyRunning => return Err(ServerError::ClaimRefused(ClaimResult::AlreadyRunning)), + ClaimResult::VersionMismatch => yield_and_takeover(config)?, + ClaimResult::RefusedAmbiguous => return Err(ServerError::ClaimRefused(ClaimResult::RefusedAmbiguous)), + } } // 关闭标志提前创建并注册信号处理器:在绑定 socket 之前,socket 文件 @@ -251,14 +263,14 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { shutdown: Arc::clone(&shutdown), in_flight: Arc::new(AtomicUsize::new(0)), pool_counters: counters, - socket_path: config.socket_path.to_string_lossy().into_owned(), + socket_path: config.endpoint.describe(), }); // accept 循环:非阻塞 accept 加 10 毫秒轮询加原子关闭标志。 let accept_result = accept_loop(&listener, &ctx, &router); // 关闭序列:在途归零、清空发送端、join worker、drop 内核、删文件。 - finish_shutdown(&ctx, &router, worker_handles, kernel, &config.socket_path, Some(&config.sidecar_path)) + finish_shutdown(&ctx, &router, worker_handles, kernel, &config.endpoint, Some(&config.sidecar_path)) .and(accept_result) } @@ -275,7 +287,7 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { /// 7. 轮询关闭标志(10 毫秒间隔):触发源有四个——连接线程 EOF、连接线程 /// 读超时、信号、控制请求(关闭与 yield 臂经共享分派置位)。 /// 8. 走共享关闭序列(在途归零、清空发送端、join worker、drop 内核、 -/// 删 socket 文件;不删 sidecar,因为从未写过)。 +/// 删 socket 文件(Unix 臂);不删 sidecar,因为从未写过)。 /// /// # 边界与前提 /// - 空闲超时内无连接时返回 `Ok(())`:客户端未来连接是预期场景,退出是 @@ -283,7 +295,7 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { /// - 首个连接被接受后第二个连接得连接被拒绝(监听器已丢弃)。 /// /// # 副作用 -/// - 创建 socket 文件,正常退出时删除;不写 sidecar。 +/// - 创建 socket 文件(Unix 臂),正常退出时删除;不写 sidecar。 /// - 注册进程级信号处理器(SIGINT、SIGTERM,termination 特性下含 SIGHUP)。 /// - 派生 N 个常驻 worker 线程与每连接一个 I/O 线程。 /// @@ -300,7 +312,7 @@ pub fn run_server_rm(config: &RmServerConfig) -> Result<(), ServerError> { // 绑定监听器:进程唯一路径不存在 stale 文件竞争,与持久路径共用同一 // 绑定函数(删除重试逻辑保留只为共用)。 - let listener = bind_listener_path(&config.socket_path)?; + let listener = bind_listener_endpoint(&config.endpoint)?; // 建内核(标准配置)与共享聚合计数器(全体 worker 池的增减累加到同一聚合)。 let kernel = KernelCore::new(KernelConfig::standard()); @@ -322,13 +334,13 @@ pub fn run_server_rm(config: &RmServerConfig) -> Result<(), ServerError> { shutdown: Arc::clone(&shutdown), in_flight: Arc::new(AtomicUsize::new(0)), pool_counters: counters, - socket_path: config.socket_path.to_string_lossy().into_owned(), + socket_path: config.endpoint.describe(), }); // 空闲窗口一:等首个连接;空闲超时内无连接即干净退出。 let stream = match rm_accept_first(&listener, config.idle_timeout)? { Some(stream) => stream, - None => return finish_shutdown(&ctx, &router, worker_handles, kernel, &config.socket_path, None), + None => return finish_shutdown(&ctx, &router, worker_handles, kernel, &config.endpoint, None), }; // 单连接模型:首个连接被接受后丢弃监听器,后续 connect 得连接被拒绝, @@ -353,7 +365,7 @@ pub fn run_server_rm(config: &RmServerConfig) -> Result<(), ServerError> { std::thread::sleep(Duration::from_millis(10)); } - finish_shutdown(&ctx, &router, worker_handles, kernel, &config.socket_path, None) + finish_shutdown(&ctx, &router, worker_handles, kernel, &config.endpoint, None) } /// 等首个连接(--rm 空闲窗口一):非阻塞 accept 加 10 毫秒轮询加截止时间。 @@ -367,7 +379,7 @@ pub fn run_server_rm(config: &RmServerConfig) -> Result<(), ServerError> { /// /// # 边界与前提 /// - 截止时间是 `Instant` 单调时钟,无时钟回拨问题。 -fn rm_accept_first(listener: &UnixListener, timeout: Duration) -> Result, ServerError> { +fn rm_accept_first(listener: &Listener, timeout: Duration) -> Result, ServerError> { listener .set_nonblocking(true) .map_err(|e| ServerError::AcceptFailed(e.to_string()))?; @@ -375,7 +387,7 @@ fn rm_accept_first(listener: &UnixListener, timeout: Duration) -> Result return Ok(Some(stream)), + Ok(stream) => return Ok(Some(stream)), Err(e) if e.kind() == io::ErrorKind::WouldBlock => { if Instant::now() >= deadline { return Ok(None); @@ -388,7 +400,7 @@ fn rm_accept_first(listener: &UnixListener, timeout: Duration) -> Result Result, router: &Arc, worker_handles: Vec>, kernel: Arc, - socket_path: &Path, sidecar_path: Option<&Path>, + endpoint: &Endpoint, sidecar_path: Option<&Path>, ) -> Result<(), ServerError> { // 在途归零:轮询至零,5 秒安全超时后记警告继续(余留连接线程持内核 // 与路由器的共享句柄,不悬垂;其后续投递经清空的路由立即失败)。 @@ -430,8 +443,10 @@ fn finish_shutdown( // 先删 socket 文件、后删 sidecar(尽力而为,文件已不存在不视为错误): // 新 server 以 sidecar 消失为接管信号,届时 socket 文件必已删除, - // 接管绑定无竞态。 - let _ = fs::remove_file(socket_path); + // 接管绑定无竞态。TCP 端点不是文件,不删。 + if let Endpoint::Unix(path) = endpoint { + let _ = fs::remove_file(path); + } if let Some(sidecar) = sidecar_path { let _ = fs::remove_file(sidecar); } @@ -461,10 +476,16 @@ fn finish_shutdown( /// # 注意事项 /// - 纯函数式入口:版本管理路径(版本不匹配强制交接)与显式交接路径调用 /// 同一入口。 +/// - 交接协议在 Unix 文件面上;TCP 端点是防御性分支,不参与交接。 pub fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> { + let socket_path = match &config.endpoint { + Endpoint::Unix(path) => path, + Endpoint::Tcp(_) => return Err(ServerError::YieldFailed("TCP 端点不支持交接".into())), + }; + // 连接旧 server:连接成功是交接的前提。 - let mut stream = UnixStream::connect(&config.socket_path) - .map_err(|e| ServerError::YieldFailed(format!("连接旧 server 失败:{e}")))?; + let mut stream = + UnixStream::connect(socket_path).map_err(|e| ServerError::YieldFailed(format!("连接旧 server 失败:{e}")))?; // 5 秒读超时:旧 server 挂起(不回应 yield)不无限阻塞。 let _ = stream.set_read_timeout(Some(Duration::from_secs(5))); @@ -496,7 +517,7 @@ pub fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> { } // 重新认领:sidecar 已消失,O_EXCL 创建应成功;其余态判交接失败。 - match sidecar::claim_sidecar(&config.sidecar_path, &config.socket_path, &config.version) { + match sidecar::claim_sidecar(&config.sidecar_path, socket_path, &config.version) { ClaimResult::Registered => Ok(()), other => Err(ServerError::YieldFailed(format!("交接后重新认领未通过:{other:?}"))), } @@ -552,22 +573,30 @@ fn panic_payload_str(payload: &Box) -> String { "non-string payload".into() } -/// 绑定监听器:失败且路径存在时删 stale socket 文件重试一次。 +/// 绑定监听器:Unix 臂失败且路径存在时删 stale socket 文件重试一次; +/// TCP 臂直接绑地址(无文件、无 stale、无重试)。 /// /// # 边界与前提 /// - stale socket 文件是崩溃残留(文件存在但无监听者)。 /// /// # 副作用 -/// - 可能删除 socket 文件。 -fn bind_listener_path(path: &Path) -> Result { - match UnixListener::bind(path) { - Ok(listener) => Ok(listener), - Err(_first) if path.exists() => { - // 路径存在:崩溃残留的 stale socket 文件,删一次重试。 - let _ = fs::remove_file(path); - UnixListener::bind(path).map_err(|second| ServerError::BindFailed(format!("socket 绑定失败:{second}"))) - } - Err(first) => Err(ServerError::BindFailed(format!("socket 绑定失败:{first}"))), +/// - Unix 臂可能删除 socket 文件。 +fn bind_listener_endpoint(endpoint: &Endpoint) -> Result { + match endpoint { + Endpoint::Unix(path) => match UnixListener::bind(path) { + Ok(listener) => Ok(Listener::Unix(listener)), + Err(_first) if path.exists() => { + // 路径存在:崩溃残留的 stale socket 文件,删一次重试。 + let _ = fs::remove_file(path); + UnixListener::bind(path) + .map(Listener::Unix) + .map_err(|second| ServerError::BindFailed(format!("socket 绑定失败:{second}"))) + } + Err(first) => Err(ServerError::BindFailed(format!("socket 绑定失败:{first}"))), + }, + Endpoint::Tcp(addr) => TcpListener::bind(addr) + .map(Listener::Tcp) + .map_err(|e| ServerError::BindFailed(format!("socket 绑定失败:{e}"))), } } @@ -577,9 +606,9 @@ fn bind_listener_path(path: &Path) -> Result { /// - stale socket 文件是崩溃残留(文件存在但无监听者)。 /// /// # 副作用 -/// - 可能删除 socket 文件;重试仍失败时删除 sidecar(回滚)。 -fn bind_listener(config: &ServerConfig) -> Result { - match bind_listener_path(&config.socket_path) { +/// - Unix 臂可能删除 socket 文件;重试仍失败时删除 sidecar(回滚)。 +fn bind_listener(config: &ServerConfig) -> Result { + match bind_listener_endpoint(&config.endpoint) { Ok(listener) => Ok(listener), Err(err) => { let _ = fs::remove_file(&config.sidecar_path); @@ -597,9 +626,7 @@ fn bind_listener(config: &ServerConfig) -> Result { /// /// # 副作用 /// - 每连接派生一个标准库线程(只做 I/O),线程结束时减一在途计数。 -fn accept_loop( - listener: &UnixListener, ctx: &Arc, router: &Arc, -) -> Result<(), ServerError> { +fn accept_loop(listener: &Listener, ctx: &Arc, router: &Arc) -> Result<(), ServerError> { listener .set_nonblocking(true) .map_err(|e| ServerError::AcceptFailed(e.to_string()))?; @@ -611,7 +638,7 @@ fn accept_loop( } match listener.accept() { - Ok((stream, _addr)) => { + Ok(stream) => { ctx.in_flight.fetch_add(1, Ordering::SeqCst); let ctx = Arc::clone(ctx); let router = Arc::clone(router); @@ -645,7 +672,7 @@ fn accept_loop( /// /// # 副作用 /// - 关闭请求会置位关闭标志。 -fn handle_connection(stream: UnixStream, ctx: &ServerContext, router: &WorkerRouter, read_timeout: Duration) { +fn handle_connection(stream: Conn, ctx: &ServerContext, router: &WorkerRouter, read_timeout: Duration) { // 克隆写端:原流交给帧读取器,写端独立用于响应写回。 let mut writer = match stream.try_clone() { Ok(writer) => writer, @@ -785,15 +812,21 @@ fn wait_in_flight_drained(in_flight: &AtomicUsize, timeout: Duration) { #[cfg(test)] mod tests { use super::*; + use std::net::TcpStream; use std::thread::JoinHandle; + /// 测试全部用 Unix 端点;直接取 socket 路径。 + fn unix_path(endpoint: &Endpoint) -> &Path { + endpoint.socket_path().expect("测试应全部为 Unix 端点") + } + /// 唯一临时路径(进程号加测试名):单进程内唯一,进程间以进程号隔离。 /// 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"), + endpoint: Endpoint::Unix(dir.join("server.sock")), sidecar_path: dir.join("sidecar.json"), worker_count: 2, version: env!("CARGO_PKG_VERSION").to_string(), @@ -805,20 +838,23 @@ mod tests { let server_config = config.clone(); let handle = std::thread::spawn(move || run_server(&server_config)); let deadline = Instant::now() + Duration::from_secs(10); - while !config.socket_path.exists() { + while !unix_path(&config.endpoint).exists() { assert!(Instant::now() < deadline, "socket 文件未出现"); std::thread::sleep(Duration::from_millis(10)); } handle } - /// 连接 server 并返回流。 - fn connect(config: &ServerConfig) -> UnixStream { - UnixStream::connect(&config.socket_path).expect("连接 server 应成功") + /// 连接 server 并返回连接。 + fn connect(config: &ServerConfig) -> Conn { + match &config.endpoint { + Endpoint::Unix(path) => Conn::Unix(UnixStream::connect(path).expect("连接 server 应成功")), + Endpoint::Tcp(addr) => Conn::Tcp(TcpStream::connect(addr).expect("连接 server 应成功")), + } } /// 发送请求帧并读回一帧响应。 - fn send_and_recv(stream: &mut UnixStream, request: &ServerRequest) -> ServerResponse { + fn send_and_recv(stream: &mut Conn, request: &ServerRequest) -> ServerResponse { stream .write_all(protocol::encode_request(request).as_bytes()) .expect("写请求应成功"); @@ -828,7 +864,7 @@ mod tests { } /// 停 server:发关闭请求、等线程退出、断言正常退出。 - fn stop_server(stream: &mut UnixStream, handle: JoinHandle>) { + fn stop_server(stream: &mut Conn, handle: JoinHandle>) { let response = send_and_recv(stream, &ServerRequest::Shutdown); assert!(matches!(response, ServerResponse::Shutdown), "应得关闭确认帧:{response:?}"); let result = handle.join().expect("server 线程应正常退出"); @@ -841,7 +877,7 @@ mod tests { let dir = std::env::temp_dir().join(format!("oxide_server_rm_test_{}_{}", std::process::id(), test_name)); fs::create_dir_all(&dir).expect("测试目录创建应成功"); RmServerConfig { - socket_path: dir.join("rm.sock"), + endpoint: Endpoint::Unix(dir.join("rm.sock")), worker_count: 1, idle_timeout: Duration::from_millis(300), } @@ -852,7 +888,7 @@ mod tests { let server_config = config.clone(); let handle = std::thread::spawn(move || run_server_rm(&server_config)); let deadline = Instant::now() + Duration::from_secs(10); - while !config.socket_path.exists() { + while !unix_path(&config.endpoint).exists() { assert!(Instant::now() < deadline, "socket 文件未出现"); std::thread::sleep(Duration::from_millis(10)); } @@ -949,7 +985,7 @@ mod tests { match response { ServerResponse::Info { version, socket_path, pid } => { assert_eq!(version, env!("CARGO_PKG_VERSION"), "版本应与构建期版本一致"); - assert_eq!(socket_path, config.socket_path.to_string_lossy().into_owned(), "socket 路径应与配置一致"); + assert_eq!(socket_path, config.endpoint.describe(), "socket 路径应与配置一致"); assert_eq!(pid, std::process::id(), "进程号应为当前进程号"); } other => panic!("应得 Info 帧,实得 {other:?}"), @@ -965,7 +1001,7 @@ mod tests { let mut stream = connect(&config); stop_server(&mut stream, handle); assert!(!config.sidecar_path.exists(), "sidecar 文件应被删除"); - assert!(!config.socket_path.exists(), "socket 文件应被删除"); + assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); } /// 畸形帧得 Error 帧,连接保持可用(后续请求仍正常应答)。 @@ -1045,7 +1081,7 @@ mod tests { let response = send_and_recv(&mut stream, &ServerRequest::Version); assert!(matches!(response, ServerResponse::Version { .. }), "第一 server 应继续应答:{response:?}"); assert!(config.sidecar_path.exists(), "第一 server 的 sidecar 应原样"); - assert!(config.socket_path.exists(), "第一 server 的 socket 应原样"); + assert!(unix_path(&config.endpoint).exists(), "第一 server 的 socket 应原样"); stop_server(&mut stream, handle); } @@ -1195,7 +1231,7 @@ mod tests { let result = handle.join().expect("server 线程应正常退出"); assert!(result.is_ok(), "server 应正常退出:{result:?}"); assert!(!config.sidecar_path.exists(), "sidecar 文件应被删除"); - assert!(!config.socket_path.exists(), "socket 文件应被删除"); + assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); } /// 交接端到端:server A(版本 0.0.1)运行中,server B(版本 0.0.2)以同 @@ -1230,7 +1266,7 @@ mod tests { // B 已接管:等 socket 文件出现(B 的绑定),再核对 sidecar 身份。 let deadline = Instant::now() + Duration::from_secs(30); - while !config.socket_path.exists() { + while !unix_path(&config.endpoint).exists() { assert!(Instant::now() < deadline, "B 的 socket 文件应出现"); std::thread::sleep(Duration::from_millis(10)); } @@ -1256,9 +1292,9 @@ mod tests { // 模拟旧 server:绑定 socket 并写本进程身份加异版本(0.0.0)的 // sidecar(PID 存活且启动时刻刻度匹配,版本不匹配,认领判 // VersionMismatch)。 - let listener = UnixListener::bind(&config.socket_path).expect("绑定模拟旧 server 应成功"); + let listener = UnixListener::bind(unix_path(&config.endpoint)).expect("绑定模拟旧 server 应成功"); sidecar::write_exclusive( - &sidecar::ServerIdentity::new(config.socket_path.to_str().unwrap(), "0.0.0"), + &sidecar::ServerIdentity::new(unix_path(&config.endpoint).to_str().unwrap(), "0.0.0"), &config.sidecar_path, ) .expect("写模拟 sidecar 应成功"); @@ -1280,7 +1316,7 @@ mod tests { let err = run_server(&config).unwrap_err(); assert!(matches!(err, ServerError::YieldFailed(_)), "应判 YieldFailed:{err:?}"); assert!(config.sidecar_path.exists(), "模拟 sidecar 文件不应被删除"); - assert!(config.socket_path.exists(), "模拟 socket 文件不应被删除"); + assert!(unix_path(&config.endpoint).exists(), "模拟 socket 文件不应被删除"); } /// 旧 server 挂起:模拟旧 server 接受连接后不回应;run_server 在约 5 秒 @@ -1289,9 +1325,9 @@ mod tests { fn yield_failed_when_old_server_hangs() { let config = unique_paths("yield_hang"); - let listener = UnixListener::bind(&config.socket_path).expect("绑定模拟旧 server 应成功"); + let listener = UnixListener::bind(unix_path(&config.endpoint)).expect("绑定模拟旧 server 应成功"); sidecar::write_exclusive( - &sidecar::ServerIdentity::new(config.socket_path.to_str().unwrap(), "0.0.0"), + &sidecar::ServerIdentity::new(unix_path(&config.endpoint).to_str().unwrap(), "0.0.0"), &config.sidecar_path, ) .expect("写模拟 sidecar 应成功"); @@ -1310,7 +1346,7 @@ mod tests { assert!(matches!(err, ServerError::YieldFailed(_)), "应判 YieldFailed:{err:?}"); assert!(elapsed >= Duration::from_secs(5), "应命中 5 秒读超时:{elapsed:?}"); assert!(config.sidecar_path.exists(), "模拟 sidecar 文件不应被删除"); - assert!(config.socket_path.exists(), "模拟 socket 文件不应被删除"); + assert!(unix_path(&config.endpoint).exists(), "模拟 socket 文件不应被删除"); } /// --rm 加断开 EOF:eval 得完成值后客户端断开,server 在有界时间内 @@ -1327,7 +1363,7 @@ mod tests { let socket_before = well_known_socket.exists(); let handle = start_rm_server(config.clone()); - let mut stream = UnixStream::connect(&config.socket_path).expect("连接 rm server 应成功"); + let mut stream = Conn::Unix(UnixStream::connect(unix_path(&config.endpoint)).expect("连接 rm server 应成功")); let response = send_and_recv( &mut stream, &ServerRequest::Eval { @@ -1349,7 +1385,7 @@ mod tests { } let result = handle.join().expect("rm server 线程应正常退出"); assert!(result.is_ok(), "rm server 应正常退出:{result:?}"); - assert!(!config.socket_path.exists(), "socket 文件应被删除"); + assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); assert_eq!(well_known_sidecar.exists(), sidecar_before, "全局 well-known sidecar 应前后不变"); assert_eq!(well_known_socket.exists(), socket_before, "全局 well-known socket 路径应前后不变"); } @@ -1371,7 +1407,7 @@ mod tests { "应在空闲超时后退出(留就绪探测裕量):{:?}", started.elapsed() ); - assert!(!config.socket_path.exists(), "socket 文件应被删除"); + assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); } /// --rm 空闲窗口二:连接建立后不发任何数据,读超时后 server 正常退出, @@ -1384,7 +1420,7 @@ mod tests { fn rm_idle_kill_silent_client() { let config = rm_config("idle_silent"); let handle = start_rm_server(config.clone()); - let stream = UnixStream::connect(&config.socket_path).expect("连接 rm server 应成功"); + let stream = UnixStream::connect(unix_path(&config.endpoint)).expect("连接 rm server 应成功"); // 保持连接打开但不发数据:连接线程在读超时后退出。 let started = Instant::now(); @@ -1395,7 +1431,7 @@ mod tests { "应在读超时后退出(留就绪探测裕量):{:?}", started.elapsed() ); - assert!(!config.socket_path.exists(), "socket 文件应被删除"); + assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); drop(stream); } @@ -1405,7 +1441,7 @@ mod tests { fn rm_second_connection_refused() { let config = rm_config("second_refused"); let handle = start_rm_server(config.clone()); - let mut stream = UnixStream::connect(&config.socket_path).expect("首个连接应成功"); + let mut stream = Conn::Unix(UnixStream::connect(unix_path(&config.endpoint)).expect("首个连接应成功")); let response = send_and_recv(&mut stream, &ServerRequest::Health); assert!( matches!(response, ServerResponse::Health { healthy: true }), @@ -1415,11 +1451,11 @@ mod tests { // 监听器已丢弃:第二个连接得连接被拒绝。轮询至拒绝,容忍连接线程 // 建立与监听器丢弃之间的极小时序窗口(并发套件下调度延迟放大窗口)。 let deadline = Instant::now() + Duration::from_secs(5); - let mut second = UnixStream::connect(&config.socket_path); + let mut second = UnixStream::connect(unix_path(&config.endpoint)); while second.is_ok() { assert!(Instant::now() < deadline, "第二个连接应被拒绝"); std::thread::sleep(Duration::from_millis(10)); - second = UnixStream::connect(&config.socket_path); + second = UnixStream::connect(unix_path(&config.endpoint)); } assert!(second.is_err(), "第二个连接应被拒绝:{second:?}"); @@ -1433,8 +1469,8 @@ mod tests { fn rm_shutdown_request_exits() { let config = rm_config("shutdown_req"); let handle = start_rm_server(config.clone()); - let mut stream = UnixStream::connect(&config.socket_path).expect("连接 rm server 应成功"); + let mut stream = Conn::Unix(UnixStream::connect(unix_path(&config.endpoint)).expect("连接 rm server 应成功")); stop_server(&mut stream, handle); - assert!(!config.socket_path.exists(), "socket 文件应被删除"); + assert!(!unix_path(&config.endpoint).exists(), "socket 文件应被删除"); } } diff --git a/crates/oxide_cli/tests/server_client.rs b/crates/oxide_cli/tests/server_client.rs index 2f201d5b..098eeaf1 100644 --- a/crates/oxide_cli/tests/server_client.rs +++ b/crates/oxide_cli/tests/server_client.rs @@ -10,13 +10,14 @@ use std::time::{Duration, Instant}; use oxide_cli::server::client::send_control_request_from; use oxide_cli::server::protocol::{ServerRequest, ServerResponse}; use oxide_cli::server::server::{run_server, ServerConfig}; +use oxide_net::Endpoint; /// 临时路径配置:进程号加测试名唯一,worker 数 2,版本取构建期版本。 fn test_config(test_name: &str) -> ServerConfig { let dir = std::env::temp_dir().join(format!("oxide_server_client_test_{}_{}", std::process::id(), test_name)); fs::create_dir_all(&dir).expect("测试目录创建应成功"); ServerConfig { - socket_path: dir.join("server.sock"), + endpoint: Endpoint::Unix(dir.join("server.sock")), sidecar_path: dir.join("sidecar.json"), worker_count: 2, version: env!("CARGO_PKG_VERSION").to_string(), @@ -28,7 +29,7 @@ fn start_server(config: ServerConfig) -> std::thread::JoinHandle { - assert_eq!(socket_path, config.socket_path.to_string_lossy().into_owned(), "socket 路径应与配置一致"); + assert_eq!(socket_path, config.endpoint.describe(), "socket 路径应与配置一致"); assert_eq!(pid, std::process::id(), "进程号应为当前进程号"); } other => panic!("应得 Info 帧,实得 {other:?}"), @@ -92,6 +93,9 @@ fn control_client_end_to_end() { assert!(matches!(response, ServerResponse::Shutdown), "应得关闭确认帧:{response:?}"); let result = handle.join().expect("server 线程应正常退出"); assert!(result.is_ok(), "server 应正常退出:{result:?}"); - assert!(!config.socket_path.exists(), "socket 文件应被删除"); + assert!( + !config.endpoint.socket_path().expect("Unix 端点应含 socket 路径").exists(), + "socket 文件应被删除" + ); assert!(!config.sidecar_path.exists(), "sidecar 文件应被删除"); } diff --git a/crates/oxide_cli/tests/server_signal_int.rs b/crates/oxide_cli/tests/server_signal_int.rs index 62946ba9..c50f3b53 100644 --- a/crates/oxide_cli/tests/server_signal_int.rs +++ b/crates/oxide_cli/tests/server_signal_int.rs @@ -9,6 +9,7 @@ use std::thread::JoinHandle; use std::time::{Duration, Instant}; use oxide_cli::server::server::{run_server, ServerConfig, ServerError}; +use oxide_net::Endpoint; /// 唯一临时路径(进程号加测试名):单进程内唯一,进程间以进程号隔离。 /// worker 数取小值 2:与既有单测同口径,大值徒增预热成本。 @@ -16,7 +17,7 @@ fn unique_paths(test_name: &str) -> ServerConfig { let dir = std::env::temp_dir().join(format!("oxide_server_signal_int_{}_{}", std::process::id(), test_name)); std::fs::create_dir_all(&dir).expect("测试目录创建应成功"); ServerConfig { - socket_path: dir.join("server.sock"), + endpoint: Endpoint::Unix(dir.join("server.sock")), sidecar_path: dir.join("sidecar.json"), worker_count: 2, version: env!("CARGO_PKG_VERSION").to_string(), @@ -29,7 +30,7 @@ fn start_server(config: ServerConfig) -> JoinHandle> { let server_config = config.clone(); let handle = std::thread::spawn(move || run_server(&server_config)); let deadline = Instant::now() + Duration::from_secs(10); - while !config.socket_path.exists() { + while !config.endpoint.socket_path().expect("Unix 端点应含 socket 路径").exists() { assert!(Instant::now() < deadline, "socket 文件未出现"); std::thread::sleep(Duration::from_millis(10)); } @@ -55,7 +56,10 @@ fn wait_shutdown(config: &ServerConfig, handle: JoinHandle ServerConfig { let dir = std::env::temp_dir().join(format!("oxide_server_signal_term_{}_{}", std::process::id(), test_name)); std::fs::create_dir_all(&dir).expect("测试目录创建应成功"); ServerConfig { - socket_path: dir.join("server.sock"), + endpoint: Endpoint::Unix(dir.join("server.sock")), sidecar_path: dir.join("sidecar.json"), worker_count: 2, version: env!("CARGO_PKG_VERSION").to_string(), @@ -29,7 +30,7 @@ fn start_server(config: ServerConfig) -> JoinHandle> { let server_config = config.clone(); let handle = std::thread::spawn(move || run_server(&server_config)); let deadline = Instant::now() + Duration::from_secs(10); - while !config.socket_path.exists() { + while !config.endpoint.socket_path().expect("Unix 端点应含 socket 路径").exists() { assert!(Instant::now() < deadline, "socket 文件未出现"); std::thread::sleep(Duration::from_millis(10)); } @@ -55,7 +56,10 @@ fn wait_shutdown(config: &ServerConfig, handle: JoinHandle Option<&Path> { + match self { + Endpoint::Unix(path) => Some(path), + Endpoint::Tcp(_) => None, + } + } + + /// 返回端点文本形态:Unix 臂为路径字符串,TCP 臂为 host:port 字符串。 + pub fn describe(&self) -> String { + match self { + Endpoint::Unix(path) => path.to_string_lossy().into_owned(), + Endpoint::Tcp(addr) => addr.to_string(), + } + } +} + impl Listener { /// 设置监听者非阻塞模式。 /// @@ -104,6 +135,34 @@ impl Conn { } } +impl Read for Conn { + /// 从连接读取字节,委托内臂。 + fn read(&mut self, buf: &mut [u8]) -> io::Result { + match self { + Conn::Unix(s) => s.read(buf), + Conn::Tcp(s) => s.read(buf), + } + } +} + +impl Write for Conn { + /// 向连接写入字节,委托内臂。 + fn write(&mut self, buf: &[u8]) -> io::Result { + match self { + Conn::Unix(s) => s.write(buf), + Conn::Tcp(s) => s.write(buf), + } + } + + /// 刷新连接,委托内臂。 + fn flush(&mut self) -> io::Result<()> { + match self { + Conn::Unix(s) => s.flush(), + Conn::Tcp(s) => s.flush(), + } + } +} + #[cfg(test)] mod tests { use super::*;