368 lines
16 KiB
Rust
368 lines
16 KiB
Rust
//! 后台文件/网络 provider。线程池只携带纯数据、取消控制与弱 Task 身份，不持有 VM。
|
||
//! Lease 由托管任务拥有，释放会撤销交付资格并取消可取消的后台工作；已经完成的结果还要由 VM worker 检查归属后才能交付。
|
||
//! 并发、队列、读取字节数、超时和 URL 协议均有边界，避免 provider 绕过运行时资源限制。
|
||
//!
|
||
//! Background providers carry plain data and weak task identities, never a VM.
|
||
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;
|
||
const WORKERS: usize = 4;
|
||
const QUEUE: usize = 64;
|
||
type Completion = Result<String, String>;
|
||
|
||
#[derive(Debug, Default)]
|
||
/// provider 共享的就绪结果与并发计数；内容只有纯数据和弱任务身份。
|
||
struct Shared {
|
||
ready: Mutex<HashMap<WeakObjectRef, Completion>>,
|
||
// 统计当前活租约而非正在执行的 OS 操作；完成结果被取走前，租约仍可能占用名额。
|
||
active: AtomicUsize,
|
||
}
|
||
#[derive(Clone, Debug, Default)]
|
||
/// 某个运行时拥有的后台 provider 能力，负责提交和取走完成结果。
|
||
pub struct Provider {
|
||
shared: Arc<Shared>,
|
||
}
|
||
#[derive(Debug, Default)]
|
||
/// 后台工作取消标记与可取消子进程，不拥有托管 VM。
|
||
struct Control {
|
||
cancelled: AtomicBool,
|
||
process: Mutex<Option<Child>>,
|
||
}
|
||
impl Control {
|
||
/// 撤销当前操作的后续交付资格，并执行对应取消清理。
|
||
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 {
|
||
ReadText(String),
|
||
FetchText(String),
|
||
}
|
||
/// Only Task payloads own this lease. Provider threads hold plain control data.
|
||
#[derive(Debug)]
|
||
/// 仅托管 Task 拥有的交付租约；释放它会取消后台资格并清理晚到结果。
|
||
pub struct Lease {
|
||
shared: Arc<Shared>,
|
||
task: WeakObjectRef,
|
||
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() {
|
||
ready.remove(&self.task);
|
||
}
|
||
// Lease 包在 Arc 中，只有最后一个拥有者释放才析构并归还一次额度；Job 不持有 Lease。
|
||
self.shared.active.fetch_sub(1, Ordering::AcqRel);
|
||
}
|
||
}
|
||
/// 线程池队列中的纯数据工作，关联弱任务身份、控制和共享完成表。
|
||
struct Job {
|
||
request: Request,
|
||
task: WeakObjectRef,
|
||
control: Arc<Control>,
|
||
shared: Arc<Shared>,
|
||
}
|
||
impl Provider {
|
||
/// 检查并发额度并取得任务 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| {
|
||
(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()),
|
||
});
|
||
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)
|
||
}
|
||
/// 一次取走当前邮箱内全部 (弱任务身份, 纯文本结果)，实际托管交付仍由 VM worker 校验后执行。
|
||
// 一次 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())
|
||
}
|
||
}
|
||
/// 懒初始化共享工作线程和有界任务队列；初始化失败反馈给提交方。
|
||
fn pool() -> Result<&'static SyncSender<Job>, &'static str> {
|
||
// 全进程只初始化一次，初始化失败也会记住失败结果；后续 submit 返回 unavailable，不反复创建线程。
|
||
static POOL: OnceLock<Result<SyncSender<Job>, String>> = OnceLock::new();
|
||
POOL.get_or_init(|| {
|
||
let (sender, receiver) = sync_channel::<Job>(QUEUE);
|
||
let receiver = Arc::new(Mutex::new(receiver));
|
||
for index in 0..WORKERS {
|
||
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 {
|
||
break;
|
||
};
|
||
// 执行前先跳过已取消队列项，避免启动无用文件读取或 curl。
|
||
if job.control.cancelled.load(Ordering::Acquire) {
|
||
continue;
|
||
}
|
||
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)
|
||
{
|
||
ready.insert(job.task, result);
|
||
}
|
||
}
|
||
})
|
||
.map_err(|error| error.to_string())?;
|
||
}
|
||
Ok(sender)
|
||
})
|
||
.as_ref()
|
||
.map_err(|_| "I/O provider unavailable")
|
||
}
|
||
/// 读取受字节上限限制的 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());
|
||
}
|
||
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())
|
||
}
|
||
|
||
/// No shell interpretation, curlrc, URL globbing, or non-HTTP redirects.
|
||
/// Authentication/custom headers are deliberately outside this text GET API.
|
||
/// 检查网络请求的 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 {
|
||
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)
|
||
}
|
||
|
||
/// Created only on an I/O worker. All early exits kill and reap the child.
|
||
/// 子进程作用域收尾保护，避免取消/失败路径遗留未回收进程。
|
||
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();
|
||
}
|
||
}
|
||
}
|
||
/// 在取消、超时、协议和大小限制下取得 HTTP 文本；不把网络执行放在 VM 线程。
|
||
fn fetch_text(url: &str, control: &Control) -> Completion {
|
||
if !valid_url(url) {
|
||
return Err("invalid HTTP(S) URL".into());
|
||
}
|
||
#[cfg(unix)]
|
||
let executable = "/usr/bin/curl";
|
||
#[cfg(not(unix))]
|
||
let executable = "curl.exe";
|
||
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 覆盖其后的所有提前返回。
|
||
let mut input = child.stdin.take().expect("piped stdin");
|
||
let output = child.stdout.take().expect("piped stdout");
|
||
match control.process.lock() {
|
||
Ok(mut slot) => *slot = Some(child),
|
||
Err(_) => {
|
||
let _ = child.kill();
|
||
let _ = child.wait();
|
||
return Err("HTTP process lock poisoned".into());
|
||
}
|
||
}
|
||
let _reaper = Reaper(control);
|
||
if control.cancelled.load(Ordering::Acquire) {
|
||
return Err("request cancelled".into());
|
||
}
|
||
// URL stays out of process arguments. Reject controls before config escaping.
|
||
// 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);
|
||
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 {
|
||
let mut slot = control
|
||
.process
|
||
.lock()
|
||
.map_err(|_| "HTTP process lock poisoned".to_owned())?;
|
||
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) {
|
||
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() {
|
||
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())
|
||
}
|