feat: server forge 子命令,四张共享 forge 状态查询加三行动旗标

This commit is contained in:
2026-10-06 18:46:44 +00:00
parent 83e41e10ab
commit e199ebeaa1
12 changed files with 635 additions and 54 deletions
Generated
+18 -18
View File
@@ -1160,7 +1160,7 @@ dependencies = [
[[package]]
name = "oxide_builtins"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"chrono",
"fast-float",
@@ -1185,7 +1185,7 @@ dependencies = [
[[package]]
name = "oxide_bytecode"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"num-bigint",
"oxide_log",
@@ -1193,7 +1193,7 @@ dependencies = [
[[package]]
name = "oxide_cfg"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxide_bytecode",
"oxide_emit",
@@ -1204,7 +1204,7 @@ dependencies = [
[[package]]
name = "oxide_cli"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"ansi_term",
"bumpalo",
@@ -1225,7 +1225,7 @@ dependencies = [
[[package]]
name = "oxide_code_cache"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"lru",
"oxide_bytecode",
@@ -1234,7 +1234,7 @@ dependencies = [
[[package]]
name = "oxide_compiler"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxide_bytecode",
"oxide_cfg",
@@ -1253,7 +1253,7 @@ dependencies = [
[[package]]
name = "oxide_dce"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxide_bytecode",
"oxide_cfg",
@@ -1267,7 +1267,7 @@ dependencies = [
[[package]]
name = "oxide_emit"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"num-bigint",
"oxc_allocator",
@@ -1282,7 +1282,7 @@ dependencies = [
[[package]]
name = "oxide_ir"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxide_bytecode",
"oxide_log",
@@ -1291,7 +1291,7 @@ dependencies = [
[[package]]
name = "oxide_kernel"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"bumpalo",
"dashmap",
@@ -1306,7 +1306,7 @@ dependencies = [
[[package]]
name = "oxide_liveness"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxide_bytecode",
"oxide_cfg",
@@ -1318,7 +1318,7 @@ dependencies = [
[[package]]
name = "oxide_log"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"tracing",
"tracing-appender",
@@ -1327,7 +1327,7 @@ dependencies = [
[[package]]
name = "oxide_parser"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxc_allocator",
"oxc_ast",
@@ -1342,7 +1342,7 @@ dependencies = [
[[package]]
name = "oxide_regalloc"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxide_bytecode",
"oxide_cfg",
@@ -1358,7 +1358,7 @@ dependencies = [
[[package]]
name = "oxide_runtime_api"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"num-bigint",
"num-traits",
@@ -1370,7 +1370,7 @@ dependencies = [
[[package]]
name = "oxide_test262"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"oxide_bytecode",
"oxide_compiler",
@@ -1387,7 +1387,7 @@ dependencies = [
[[package]]
name = "oxide_types"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"bumpalo",
"hashbrown 0.14.5",
@@ -1399,7 +1399,7 @@ dependencies = [
[[package]]
name = "oxide_vm"
version = "0.8.40"
version = "0.8.41"
dependencies = [
"bumpalo",
"chrono",
+1 -1
View File
@@ -22,7 +22,7 @@ members = [
]
[workspace.package]
version = "0.8.40"
version = "0.8.41"
edition = "2021"
license = "MIT"
rust-version = "1.80"
+104 -6
View File
@@ -12,7 +12,7 @@ use oxide_cli::format_js_value;
use oxide_cli::server::cleanup::{well_known_cleanup, CleanupOutcome};
use oxide_cli::server::client::send_control_request;
use oxide_cli::server::log as server_log;
use oxide_cli::server::protocol::{ServerRequest, ServerResponse};
use oxide_cli::server::protocol::{ForgeTarget, LookupResult, ServerRequest, ServerResponse};
use oxide_cli::server::server::{run_server, run_server_rm, RmServerConfig, ServerConfig};
use oxide_cli::server::sidecar;
use oxide_compiler::compiler::{compiled_module_hash, Compiler};
@@ -154,8 +154,46 @@ enum ServerCommands {
#[arg(long)]
lines: Option<usize>,
},
/// 查询内部状态(代码、对象、字符串、属性)。
Forge,
/// 查询内部 forge 状态(code/object/string/property),加三个行动旗标。
Forge {
/// 目标 forge:code(字节码缓存)/ object(shape 表)/ string(字符串
/// intern)/ property(属性模板缓存)。
target: ForgeTargetCli,
/// 触发垃圾回收(对池内全体空闲 VM 执行完整 session GC)。
#[arg(long)]
gc: bool,
/// 清空字节码缓存(LRU)。
#[arg(long)]
clear_cache: bool,
/// 在字符串 intern 表纯查询一个键的 id(不插入)。
#[arg(long)]
lookup: Option<String>,
},
}
/// forge 查询目标(CLI 侧,映射到协议侧 `ForgeTarget`)。
#[derive(Clone, Copy, Debug, PartialEq, Eq, clap::ValueEnum)]
enum ForgeTargetCli {
/// 字节码 LRU 缓存。
Code,
/// 隐藏类 shape 表。
Object,
/// 字符串 intern 表。
String,
/// 属性模板缓存。
Property,
}
impl ForgeTargetCli {
/// 映射到协议侧 `ForgeTarget`。
fn to_protocol(self) -> ForgeTarget {
match self {
ForgeTargetCli::Code => ForgeTarget::Code,
ForgeTargetCli::Object => ForgeTarget::Object,
ForgeTargetCli::String => ForgeTarget::String,
ForgeTargetCli::Property => ForgeTarget::Property,
}
}
}
/// 日志级别阈值(读取侧过滤,映射到 `oxide_log::Level`)。
@@ -246,7 +284,7 @@ fn main() -> ExitCode {
/// server 子命令分派:start(两形态)与 cleanup 为最小实现,
/// version/status/health/info/stop/restart 六臂为真实实现,
/// log 为真实实现(读 well-known 日志文件,支持 --follow/--level/--lines),
/// 其余一臂(forge)为 not_implemented 占位。
/// forge 为真实实现(查询四张共享 forge 状态,加三行动旗标)。
///
/// # 步骤
/// 1. Start:rm 为真调独立模式入口(进程唯一路径);foreground 为真当前进程
@@ -261,7 +299,8 @@ fn main() -> ExitCode {
/// (详见 restart_command)。
/// 5. Log:读 well-known 日志文件,按级别阈值与最后 N 行过滤后打印,
/// `--follow` 时进入阻塞跟踪循环(详见 log_command)。
/// 6. 其余一臂(forge):not_implemented 占位(退出码 2)。
/// 6. Forge:发 forge 查询请求(目标加三行动旗标),成功打印渲染文本退 0,
/// 失败打印错误消息退 1。
///
/// # 边界与前提
/// - `--idle-timeout` 不带 --rm 时静默忽略(语义门控归后续任务)。
@@ -341,10 +380,69 @@ fn server_command(command: ServerCommands) -> ExitCode {
}),
ServerCommands::Restart => restart_command(),
ServerCommands::Log { follow, level, lines } => log_command(follow, level.map(LogLevel::as_level), lines),
ServerCommands::Forge => not_implemented("server forge"),
ServerCommands::Forge {
target,
gc,
clear_cache,
lookup,
} => {
let request = ServerRequest::ForgeQuery {
target: target.to_protocol(),
gc,
clear_cache,
lookup,
};
control_arm(&request, |response| match response {
ServerResponse::ForgeStatus {
target,
entries,
capacity,
gc_collected,
cache_cleared,
lookup,
} => Some(render_forge_status(
*target,
*entries,
*capacity,
*gc_collected,
*cache_cleared,
lookup.clone(),
)),
_ => None,
})
}
}
}
/// 渲染 `ForgeStatus` 为多行文本:首行条目数,容量 / gc / 清缓存 / lookup
/// 行按旗标结果附加。
fn render_forge_status(
target: ForgeTarget,
entries: usize,
capacity: usize,
gc_collected: Option<usize>,
cache_cleared: bool,
lookup: Option<LookupResult>,
) -> String {
let mut lines = vec![format!("{} forge: {entries} entries", target.as_str())];
if capacity != 0 {
lines.push(format!("capacity: {capacity}"));
}
if let Some(n) = gc_collected {
lines.push(format!("gc: collected {n} idle VMs"));
}
if cache_cleared {
lines.push("cache cleared".to_string());
}
if let Some(result) = lookup {
lines.push(match result {
LookupResult::Interned { id } => format!("lookup: interned id={id}"),
LookupResult::Absent => "lookup: not interned".to_string(),
});
}
lines.join("\n")
}
/// 控制臂公共入口:发请求,成功打印渲染文本退 0,失败打印消息退 1。
/// 渲染函数返回 None 表示响应变体与请求不匹配(防御性兜底)。
fn control_arm(request: &ServerRequest, render: fn(&ServerResponse) -> Option<String>) -> ExitCode {
+153
View File
@@ -0,0 +1,153 @@
//! forge 状态查询:读四张共享 forge 的条目数与容量,执行三个行动旗标
//! (gc / clear-cache / lookup)。
//!
//! `handle_forge_query` 是纯函数,参数全注入、无全局状态;worker 线程调用
//! (gc 需触及本线程自有池),forge 状态读是跨线程安全的只读操作。
use std::sync::Arc;
use oxide_kernel::kernel::KernelCore;
use oxide_vm::vm_pool::VmPool;
use super::protocol::{ForgeTarget, LookupResult, ServerResponse};
/// forge 查询处理:读目标 forge 状态,按旗标执行 gc / clear-cache / lookup,
/// 组装 ForgeStatus 响应帧。
///
/// # 步骤
/// 1. 读目标 forge 条目数与容量(code 目标容量取 max_cached_modules,其余 0)。
/// 2. --gc 时对池内全体空闲 VM 执行完整 session GC,记收集数。
/// 3. --clear-cache 时清空字节码 LRU 缓存。
/// 4. --lookup 时在字符串 intern 表纯查询键的 id(不插入)。
/// 5. 组装 ForgeStatus 响应帧返回。
///
/// # 边界与前提
/// - 在 worker 线程调用(gc 需触及本线程自有池);forge 状态读是跨线程安全
/// 的只读操作。
/// - gc 只触及空闲队列,在借 VM(在途 eval)不受影响;clear-cache 清 LRU
/// 不使在借模块失效(Arc 保活)。
///
/// # 副作用
/// - --gc 重置各空闲 VM 的 session_bytes_allocated 为清扫后存活字节。
/// - --clear-cache 清空字节码缓存(后续 eval 重编译)。
pub fn handle_forge_query(
target: ForgeTarget,
gc: bool,
clear_cache: bool,
lookup: Option<String>,
kernel: &Arc<KernelCore>,
pool: &Arc<VmPool>,
) -> ServerResponse {
// 读目标 forge 条目数与容量。
let (entries, capacity) = match target {
ForgeTarget::Code => (kernel.code_forge().len(), kernel.max_cached_modules()),
ForgeTarget::Object => (kernel.shape_forge().len(), 0),
ForgeTarget::String => (kernel.perm_interner().entry_count() as usize, 0),
ForgeTarget::Property => (kernel.prop_forge().len(), 0),
};
// --gc:全体空闲 VM 完整 session GC。
let gc_collected = gc.then(|| pool.collect_idle_gc());
// --clear-cache:清空字节码 LRU。
let cache_cleared = if clear_cache {
kernel.code_forge().clear();
true
} else {
false
};
// --lookup:字符串 intern 表纯查询(不插入)。
let lookup_result = lookup.map(|key| match kernel.perm_interner().lookup_id(&key) {
Some(id) => LookupResult::Interned { id },
None => LookupResult::Absent,
});
ServerResponse::ForgeStatus {
target,
entries,
capacity,
gc_collected,
cache_cleared,
lookup: lookup_result,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::server::eval;
use oxide_kernel::kernel::KernelConfig;
/// 测试环境:minimal 内核加单 VM 池(与 eval 测试同口径)。
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)
}
/// code 目标:容量等于 max_cached_modules,条目数 0(新内核),无旗标无行动结果。
#[test]
fn forge_query_code_entries() {
let (kernel, pool) = test_env();
let response = handle_forge_query(ForgeTarget::Code, false, false, None, &kernel, &pool);
match response {
ServerResponse::ForgeStatus {
target,
entries,
capacity,
gc_collected,
cache_cleared,
lookup,
} => {
assert_eq!(target, ForgeTarget::Code, "目标应为 code");
assert_eq!(entries, 0, "新内核的字节码缓存应为空");
assert_eq!(capacity, kernel.max_cached_modules(), "容量应等于 max_cached_modules");
assert!(gc_collected.is_none(), "未设 gc 旗标应无收集数");
assert!(!cache_cleared, "未设 clear-cache 旗标不应清缓存");
assert!(lookup.is_none(), "未设 lookup 旗标应无结果");
}
other => panic!("应得 ForgeStatus 帧,实得 {other:?}"),
}
}
/// gc 旗标:先 eval 制造 session 状态归还池,再 gc 得收集数 Some(1)。
#[test]
fn forge_query_gc_collects() {
let (kernel, pool) = test_env();
eval::handle_eval("1 + 1", None, &kernel, &pool);
let response = handle_forge_query(ForgeTarget::Code, true, false, None, &kernel, &pool);
match response {
ServerResponse::ForgeStatus { gc_collected, .. } => {
assert_eq!(gc_collected, Some(1), "单 VM 池应收集 1 个空闲 VM");
}
other => panic!("应得 ForgeStatus 帧,实得 {other:?}"),
}
}
/// lookup:已 intern 键得 Interned,未 intern 键得 Absent,且纯查询不插入。
#[test]
fn forge_query_lookup() {
let (kernel, pool) = test_env();
let (id, _) = kernel.perm_interner().intern("forge-key");
let before = kernel.perm_interner().entry_count();
let response = handle_forge_query(ForgeTarget::String, false, false, Some("forge-key".into()), &kernel, &pool);
match response {
ServerResponse::ForgeStatus { lookup, .. } => {
assert_eq!(lookup, Some(LookupResult::Interned { id }), "已 intern 键应得该 id");
}
other => panic!("应得 ForgeStatus 帧,实得 {other:?}"),
}
let response = handle_forge_query(ForgeTarget::String, false, false, Some("not-interned".into()), &kernel, &pool);
match response {
ServerResponse::ForgeStatus { lookup, .. } => {
assert_eq!(lookup, Some(LookupResult::Absent), "未 intern 键应得 Absent");
}
other => panic!("应得 ForgeStatus 帧,实得 {other:?}"),
}
assert_eq!(kernel.perm_interner().entry_count(), before, "纯查询不应插入新条目");
}
}
+6 -2
View File
@@ -2,13 +2,14 @@
//!
//! 持久 server 是常驻进程,复用预热池、近零 spawn 成本;
//! 通信走 Unix socket + NDJSON(换行分隔 JSON)。
//! 模块分七层:协议数据面(帧格式与 serde 结构体)、身份注册(sidecar 文件
//! 模块分八层:协议数据面(帧格式与 serde 结构体)、身份注册(sidecar 文件
//! 与 liveness 检查原语)、控制客户端(读 sidecar、连接 server、发控制请求
//! 帧、读回响应帧、三分支错误)、执行请求路径(parse → compile → spawn →
//! run → format → drop 六段纯函数)、worker 池(固定 N 常驻 worker、各持
//! 自有 VM 池、mpsc 轮询路由)、server 进程主体(accept 循环、控制请求直接
//! 处理、执行请求路由、优雅退出)、日志读取(行级别解析、阈值加最后 N 行
//! 过滤、阻塞式跟踪)。
//! 过滤、阻塞式跟踪)、forge 状态查询(读四张共享 forge 条目数与容量、执行
//! gc / clear-cache / lookup 三旗标)。
/// 人工兜底清理命令:扫 sidecar、探活、杀 PID、删孤儿文件。
pub mod cleanup;
@@ -16,6 +17,9 @@ pub mod cleanup;
pub mod client;
/// 执行请求路径:parse → compile → spawn → run → format → drop 六段纯函数。
pub mod eval;
/// forge 状态查询:读四张共享 forge 条目数与容量,执行 gc / clear-cache /
/// lookup 三旗标。
pub mod forge;
/// 日志读取:行级别解析、阈值加最后 N 行过滤、整文件读取、阻塞式跟踪。
pub mod log;
/// 协议数据面:NDJSON 帧格式与 serde 结构体。
+153 -3
View File
@@ -48,7 +48,46 @@ impl ProtocolError {
}
}
/// server 请求帧:eval 执行请求加六类控制请求。
/// forge 查询目标:四张共享 forge 之一。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ForgeTarget {
/// 代码对象(字节码 LRU 缓存)。
Code,
/// 对象表(隐藏类 shape 表,含根节点)。
Object,
/// 字符串驻留(intern 键表)。
String,
/// 属性表(属性模板缓存)。
Property,
}
impl ForgeTarget {
/// 目标名(小写),供渲染与展示。
pub fn as_str(self) -> &'static str {
match self {
ForgeTarget::Code => "code",
ForgeTarget::Object => "object",
ForgeTarget::String => "string",
ForgeTarget::Property => "property",
}
}
}
/// lookup 结果:已 intern(带稳定 id)或未 intern。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum LookupResult {
/// 已 intern,附稳定 id。
Interned {
/// intern 键的稳定 id。
id: u32,
},
/// 未 intern。
Absent,
}
/// server 请求帧:eval 执行请求、forge 状态查询加六类控制请求。
///
/// 序列化为 tagged 对象,`type` 字段区分类型(snake_case)。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -61,6 +100,20 @@ pub enum ServerRequest {
#[serde(default, skip_serializing_if = "Option::is_none")]
max_steps: Option<u64>,
},
/// forge 状态查询:目标 forge 加三个行动旗标(gc / clear_cache / lookup)。
ForgeQuery {
/// 目标 forge:code / object / string / property。
target: ForgeTarget,
/// 触发垃圾回收(对池内全体空闲 VM 执行完整 session GC)。
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
gc: bool,
/// 清空字节码 LRU 缓存。
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
clear_cache: bool,
/// 在字符串 intern 表纯查询一个键的 id(不插入)。
#[serde(default, skip_serializing_if = "Option::is_none")]
lookup: Option<String>,
},
/// 查询 server 版本。
Version,
/// 查询 server 状态(池与运行时长)。
@@ -90,6 +143,24 @@ pub enum ServerResponse {
#[serde(default, skip_serializing_if = "Option::is_none")]
error: Option<String>,
},
/// forge 状态响应:目标 forge 条目数与容量,加三个行动结果。
ForgeStatus {
/// 目标 forge:code / object / string / property。
target: ForgeTarget,
/// 目标 forge 的条目数(object 目标含根节点)。
entries: usize,
/// 容量上限(code 目标为 LRU 上限,其余为 0 表示无界)。
capacity: usize,
/// --gc 收集的空闲 VM 数;未设旗标为 None。
#[serde(default, skip_serializing_if = "Option::is_none")]
gc_collected: Option<usize>,
/// --clear-cache 已执行时为真。
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
cache_cleared: bool,
/// --lookup 结果;未设旗标为 None。
#[serde(default, skip_serializing_if = "Option::is_none")]
lookup: Option<LookupResult>,
},
/// 版本响应。
Version { version: String },
/// 状态响应:池可用/总数、峰值借出数与运行时长。
@@ -130,10 +201,12 @@ impl ServerResponse {
}
/// 请求帧可取的 `type` 值(须与 `ServerRequest` 变体保持同步)。
const REQUEST_TAGS: &[&str] = &["eval", "version", "status", "health", "info", "shutdown", "yield"];
const REQUEST_TAGS: &[&str] =
&["eval", "forge_query", "version", "status", "health", "info", "shutdown", "yield"];
/// 响应帧可取的 `type` 值(须与 `ServerResponse` 变体保持同步)。
const RESPONSE_TAGS: &[&str] = &["eval_result", "version", "status", "health", "info", "shutdown", "yield", "error"];
const RESPONSE_TAGS: &[&str] =
&["eval_result", "forge_status", "version", "status", "health", "info", "shutdown", "yield", "error"];
/// 序列化请求为帧字符串(含结尾换行)。
///
@@ -357,6 +430,83 @@ mod tests {
}
}
/// forge 查询请求 round-trip:四目标各一轮,带与不带三旗标两形态。
#[test]
fn forge_query_roundtrip() {
for target in [ForgeTarget::Code, ForgeTarget::Object, ForgeTarget::String, ForgeTarget::Property] {
roundtrip_request(&ServerRequest::ForgeQuery {
target,
gc: false,
clear_cache: false,
lookup: None,
});
roundtrip_request(&ServerRequest::ForgeQuery {
target,
gc: true,
clear_cache: true,
lookup: Some("key".into()),
});
}
}
/// forge 状态响应 round-trip:含 gc_collected 与 lookup 两 Some 形态与全省略形态。
#[test]
fn forge_status_roundtrip() {
for resp in [
ServerResponse::ForgeStatus {
target: ForgeTarget::Code,
entries: 0,
capacity: 512,
gc_collected: None,
cache_cleared: false,
lookup: None,
},
ServerResponse::ForgeStatus {
target: ForgeTarget::String,
entries: 42,
capacity: 0,
gc_collected: Some(3),
cache_cleared: true,
lookup: Some(LookupResult::Interned { id: 7 }),
},
ServerResponse::ForgeStatus {
target: ForgeTarget::Object,
entries: 1,
capacity: 0,
gc_collected: None,
cache_cleared: false,
lookup: Some(LookupResult::Absent),
},
] {
roundtrip_response(&resp);
}
}
/// 序列化形态:forge_query 省略空旗标,lookup 省略 None。
#[test]
fn forge_query_serialization_shape() {
assert_eq!(
serde_json::to_string(&ServerRequest::ForgeQuery {
target: ForgeTarget::Code,
gc: false,
clear_cache: false,
lookup: None,
})
.unwrap(),
r#"{"type":"forge_query","target":"code"}"#
);
assert_eq!(
serde_json::to_string(&ServerRequest::ForgeQuery {
target: ForgeTarget::String,
gc: true,
clear_cache: false,
lookup: Some("k".into()),
})
.unwrap(),
r#"{"type":"forge_query","target":"string","gc":true,"lookup":"k"}"#
);
}
/// 序列化形态:eval 请求省略空 max_steps,控制请求无载荷字段。
#[test]
fn request_serialization_shape() {
+41 -7
View File
@@ -10,8 +10,9 @@
//! - accept 循环为非阻塞 accept 加 10 毫秒轮询加原子关闭标志:唤醒路径
//! 确定、无锁、无唤醒丢失。
//! - 并发模型:主线程只跑 accept 循环;每连接一个标准库线程只做 I/O
//! (帧读写);六类控制请求由连接线程直接应答、不占 worker;执行请求
//! 经 mpsc 路由到固定 N 个常驻 worker 执行(各持自己线程上的 VM 池)。
//! (帧读写);六类控制请求由连接线程直接应答、不占 worker;执行请求与
//! forge 查询经 mpsc 路由到固定 N 个常驻 worker 执行(各持自己线程上的
//! VM 池)。
//! - 优雅退出顺序:关闭请求、SIGINT/SIGTERM 信号或 yield 请求置位关闭标志
//! → 在途归零 → 清空发送端 → join worker → drop 内核 → 先删 socket 文件
//! 后删 sidecar(新 server 以 sidecar 消失为接管信号)。
@@ -32,7 +33,7 @@ use oxide_kernel::kernel::{KernelConfig, KernelCore};
use oxide_log::{Level, LogConfig, Output, SUBSYSTEM_COUNT};
use oxide_vm::vm_pool::PoolCounters;
use super::protocol::{self, FrameReader, ProtocolError, ServerRequest, ServerResponse};
use super::protocol::{self, ForgeTarget, FrameReader, ProtocolError, ServerRequest, ServerResponse};
use super::sidecar::{self, ClaimResult};
use super::workers::{self, WorkerRouter};
@@ -617,8 +618,8 @@ fn accept_loop(
/// # 步骤
/// 1. 克隆写端,原流交给帧读取器,设读超时(`read_timeout` 参数)。
/// 2. 循环读帧:EOF、读超时、非法 UTF-8 退出;帧长超限按协议约定关连接。
/// 3. 畸形帧以 `Error` 帧答复后继续;执行请求路由到 worker 并阻塞等回复;
/// 控制请求直接应答并写回。
/// 3. 畸形帧以 `Error` 帧答复后继续;执行请求与 forge 查询经 worker 路由
/// 并阻塞等回复;控制请求直接应答并写回。
/// 4. 关闭或 yield 确认帧写回即退出循环。
///
/// # 边界与前提
@@ -645,6 +646,12 @@ fn handle_connection(stream: UnixStream, ctx: &ServerContext, router: &WorkerRou
// 执行请求走 worker 路由,控制请求直接应答。
let response = match protocol::parse_request(&frame) {
Ok(ServerRequest::Eval { code, max_steps }) => dispatch_eval(&code, max_steps, router),
Ok(ServerRequest::ForgeQuery {
target,
gc,
clear_cache,
lookup,
}) => dispatch_forge(target, gc, clear_cache, lookup, router),
Ok(request) => dispatch_control(&request, ctx),
Err(ProtocolError::FrameTooLarge) => break,
Err(error) => error.to_error_frame(),
@@ -684,6 +691,32 @@ fn dispatch_eval(code: &str, max_steps: Option<u64>, router: &WorkerRouter) -> S
}
}
/// forge 查询分派:路由到 worker 执行(读 forge 状态加可选 gc / clear-cache /
/// lookup),阻塞在一次性回复通道上取回响应帧。
///
/// # 步骤
/// 1. 建一次性回复通道,发送端随任务进 worker。
/// 2. 轮询投递到 worker;发送端已清空(关闭序列进行中)时立即以错误帧答复。
/// 3. 阻塞 recv 至回复;worker 已退出(通道断开)时以错误帧答复,不挂起。
fn dispatch_forge(
target: ForgeTarget, gc: bool, clear_cache: bool, lookup: Option<String>, router: &WorkerRouter,
) -> ServerResponse {
let (reply_tx, reply_rx) = std::sync::mpsc::channel();
let task = workers::WorkerTask::ForgeQuery {
target,
gc,
clear_cache,
lookup,
reply: reply_tx,
};
match router.route(task) {
Ok(()) => reply_rx
.recv()
.unwrap_or_else(|_| ServerResponse::eval_err("worker 已退出,forge 查询未处理")),
Err(_) => ServerResponse::eval_err("server 正在关闭,forge 查询未受理"),
}
}
/// 控制请求分派:六类控制请求直接应答,不占 worker。
///
/// Status 响应读全体 worker 池的共享聚合计数器。
@@ -714,9 +747,10 @@ fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerRespo
ctx.shutdown.store(true, Ordering::SeqCst);
ServerResponse::Yield
}
// 防御性兜底:正常路径 eval 在调用本函数前已被拦截走 worker 路由;
// 若到达此臂说明上游分流缺失,以错误帧答复而非静默执行。
// 防御性兜底:正常路径 eval 与 forge 查询在调用本函数前已被拦截走
// worker 路由;若到达此臂说明上游分流缺失,以错误帧答复而非静默执行。
ServerRequest::Eval { .. } => ServerResponse::eval_err("执行请求应经 worker 路由,不应到达控制分派"),
ServerRequest::ForgeQuery { .. } => ServerResponse::eval_err("forge 查询应经 worker 路由,不应到达控制分派"),
}
}
+20 -1
View File
@@ -17,7 +17,8 @@ use oxide_kernel::kernel::KernelCore;
use oxide_vm::vm_pool::{PoolCounters, VmPool};
use super::eval;
use super::protocol::ServerResponse;
use super::forge;
use super::protocol::{ForgeTarget, ServerResponse};
/// worker 任务:执行请求加一次性回复通道。
#[derive(Debug)]
@@ -28,6 +29,14 @@ pub enum WorkerTask {
max_steps: Option<u64>,
reply: mpsc::Sender<ServerResponse>,
},
/// forge 查询:目标 forge 加三个行动旗标,一次性回复通道。
ForgeQuery {
target: ForgeTarget,
gc: bool,
clear_cache: bool,
lookup: Option<String>,
reply: mpsc::Sender<ServerResponse>,
},
}
/// worker 路由器:N 条独立任务通道的发送端加轮询计数。
@@ -125,6 +134,16 @@ fn worker_loop(receiver: mpsc::Receiver<WorkerTask>, kernel: Arc<KernelCore>, po
let response = eval::handle_eval(&code, max_steps, &kernel, &pool);
let _ = reply.send(response);
}
WorkerTask::ForgeQuery {
target,
gc,
clear_cache,
lookup,
reply,
} => {
let response = forge::handle_forge_query(target, gc, clear_cache, lookup, &kernel, &pool);
let _ = reply.send(response);
}
}
}
}
+81 -16
View File
@@ -1,5 +1,5 @@
//! server CLI 集成测试:真实二进制的分派面(help 列子命令、占位臂退出码、
//! cleanup 幂等、start 与 restart 端到端)。
//! server CLI 集成测试:真实二进制的分派面(help 列子命令、无 server 时
//! 控制臂与 forge 退 1、cleanup 幂等、start 与 restart 端到端、forge 端到端)。
//!
//! start / restart 端到端与 cleanup 走 well-known 全局路径(每用户单例):
//! 测试先探活,socket 存活即 panic 不抢占存活 server;三枚触碰全局路径的
@@ -48,20 +48,6 @@ fn server_help_lists_subcommands() {
}
}
/// 占位臂退 2,stderr 含 not yet implemented。
///
/// version/status/health/info/stop/restart/log 七臂已是真实实现(无 server
/// 时退 1),不在此列;forge 仍是占位。
#[test]
fn server_stub_subcommands_exit_2() {
for sub in ["forge"] {
let output = oxide(&["server", sub]);
assert_eq!(output.status.code(), Some(2), "{sub} 应退 2");
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(stderr.contains("not yet implemented"), "{sub} 应打印未实现提示:{stderr}");
}
}
/// log 子命令无日志文件时退 1,stderr 含「日志文件不存在」。
///
/// 走 well-known 全局路径:先探活,存活 server 占用时 panic 不抢占;删除
@@ -105,6 +91,26 @@ fn server_control_arms_no_server_exit_1() {
}
}
/// forge 子命令无 server 时退 1,stderr 含「无已注册的 server」提示。
///
/// 走 well-known 全局路径:先探活,存活 server 占用时 panic 不抢占。
#[test]
fn server_forge_no_server_exit_1() {
let _guard = WELL_KNOWN_LOCK.lock().expect("全局路径锁不应中毒");
assert!(
!sidecar::is_server_alive(&sidecar::well_known_socket_path()),
"存活 server 占用 well-known 路径,测试不抢占"
);
let output = oxide(&["server", "forge", "code"]);
assert_eq!(output.status.code(), Some(1), "forge 应退 1");
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(
stderr.contains("无已注册的 server"),
"forge 应打印无 server 提示:{stderr}"
);
}
/// cleanup 退 0(幂等:有残留则清理、无残留则报无残留),事后全局路径无文件。
#[test]
fn server_cleanup_exit_0() {
@@ -367,3 +373,62 @@ fn server_log_e2e() {
// 收尾:删除日志文件恢复状态。
let _ = std::fs::remove_file(&log_path);
}
/// forge 端到端:守护形态 start 后,`forge code` 打印条目数与容量行,
/// `forge string --lookup` 打印 lookup 行,`forge code --gc --clear-cache`
/// 打印 gc 与清缓存行;关闭后 socket 与 sidecar 文件被删。
#[test]
fn server_forge_e2e() {
let _guard = WELL_KNOWN_LOCK.lock().expect("全局路径锁不应中毒");
let socket = sidecar::well_known_socket_path();
let sidecar_path = sidecar::well_known_sidecar_path();
// 探活先行:存活 server 占用全局路径时 panic,不抢占。
assert!(!sidecar::is_server_alive(&socket), "存活 server 占用 well-known 路径,测试不抢占");
// start 为守护形态:CLI 即刻返回,server 是脱离的孙进程;轮询 socket 文件
// 出现判就绪(worker 数压到 2 避免按宿主核数预热的成本)。
let output = oxide(&["server", "start", "--workers", "2"]);
assert_eq!(output.status.code(), Some(0), "start 应退 0");
let deadline = Instant::now() + Duration::from_secs(10);
while !socket.exists() {
assert!(Instant::now() < deadline, "socket 文件未出现");
std::thread::sleep(Duration::from_millis(10));
}
// forge code:退 0,stdout 含条目数行与容量行。
let output = oxide(&["server", "forge", "code"]);
assert_eq!(output.status.code(), Some(0), "forge code 应退 0");
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(stdout.contains("code forge:"), "forge code 应打印条目数行:{stdout}");
assert!(stdout.contains("capacity: 512"), "forge code 应打印容量行:{stdout}");
// forge string --lookup:退 0,stdout 含 lookup 行。
let output = oxide(&["server", "forge", "string", "--lookup", "forge-e2e-key"]);
assert_eq!(output.status.code(), Some(0), "forge string --lookup 应退 0");
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(stdout.contains("lookup:"), "forge --lookup 应打印 lookup 行:{stdout}");
// forge code --gc --clear-cache:退 0,stdout 含 gc 与清缓存行。
let output = oxide(&["server", "forge", "code", "--gc", "--clear-cache"]);
assert_eq!(output.status.code(), Some(0), "forge code --gc --clear-cache 应退 0");
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(stdout.contains("gc: collected"), "forge --gc 应打印 gc 行:{stdout}");
assert!(stdout.contains("cache cleared"), "forge --clear-cache 应打印清缓存行:{stdout}");
// 关闭请求帧:读回关闭确认帧。
let mut stream = UnixStream::connect(&socket).expect("连接 server 应成功");
let response = send_and_recv(&mut stream, &ServerRequest::Shutdown);
assert!(matches!(response, ServerResponse::Shutdown), "应得关闭确认帧:{response:?}");
drop(stream);
// 轮询 sidecar 文件消失(30 秒截止、50 毫秒间隔)判 server 进程已退出。
let deadline = Instant::now() + Duration::from_secs(30);
while sidecar_path.exists() {
assert!(Instant::now() < deadline, "server 应在有界时间内退出");
std::thread::sleep(Duration::from_millis(50));
}
assert!(!socket.exists(), "socket 文件应被删除");
assert!(!sidecar_path.exists(), "sidecar 文件应被删除");
}
+5
View File
@@ -87,6 +87,11 @@ impl CodeForge {
Ok(module)
}
/// 清空全部缓存模块(LRU 复位,后续 eval 重编译)。
pub fn clear(&self) {
self.map.lock().unwrap().clear();
}
/// 当前缓存条目数。
pub fn len(&self) -> usize {
self.map.lock().unwrap().len()
+27
View File
@@ -324,6 +324,16 @@ impl PermInterner {
entries.get(id as usize).map(|e| e.data)
}
/// 纯查询:键已 intern 时返回其稳定 id,未 intern 返回 None。
///
/// 不插入新条目(区别于会插入的 `intern`,只读语义,供诊断查询)。
pub fn lookup_id(&self, s: &str) -> Option<u32> {
let hash = hash64(s);
let candidates = self.hash_map.get(&hash)?;
let entries = self.entries.read().unwrap();
candidates.iter().find(|&&id| entries[id as usize].data == s).copied()
}
/// 全部唯一 intern 键的数量。
pub fn entry_count(&self) -> u32 {
self.entries.read().unwrap().len() as u32
@@ -509,3 +519,20 @@ pub fn const_string_ptr(kind: u8) -> *const JsString {
}
}
}
#[cfg(test)]
mod tests {
use super::PermInterner;
/// `lookup_id` 纯查询:已 intern 键得该 id,未 intern 键得 None,且不增条目数。
#[test]
fn lookup_id_pure_query_no_insert() {
let interner = PermInterner::new();
let (id, _) = interner.intern("probe-key");
let before = interner.entry_count();
assert_eq!(interner.lookup_id("probe-key"), Some(id), "已 intern 键应得该 id");
assert_eq!(interner.lookup_id("absent-key"), None, "未 intern 键应得 None");
assert_eq!(interner.entry_count(), before, "纯查询不应插入新条目");
}
}
+26
View File
@@ -228,6 +228,23 @@ impl VmPool {
pub fn counters(&self) -> Arc<PoolCounters> {
Arc::clone(&self.counters)
}
/// 对池内全体空闲 VM 执行一次完整 session GC,返回收集的 VM 数。
///
/// # 边界与前提
/// - 只触及空闲队列(available);在借 VM(在途 eval)不在范围(非安全点)。
/// - 锁池内互斥锁,与 spawn 原子互斥,无竞态。
///
/// # 副作用
/// - 各空闲 VM 的 session_bytes_allocated 重置为清扫后存活字节。
pub fn collect_idle_gc(&self) -> usize {
let mut inner = self.inner.lock().unwrap();
let n = inner.available.len();
for vm in inner.available.iter_mut() {
vm.collect_session_gc();
}
n
}
}
impl VmGuard {
@@ -428,4 +445,13 @@ mod tests {
let peak = counters.peak();
assert!((1..=4).contains(&peak), "并发峰值应落在 1 到 4 之间:{peak}");
}
#[test]
fn test_collect_idle_gc_returns_idle_count() {
let kernel = test_kernel();
let pool = VmPool::new(kernel, 1, None);
let guard = pool.spawn();
drop(guard);
assert_eq!(pool.collect_idle_gc(), 1, "归还一个空闲 VM 后收集数应为 1");
}
}