ITStudio_OUC/OxideJS-New
0
1
Fork 0
mirror of https://github.com/pjh456/OxideJS.git synced 2026-10-23 18:21:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: server 信号处理，SIGINT/SIGTERM 置位关闭标志走优雅退出

Browse Source
This commit is contained in:
彭俊皓
2026-10-06 15:53:21 +00:00
parent a08652252b
commit 20e71417ee
9 changed files with 423 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/oxide_cli/src/lib.rs
+104
View File
@@ -0,0 +1,104 @@
//! oxide_cli 库目标：暴露持久 server 子系统与完成值渲染，供集成测试跨进程
//! 引用、供二进制与 server 执行路径共用。
//!
//! 二进制目标（`src/main.rs`）承载 CLI 分派（eval / run / compile / bench /
//! REPL）；库目标暴露 server 模块与完成值渲染函数，集成测试（`tests/`）经
//! `oxide_cli::server` 调用 `run_server` 起真实 server 进程主体，eval 臂与
//! server 执行路径共用 `format_js_value` 渲染完成值。
use oxide_kernel::shape_forge::{ShapeForge, EMPTY_SHAPE_ID};
use oxide_kernel::string_forge::PermInterner;
use oxide_types::object::JsObject;
use oxide_types::private_key::{is_private_name_key, is_symbol_key};
use oxide_vm::vm::Vm;
use oxide_vm::JsValue;
pub mod server;
/// 完成值渲染：把 JsValue 递归渲染为可读文本（字符串加引号、Promise 按结算态
/// 展开、函数 `[Function]`、对象走 shape 链、数组按元素区）。
///
/// 二进制 eval 臂与 server 执行路径共用。
pub fn format_js_value(vm: &Vm, string_forge: &PermInterner, shape_forge: &ShapeForge, val: JsValue) -> String {
if val.is_string() {
// SAFETY: val 已确认是字符串值。
let s = unsafe { (*val.as_string_ptr()).to_owned_string() };
format!("\"{s}\"")
} else if val.is_bigint() {
format!("{}", vm.bigint_value(val))
} else if val.is_object() {
let obj = unsafe { &*val.as_js_object_ptr() };
if obj.is_promise_obj() {
// 已 settle 的 Promise 打印其结算值（便于 eval 观察微任务结果）。
return match oxide_vm::promise::promise_settled_value(obj) {
Some((true, v)) => format_js_value(vm, string_forge, shape_forge, v),
Some((false, v)) => {
format!("Promise {{ <rejected> {} }}", format_js_value(vm, string_forge, shape_forge, v))
}
None => "Promise { <pending> }".to_string(),
};
}
if obj.is_function() {
"[Function]".to_string()
} else if obj.is_array() {
format_array(vm, string_forge, shape_forge, obj)
} else {
format_object(vm, string_forge, shape_forge, obj)
}
} else if val.is_undefined() {
"undefined".to_string()
} else {
format!("{val}")
}
}
/// 对象渲染：沿 shape 父链收集非空属性名（跳过 Symbol/私有名键），按序渲染。
fn format_object(vm: &Vm, string_forge: &PermInterner, shape_forge: &ShapeForge, obj: &JsObject) -> String {
let mut entries = Vec::new();
let shape_id = obj.shape_id();
let mut shape_ids = Vec::new();
let mut cursor = Some(shape_id);
while let Some(id) = cursor {
if id == EMPTY_SHAPE_ID {
break;
}
if let Some(shape) = shape_forge.get_shape(id) {
cursor = shape.parent;
if shape.property_name != u32::MAX {
shape_ids.push(id);
}
} else {
break;
}
}
let mut pos: u32 = 0;
for id in shape_ids.iter().rev() {
if let Some(shape) = shape_forge.get_shape(*id) {
// 跳过 Symbol/私有名键，仅展示字符串属性名。
if !is_symbol_key(shape.property_name) && !is_private_name_key(shape.property_name) {
let prop_val = obj.get_prop_at(pos);
if prop_val.is_undefined() {
pos += 1;
continue;
}
let name = string_forge.lookup(shape.property_name).unwrap_or_default();
let val_str = format_js_value(vm, string_forge, shape_forge, prop_val);
entries.push(format!("\"{name}\": {val_str}"));
}
}
pos += 1;
}
format!("{{{}}}", entries.join(", "))
}
/// 数组渲染：按元素区长度遍历，逐元素渲染。
fn format_array(vm: &Vm, string_forge: &PermInterner, shape_forge: &ShapeForge, obj: &JsObject) -> String {
// 数组元素在独立元素区，命名属性区长度恒 0；元素区未分配时 get_prop_at 自然返回 undefined。
let len = obj.array_prop_count as usize;
let mut items = Vec::new();
for i in 0..len {
let val = obj.get_prop_at(i);
items.push(format_js_value(vm, string_forge, shape_forge, val));
}
format!("[{}]", items.join(", "))
}
crates/oxide_cli/src/main.rs
+2 -92
View File
@@ -10,21 +10,19 @@ use ansi_term::Colour::Red;
use clap::{Parser, Subcommand};
use oxide_compiler::compiler::{compiled_module_hash, Compiler};
use oxide_compiler::compiler_error;
use oxide_cli::format_js_value;
use oxide_kernel::kernel::{KernelConfig, KernelCore};
use oxide_kernel::shape_forge::{ShapeForge, EMPTY_SHAPE_ID};
use oxide_kernel::shape_forge::ShapeForge;
use oxide_kernel::string_forge::PermInterner;
use oxide_kernel::{kernel_error, kernel_info};
use oxide_log::{Level, SUBSYSTEM_COUNT};
use oxide_parser::Allocator;
use oxide_types::object::JsObject;
use oxide_types::private_key::{is_private_name_key, is_symbol_key};
use oxide_vm::vm::Vm;
use oxide_vm::vm_error;
use oxide_vm::vm_pool::VmPool;
use oxide_vm::JsValue;
mod bench;
mod server;
#[derive(Parser)]
#[command(
@@ -234,94 +232,6 @@ fn format_result(vm: &oxide_vm::vm::Vm, string_forge: &PermInterner, shape_forge
println!("{}", format_js_value(vm, string_forge, shape_forge, val));
}
fn format_js_value(
vm: &oxide_vm::vm::Vm, string_forge: &PermInterner, shape_forge: &ShapeForge, val: JsValue,
) -> String {
if val.is_string() {
// SAFETY: val 已确认是字符串值。
let s = unsafe { (*val.as_string_ptr()).to_owned_string() };
format!("\"{s}\"")
} else if val.is_bigint() {
format!("{}", vm.bigint_value(val))
} else if val.is_object() {
let obj = unsafe { &*val.as_js_object_ptr() };
if obj.is_promise_obj() {
// 已 settle 的 Promise 打印其结算值（便于 eval 观察微任务结果）。
return match oxide_vm::promise::promise_settled_value(obj) {
Some((true, v)) => format_js_value(vm, string_forge, shape_forge, v),
Some((false, v)) => {
format!("Promise {{ <rejected> {} }}", format_js_value(vm, string_forge, shape_forge, v))
}
None => "Promise { <pending> }".to_string(),
};
}
if obj.is_function() {
"[Function]".to_string()
} else if obj.is_array() {
format_array(vm, string_forge, shape_forge, obj)
} else {
format_object(vm, string_forge, shape_forge, obj)
}
} else if val.is_undefined() {
"undefined".to_string()
} else {
format!("{val}")
}
}
fn format_object(
vm: &oxide_vm::vm::Vm, string_forge: &PermInterner, shape_forge: &ShapeForge, obj: &JsObject,
) -> String {
let mut entries = Vec::new();
let shape_id = obj.shape_id();
let mut shape_ids = Vec::new();
let mut cursor = Some(shape_id);
while let Some(id) = cursor {
if id == EMPTY_SHAPE_ID {
break;
}
if let Some(shape) = shape_forge.get_shape(id) {
cursor = shape.parent;
if shape.property_name != u32::MAX {
shape_ids.push(id);
}
} else {
break;
}
}
let mut pos: u32 = 0;
for id in shape_ids.iter().rev() {
if let Some(shape) = shape_forge.get_shape(*id) {
// 跳过 Symbol/私有名键，仅展示字符串属性名。
if !is_symbol_key(shape.property_name) && !is_private_name_key(shape.property_name) {
let prop_val = obj.get_prop_at(pos);
if prop_val.is_undefined() {
pos += 1;
continue;
}
let name = string_forge.lookup(shape.property_name).unwrap_or_default();
let val_str = format_js_value(vm, string_forge, shape_forge, prop_val);
entries.push(format!("\"{name}\": {val_str}"));
}
}
pos += 1;
}
format!("{{{}}}", entries.join(", "))
}
fn format_array(
vm: &oxide_vm::vm::Vm, string_forge: &PermInterner, shape_forge: &ShapeForge, obj: &JsObject,
) -> String {
// 数组元素在独立元素区，命名属性区长度恒 0；元素区未分配时 get_prop_at 自然返回 undefined。
let len = obj.array_prop_count as usize;
let mut items = Vec::new();
for i in 0..len {
let val = obj.get_prop_at(i);
items.push(format_js_value(vm, string_forge, shape_forge, val));
}
format!("[{}]", items.join(", "))
}
fn run(file: &str, kernel: &Arc<KernelCore>, pool: &Arc<VmPool>, trace: bool, profile: bool) -> ExitCode {
match fs::read_to_string(file) {
Ok(source) => eval(&source, kernel, pool, trace, profile, false),
crates/oxide_cli/src/server/mod.rs
+2 -1
View File
@@ -12,7 +12,8 @@
pub mod eval;
/// 协议数据面：NDJSON 帧格式与 serde 结构体。
pub mod protocol;
/// server 进程主体：accept 循环、控制请求分派、执行请求路由与优雅退出。
/// server 进程主体：accept 循环、控制请求分派、执行请求路由、信号处理
/// （SIGINT/SIGTERM 置位关闭标志）与优雅退出。
#[allow(clippy::module_inception)]
pub mod server;
/// 身份注册：sidecar 文件、liveness 检查与陈旧清理。
crates/oxide_cli/src/server/server.rs
+84 -12
View File
@@ -8,8 +8,8 @@
//! - 并发模型：主线程只跑 accept 循环；每连接一个标准库线程只做 I/O
//! （帧读写）；五类控制请求由连接线程直接应答、不占 worker；执行请求
//! 经 mpsc 路由到固定 N 个常驻 worker 执行（各持自己线程上的 VM 池）。
//! - 优雅退出顺序：关闭请求 → 在途归零 → 清空发送端 → join worker →
//! drop 内核 → 删 sidecar 与 socket 文件。
//! - 优雅退出顺序：关闭请求或 SIGINT/SIGTERM 信号置位关闭标志 → 在途归零
//! → 清空发送端 → join worker → drop 内核 → 删 sidecar 与 socket 文件。
//! - 池本体不是 Send/Sync（Vm 不跨线程），上下文只持池的原子状态句柄；
//! 状态响应读全体 worker 池的共享聚合计数器。
@@ -29,7 +29,8 @@ use super::protocol::{self, FrameReader, ProtocolError, ServerRequest, ServerRes
use super::sidecar::{self, ClaimResult};
use super::workers::{self, WorkerRouter};
/// server 启动与运行错误：绑定失败、认领被拒、接受连接失败、读取身份失败。
/// server 启动与运行错误：绑定失败、认领被拒、接受连接失败、读取身份失败、
/// 信号处理器注册失败。
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ServerError {
/// sidecar 认领未通过（已有存活 server 或存活证据不足以裁决）。
@@ -42,6 +43,8 @@ pub enum ServerError {
IdentityReadFailed(String),
/// worker 线程异常退出（panic 载荷文本）。
WorkerPanic(String),
/// 信号处理器注册失败（系统级错误，非"已注册"的幂等情形）。
SignalInstallFailed(String),
}
impl std::fmt::Display for ServerError {
@@ -52,6 +55,7 @@ impl std::fmt::Display for ServerError {
ServerError::AcceptFailed(msg) => write!(f, "接受连接失败：{msg}"),
ServerError::IdentityReadFailed(msg) => write!(f, "读取身份失败：{msg}"),
ServerError::WorkerPanic(msg) => write!(f, "worker 线程异常退出：{msg}"),
ServerError::SignalInstallFailed(msg) => write!(f, "信号处理器注册失败：{msg}"),
}
}
}
@@ -93,6 +97,9 @@ fn default_worker_count() -> usize {
/// 内核跨线程共享（`Arc<KernelCore>` 是 Send/Sync）；池本体不进上下文
/// （Vm 不跨线程，池归各 worker 线程所有），状态响应读共享聚合计数器。
struct ServerContext {
// 只持不读：余留连接线程（在途排水超时后仍存活）持本上下文，内核句柄
// 随上下文存活，不悬垂。
#[allow(dead_code)]
kernel: Arc<KernelCore>,
started_at: Instant,
shutdown: Arc<AtomicBool>,
@@ -101,28 +108,31 @@ struct ServerContext {
socket_path: String,
}
/// server 进程主体：认领 sidecar → 绑定 socket → 建内核 → 派生 worker →
/// accept 循环 → 在途归零 → 清空发送端 → join worker → drop 内核 →
/// 删 sidecar 与 socket 文件。
/// server 进程主体：认领 sidecar → 注册信号处理器 → 绑定 socket → 建内核 →
/// 派生 worker → accept 循环 → 在途归零 → 清空发送端 → join worker →
/// drop 内核 → 删 sidecar 与 socket 文件。
///
/// # 步骤
/// 1. 认领 sidecar（启动顺序契约：先写 sidecar 再 bind socket）。
/// 2. 绑定监听器；失败且路径存在时删 stale socket 文件重试一次。
/// 3. 建内核（标准配置）与共享聚合计数器。
/// 4. 派生 N 个常驻 worker（各在自己线程上建自有池）。
/// 5. 进入 accept 循环；关闭标志置位后按关闭序列退出。
/// 2. 注册信号处理器（绑定 socket 之前；失败回滚 sidecar，此时无存活线程）。
/// 3. 绑定监听器；失败且路径存在时删 stale socket 文件重试一次。
/// 4. 建内核（标准配置）与共享聚合计数器。
/// 5. 派生 N 个常驻 worker（各在自己线程上建自有池）。
/// 6. 进入 accept 循环；关闭标志置位后按关闭序列退出。
///
/// # 边界与前提
/// - 认领被拒（AlreadyRunning / RefusedAmbiguous）立即返回错误，不双注册。
/// - 信号处理器注册失败（系统级错误）回滚 sidecar 并以错误返回。
/// - stale socket 文件删除重试后再失败时回滚删除 sidecar 并以错误返回。
///
/// # 副作用
/// - 创建 sidecar 与 socket 文件；正常退出时删除两者。
/// - 注册进程级信号处理器（SIGINT、SIGTERM，termination 特性下含 SIGHUP）。
/// - 派生 N 个常驻 worker 线程与每连接一个 I/O 线程。
///
/// # 注意事项
/// - 在调用线程上运行（后续 CLI 骨架任务在主线程分派到它）。
/// - 关闭标志由关闭请求置位；信号处理由后续任务在同一标志上接线。
/// - 关闭标志由关闭请求或 SIGINT/SIGTERM 信号置位。
/// - 关闭序列固定：清空发送端 → join 全部 worker → drop 内核（内核 drop
/// 的调试断言要求存活 VM 数为零，worker 池随线程退出先释放，前提满足）。
pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> {
@@ -132,6 +142,15 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> {
other => return Err(ServerError::ClaimRefused(other)),
}
// 关闭标志提前创建并注册信号处理器：在绑定 socket 之前，socket 文件
// 出现即蕴含处理器已就位（集成测试的信号投递无竞态）；失败回滚
// sidecar，此时进程内没有存活线程。
let shutdown = Arc::new(AtomicBool::new(false));
if let Err(err) = install_signal_handler(&shutdown) {
let _ = fs::remove_file(&config.sidecar_path);
return Err(err);
}
// 绑定监听器：stale socket 文件删一次重试，失败回滚 sidecar。
let listener = bind_listener(config)?;
@@ -152,7 +171,7 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> {
let ctx = Arc::new(ServerContext {
kernel: Arc::clone(&kernel),
started_at: Instant::now(),
shutdown: Arc::new(AtomicBool::new(false)),
shutdown: Arc::clone(&shutdown),
in_flight: Arc::new(AtomicUsize::new(0)),
pool_counters: counters,
socket_path: config.socket_path.to_string_lossy().into_owned(),
@@ -184,6 +203,44 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> {
accept_result
}
/// 信号处理器体：置位关闭标志。
///
/// # 边界与前提
/// - 只做一次原子写，不分配、不取锁、不做 I/O；即使按最严格的信号上下文
/// 标准衡量也是安全的。
fn signal_shutdown(shutdown: &Arc<AtomicBool>) {
shutdown.store(true, Ordering::SeqCst);
}
/// 注册 SIGINT/SIGTERM 信号处理器：信号到达时置位关闭标志。
///
/// # 步骤
/// 1. 闭包捕获关闭标志的共享句柄，转调 `signal_shutdown`。
/// 2. `ctrlc::set_handler` 注册；"已注册"的幂等情形视为成功，其余系统
/// 错误映射为 `SignalInstallFailed`。
///
/// # 边界与前提
/// - 处理器是进程级全局的（同一进程只允许一次注册）；生产路径一个进程
/// 只有一个 server，注册一次即成功。
///
/// # 副作用
/// - 注册进程级信号处理器（SIGINT、SIGTERM，termination 特性下含 SIGHUP）。
///
/// # 注意事项
/// - 须在绑定 socket 之前调用：socket 文件出现即蕴含处理器已就位，
/// 集成测试的信号投递无竞态。
/// - 测试进程内多 server 并存时，仅首个注册者接上信号，其余返回"已注册"
/// 幂等成功；单测不发真实信号，无干扰。
fn install_signal_handler(shutdown: &Arc<AtomicBool>) -> Result<(), ServerError> {
let flag = Arc::clone(shutdown);
match ctrlc::set_handler(move || signal_shutdown(&flag)) {
Ok(()) => Ok(()),
// 进程级处理器已注册（测试进程内多 server 并存的幂等情形）。
Err(ctrlc::Error::MultipleHandlers) => Ok(()),
Err(e) => Err(ServerError::SignalInstallFailed(e.to_string())),
}
}
/// 从 `catch_unwind` 的 panic payload 提取可读文本：`&str` 与 `String` 两种
/// 常见形态，其余返回占位文本。
fn panic_payload_str(payload: &Box<dyn std::any::Any + Send>) -> String {
@@ -430,6 +487,21 @@ mod tests {
assert!(result.is_ok(), "server 应正常退出：{result:?}");
}
/// 信号处理器体：置位关闭标志。
#[test]
fn signal_shutdown_sets_flag() {
let shutdown = Arc::new(AtomicBool::new(false));
signal_shutdown(&shutdown);
assert!(shutdown.load(Ordering::SeqCst), "标志应被置位");
}
/// 注册信号处理器：返回成功（"已注册"的幂等情形同样视为成功）。
#[test]
fn install_signal_handler_registers() {
let shutdown = Arc::new(AtomicBool::new(false));
install_signal_handler(&shutdown).expect("信号处理器注册应成功");
}
/// 版本请求得 Version 帧，取值与构建期版本一致。
#[test]
fn version_response() {