mirror of
https://github.com/pjh456/OxideJS.git
synced 2026-10-23 18:21:13 +08:00
feat: server 执行请求路径,六段纯函数加每请求步数覆盖
This commit is contained in:
@@ -0,0 +1,181 @@
|
||||
//! 执行请求路径:parse → compile → spawn → run → format → drop。
|
||||
//!
|
||||
//! `handle_eval` 是纯函数,参数全注入、无全局状态;每请求用单一
|
||||
//! `catch_unwind` 包裹整条路径,panic 载荷转成错误帧,不 panic、不退出。
|
||||
//!
|
||||
//! # release 边界
|
||||
//! 根 `Cargo.toml` 的 `[profile.release]` 为 `panic=abort`:release 下
|
||||
//! `catch_unwind` 是空操作、panic 直接终止进程,watchdog 是 release 崩溃后
|
||||
//! 恢复的唯一自动手段;debug 下本路径完整生效,是日常验证环境。
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use oxide_compiler::compiler::{compiled_module_hash, Compiler};
|
||||
use oxide_kernel::kernel::KernelCore;
|
||||
use oxide_parser::Allocator;
|
||||
use oxide_vm::vm_pool::VmPool;
|
||||
|
||||
use super::protocol::ServerResponse;
|
||||
|
||||
/// 执行请求处理:解析 → 编译 → 取 VM → 执行 → 渲染完成值 → 归还池的六段路径。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. parse:局部分配器加 `oxide_parser::parse`,失败时各条错误换行拼接成错误帧。
|
||||
/// 2. compile:`Compiler` 加 `CodeForge` 结构哈希缓存,失败以原文成错误帧。
|
||||
/// 3. spawn:从池取 `VmGuard`(闭包内创建,panic 时随展开被 drop)。
|
||||
/// 4. run:设每请求步数覆盖(含 None)后 `run`,`Err` 即错误帧。
|
||||
/// 5. format:成功值经 `crate::format_js_value` 渲染成完成值文本,`eval_ok` 封装。
|
||||
/// 6. drop:guard 在闭包出口 drop,干净 VM 经 `full_reset` 回池复用、被污染 VM 丢弃并新建替补。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - guard 在 `catch_unwind` 闭包内创建:panic 时 guard 随展开在闭包边界被 drop,
|
||||
/// `VmGuard::drop` 的 `thread::panicking()` 检查此刻为真,VM 按 dirty 丢弃。
|
||||
///
|
||||
/// # 副作用
|
||||
/// - 无(纯函数,参数全注入)。
|
||||
///
|
||||
/// # 注意事项
|
||||
/// - release 下 `panic=abort` 使 `catch_unwind` 空操作,watchdog 是崩溃后恢复的唯一自动手段。
|
||||
pub fn handle_eval(code: &str, max_steps: Option<u64>, kernel: &Arc<KernelCore>, pool: &Arc<VmPool>) -> ServerResponse {
|
||||
let response = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
// parse:局部分配器,解析失败各条换行拼接成错误帧。
|
||||
let allocator = Allocator::default();
|
||||
let program = match oxide_parser::parse(&allocator, code) {
|
||||
Ok(program) => program,
|
||||
Err(errors) => {
|
||||
let message = errors.iter().map(|e| e.to_string()).collect::<Vec<_>>().join("\n");
|
||||
return ServerResponse::eval_err(message);
|
||||
}
|
||||
};
|
||||
|
||||
// compile:结构哈希缓存,失败以原文成错误帧。
|
||||
let compiler = Compiler::new();
|
||||
let hash = compiled_module_hash(&program);
|
||||
let module = match kernel.code_forge().get_or_insert_with(hash, || compiler.compile(&program)) {
|
||||
Ok(module) => module,
|
||||
Err(message) => return ServerResponse::eval_err(message),
|
||||
};
|
||||
|
||||
// spawn:从池取 VM;guard 在闭包内创建,panic 时随展开被 drop。
|
||||
let mut guard = pool.spawn();
|
||||
|
||||
// run:设每请求步数覆盖(含 None)后执行。
|
||||
guard.vm_mut().set_max_steps(max_steps);
|
||||
let result = guard.vm_mut().run(&module);
|
||||
|
||||
// format:渲染完成值,`Err` 即错误帧。
|
||||
match result {
|
||||
Ok(value) => {
|
||||
let text = crate::format_js_value(
|
||||
guard.vm(),
|
||||
kernel.perm_interner().as_ref(),
|
||||
kernel.shape_forge().as_ref(),
|
||||
value,
|
||||
);
|
||||
ServerResponse::eval_ok(text)
|
||||
}
|
||||
Err(message) => ServerResponse::eval_err(message),
|
||||
}
|
||||
// drop:guard 在闭包出口 drop,干净 VM 经 full_reset 回池复用、dirty 丢弃。
|
||||
}));
|
||||
|
||||
match response {
|
||||
Ok(resp) => resp,
|
||||
Err(payload) => ServerResponse::eval_err(format!("internal panic: {}", panic_payload_str(&payload))),
|
||||
}
|
||||
}
|
||||
|
||||
/// 从 `catch_unwind` 的 panic payload 提取可读文本:`&str` 与 `String` 两种
|
||||
/// 常见形态,其余返回占位文本。
|
||||
fn panic_payload_str(payload: &Box<dyn std::any::Any + Send>) -> String {
|
||||
if let Some(s) = payload.downcast_ref::<&str>() {
|
||||
return (*s).to_string();
|
||||
}
|
||||
if let Some(s) = payload.downcast_ref::<String>() {
|
||||
return s.clone();
|
||||
}
|
||||
"non-string payload".into()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use oxide_kernel::kernel::KernelConfig;
|
||||
|
||||
/// 测试环境:minimal 内核加单 VM 池(与池既有测试同口径)。
|
||||
fn test_env() -> (Arc<KernelCore>, Arc<VmPool>) {
|
||||
let kernel = KernelCore::new(KernelConfig::minimal());
|
||||
let pool = VmPool::new(Arc::clone(&kernel), 1, Some(2));
|
||||
(kernel, pool)
|
||||
}
|
||||
|
||||
/// 成功渲染:"1 + 1" 得完成值 "2",无错误。
|
||||
#[test]
|
||||
fn eval_success_renders_value() {
|
||||
let (kernel, pool) = test_env();
|
||||
let response = handle_eval("1 + 1", None, &kernel, &pool);
|
||||
match response {
|
||||
ServerResponse::EvalResult { ref value, ref error } => {
|
||||
assert_eq!(value.as_deref(), Some("2"), "完成值应为 2:{response:?}");
|
||||
assert!(error.is_none(), "不应有错误:{response:?}");
|
||||
}
|
||||
other => panic!("应得 EvalResult 帧,实得 {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// 解析错误帧:"function(" 得非空错误,无完成值。
|
||||
#[test]
|
||||
fn eval_parse_error_frame() {
|
||||
let (kernel, pool) = test_env();
|
||||
let response = handle_eval("function(", None, &kernel, &pool);
|
||||
match response {
|
||||
ServerResponse::EvalResult { ref value, ref error } => {
|
||||
assert!(value.is_none(), "不应有完成值:{response:?}");
|
||||
assert!(error.as_deref().map(|e| !e.is_empty()).unwrap_or(false), "错误应非空:{response:?}");
|
||||
}
|
||||
other => panic!("应得 EvalResult 帧,实得 {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// 运行时错误帧:throw 得含 uncaught 与 boom 的错误。
|
||||
#[test]
|
||||
fn eval_runtime_error_frame() {
|
||||
let (kernel, pool) = test_env();
|
||||
let response = handle_eval("throw new Error('boom')", None, &kernel, &pool);
|
||||
match response {
|
||||
ServerResponse::EvalResult { ref value, ref error } => {
|
||||
assert!(value.is_none(), "不应有完成值:{response:?}");
|
||||
let err = error.as_deref().expect("应有错误");
|
||||
assert!(err.contains("uncaught"), "错误应含 uncaught:{err}");
|
||||
assert!(err.contains("boom"), "错误应含 boom:{err}");
|
||||
}
|
||||
other => panic!("应得 EvalResult 帧,实得 {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// 步数超限帧:for(;;){} 加 Some(1000) 得含 step limit 的错误。
|
||||
#[test]
|
||||
fn eval_step_limit_frame() {
|
||||
let (kernel, pool) = test_env();
|
||||
let response = handle_eval("for(;;){}", Some(1000), &kernel, &pool);
|
||||
match response {
|
||||
ServerResponse::EvalResult { ref value, ref error } => {
|
||||
assert!(value.is_none(), "不应有完成值:{response:?}");
|
||||
let err = error.as_deref().expect("应有错误");
|
||||
assert!(err.contains("step limit"), "错误应含 step limit:{err}");
|
||||
}
|
||||
other => panic!("应得 EvalResult 帧,实得 {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// 池复用:连续两次成功 eval 后池可用数不变。
|
||||
#[test]
|
||||
fn eval_pool_reuse() {
|
||||
let (kernel, pool) = test_env();
|
||||
let before = pool.available_count();
|
||||
handle_eval("1 + 1", None, &kernel, &pool);
|
||||
handle_eval("2 + 2", None, &kernel, &pool);
|
||||
let after = pool.available_count();
|
||||
assert_eq!(before, after, "池可用数应不变:before={before} after={after}");
|
||||
}
|
||||
}
|
||||
@@ -1,11 +1,14 @@
|
||||
//! 持久 server 子系统:协议数据面、身份注册与 server 进程主体。
|
||||
//! 持久 server 子系统:协议数据面、身份注册、执行路径与 server 进程主体。
|
||||
//!
|
||||
//! 持久 server 是常驻进程,复用预热池、近零 spawn 成本;
|
||||
//! 通信走 Unix socket + NDJSON(换行分隔 JSON)。
|
||||
//! 模块分三层:协议数据面(帧格式与 serde 结构体)、身份注册(sidecar 文件
|
||||
//! 与 liveness 检查原语)、server 进程主体(accept 循环、控制请求直接处理、
|
||||
//! 优雅退出)。执行路径与并发模型在后续任务。
|
||||
//! 模块分四层:协议数据面(帧格式与 serde 结构体)、身份注册(sidecar 文件
|
||||
//! 与 liveness 检查原语)、执行请求路径(parse → compile → spawn → run →
|
||||
//! format → drop 六段纯函数)、server 进程主体(accept 循环、控制请求直接
|
||||
//! 处理、优雅退出)。并发模型在后续任务。
|
||||
|
||||
/// 执行请求路径:parse → compile → spawn → run → format → drop 六段纯函数。
|
||||
pub mod eval;
|
||||
/// 协议数据面:NDJSON 帧格式与 serde 结构体。
|
||||
pub mod protocol;
|
||||
/// server 进程主体:accept 循环、控制请求分派与优雅退出。
|
||||
|
||||
@@ -22,8 +22,9 @@ use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use oxide_kernel::kernel::{KernelConfig, KernelCore};
|
||||
use oxide_vm::vm_pool::{PoolCounters, VmPool};
|
||||
use oxide_vm::vm_pool::VmPool;
|
||||
|
||||
use super::eval;
|
||||
use super::protocol::{self, FrameReader, ProtocolError, ServerRequest, ServerResponse};
|
||||
use super::sidecar::{self, ClaimResult};
|
||||
|
||||
@@ -72,12 +73,12 @@ impl ServerConfig {
|
||||
}
|
||||
}
|
||||
|
||||
/// server 运行上下文:池状态、启动时刻、关闭标志、在途计数。
|
||||
/// server 运行上下文:共享内核、启动时刻、关闭标志、在途计数。
|
||||
///
|
||||
/// 池本体不进上下文(Vm 不跨线程),只持池的原子状态句柄;
|
||||
/// 内核不跨线程共享(执行路径由后续任务接线)。
|
||||
/// 内核跨线程共享(`Arc<KernelCore>` 是 Send/Sync);池本体不进上下文
|
||||
/// (Vm 不跨线程),每连接线程自建池(见 `handle_connection`)。
|
||||
struct ServerContext {
|
||||
pool_counters: Arc<PoolCounters>,
|
||||
kernel: Arc<KernelCore>,
|
||||
started_at: Instant,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
in_flight: Arc<AtomicUsize>,
|
||||
@@ -114,12 +115,11 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> {
|
||||
// 绑定监听器:stale socket 文件删一次重试,失败回滚 sidecar。
|
||||
let listener = bind_listener(config)?;
|
||||
|
||||
// 建内核与池:标准配置,池按 min_pool_size 预热。
|
||||
// 建内核(标准配置);池每连接线程自建(Vm 不跨线程,池本体不共享)。
|
||||
let kernel = KernelCore::new(KernelConfig::standard());
|
||||
let pool = VmPool::new(Arc::clone(&kernel), kernel.config.min_pool_size, kernel.config.max_pool_size);
|
||||
|
||||
let ctx = Arc::new(ServerContext {
|
||||
pool_counters: pool.counters(),
|
||||
kernel: Arc::clone(&kernel),
|
||||
started_at: Instant::now(),
|
||||
shutdown: Arc::new(AtomicBool::new(false)),
|
||||
in_flight: Arc::new(AtomicUsize::new(0)),
|
||||
@@ -132,9 +132,8 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> {
|
||||
// 在途归零:轮询至零,5 秒安全超时后记警告继续。
|
||||
wait_in_flight_drained(&ctx.in_flight, Duration::from_secs(5));
|
||||
|
||||
// 按序释放:上下文(持池状态句柄)→ 池 → 内核。
|
||||
// 按序释放:上下文(持内核句柄)→ 内核。
|
||||
drop(ctx);
|
||||
drop(pool);
|
||||
drop(kernel);
|
||||
|
||||
// 删 sidecar 与 socket 文件(尽力而为,文件已不存在不视为错误)。
|
||||
@@ -232,12 +231,15 @@ 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 帧答复后继续。
|
||||
let response = match protocol::parse_request(&frame) {
|
||||
Ok(request) => dispatch_control(&request, ctx),
|
||||
Ok(request) => dispatch_control(&request, ctx, &pool),
|
||||
Err(ProtocolError::FrameTooLarge) => break,
|
||||
Err(error) => error.to_error_frame(),
|
||||
};
|
||||
@@ -255,17 +257,17 @@ fn handle_connection(stream: UnixStream, ctx: &ServerContext) {
|
||||
}
|
||||
}
|
||||
|
||||
/// 控制请求分派:五类控制请求直接应答,不 spawn 虚拟机。
|
||||
/// 控制请求分派:五类控制请求直接应答;执行请求走 `handle_eval`。
|
||||
///
|
||||
/// 执行请求暂以协议错误帧答复(执行路径由后续任务补齐)。
|
||||
fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerResponse {
|
||||
/// 池是每连接资产(`handle_connection` 创建),Status 响应读本连接池状态。
|
||||
fn dispatch_control(request: &ServerRequest, ctx: &ServerContext, pool: &Arc<VmPool>) -> ServerResponse {
|
||||
match request {
|
||||
ServerRequest::Version => ServerResponse::Version {
|
||||
version: env!("CARGO_PKG_VERSION").to_string(),
|
||||
},
|
||||
ServerRequest::Status => ServerResponse::Status {
|
||||
pool_available: ctx.pool_counters.available(),
|
||||
pool_total: ctx.pool_counters.total(),
|
||||
pool_available: pool.available_count(),
|
||||
pool_total: pool.total_count(),
|
||||
uptime_ms: ctx.started_at.elapsed().as_millis() as u64,
|
||||
},
|
||||
ServerRequest::Health => ServerResponse::Health { healthy: true },
|
||||
@@ -279,9 +281,7 @@ fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerRespo
|
||||
ctx.shutdown.store(true, Ordering::SeqCst);
|
||||
ServerResponse::Shutdown
|
||||
}
|
||||
ServerRequest::Eval { .. } => ServerResponse::Error {
|
||||
message: "执行路径尚未实现".to_string(),
|
||||
},
|
||||
ServerRequest::Eval { code, max_steps } => eval::handle_eval(code, *max_steps, &ctx.kernel, pool),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -449,9 +449,9 @@ mod tests {
|
||||
stop_server(&mut stream, handle);
|
||||
}
|
||||
|
||||
/// 执行请求得 Error 帧(尚未实现),不 panic。
|
||||
/// 执行请求得 EvalResult 帧:"1 + 1" 得完成值 "2",不 panic。
|
||||
#[test]
|
||||
fn eval_not_implemented() {
|
||||
fn eval_request_returns_value() {
|
||||
let config = unique_paths("eval");
|
||||
let handle = start_server(config.clone());
|
||||
let mut stream = connect(&config);
|
||||
@@ -462,7 +462,10 @@ mod tests {
|
||||
max_steps: None,
|
||||
},
|
||||
);
|
||||
assert!(matches!(response, ServerResponse::Error { .. }), "应得 Error 帧:{response:?}");
|
||||
assert!(
|
||||
matches!(response, ServerResponse::EvalResult { ref value, .. } if value.as_deref() == Some("2")),
|
||||
"应得完成值为 2 的 EvalResult 帧:{response:?}"
|
||||
);
|
||||
stop_server(&mut stream, handle);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user