406 lines
21 KiB
Rust
406 lines
21 KiB
Rust
//! 后台文件/网络 provider。线程池只携带纯数据、取消控制与弱 Task 身份，不持有 VM。
|
||
//! Lease 由托管任务拥有，释放会撤销交付资格并取消可取消的后台工作；已经完成的结果还要由 VM worker 检查归属后才能交付。
|
||
//! 并发、队列、读取字节数、超时和 URL 协议均有边界，避免 provider 绕过运行时资源限制。
|
||
//!
|
||
//! runtime::tasks 为 readText/fetchText 创建 Task 时调用 Provider::submit，并把返回 Lease 放进 Task payload。
|
||
//! Scheduler 轮询 Provider::take_ready 后在 owning VM 中核验弱身份、owner/key 与取消状态，再调用源回调。
|
||
//! 所有 Provider 共用一个全进程有界线程池；每个运行时仍有自己的结果邮箱与租约额度，不互读 Task 堆。
|
||
use endor_core::{VmError, WeakObjectRef};
|
||
use std::{
|
||
collections::HashMap,
|
||
fs::File,
|
||
io::{Read, Write},
|
||
process::{Child, Command, Stdio},
|
||
sync::{
|
||
Arc, Mutex, OnceLock,
|
||
atomic::{AtomicBool, AtomicUsize, Ordering},
|
||
mpsc::{SyncSender, sync_channel},
|
||
},
|
||
thread,
|
||
time::Duration,
|
||
};
|
||
|
||
// 文件与 HTTP 返回文本最多 1 MiB，最终必须为有效 UTF-8；长度按字节计算。
|
||
const MAX_TEXT: usize = 1024 * 1024;
|
||
// 每个运行时的租约额度，独立于全进程 4 个执行线程和 64 项工作队列。
|
||
const MAX_ACTIVE: usize = 16;
|
||
// 全进程实际执行 OS I/O 的线程数；排队不等于已占一个执行线程。
|
||
const WORKERS: usize = 4;
|
||
// 工作队列全局上限，try_send 失败直接反馈给提交方，不让 VM 阻塞等空位。
|
||
const QUEUE: usize = 64;
|
||
// 成功/失败都只传字符串；不能把 VmError 内部可能关联的托管值或 VM 句柄送入后台线程。
|
||
type Completion = Result<String, String>;
|
||
|
||
#[derive(Debug, Default)]
|
||
/// provider 共享的就绪结果与并发计数；内容只有纯数据和弱任务身份。
|
||
struct Shared {
|
||
// worker 写入、take_ready 排空，Lease::drop 也会移除本任务的晚到/未取结果。
|
||
ready: Mutex<HashMap<WeakObjectRef, Completion>>,
|
||
// 统计当前活租约而非正在执行的 OS 操作；完成结果被取走前，租约仍可能占用名额。
|
||
active: AtomicUsize,
|
||
}
|
||
#[derive(Clone, Debug, Default)]
|
||
/// 某个运行时拥有的后台 provider 能力，负责提交和取走完成结果。
|
||
pub struct Provider {
|
||
// Arc 可供后台 Job 访问邮箱，不代表 Job 拥有 Task 或能延长交付租约。
|
||
shared: Arc<Shared>,
|
||
}
|
||
#[derive(Debug, Default)]
|
||
/// 后台工作取消标记与可取消子进程，不拥有托管 VM。
|
||
struct Control {
|
||
// Release/Acquire 传播取消资格；最终写邮箱时还需在 ready 锁内再次检查。
|
||
cancelled: AtomicBool,
|
||
process: Mutex<Option<Child>>,
|
||
}
|
||
impl Control {
|
||
/// Lease 析构调用，可在 VM/租约线程执行；只撤资格/kill，不在调用线程 wait 进程退出。
|
||
fn cancel(&self) {
|
||
// 先撤销结果交付资格，再尝试终止当前子进程；文件读取无法用此标记中断，只能丢弃晚到结果。
|
||
self.cancelled.store(true, Ordering::Release);
|
||
if let Ok(mut process) = self.process.lock()
|
||
&& let Some(child) = process.as_mut()
|
||
{
|
||
// Reaping belongs to the provider worker, never the VM thread.
|
||
// VM/租约线程这里只 kill，不 wait；子进程回收由持有 Reaper 的 I/O worker 完成，避免阻塞 VM。
|
||
let _ = child.kill();
|
||
}
|
||
}
|
||
}
|
||
#[derive(Debug)]
|
||
/// 后台工作种类及其纯文本参数，源调用的类型检查在提交前完成。
|
||
pub enum Request {
|
||
// 本地路径已在任务创建时做类型/长度校验，worker 再检查实际文件类型和大小。
|
||
ReadText(String),
|
||
// URL 会由 fetch_text 再验证 HTTP(S)/authority，传输不用 shell 解释。
|
||
FetchText(String),
|
||
}
|
||
#[derive(Debug)]
|
||
/// 仅托管 Task 拥有的交付租约；释放它会取消后台资格并清理晚到结果。
|
||
pub struct Lease {
|
||
// 归还名额与移除结果所需共享状态；不持有任务本体，避免后台根反向保活源程序。
|
||
shared: Arc<Shared>,
|
||
// 带代际的弱身份用于邮箱键，Task 死亡后不能让旧结果命中复用槽位的新任务。
|
||
task: WeakObjectRef,
|
||
// 取消后台进程/晚到交付的纯原生控制对象，后台 Job 只共享它而不共享 Lease。
|
||
control: Arc<Control>,
|
||
}
|
||
impl Drop for Lease {
|
||
/// 最后一个 Arc<Lease> 释放时撤销交付资格、删除待交付结果并归还一次并发额度；后台 Job 不延长租约。
|
||
fn drop(&mut self) {
|
||
// 租约释放立即取消交付资格；后台 OS 读取即使稍后结束，也不能再次唤醒旧 Task。
|
||
self.control.cancel();
|
||
if let Ok(mut ready) = self.shared.ready.lock() {
|
||
// 删除已到邮箱的结果；worker 同锁复查 cancelled，所以取消之后也不能重新留下结果。
|
||
ready.remove(&self.task);
|
||
}
|
||
// Lease 包在 Arc 中，只有最后一个拥有者释放才析构并归还一次额度；Job 不持有 Lease。
|
||
self.shared.active.fetch_sub(1, Ordering::AcqRel);
|
||
}
|
||
}
|
||
/// 线程池队列中的纯数据工作，关联弱任务身份、控制和共享完成表。
|
||
struct Job {
|
||
// 操作种类及参数只含拥有字符串，不借用 VM 对象属性或源参数切片。
|
||
request: Request,
|
||
// 只用于交付结果地址，后台从不尝试升级弱引用。
|
||
task: WeakObjectRef,
|
||
// 共享取消状态/进程槽，让租约销毁可终止传输；Job 不持 Lease，因此不阻止取消。
|
||
control: Arc<Control>,
|
||
// 结果写入对应 Provider 的 mailbox，不把不同运行时结果混入一个全局表。
|
||
shared: Arc<Shared>,
|
||
}
|
||
impl Provider {
|
||
/// tasks 的 I/O 创建路径调用，原子预占额度、建立 RAII Lease 并非阻塞入队，返回交付资格而非执行结果。
|
||
pub fn submit(
|
||
&self,
|
||
task: WeakObjectRef,
|
||
request: Request,
|
||
replacing: bool,
|
||
) -> Result<Arc<Lease>, VmError> {
|
||
// 原子更新同时完成检查和预占，防止并发提交都观察到剩余名额而超配。
|
||
// replacing 允许替换期间临时多一个租约，旧任务取消后回到正常上限。
|
||
self.shared
|
||
.active
|
||
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| {
|
||
// replacing 只放宽一次替换的瞬时名额，不为普通提交永久提升每运行时上限。
|
||
(active < MAX_ACTIVE + usize::from(replacing)).then_some(active + 1)
|
||
})
|
||
.map_err(|_| VmError::ExecutionLimit("16 I/O tasks per runtime"))?;
|
||
// 预占后立即建立 RAII 租约；线程池不可用或队列发送失败时，局部 lease 析构会归还名额。
|
||
let lease = Arc::new(Lease {
|
||
shared: self.shared.clone(),
|
||
task,
|
||
control: Arc::new(Control::default()),
|
||
});
|
||
// 池初始化也可能失败；lease 已建立，任何后续 ? 都会归还预占额度。
|
||
let sender = pool().map_err(|_| VmError::InvalidArgument("I/O provider unavailable"))?;
|
||
// 非阻塞入队，不让调用 VM 等待后台吞吐；Job 只带参数、弱任务身份和原生取消/结果状态。
|
||
sender
|
||
.try_send(Job {
|
||
request,
|
||
task,
|
||
control: lease.control.clone(),
|
||
shared: self.shared.clone(),
|
||
})
|
||
.map_err(|_| VmError::ExecutionLimit("I/O provider queue full or unavailable"))?;
|
||
Ok(lease)
|
||
}
|
||
/// Scheduler::poll 在 VM worker 调用，排空此 Provider 的已就绪纯数据，不在邮箱锁中执行源回调。
|
||
// 一次 drain 所有已就绪结果；移走纯数据不代表任务仍可交付，VM worker 还需检查弱身份和归属。
|
||
pub fn take_ready(&self) -> Result<Vec<(WeakObjectRef, Completion)>, VmError> {
|
||
Ok(self
|
||
.shared
|
||
.ready
|
||
.lock()
|
||
.map_err(|_| VmError::InvalidArgument("I/O mailbox poisoned"))?
|
||
.drain()
|
||
.collect())
|
||
}
|
||
}
|
||
/// Provider::submit 调用，懒初始化全进程线程池；返回借用的 SyncSender，不为每个 VM 新建 I/O 线程。
|
||
fn pool() -> Result<&'static SyncSender<Job>, &'static str> {
|
||
// 全进程只初始化一次，初始化失败也会记住失败结果；后续 submit 返回 unavailable，不反复创建线程。
|
||
static POOL: OnceLock<Result<SyncSender<Job>, String>> = OnceLock::new();
|
||
POOL.get_or_init(|| {
|
||
// 队列只承载 Job；容量限制排队任务，不代替 Provider 自己的 MAX_ACTIVE 租约预算。
|
||
let (sender, receiver) = sync_channel::<Job>(QUEUE);
|
||
// std Receiver 不可 Clone，短锁共享接收端；拿到一个 Job 后放锁再并行执行 I/O。
|
||
let receiver = Arc::new(Mutex::new(receiver));
|
||
for index in 0..WORKERS {
|
||
// 每线程持接收端 Arc，不持任何 Provider/Task 的租约，空闲时阻塞 recv 而非轮询。
|
||
let receiver = receiver.clone();
|
||
thread::Builder::new()
|
||
.name(format!("tessera-io-{index}"))
|
||
.spawn(move || {
|
||
loop {
|
||
// std mpsc Receiver 通过 Mutex 共享；锁只覆盖取任务，实际 I/O 在离开该 match 后并行执行。
|
||
let job = match receiver.lock() {
|
||
Ok(receiver) => receiver.recv(),
|
||
// 接收锁损坏时终止该线程，不在不可靠队列状态上继续读任务。
|
||
Err(_) => break,
|
||
};
|
||
let Ok(job) = job else {
|
||
// 所有 sender 已断开就退出；这不是一个可重试的临时“队列为空”。
|
||
break;
|
||
};
|
||
// 执行前先跳过已取消队列项，避免启动无用文件读取或 curl。
|
||
if job.control.cancelled.load(Ordering::Acquire) {
|
||
continue;
|
||
}
|
||
// 只进行原生文件/HTTP 操作，失败也编码成纯文本 Completion，留给 VM 线程交付。
|
||
let result = match &job.request {
|
||
Request::ReadText(path) => read_text(path),
|
||
Request::FetchText(url) => fetch_text(url, &job.control),
|
||
};
|
||
// 持有 ready 锁后再次检查取消，与 Lease::drop 的取消/移除配合，防止取消后重新留下晚到结果。
|
||
if let Ok(mut ready) = job.shared.ready.lock()
|
||
&& !job.control.cancelled.load(Ordering::Acquire)
|
||
{
|
||
// 与 Lease::drop 用同一 ready 锁封住取消/插入竞争，晚到结果不能重新激活旧 Task。
|
||
ready.insert(job.task, result);
|
||
}
|
||
}
|
||
})
|
||
.map_err(|error| error.to_string())?;
|
||
}
|
||
Ok(sender)
|
||
})
|
||
.as_ref()
|
||
.map_err(|_| "I/O provider unavailable")
|
||
}
|
||
/// pool 的 ReadText 分支在后台调用，返回受字节预算限制的 UTF-8 文本或可交付错误字符串。
|
||
fn read_text(path: &str) -> Completion {
|
||
// 先按路径检查普通文件和已知长度，及早拒绝目录/超大源；实际打开后还会复核文件类型。
|
||
let metadata = std::fs::metadata(path).map_err(|error| error.to_string())?;
|
||
if !metadata.is_file() {
|
||
return Err("text source must be a regular file".into());
|
||
}
|
||
if metadata.len() > MAX_TEXT as u64 {
|
||
return Err("text source exceeds 1 MiB".into());
|
||
}
|
||
// 路径检查与打开之间可能变化，所以打开后再次检查句柄实际指向的对象类型。
|
||
let file = File::open(path).map_err(|error| error.to_string())?;
|
||
if !file
|
||
.metadata()
|
||
.map_err(|error| error.to_string())?
|
||
.is_file()
|
||
{
|
||
return Err("text source must be a regular file".into());
|
||
}
|
||
// 原始字节最后才严格转 UTF-8，不用有损替换悄悄改写用户文件内容。
|
||
let mut bytes = Vec::new();
|
||
// 再用流读取硬上限检测增长中的文件：多读一个字节即可判断超限，不能只依赖之前 metadata 长度。
|
||
file.take(MAX_TEXT as u64 + 1)
|
||
.read_to_end(&mut bytes)
|
||
.map_err(|error| error.to_string())?;
|
||
if bytes.len() > MAX_TEXT {
|
||
return Err("text source exceeds 1 MiB".into());
|
||
}
|
||
String::from_utf8(bytes).map_err(|_| "text source must be UTF-8".into())
|
||
}
|
||
|
||
/// tasks 创建 fetchText 与 fetch_text 真正传输前都会调用，拒绝错误协议、过长 URL、控制字符和 userinfo。
|
||
/// 这是受限文本 GET 接口的基本格式检查，不负责完整 URL 规范化，也不声明网络地址访问授权。
|
||
pub fn valid_url(url: &str) -> bool {
|
||
// 控制字符必须在写 curl config 前拒绝，否则换行等可能破坏单个 url 配置项边界。
|
||
if url.len() > 8192 || url.chars().any(char::is_control) {
|
||
return false;
|
||
}
|
||
let Some((scheme, rest)) = url.split_once("://") else {
|
||
// 只接受有显式 scheme 的 URL，不让 curl 猜协议或把路径当作另一类资源地址。
|
||
return false;
|
||
};
|
||
if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
|
||
return false;
|
||
}
|
||
// 仅检查协议与 authority 基本约束；@ 禁止 userinfo 凭据，空白禁止模糊主机部分。
|
||
// 这不是完整 URL 解析器，具体 HTTP 解析仍由 curl 负责。
|
||
let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
|
||
!authority.is_empty() && !authority.contains('@') && !authority.chars().any(char::is_whitespace)
|
||
}
|
||
|
||
/// fetch_text 仅在 I/O worker 上创建的清理守卫；任何提前返回都取走、kill/wait 尚未回收的子进程。
|
||
struct Reaper<'a>(&'a Control);
|
||
impl Drop for Reaper<'_> {
|
||
/// 取走残留 curl 子进程后尝试 kill/wait，保证失败或取消路径也回收进程，且不持控制锁等待。
|
||
fn drop(&mut self) {
|
||
// 从控制槽取出子进程后释放锁，再 kill/wait；只有实际留下的进程才需要回收。
|
||
let child = self.0.process.lock().ok().and_then(|mut slot| slot.take());
|
||
if let Some(mut child) = child {
|
||
let _ = child.kill();
|
||
let _ = child.wait();
|
||
}
|
||
}
|
||
}
|
||
/// pool 的 FetchText 分支调用，用系统 curl 做有界 HTTP GET；所有等待与进程回收都发生在后台线程。
|
||
fn fetch_text(url: &str, control: &Control) -> Completion {
|
||
if !valid_url(url) {
|
||
return Err("invalid HTTP(S) URL".into());
|
||
}
|
||
#[cfg(unix)]
|
||
// Unix 固定系统路径，不从用户 PATH 选择任意同名可执行文件。
|
||
let executable = "/usr/bin/curl";
|
||
#[cfg(not(unix))]
|
||
// 非 Unix 使用系统环境可解析的 curl.exe，找不到时反馈明确的传输不可用错误。
|
||
let executable = "curl.exe";
|
||
// 参数数组直接交给进程 API，URL 另外经 stdin 配置传入，全程不经 shell 拼接。
|
||
let mut child = Command::new(executable)
|
||
// --disable 放首项关闭 curlrc，--globoff 禁用 URL 展开，首次请求与重定向均限制 HTTP(S)。
|
||
// 限制重定向数、连接/总超时和下载大小；--compressed 后仍会用下面字节读取上限限制解压文本。
|
||
.args([
|
||
"--disable",
|
||
"--silent",
|
||
"--show-error",
|
||
"--globoff",
|
||
"--location",
|
||
"--max-redirs",
|
||
"5",
|
||
"--proto",
|
||
"=http,https",
|
||
"--proto-redir",
|
||
"=http,https",
|
||
"--max-time",
|
||
"30",
|
||
"--connect-timeout",
|
||
"10",
|
||
"--max-filesize",
|
||
"1048576",
|
||
"--compressed",
|
||
"--fail",
|
||
"--write-out",
|
||
"\n%{http_code}",
|
||
"--config",
|
||
"-",
|
||
])
|
||
.stdin(Stdio::piped())
|
||
.stdout(Stdio::piped())
|
||
.stderr(Stdio::null())
|
||
.spawn()
|
||
.map_err(|_| "HTTP transport unavailable: system curl is required".to_owned())?;
|
||
// 取走管道后把 Child 放入共享 Control，取消线程可随时 kill；Reaper 覆盖其后的所有提前返回。
|
||
// 管道配置是上方 spawn 的内部前提，take 后分别归配置写入和正文读取负责。
|
||
let mut input = child.stdin.take().expect("piped stdin");
|
||
// 取走 stdout 后 Child 仍可放入共享进程槽供取消线程 kill。
|
||
let output = child.stdout.take().expect("piped stdout");
|
||
match control.process.lock() {
|
||
Ok(mut slot) => *slot = Some(child),
|
||
Err(_) => {
|
||
// 尚未安装 Reaper 且不能存入共享槽，当前线程必须亲自 kill/wait 清理已启动子进程。
|
||
let _ = child.kill();
|
||
let _ = child.wait();
|
||
return Err("HTTP process lock poisoned".into());
|
||
}
|
||
}
|
||
// 从这里起即使写配置/读正文/解析状态失败，也会通过 Drop 回收进程而不留下僵尸。
|
||
let _reaper = Reaper(control);
|
||
if control.cancelled.load(Ordering::Acquire) {
|
||
return Err("request cancelled".into());
|
||
}
|
||
// URL 经 curl 配置的引号/反斜杠转义后走 stdin，避免出现在进程参数；这里不经过 shell。
|
||
let escaped = url.replace('\\', "\\\\").replace('"', "\\\"");
|
||
writeln!(input, "url = \"{escaped}\"").map_err(|_| "HTTP transport input failed".to_owned())?;
|
||
// 关闭 stdin 表示配置结束，否则 curl 可能继续等待输入，无法开始/完成请求。
|
||
drop(input);
|
||
// stdout 同时含正文与 curl --write-out 状态尾，预算须为尾部保留四字节。
|
||
let mut bytes = Vec::new();
|
||
output
|
||
// 预留结尾换行和三位状态码共 4 字节，再额外读 1 字节探测正文超限。
|
||
.take(MAX_TEXT as u64 + 5)
|
||
.read_to_end(&mut bytes)
|
||
.map_err(|_| "HTTP response read failed".to_owned())?;
|
||
if bytes.len() > MAX_TEXT + 4 {
|
||
return Err("HTTP text exceeds 1 MiB".into());
|
||
}
|
||
// 读完输出后用 try_wait 检查退出；等待间隔先释放 process 锁，让其他线程仍可取消进程。
|
||
let status = loop {
|
||
// 短暂持锁查询 child 状态；睡眠前释放槽锁，让 Lease::drop 仍可及时 kill。
|
||
let mut slot = control
|
||
.process
|
||
.lock()
|
||
.map_err(|_| "HTTP process lock poisoned".to_owned())?;
|
||
// 槽内 Child 在本请求结束前应存在；缺项是状态错误，不继续对另一进程等待。
|
||
let child = slot.as_mut().ok_or("HTTP process missing")?;
|
||
if let Some(status) = child
|
||
.try_wait()
|
||
.map_err(|_| "HTTP transport wait failed".to_owned())?
|
||
{
|
||
// try_wait 已回收退出状态，移除 Child 后 Reaper 析构不会再次 kill/wait 同一个进程。
|
||
slot.take();
|
||
break status;
|
||
}
|
||
drop(slot);
|
||
thread::sleep(Duration::from_millis(2));
|
||
};
|
||
if control.cancelled.load(Ordering::Acquire) {
|
||
// 完成传输并不自动取得交付资格，取消优先于刚读到的正常 HTTP 返回值。
|
||
return Err("request cancelled".into());
|
||
}
|
||
// --write-out 的末尾四字节应是换行加三位状态码；checked_sub 处理空/短输出，不直接倒数切片。
|
||
let code = bytes
|
||
.len()
|
||
.checked_sub(4)
|
||
.and_then(|offset| (bytes[offset] == b'\n').then_some(&bytes[offset + 1..]))
|
||
.and_then(|digits| std::str::from_utf8(digits).ok())
|
||
.and_then(|digits| digits.parse::<u16>().ok());
|
||
// 已知 HTTP 3xx/4xx/5xx 优先以 HTTP 错误报告，其他 curl 非零状态再映射为超时、大小或传输失败。
|
||
if let Some(code) = code
|
||
&& code >= 300
|
||
{
|
||
return Err(format!("HTTP {code}"));
|
||
}
|
||
if !status.success() {
|
||
// 区分 curl 的超时/大小错误，其余保留退出码供诊断；信号终止没有普通退出码。
|
||
return Err(match status.code() {
|
||
Some(28) => "HTTP request timed out".into(),
|
||
Some(63) => "HTTP text exceeds 1 MiB".into(),
|
||
Some(code) => format!("HTTP transport failed (curl {code})"),
|
||
None => "HTTP transport terminated".into(),
|
||
});
|
||
}
|
||
if !code.is_some_and(|code| (200..300).contains(&code)) {
|
||
return Err("HTTP response status unavailable".into());
|
||
}
|
||
// 确认 2xx 与成功退出后才剥去协议附加状态尾部，再严格解码 UTF-8 正文。
|
||
bytes.truncate(bytes.len() - 4);
|
||
String::from_utf8(bytes).map_err(|_| "HTTP text must be UTF-8".into())
|
||
}
|