1306 lines
69 KiB
Rust
1306 lines
69 KiB
Rust
//! owner/key 管理的异步任务与协作式源函数调度。弱索引用于查找，回调、等待对象和协程帧是托管图中的真实边。
|
||
//! 任务可能处于等待 I/O、定时器、可执行协程或等待另一任务的状态；取消 owner/key 不能让晚到结果再次执行。
|
||
//! 调度使用有界量子和公平推进；await 挂起后保留操作数与帧，不重放此前副作用。失败和依赖环有明确处理路径。
|
||
//!
|
||
//! ui_bridge::bootstrap 调用 install，注册 ui.after/sleep/readText/fetchText/spawn 并取得每 VM 的 Scheduler。
|
||
//! 源 API 经 After 创建 Task；UiBridge/endor_host 的轮询调用 Scheduler::poll，才在所属 VM 交付结果或推进协程。
|
||
//! 页面/窗口关闭调用 cancel_owner，运行时退出调用 quiesce；只取消身份和执行边，不允许旧 I/O 结果复活它们。
|
||
//! 这里同时含通用调度和 Tessera 展示资格检查，未来迁移 Nexus 时需将 scope/eligible 的 UI 规则留在宿主适配中。
|
||
use std::{
|
||
sync::{
|
||
Arc, Mutex,
|
||
atomic::{AtomicBool, Ordering},
|
||
},
|
||
time::{Duration, Instant},
|
||
};
|
||
|
||
use anyhow::{Result, bail};
|
||
use endor_core::{
|
||
ContextNativeFunction, NativeCallable, NativeContext, NativeResult, ObjectPayload, ObjectRef,
|
||
ObjectStore, Value, Vm, VmError, WeakObjectRef,
|
||
};
|
||
|
||
use crate::{block::Block, mount::Mount, ui_bridge::desktop_type, window::ManagedWindow};
|
||
|
||
// 普通对象的私有任务字典属性名；Block/Mount/Window 则把同一拥有关系保存在宿主 payload。
|
||
const SCOPE: &str = "\0tessera.tasks";
|
||
// 单 VM 弱调度索引的活任务预算；已完成任务可被源变量持有，但不继续占调度槽。
|
||
const MAX_TASKS: usize = 1024;
|
||
// 单 owner 作用域字典的项数上限；同 key 替换不新增一项，允许在满额度时进行。
|
||
const MAX_OWNER_TASKS: usize = 64;
|
||
// 一轮 poll 最多完成/推进/取消的工作数，避免任务队列独占 VM worker。
|
||
const MAX_READY: usize = 64;
|
||
|
||
// 调度器只保存弱任务身份，不因排队把任务、回调或 owner 隐式保活。
|
||
type Index = Arc<Mutex<Vec<WeakObjectRef>>>;
|
||
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
/// 任务对源代码可见的生命周期状态，完成/取消/失败各有不同 await 语义。
|
||
enum State {
|
||
// 等截止时间、I/O 或首次推进；不表示后台操作尚未开始。
|
||
Pending,
|
||
// 普通回调正在执行，或协作任务已有保存的协程，仍可能处于 await 等待。
|
||
Running,
|
||
// 正常终态，允许 await/result 读取保存的结果或 Nil。
|
||
Completed,
|
||
// 失去 owner/key 资格或主动取消的终态，await 应报告 AwaitCancelled。
|
||
Cancelled,
|
||
// provider/回调/协程失败，error 保存诊断而非可当作成功值交付的文字。
|
||
Failed,
|
||
}
|
||
impl State {
|
||
/// Task.status 方法调用，返回稳定的小写 API 状态字符串，不用于控制调度分支。
|
||
fn name(self) -> &'static str {
|
||
match self {
|
||
Self::Pending => "pending",
|
||
Self::Running => "running",
|
||
Self::Completed => "completed",
|
||
Self::Cancelled => "cancelled",
|
||
Self::Failed => "failed",
|
||
}
|
||
}
|
||
}
|
||
|
||
#[derive(Debug, Clone)]
|
||
/// 托管异步对象，保存 owner/key、结果与真正拥有的回调/协程/等待引用。
|
||
struct Task {
|
||
// owner 通过 scope 字典强持有 Task，Task 反向弱引用 owner，避免任务把已关闭页面永久保活。
|
||
owner: WeakObjectRef,
|
||
// owner 内唯一键，作用域字典决定它是否仍是当前任务；不是跨运行时的全局 ID。
|
||
key: String,
|
||
// 单调时钟截止点；ready 与到期是两个独立条件，I/O 完成不受系统日期调整影响。
|
||
due: Instant,
|
||
// 定时/I/O 的零参回调或 spawn 的初始函数，由 trace 声明真实强边。
|
||
callback: Option<ObjectRef>,
|
||
// 真实执行状态；只读 status 可能因失去资格投影为 cancelled，而不立刻改写此字段。
|
||
state: State,
|
||
// 定时/协作任务创建即 ready，但仍等 due；I/O 任务等 provider 结果写回后才 ready。
|
||
ready: bool,
|
||
// I/O 成功纯文本，普通回调执行期间也可通过 Task.result 查询；非托管字符串不需 trace。
|
||
text: Option<Arc<str>>,
|
||
// provider/执行失败的有界预览，Completed 的成功值不能与此混用。
|
||
error: Option<Arc<str>>,
|
||
// 仅任务 payload 持有 I/O 租约；取消/终态清除此边即可撤销后台交付并归还额度。
|
||
lease: Option<Arc<crate::io_tasks::Lease>>,
|
||
// 仅 spawn 为 true：使用可恢复 Coroutine 和软量子，其余回调一次有界调用完成。
|
||
cooperative: bool,
|
||
// spawn 首次推进创建的托管协程帧，后续 resume 保留此前副作用与累计硬预算。
|
||
coroutine: Option<ObjectRef>,
|
||
// 协程 await 的对象是真实托管强边；等待期间由它保活被等待值，不以弱调度索引代替。
|
||
waiting: Option<ObjectRef>,
|
||
// 协程最终返回的任意语言值；对象结果也经 trace 保活，取消时清掉。
|
||
result: Option<Value>,
|
||
}
|
||
impl ObjectPayload for Task {
|
||
/// 提供受检查的 Any 下转型入口，让宿主按具体 payload 类型读取数据。
|
||
fn as_any(&self) -> &dyn std::any::Any {
|
||
self
|
||
}
|
||
/// 返回运行时诊断/类型检查使用的稳定 payload 名称。
|
||
fn kind_name(&self) -> &'static str {
|
||
"tessera-task"
|
||
}
|
||
/// 声明此 payload 的赋值遵循引用身份语义，避免把内部宿主状态当作可复制值。
|
||
fn reference_assignment_only(&self) -> bool {
|
||
true
|
||
}
|
||
/// Aster 的 await 执行和 Scheduler 的等待检查调用，查询状态而不递归推进任务或 VM。
|
||
fn poll_await(&self) -> Result<endor_core::AwaitPoll, VmError> {
|
||
match self.state {
|
||
// I/O 已收到 ready 文本但尚未真正处理到终态时，await 仍不能抢先越过回调。
|
||
State::Pending | State::Running => Ok(endor_core::AwaitPoll::Pending),
|
||
// 协作任务优先返回保存的任意 Value，I/O 返回 text，普通定时器默认为 Nil。
|
||
State::Completed => Ok(endor_core::AwaitPoll::Ready(
|
||
self.result.clone().unwrap_or_else(|| {
|
||
// 协作返回值优先，否则 I/O 文本转语言字符串，纯计时器/sleep 无值则返回 Nil。
|
||
self.text
|
||
.as_ref()
|
||
.map(|text| Value::Text(text.to_string()))
|
||
.unwrap_or(Value::Nil)
|
||
}),
|
||
)),
|
||
State::Cancelled => Err(VmError::AwaitCancelled),
|
||
State::Failed => Err(VmError::AwaitFailed(
|
||
self.error.as_deref().unwrap_or("task failed").into(),
|
||
)),
|
||
}
|
||
}
|
||
/// 枚举本对象真正拥有的托管强引用，供 Aster 引用追踪和循环回收使用。
|
||
fn trace(&self, visit: &mut dyn FnMut(ObjectRef)) {
|
||
// Aster GC 只追踪真实拥有边，owner/调度索引的弱身份均不得在这里访问。
|
||
if let Some(callback) = self.callback {
|
||
visit(callback);
|
||
}
|
||
if let Some(coroutine) = self.coroutine {
|
||
visit(coroutine);
|
||
}
|
||
if let Some(waiting) = self.waiting {
|
||
visit(waiting);
|
||
}
|
||
if let Some(Value::Object(result)) = self.result {
|
||
visit(result);
|
||
}
|
||
}
|
||
}
|
||
/// API 解码和 Task 方法检查共用错误构造器，调用位置与源栈由外层 VM 诊断补充。
|
||
fn invalid(message: &'static str) -> VmError {
|
||
VmError::InvalidArgument(message)
|
||
}
|
||
/// After/Method/作用域遍历解码对象值时调用，只复制句柄，不建立隐藏托管强边。
|
||
fn object(value: &Value) -> Result<ObjectRef, VmError> {
|
||
match value {
|
||
Value::Object(object) => Ok(*object),
|
||
_ => Err(invalid("task owner and callback must be managed objects")),
|
||
}
|
||
}
|
||
/// After、current、cancel_owner/finish 调用，读取 owner 真正拥有 Task 的字典；不创建新作用域。
|
||
fn scope(store: &ObjectStore, owner: ObjectRef) -> Result<Option<ObjectRef>, VmError> {
|
||
// scope 的位置依 owner 类别而异；普通对象即使使用私有属性，也要验证该值确为字典。
|
||
let payload = store.object(owner)?.payload();
|
||
if let Some(value) = payload.as_any().downcast_ref::<Block>() {
|
||
// Block 的宿主任务边保存在 payload，不能因普通属性同名而被覆盖或伪造。
|
||
return Ok(value.tasks);
|
||
}
|
||
if let Some(value) = payload.as_any().downcast_ref::<Mount>() {
|
||
// 同业务对象的不同出现可分别拥有任务，Mount 作用域不是 Block 作用域别名。
|
||
return Ok(value.tasks);
|
||
}
|
||
if let Some(value) = payload.as_any().downcast_ref::<ManagedWindow>() {
|
||
// 窗口级任务跟资源生命周期，不随某一页切换自动转移到另一页。
|
||
return Ok(value.tasks);
|
||
}
|
||
match store.get_attr(owner, SCOPE) {
|
||
Ok(Value::Object(scope)) => {
|
||
// 普通对象的私有属性也必须真为字典，不能把任意 Object 当作 owner/key 拥有表。
|
||
store.dict_items(scope)?;
|
||
Ok(Some(scope))
|
||
}
|
||
// 尚未提交过任务是正常 None，其他读取错误必须传播而非被当作缺省空字典。
|
||
Err(VmError::AttributeNotFound(_)) => Ok(None),
|
||
Ok(_) => Err(invalid("invalid task scope")),
|
||
Err(error) => Err(error),
|
||
}
|
||
}
|
||
/// After 首次给 owner 创建字典时调用，按 owner 种类建立可追踪强边，不修改其他任务内容。
|
||
fn install_scope(
|
||
cx: &mut NativeContext<'_>,
|
||
owner: ObjectRef,
|
||
scope: ObjectRef,
|
||
) -> Result<(), VmError> {
|
||
// 宿主 payload 通过 replace_payload 建立受 GC 追踪的强边，普通对象通过 set_attr；均不能绕过写屏障。
|
||
let payload = cx.store().object(owner)?.payload();
|
||
if let Some(value) = payload.as_any().downcast_ref::<Block>() {
|
||
// 克隆 payload 后改任务边，replace_payload 负责引用增减与版本记录，不能直接改 store 借用值。
|
||
let mut value = value.clone();
|
||
value.tasks = Some(scope);
|
||
return cx.replace_payload(owner, value);
|
||
}
|
||
if let Some(value) = payload.as_any().downcast_ref::<Mount>() {
|
||
// 出现位置持字典，不把它挂到共享 Block 导致关闭一个出现时取消另一个窗口的任务。
|
||
let mut value = value.clone();
|
||
value.tasks = Some(scope);
|
||
return cx.replace_payload(owner, value);
|
||
}
|
||
if let Some(value) = payload.as_any().downcast_ref::<ManagedWindow>() {
|
||
// 复用同一写屏障保持窗口资源对象的 GC 关系，不另建 Rust HashMap 持 Task 根。
|
||
let mut value = value.clone();
|
||
value.tasks = Some(scope);
|
||
return cx.replace_payload(owner, value);
|
||
}
|
||
cx.set_attr(owner, SCOPE, Value::Object(scope))
|
||
}
|
||
/// After 创建、Task.active/status 和 Scheduler 交付前调用，验证 owner 的 Tessera 展示资格，不检查 key 身份。
|
||
fn eligible(store: &ObjectStore, owner: ObjectRef, desktop: ObjectRef) -> Result<bool, VmError> {
|
||
// 本函数只做同步借用查询，不升级额外托管根；真正调用者负责在检查/交付期间持有 owner。
|
||
let payload = store.object(owner)?.payload();
|
||
if let Some(mount) = payload.as_any().downcast_ref::<Mount>() {
|
||
if !mount.mounted {
|
||
// Mount 对象可被源变量持有，但卸载后任务不能再有展示交付资格。
|
||
return Ok(false);
|
||
}
|
||
// Mount 仅 mounted 不足以交付，所属 Page 还必须在某个活动窗口页面注册表里，关闭再打开不能恢复旧任务。
|
||
if let Some(page) = mount.page {
|
||
// desktop 指主资源域；在其真实 windows 注册表中找页，不只检查当前 active 页面。
|
||
let windows = object(&store.get_attr(desktop, "windows")?)?;
|
||
for (_, value) in store.dict_items(windows)? {
|
||
// 各窗口所有已登记 Page 都算有效，未选中标签不等于页面已关闭。
|
||
for candidate in crate::window::pages(store, object(value)?)? {
|
||
if store.downgrade(*candidate)? == page {
|
||
// 比较弱代际身份，避免堆槽复用后把新 Page 误认成原出现的所属页。
|
||
return Ok(true);
|
||
}
|
||
}
|
||
}
|
||
return Ok(false);
|
||
}
|
||
// 没有所属 Page 的 Mount 不具备异步交付资格，不能因 mounted=true 就放行孤立出现。
|
||
return Ok(false);
|
||
}
|
||
// 窗口 owner 要求未 closed；Page owner 要求仍注册，普通非页面业务 owner 没有额外展示资格要求。
|
||
if let Some(window) = payload.as_any().downcast_ref::<ManagedWindow>() {
|
||
return Ok(!window.closed);
|
||
}
|
||
if payload
|
||
.as_any()
|
||
.downcast_ref::<Block>()
|
||
.is_some_and(|block| block.page)
|
||
{
|
||
// Page 需要仍在任一窗口注册表，不能只靠对象可达/标题存在判断是否还可交付。
|
||
let windows = object(&store.get_attr(desktop, "windows")?)?;
|
||
for (_, value) in store.dict_items(windows)? {
|
||
// 从字典值解码托管 Window，再按页面身份查登记关系，而非按字符串标题查找。
|
||
let window = object(value)?;
|
||
if crate::window::pages(store, window)?.contains(&owner) {
|
||
return Ok(true);
|
||
}
|
||
}
|
||
return Ok(false);
|
||
}
|
||
// 非展示类普通业务 owner 仅需可达；其作用域/当前 key 条件由 current 另行检查。
|
||
Ok(true)
|
||
}
|
||
/// After 替换、Task 方法、Scheduler 和 finish 调用，验证 owner/key 字典仍精确指向本任务，防止旧代际交付。
|
||
fn current(
|
||
store: &ObjectStore,
|
||
owner: ObjectRef,
|
||
reference: ObjectRef,
|
||
task: &Task,
|
||
) -> Result<bool, VmError> {
|
||
// key 指向的对象必须恰是当前 Task，替换后旧 Task 即使还被外部持有，也不能接收结果或执行回调。
|
||
Ok(match scope(store, owner)? {
|
||
Some(scope) => {
|
||
// 外部仍持有旧 Task token 也不能绕过同键替换，必须精确比对 Value::Object 身份。
|
||
store.dict_get(scope, &Value::Text(task.key.clone()))? == Some(Value::Object(reference))
|
||
}
|
||
None => false,
|
||
})
|
||
}
|
||
|
||
/// window/页面关闭与 Mount 退役边界调用，取消 owner 全部任务；同一输入 close 后再 open 也不能复活旧任务。
|
||
pub fn cancel_owner(cx: &mut NativeContext<'_>, owner: ObjectRef) -> Result<(), VmError> {
|
||
// 无作用域表示从未拥有任务，关闭仍正常成功，不为取消动作反而分配一个新字典。
|
||
let Some(scope) = scope(cx.store(), owner)? else {
|
||
return Ok(());
|
||
};
|
||
// 克隆条目后逐个变更字典；取消 pending/running 的执行边，再删除所有 owner/key 拥有边。
|
||
for (key, value) in cx.store().dict_items(scope)?.to_vec() {
|
||
// 复制字典项使删除不会使迭代借用失效；每个值必须是当前 VM 的 Task。
|
||
let reference = object(&value)?;
|
||
// 取消修改副本并经写屏障提交，释放 callback/coroutine/waiting 的真实托管强边。
|
||
let mut task = cx.store().payload::<Task>(reference, "Task")?.clone();
|
||
if matches!(task.state, State::Pending | State::Running) {
|
||
// 终态内容不重新改成 Cancelled，但所有 owner/key 字典边最后都会解除。
|
||
task.state = State::Cancelled;
|
||
task.callback = None;
|
||
task.lease = None;
|
||
task.coroutine = None;
|
||
task.waiting = None;
|
||
task.text = None;
|
||
task.result = None;
|
||
cx.replace_payload(reference, task)?;
|
||
}
|
||
cx.dict_delete(scope, &key)?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
/// 当前宿主 API 的操作分类，分派代码负责参数校验与状态更新。
|
||
enum Operation {
|
||
Timer,
|
||
Sleep,
|
||
ReadText,
|
||
FetchText,
|
||
Spawn,
|
||
}
|
||
#[derive(Debug)]
|
||
/// 创建定时/I/O/协作任务时的原生调用适配器。
|
||
struct After {
|
||
// 区分计时、无回调 sleep、后台读取与协作 spawn，决定参数形状和初始 ready 状态。
|
||
operation: Operation,
|
||
// 源 API 和轮询 Scheduler 共用同一个 Provider mailbox，I/O 线程不接触此 VM。
|
||
provider: crate::io_tasks::Provider,
|
||
// install 按具体 API 构造的完整数量/类型/效果契约，进入 invoke 前由 VM 检查。
|
||
signature: endor_core::CallSignature,
|
||
// 弱调度索引，不是 Task 的拥有者；真正拥有边在 owner 的作用域字典。
|
||
index: Index,
|
||
// spawn 创建后置位，让 worker 知道有可推进协程，prune 在轮末重新计算。
|
||
runnable: Arc<AtomicBool>,
|
||
// shutdown/quiesce 置位的单向门闩，清理回调不得再提交新任务。
|
||
quiescing: Arc<AtomicBool>,
|
||
// 主资源域用于 Task 类型和跨窗口 owner 登记校验，不把 Task 固定到某个 current 页。
|
||
desktop: ObjectRef,
|
||
}
|
||
impl NativeCallable for After {
|
||
/// 给出参数与返回值的托管类型契约；实际操作在 invoke 中执行。
|
||
fn signature(&self) -> Option<endor_core::CallSignature> {
|
||
Some(self.signature.clone())
|
||
}
|
||
/// 直接使用创建时准备的 after/sleep/readText/fetchText/spawn 签名数量，和具体 API 一致。
|
||
fn arity(&self) -> Option<endor_core::CallArity> {
|
||
Some(self.signature.arity)
|
||
}
|
||
|
||
/// VM 执行 ui.after/sleep/readText/fetchText/spawn 时调用，验证全部输入再建立新 owner/key 身份，最后取消旧项。
|
||
fn invoke(&self, cx: &mut NativeContext<'_>, args: &[Value]) -> Result<NativeResult, VmError> {
|
||
// 退出域先拒绝新任务；只读渲染也禁止异步提交，即使某构造操作只分配新的任务对象。
|
||
if self.quiescing.load(Ordering::Acquire) {
|
||
// 退出门闩优先于任何分配/外部 I/O，卸载回调不能重建任务延长资源域寿命。
|
||
return Err(invalid("runtime is exiting; new tasks are forbidden"));
|
||
}
|
||
if cx.store().is_read_only() {
|
||
// 提交任务会改作用域/调度资格，即使只分配新 Task 也不是合法的纯测量/绘制操作。
|
||
return Err(invalid(
|
||
"asynchronous work is forbidden during read-only rendering",
|
||
));
|
||
}
|
||
// 统一解码不同 API 参数：spawn 用零延时与回调，sleep 无回调，I/O 第三个位置是路径/URL。
|
||
// spawn 没有延时参数，借一个局部零值复用后面的毫秒校验，不扩展源 API 参数表。
|
||
let zero = Value::Number(0_i64.into());
|
||
// 统一得到 owner/key/输入/可选回调四部分；此处只借源实参，不在尚未验证时分配 Task。
|
||
let (owner, key, delay, callback) = match args {
|
||
[owner, key, callback] if self.operation == Operation::Spawn => {
|
||
// spawn 的第三实参是零参源函数，不能误作 delay 或路径解析。
|
||
(owner, key, &zero, Some(callback))
|
||
}
|
||
// sleep 无回调，只通过 await/Task 结果状态通知完成。
|
||
[owner, key, delay] if self.operation == Operation::Sleep => (owner, key, delay, None),
|
||
// 四参形状服务 after 和带回调 I/O；第三位稍后依据 operation 解释成毫秒或路径。
|
||
[owner, key, delay, callback] => (owner, key, delay, Some(callback)),
|
||
[owner, key, path]
|
||
if matches!(self.operation, Operation::ReadText | Operation::FetchText) =>
|
||
{
|
||
// 三参 I/O 返回可 await Task，不要求为了接收文本而人为提供一个空回调。
|
||
(owner, key, path, None)
|
||
}
|
||
_ => {
|
||
// 元数据检查外仍防御直接调用/不匹配形状，不让后面按错误的位置读取输入。
|
||
return Err(invalid(
|
||
"asynchronous request requires owner, key, delay/path and optional I/O callback",
|
||
));
|
||
}
|
||
};
|
||
// owner 必须是真正托管对象，随后 eligible 再检查其展示生命周期。
|
||
let owner = object(owner)?;
|
||
// Some(错误值) 传播类型错误，None 仅表示合法无回调 API，二者不可混同。
|
||
let callback = callback.map(object).transpose()?;
|
||
// 所有回调都按零显式参数执行，必须是 Function 且参数元数据确认接受零参，不能等执行时再碰运气。
|
||
if let Some(callback) = callback {
|
||
// 是否带 receiver 已由 VM 绑定函数体现，数量校验针对调用者仍需传的未绑定实参。
|
||
if !cx.store().object(callback)?.payload().is_function() {
|
||
return Err(invalid("task callback must be a Function"));
|
||
}
|
||
if !cx
|
||
.store()
|
||
.call_arity(callback)?
|
||
.ok_or(invalid("task callback requires positional metadata"))?
|
||
.accepts(0)
|
||
{
|
||
return Err(invalid("task callback requires zero unbound parameters"));
|
||
}
|
||
}
|
||
// key 是相同 owner 内的替换身份，不允许用任意对象 repr 当作不稳定字典键。
|
||
let Value::Text(key) = key else {
|
||
return Err(invalid("task key must be text"));
|
||
};
|
||
// key 是 owner 内替换/取消身份，限制 UTF-8 字节长度与控制字符，禁止空 key。
|
||
if key.is_empty() || key.len() > 128 || key.chars().any(char::is_control) {
|
||
return Err(invalid("invalid task key"));
|
||
}
|
||
// I/O 得到零计时延迟和后台 Request，计时/协作则得到毫秒数且没有外部请求。
|
||
let (milliseconds, request) =
|
||
if matches!(self.operation, Operation::ReadText | Operation::FetchText) {
|
||
// 原局部名 delay 此时借用的是路径/URL 字符串，按 operation 先分流再判断实际类型。
|
||
let Value::Text(path) = delay else {
|
||
return Err(invalid("I/O input must be a String"));
|
||
};
|
||
// 只创建纯数据请求，provider 提交发生在 Task 已分配并有可取消租约的位置。
|
||
let request = match self.operation {
|
||
Operation::ReadText => {
|
||
// 路径允许平台路径语义但拒绝空、过长或 NUL；文件类型/字节预算由 worker 再查。
|
||
if path.is_empty() || path.len() > 4096 || path.contains('\0') {
|
||
return Err(invalid("invalid text path"));
|
||
}
|
||
crate::io_tasks::Request::ReadText(path.clone())
|
||
}
|
||
Operation::FetchText => {
|
||
// 在创建后台工作前拒绝不支持 URL，worker 仍再次校验以维护独立边界。
|
||
if !crate::io_tasks::valid_url(path) {
|
||
return Err(invalid(
|
||
"expected HTTP(S) URL without credentials or control characters",
|
||
));
|
||
}
|
||
crate::io_tasks::Request::FetchText(path.clone())
|
||
}
|
||
Operation::Timer | Operation::Sleep | Operation::Spawn => unreachable!(),
|
||
};
|
||
(0, Some(request))
|
||
} else {
|
||
// 只有计时/sleep/spawn 进入这里，spawn 由上面借用零 Number，因此也受同一整数约束。
|
||
let Value::Number(delay) = delay else {
|
||
return Err(invalid(
|
||
"task delay must be an integer number of milliseconds",
|
||
));
|
||
};
|
||
// 定时延迟要求整数毫秒，范围为零到一天；spawn 复用零延迟路径，禁止小数/负数。
|
||
let milliseconds = delay
|
||
.as_i64()
|
||
.filter(|delay| (0..=86_400_000).contains(delay))
|
||
.ok_or(invalid(
|
||
"task delay must be 0..86400000 integer milliseconds",
|
||
))?;
|
||
(milliseconds, None)
|
||
};
|
||
if !eligible(cx.store(), owner, self.desktop)? {
|
||
// 不允许关闭/卸载对象创建新工作，即使源代码仍通过变量持有它的 ObjectRef。
|
||
return Err(invalid("task owner is closed or detached"));
|
||
}
|
||
// 在分配新任务前找同 key 旧身份，额度检查允许替换但不允许借其他 key 继续增加总量。
|
||
let existing_scope = scope(cx.store(), owner)?;
|
||
// 读取可选旧 Task，不删除旧项；新任务的参数/预算/分配失败时应保留旧身份。
|
||
let previous = existing_scope
|
||
.map(|scope| cx.store().dict_get(scope, &Value::Text(key.clone())))
|
||
.transpose()?
|
||
.flatten()
|
||
.map(|value| object(&value))
|
||
.transpose()?;
|
||
// 短锁覆盖调度索引剪枝及最终换表，过程不运行源回调，避免看到半次替换索引。
|
||
let mut index = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| invalid("task index poisoned"))?;
|
||
// 提交时压缩弱索引，只保留还存活且 pending/running 的任务；终态不占调度名额。
|
||
let mut live = Vec::new();
|
||
for weak in index.iter().copied() {
|
||
if let Some(task) = cx.upgrade(weak)?
|
||
&& matches!(
|
||
cx.store().payload::<Task>(task, "Task")?.state,
|
||
State::Pending | State::Running
|
||
)
|
||
{
|
||
// NativeContext 临时升级的根在本次调用结束释放，live 保存的仍只有弱身份。
|
||
live.push(weak);
|
||
}
|
||
}
|
||
if live.len() >= MAX_TASKS && previous.is_none() {
|
||
// 同 key 替换不是扩容；新 key 在总量上限时必须失败，不能通过长期保存 Task token 绕过。
|
||
return Err(VmError::ExecutionLimit("1024 pending tasks per runtime"));
|
||
}
|
||
if let Some(scope) = existing_scope
|
||
&& cx.store().dict_items(scope)?.len() >= MAX_OWNER_TASKS
|
||
&& previous.is_none()
|
||
{
|
||
// owner 局部额度独立于全运行时额度，同 key 仍可原位更新已有槽。
|
||
return Err(VmError::ExecutionLimit("64 pending tasks per owner"));
|
||
}
|
||
// 先分配并设置截止点/状态，再安装 I/O 租约和 owner 字典边；弱 owner 不承担保活作用。
|
||
let task = cx.allocate(Task {
|
||
owner: cx.store().downgrade(owner)?,
|
||
key: key.clone(),
|
||
due: Instant::now()
|
||
.checked_add(Duration::from_millis(milliseconds as u64))
|
||
.ok_or(invalid("task deadline overflow"))?,
|
||
callback,
|
||
state: State::Pending,
|
||
ready: matches!(
|
||
self.operation,
|
||
Operation::Timer | Operation::Sleep | Operation::Spawn
|
||
),
|
||
text: None,
|
||
error: None,
|
||
lease: None,
|
||
cooperative: self.operation == Operation::Spawn,
|
||
coroutine: None,
|
||
waiting: None,
|
||
result: None,
|
||
})?;
|
||
cx.attach_type(task, desktop_type(cx.store(), "Task", self.desktop)?)?;
|
||
if let Some(request) = request {
|
||
// 只有旧任务仍待完成且真的持有 I/O 租约才使用 provider 临时替换额度，不给普通任务扩大 I/O 上限。
|
||
let replacing = previous
|
||
.map(|previous| {
|
||
// 只有实际未完成 I/O 能归还一个旧租约，终态/纯定时器不能借来放宽 provider 上限。
|
||
cx.store().payload::<Task>(previous, "Task").map(|task| {
|
||
matches!(task.state, State::Pending | State::Running)
|
||
&& task.lease.is_some()
|
||
})
|
||
})
|
||
.transpose()?
|
||
.unwrap_or(false);
|
||
// 非阻塞提交拿到交付资格；后台即使很快完成，VM 交付仍等后续 poll 核验 owner/key。
|
||
let lease = self
|
||
.provider
|
||
.submit(cx.store().downgrade(task)?, request, replacing)?;
|
||
// 把租约写进 Task payload 才成为其生命周期资源，后台 Job 自己不持有 Lease。
|
||
let mut value = cx.store().payload::<Task>(task, "Task")?.clone();
|
||
value.lease = Some(lease);
|
||
cx.replace_payload(task, value)?;
|
||
}
|
||
|
||
// 首次提交才安装字典；以后同 owner 复用 scope，key 替换始终在同一拥有表中发生。
|
||
let task_scope = if let Some(scope) = existing_scope {
|
||
// 已有作用域保持原字典身份；替换不使其他 key 的 Task 所有权发生变化。
|
||
scope
|
||
} else {
|
||
// 第一个任务才分配字典并建立 owner→scope 强边，之后 dict_set 建立 scope→Task 强边。
|
||
let scope = cx.new_dict()?;
|
||
install_scope(cx, owner, scope)?;
|
||
scope
|
||
};
|
||
// 先保存新旧身份所需弱键，下面登记成功才取消旧 Task；此处不是向 GUI 发布画面。
|
||
let previous_weak = previous.map(|old| cx.store().downgrade(old)).transpose()?;
|
||
// 替换字典边前临时升级旧任务，避免 dict_set 删除最后拥有边后，清理旧 payload 时对象已被回收。
|
||
if let Some(weak) = previous_weak {
|
||
cx.upgrade(weak)?;
|
||
}
|
||
cx.dict_set(task_scope, Value::Text(key.clone()), Value::Object(task))?;
|
||
// 新 key 身份登记成功后再将旧任务取消，切断回调/协程/等待/结果与 I/O 租约，晚到结果不得复活。
|
||
if let Some(previous) = previous {
|
||
// 前面临时升级旧身份保护它跨过字典替换，即使用户没有保存旧 Task token 也能安全清理。
|
||
if cx.store().object(previous).is_ok() {
|
||
// 只取消仍活跃项，已完成/失败的旧 token 保留自己的终态结果供已持有者检查。
|
||
let mut old = cx.store().payload::<Task>(previous, "Task")?.clone();
|
||
if matches!(old.state, State::Pending | State::Running) {
|
||
old.state = State::Cancelled;
|
||
old.callback = None;
|
||
old.lease = None;
|
||
old.coroutine = None;
|
||
old.waiting = None;
|
||
old.text = None;
|
||
old.result = None;
|
||
cx.replace_payload(previous, old)?;
|
||
}
|
||
}
|
||
}
|
||
// 新 Task 的真正根已经在作用域字典中，调度器只需要弱索引，不附加第二条隐藏拥有边。
|
||
let weak = cx.store().downgrade(task)?;
|
||
// 弱索引也移除被替换身份；spawn 额外设置 runnable 唤醒标志，使外层知道存在可推进协程。
|
||
live.retain(|old| Some(*old) != previous_weak);
|
||
live.push(weak);
|
||
*index = live;
|
||
if self.operation == Operation::Spawn {
|
||
// cooperative 工作首次即可执行，唤醒标志让 worker 不误以为只有未来计时器而进入长等待。
|
||
self.runnable.store(true, Ordering::Release);
|
||
}
|
||
Ok(NativeResult::Return(Value::Object(task)))
|
||
}
|
||
}
|
||
|
||
#[derive(Debug, Clone, Copy)]
|
||
/// 该原生能力的查询/操作分类，方法包装用它选择具体实现。
|
||
enum Query {
|
||
Cancel,
|
||
Status,
|
||
Active,
|
||
Result,
|
||
Error,
|
||
}
|
||
#[derive(Debug)]
|
||
/// install 安装到 Task 类型的原生方法包装；VM 绑定 receiver 后调用，不为每个 Task 单独创建闭包。
|
||
struct Method {
|
||
// 包含隐式 Task receiver 的签名，查询只读，cancel 修改任务与 owner/key 字典。
|
||
signature: endor_core::CallSignature,
|
||
// 五个 API 共用 receiver/资格检查后，各按不同状态可读性规则返回结果。
|
||
query: Query,
|
||
// eligible 的跨窗口注册表检查使用所属 VM 主资源域，不能以 current 窗口替代。
|
||
desktop: ObjectRef,
|
||
}
|
||
impl NativeCallable for Method {
|
||
/// 给出参数与返回值的托管类型契约；实际操作在 invoke 中执行。
|
||
fn signature(&self) -> Option<endor_core::CallSignature> {
|
||
Some(self.signature.clone())
|
||
}
|
||
/// Task 方法无显式参数；底层调用包含唯一一个隐式 Task 接收者。
|
||
fn arity(&self) -> Option<endor_core::CallArity> {
|
||
Some(endor_core::CallArity::exact(1))
|
||
}
|
||
|
||
/// VM 执行 task.status/active/result/error/cancel 时调用，重新核验活任务资格，查询不因失效偷偷改 payload。
|
||
fn invoke(&self, cx: &mut NativeContext<'_>, args: &[Value]) -> Result<NativeResult, VmError> {
|
||
// 方法无显式参数，实际 args 唯一一项是隐式 Task 接收者。
|
||
let [value] = args else {
|
||
return Err(invalid("Task methods accept no arguments"));
|
||
};
|
||
// 解码 receiver 后验证真实 Task payload，普通对象无法靠同名方法字段伪造任务。
|
||
let reference = object(value)?;
|
||
// 状态副本用于 cancel 的写屏障提交；只读查询保留原 Task 不变。
|
||
let mut task = cx.store().payload::<Task>(reference, "Task")?.clone();
|
||
// pending 这里同时涵盖 Running 协程，不等同于严格的 State::Pending 枚举值。
|
||
let pending = matches!(task.state, State::Pending | State::Running);
|
||
// 只有活任务才需验证 owner；终态结果不因 owner 后续销毁而突然不可读。
|
||
let owner = if pending {
|
||
cx.upgrade(task.owner)?
|
||
} else {
|
||
None
|
||
};
|
||
// active 同时要求展示资格和当前 key 精确身份，单看 Task.state 会漏掉尚未轮询清理的旧项。
|
||
let active = if let Some(owner) = owner {
|
||
eligible(cx.store(), owner, self.desktop)?
|
||
&& current(cx.store(), owner, reference, &task)?
|
||
} else {
|
||
false
|
||
};
|
||
// 各方法返回类型由 install 签名对应；result 的可用性不同于 active，不能共享一个布尔快路。
|
||
let value = match self.query {
|
||
// pending/running 但已失去 owner/key 资格时，对源报告 cancelled；只读查询不因此改写 payload。
|
||
Query::Status => Value::Text(
|
||
if pending && !active {
|
||
State::Cancelled.name()
|
||
} else {
|
||
task.state.name()
|
||
}
|
||
.into(),
|
||
),
|
||
Query::Active => Value::Bool(active),
|
||
// I/O 回调执行期间可读取已保存 text；协作任务必须真正 Completed 才能读最终 result，错误任务不返回成功值。
|
||
Query::Result => {
|
||
if !matches!(task.state, State::Running | State::Completed)
|
||
|| task.error.is_some()
|
||
|| (task.cooperative && task.state != State::Completed)
|
||
{
|
||
// 协程未完成不能返回半成品；I/O 错误也不能把 text/Nil 当成功返回。
|
||
return Err(invalid(
|
||
"task result is not available; inspect status and error",
|
||
));
|
||
}
|
||
task.result.clone().unwrap_or_else(|| {
|
||
task.text
|
||
.as_ref()
|
||
.map(|text| Value::Text(text.to_string()))
|
||
.unwrap_or(Value::Nil)
|
||
})
|
||
}
|
||
Query::Error => Value::Text(
|
||
if task.state == State::Failed {
|
||
task.error.as_deref().unwrap_or_default()
|
||
} else {
|
||
""
|
||
}
|
||
.into(),
|
||
),
|
||
Query::Cancel => {
|
||
if cx.store().is_read_only() {
|
||
// 即使任务已失效也不允许在只读渲染里执行具有取消语义的方法。
|
||
return Err(invalid(
|
||
"task cancellation is forbidden during read-only rendering",
|
||
));
|
||
}
|
||
if pending {
|
||
task.state = State::Cancelled;
|
||
task.callback = None;
|
||
// 取消立即放弃 I/O 租约与执行边，下面也清掉 text/result；不保留可交付的成功结果。
|
||
task.lease = None;
|
||
task.coroutine = None;
|
||
task.waiting = None;
|
||
task.text = None;
|
||
task.result = None;
|
||
cx.replace_payload(reference, task.clone())?;
|
||
if let Some(owner) = owner
|
||
&& current(cx.store(), owner, reference, &task)?
|
||
&& let Some(scope) = scope(cx.store(), owner)?
|
||
{
|
||
// 必须仍是当前 key 的同一 Task 才删 owner 边，旧 token.cancel 不能删掉替换后的新任务。
|
||
cx.dict_delete(scope, &Value::Text(task.key))?;
|
||
}
|
||
}
|
||
// 返回本次是否取消了仍有效的待执行任务；重复 cancel 或已失去资格的任务返回 false。
|
||
Value::Bool(pending && active)
|
||
}
|
||
};
|
||
Ok(NativeResult::Return(value))
|
||
}
|
||
}
|
||
|
||
/// bootstrap 返回给 UiBridge 的每 VM 调度能力，endor_host 轮询/退出时使用，原生字段不持有 Task 强根。
|
||
pub struct Scheduler {
|
||
// 弱项顺序也是轮转顺序，实际推进过的任务在 poll 尾部移到末尾。
|
||
index: Index,
|
||
// worker 等待策略读取此标志；只表示可能存在可运行协程，不保证下次一定完成一个任务。
|
||
runnable: Arc<AtomicBool>,
|
||
// 退出后不可重新打开，新任务 API 与本调度器共享门闩。
|
||
quiescing: Arc<AtomicBool>,
|
||
// 跨窗口 owner 生命周期核验的主注册域，不持 GUI 窗口指针。
|
||
desktop: ObjectRef,
|
||
// 后台只提交 Completion 到邮箱，本调度器才有权在 VM 中写入 Task 与执行回调。
|
||
provider: crate::io_tasks::Provider,
|
||
}
|
||
impl Scheduler {
|
||
/// UiBridge::shutdown 调用，先关新任务入口再取消所有活任务，允许退出钩子执行但不能重建工作。
|
||
pub fn quiesce(&self, vm: &mut Vm) -> Result<usize> {
|
||
// 先设置不可逆的退出门闩再取消全部任务，防止清理回调间隙继续接收新工作。
|
||
self.quiescing.store(true, Ordering::Release);
|
||
self.cancel_all(vm)
|
||
}
|
||
/// UiBridge/endor_host 的等待策略查询，不扫描 VM 堆，仅读取上次创建/prune 得到的可运行标志。
|
||
pub fn has_cooperative_work(&self) -> bool {
|
||
self.runnable.load(Ordering::Acquire)
|
||
}
|
||
/// UiBridge::poll_tasks 由 endor_host 调用，在 owning VM 内交付邮箱、推进至多 limit 项工作并更新公平索引。
|
||
/// 不在空闲轮询中扫描整堆；不可达 owner/回调环由外层变更边界的 collect_cycles 清理。
|
||
pub fn poll(&self, vm: &mut Vm, limit: usize) -> Result<usize> {
|
||
// limit 是本轮工作预算而非毫秒超时，零或超过 MAX_READY 都不能进入调度。
|
||
anyhow::ensure!((1..=MAX_READY).contains(&limit), "invalid task poll limit");
|
||
// provider 只产纯数据，VM worker 此处重新核对任务 pending、租约仍在、owner 有效且 key 仍当前。
|
||
for (weak, completion) in self.provider.take_ready()? {
|
||
if let Some(reference) = vm.upgrade(weak)? {
|
||
// 保存邮箱交付 Result，外层先释放 upgrade 得到的 Task 根，再传播任意中途错误。
|
||
let result = (|| -> Result<()> {
|
||
// 任务可能在后台运行时已被源代码取消/替换，不能直接把 Completion 写进旧副本。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
if task.state != State::Pending || task.lease.is_none() {
|
||
// 已终态或放弃租约的晚到结果直接忽略，不设置 ready 把旧任务重新变为可执行。
|
||
return Ok(());
|
||
}
|
||
// 临时拥有 owner 跨过资格检查，不能在完成邮箱里保存一个长期 owner 根。
|
||
let owner = vm.upgrade(task.owner)?;
|
||
// 即使 eligible/current 报错，也先走下面 owner 根释放路径。
|
||
let active = (|| -> Result<bool> {
|
||
Ok(if let Some(owner) = owner {
|
||
eligible(vm.object_store(), owner, self.desktop)?
|
||
&& current(vm.object_store(), owner, reference, &task)?
|
||
} else {
|
||
false
|
||
})
|
||
})();
|
||
if let Some(owner) = owner {
|
||
vm.release(owner)?;
|
||
}
|
||
if !active? {
|
||
// 身份/生命周期已失效，留给后续有界扫描清理，不交付回调或保存可观察成功文本。
|
||
return Ok(());
|
||
}
|
||
// 收到后台结果只设置 ready/text/error，实际终态与回调交付由下面的有界调度扫描决定。
|
||
task.ready = true;
|
||
match completion {
|
||
// provider 的纯文本结果在 VM 线程转换为 Task 字段，后台不直接改对象版本。
|
||
Ok(text) => task.text = Some(text.into()),
|
||
Err(error) => task.error = Some(error.into()),
|
||
}
|
||
vm.replace_payload(reference, task)?;
|
||
Ok(())
|
||
})();
|
||
vm.release(reference)?;
|
||
result?;
|
||
}
|
||
}
|
||
// 复制本轮弱索引后释放锁，用户回调可在 poll 期间提交任务，不应因重入同一索引锁死锁。
|
||
let snapshot = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?
|
||
.clone();
|
||
// 本轮使用同一 now 判断到期，避免长回调执行后在同轮无限吸收后来才到期的定时器。
|
||
let now = Instant::now();
|
||
// 取消无效任务也消耗工作预算，避免大量死亡 owner 使一次轮询无限延长。
|
||
let mut work = 0;
|
||
// 记录本轮实际执行的弱身份，扫描结束后移到队尾形成 round-robin，不边迭代边改原表。
|
||
let mut advanced = Vec::new();
|
||
for weak in snapshot {
|
||
if work >= limit {
|
||
// 未扫描的任务保留原索引，下一次 poll 继续有机会运行，而非丢弃余下项。
|
||
break;
|
||
}
|
||
// Task 已不可达时不复活它，失效弱项由 prune 最后清理。
|
||
let Some(reference) = vm.upgrade(weak)? else {
|
||
continue;
|
||
};
|
||
// 一项扫描的任何 VM 错误都先回到 Task 根释放，再让外层按运行时故障规则处理。
|
||
let result = (|| -> Result<()> {
|
||
// 每轮重读当前 payload，不能沿用 mailbox 阶段或上轮协程恢复前的状态副本。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
if !matches!(task.state, State::Pending | State::Running) {
|
||
// 已终态仍可被用户变量持有，但不再消耗执行量子，也不能重复执行 callback。
|
||
return Ok(());
|
||
}
|
||
// owner 弱边可能已死亡，按取消处理而不是把当前运行时当作替代 owner。
|
||
let owner = vm.upgrade(task.owner)?;
|
||
// 内层结果暂存，使任何回调/完成错误都不会跳过 owner 根释放。
|
||
let result = (|| -> Result<()> {
|
||
// owner 展示资格和同 key 当前身份都必须满足，保留旧 Task token 不能继续取得执行权。
|
||
let active = if let Some(owner) = owner {
|
||
eligible(vm.object_store(), owner, self.desktop)?
|
||
&& current(vm.object_store(), owner, reference, &task)?
|
||
} else {
|
||
false
|
||
};
|
||
// owner 回收、页面关闭或 key 被替换都使任务取消；取消也计一次 work，避免清理无限占用一轮。
|
||
if !active {
|
||
work += 1;
|
||
return finish(vm, reference, State::Cancelled);
|
||
}
|
||
// 被等待对象未就绪则不消费量子；被取消会取消等待者，失败等其他结果交回协程 await 语义处理。
|
||
if let Some(waiting) = task.waiting {
|
||
// 仅查询 AwaitPoll，不递归执行被等待对象，公平调度仍由外层轮询负责。
|
||
match vm.object(waiting)?.payload().poll_await() {
|
||
Ok(endor_core::AwaitPoll::Pending) => return Ok(()),
|
||
Err(VmError::AwaitCancelled) => {
|
||
// 被等待任务取消会取消等待者；失败则留给恢复后的 await 按语言语义抛出。
|
||
work += 1;
|
||
return finish(vm, reference, State::Cancelled);
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
if task.due > now || !task.ready {
|
||
// 时间到了不代表 I/O 已就绪，I/O 就绪也不能越过计时截止；两条件均需满足。
|
||
return Ok(());
|
||
}
|
||
work += 1;
|
||
// 记录实际推进的任务，轮末移到索引尾部，保证一回调空闲预算下仍能轮转多个协程。
|
||
advanced.push(weak);
|
||
// spawn 使用可恢复协程而非普通 call；其余 I/O/定时器回调作为一次完整有界调用执行。
|
||
if task.cooperative {
|
||
// 只推进一次软量子，完成/await/仍可运行分别由 advance_coroutine 处理。
|
||
return advance_coroutine(vm, reference);
|
||
}
|
||
if task.error.is_some() {
|
||
// provider 失败不调用成功回调，用 Failed 终态让 await/error 观察原始文本诊断。
|
||
return finish(vm, reference, State::Failed);
|
||
}
|
||
// sleep 或无回调 I/O 直接完成，结果已保存在 Task，不需为了 await 虚构空函数。
|
||
let Some(callback) = task.callback else {
|
||
return finish(vm, reference, State::Completed);
|
||
};
|
||
// provider 已完成，回调开始前归还 I/O 名额；text 暂留给回调中的 Task.result 查询。
|
||
task.lease = None;
|
||
task.coroutine = None;
|
||
task.waiting = None;
|
||
// 先发布 Running 状态再进入用户代码，回调可查询任务或取消/替换自身。
|
||
let mut running = task.clone();
|
||
running.state = State::Running;
|
||
vm.replace_payload(reference, running)?;
|
||
// 普通回调必须一次有界完成，不是 spawn 的可恢复协程；预算不因别的任务回调重入而累加到无限。
|
||
let result = vm
|
||
.call_with_limits(
|
||
callback,
|
||
&[],
|
||
endor_core::ExecutionLimits {
|
||
max_frames: 128,
|
||
max_instructions: 100_000,
|
||
},
|
||
)
|
||
.map_err(|error| crate::endor_host::runtime_error(vm, error));
|
||
match result {
|
||
Ok(value) => {
|
||
// 普通异步回调契约要求 Unit；先释放实际返回值，再把不符合契约的回调标失败并向外报告。
|
||
let unit = matches!(value, Value::Nil);
|
||
vm.release_value(value)?;
|
||
if !unit {
|
||
// 已释放非 Unit 返回根后再标记失败并报告，不能因契约错误泄漏任意对象返回值。
|
||
fail_callback(
|
||
vm,
|
||
reference,
|
||
"asynchronous callback must return Unit",
|
||
)?;
|
||
bail!("asynchronous callback must return Unit");
|
||
}
|
||
finish(vm, reference, State::Completed)
|
||
}
|
||
Err(error) => {
|
||
// 存有界诊断并清执行资源，同时把实际错误交给 endor_host 的故障处理。
|
||
fail_callback(vm, reference, &format!("{error:#}"))?;
|
||
Err(error)
|
||
}
|
||
}
|
||
})();
|
||
if let Some(owner) = owner {
|
||
vm.release(owner)?;
|
||
}
|
||
result
|
||
})();
|
||
vm.release(reference)?;
|
||
result?;
|
||
}
|
||
// 实际推进项移到末尾，空闲 limit=1 时一个长协程不能每轮都抢在其他可执行任务前。
|
||
if !advanced.is_empty() {
|
||
// 回调可能新建任务，所以重新取得当前索引再移动已推进项，不能拿旧 snapshot 覆盖新登记。
|
||
let mut index = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?;
|
||
index.retain(|weak| !advanced.contains(weak));
|
||
index.extend(advanced);
|
||
}
|
||
// 轮末移除终态/死亡任务并重算 runnable，等待外部 I/O 的任务不应导致忙轮询协程。
|
||
self.prune(vm)?;
|
||
Ok(work)
|
||
}
|
||
/// UiBridge 取消和 Scheduler::quiesce 调用，逐项 finish 并清空弱索引，不执行用户取消回调。
|
||
pub fn cancel_all(&self, vm: &mut Vm) -> Result<usize> {
|
||
// 不在锁内调用 finish；它会变更 owner 作用域，但索引不需在整个 VM 清理过程中借用。
|
||
let snapshot = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?
|
||
.clone();
|
||
for weak in snapshot {
|
||
if let Some(task) = vm.upgrade(weak)? {
|
||
// 升级是拥有根，先保存 finish 错误再 release，避免失败时遗留额外 Task 引用。
|
||
let result = finish(vm, task, State::Cancelled);
|
||
vm.release(task)?;
|
||
result?;
|
||
}
|
||
}
|
||
// VM 清理完成后再锁索引，取原项数供诊断并一次性清表。
|
||
let mut index = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?;
|
||
// cancel_all 返回清空前索引项数，包含升级失败的弱项，不能等同于实际执行过的回调数。
|
||
let count = index.len();
|
||
index.clear();
|
||
self.runnable.store(false, Ordering::Release);
|
||
Ok(count)
|
||
}
|
||
/// poll 的轮末收尾调用，移除死亡/终态弱项并计算“是否有可运行协程”，防止等待 I/O 时忙循环。
|
||
fn prune(&self, vm: &mut Vm) -> Result<()> {
|
||
// 本函数不执行用户函数，短暂持锁扫描弱项及查询 AwaitPoll，不需把整个 VM 堆遍历一遍。
|
||
let mut index = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?;
|
||
// live 保留待处理状态任务，终态由源变量自己持有，不因调度索引延长寿命。
|
||
let mut live = Vec::new();
|
||
// 累加是否有任意协程不用等待外部事件即可推进，不把普通未来计时器视为 runnable。
|
||
let mut runnable = false;
|
||
for weak in index.iter().copied() {
|
||
if let Some(task) = vm.upgrade(weak)? {
|
||
// 先保存 payload 查询 Result，让升级临时根在返回错误之前也能被 release。
|
||
let pending = vm.payload::<Task>(task, "Task").map(|task| {
|
||
// 此局部 pending 是状态布尔，与外层承载查询结果的 Result 区分。
|
||
let pending = matches!(task.state, State::Pending | State::Running);
|
||
// 可运行标记仅针对协作任务：无等待对象可立即推进，有等待对象则看其 poll_await 是否不再 Pending。
|
||
runnable |= pending && task.cooperative && task.waiting.is_none();
|
||
pending
|
||
});
|
||
if pending.as_ref().is_ok_and(|pending| *pending) {
|
||
// 只对成功读出的活任务继续看 waiting，失败则留给释放后的 pending? 传播。
|
||
let value = vm.payload::<Task>(task, "Task")?;
|
||
if value.cooperative
|
||
&& let Some(waiting) = value.waiting
|
||
{
|
||
// 等待值已完成/失败/取消都可以恢复处理；只有明确 Pending 才继续休眠。
|
||
runnable |= !matches!(
|
||
vm.object(waiting)?.payload().poll_await(),
|
||
Ok(endor_core::AwaitPoll::Pending)
|
||
);
|
||
}
|
||
}
|
||
vm.release(task)?;
|
||
if pending? {
|
||
live.push(weak);
|
||
}
|
||
}
|
||
}
|
||
*index = live;
|
||
self.runnable.store(runnable, Ordering::Release);
|
||
Ok(())
|
||
}
|
||
}
|
||
/// advance_coroutine 准备保存 await 依赖前调用，沿 Task.waiting 判断自等/互等，非 Task 值由其自身 await 协议处理。
|
||
fn cyclic_wait(vm: &Vm, source: ObjectRef, mut target: ObjectRef) -> Result<bool> {
|
||
// seen 记录目标身份，既识别回到 source，也识别原本已有的封闭依赖环。
|
||
let mut seen = std::collections::HashSet::new();
|
||
loop {
|
||
// 等待链到达 source 或重复已有目标都表示环；不是 Task 或没有 waiting 边时链终止，不判环。
|
||
if target == source || !seen.insert(target) {
|
||
return Ok(true);
|
||
}
|
||
// 非本模块 Task 的 awaitable 没有可遍历 waiting 边，不能臆测其内部依赖结构。
|
||
let Some(task) = vm.object(target)?.payload().as_any().downcast_ref::<Task>() else {
|
||
return Ok(false);
|
||
};
|
||
// 一个尚未等待其他对象的任务是链尾，当前这次新增等待不会自行形成环。
|
||
let Some(waiting) = task.waiting else {
|
||
return Ok(false);
|
||
};
|
||
target = waiting;
|
||
}
|
||
}
|
||
|
||
/// Scheduler::poll 只为 cooperative Task 调用，首次创建协程后每轮恢复 256 指令软量子，保留累计硬预算。
|
||
fn advance_coroutine(vm: &mut Vm, reference: ObjectRef) -> Result<()> {
|
||
// 保存初始执行边用于第一次准备，运行源代码后必须重读状态，不能让旧副本覆盖自取消结果。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
// 首次推进把 callback 转成托管 Coroutine，后续复用同一帧与预算，不重放已发生的源副作用。
|
||
let coroutine = if let Some(coroutine) = task.coroutine {
|
||
// 已有协程直接继续原帧，不从 callback 起点回放写属性/I/O 等副作用。
|
||
coroutine
|
||
} else {
|
||
// 只有第一次推进才取 callback，成功保存协程后会清 callback 强边避免重复准备。
|
||
let callback = task
|
||
.callback
|
||
.ok_or_else(|| anyhow::anyhow!("spawn callback missing"))?;
|
||
// 协程自身保存硬预算，软量子每轮刷新也不会让总指令限制重新从零计算。
|
||
let prepared = vm
|
||
.new_coroutine(
|
||
callback,
|
||
&[],
|
||
endor_core::ExecutionLimits {
|
||
max_frames: 128,
|
||
max_instructions: 100_000,
|
||
},
|
||
)
|
||
.map_err(|error| crate::endor_host::runtime_error(vm, error));
|
||
// 准备可能因回调/类型/预算错误失败，统一标记任务失败并保留源诊断。
|
||
let coroutine = match prepared {
|
||
Ok(coroutine) => coroutine,
|
||
Err(error) => {
|
||
fail_callback(vm, reference, &format!("{error:#}"))?;
|
||
return Err(error);
|
||
}
|
||
};
|
||
// 准备 coroutine 期间原生代码也可能取消/替换任务，必须重读当前状态，不能把旧 task 副本覆盖回去。
|
||
// 看真实当前 payload 的状态，而不是此函数入口保存的 task.state。
|
||
let current = vm.payload::<Task>(reference, "Task")?.state;
|
||
if !matches!(current, State::Pending | State::Running) {
|
||
// 已失去执行资格，刚准备出的协程返回根仍需释放，不能写回 Task 复活它。
|
||
vm.release(coroutine)?;
|
||
return Ok(());
|
||
}
|
||
task.state = State::Running;
|
||
task.callback = None;
|
||
task.coroutine = Some(coroutine);
|
||
// 先由 payload 建立协程拥有边，再释放创建根；安装失败也尝试释放局部根。
|
||
// 先暂存安装结果，让创建根不管写屏障成功/失败都经过下面 release。
|
||
let installed = vm.replace_payload(reference, task);
|
||
vm.release(coroutine)?;
|
||
installed?;
|
||
coroutine
|
||
};
|
||
// 续体结果先保存，运行函数可能主动取消/替换 Task，交付前还要核验真实状态。
|
||
let result = vm
|
||
// 每次最多执行 256 指令软量子，达到量子时仅挂起；总 100000 指令上限仍由协程本身累计。
|
||
.resume_coroutine(coroutine, 256)
|
||
.map_err(|error| crate::endor_host::runtime_error(vm, error));
|
||
// 源代码运行期间可取消自身，取消/替换优先于本轮完成；完成值即使不交付也必须释放其拥有根。
|
||
let active = matches!(
|
||
vm.payload::<Task>(reference, "Task")?.state,
|
||
State::Pending | State::Running
|
||
);
|
||
match result {
|
||
Ok(endor_core::CoroutineProgress::Waiting { awaitable, .. }) => {
|
||
if !active {
|
||
// 等待来自已取消旧执行时不再登记 Task.waiting，让取消/替换优先。
|
||
return Ok(());
|
||
}
|
||
// 登记等待强边前检查依赖环，拒绝永不满足的互相 await，并把当前任务转失败。
|
||
if cyclic_wait(vm, reference, awaitable)? {
|
||
fail_callback(vm, reference, "cyclic task await")?;
|
||
bail!("cyclic task await");
|
||
}
|
||
// 保存强等待边，让被等待值在协程挂起期间真实可达，而不是靠弱调度表保活。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
task.waiting = Some(awaitable);
|
||
vm.replace_payload(reference, task)?;
|
||
Ok(())
|
||
}
|
||
// 软量子挂起不代表等待外部值，清 waiting 后下一轮仍视作可运行。
|
||
Ok(endor_core::CoroutineProgress::Pending { .. }) => {
|
||
if active {
|
||
// 本片量子用尽但并未 await，清旧 waiting 让 prune 把它判为下一轮可运行。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
task.waiting = None;
|
||
vm.replace_payload(reference, task)?;
|
||
}
|
||
Ok(())
|
||
}
|
||
Ok(endor_core::CoroutineProgress::Complete { value, .. }) => {
|
||
if !active {
|
||
// 完成值来自 VM 返回拥有根；即使不再交付也不能直接 drop Rust Value 而不 release。
|
||
vm.release_value(value)?;
|
||
return Ok(());
|
||
}
|
||
// 重新读可能被源函数更新过的字段，仅附加本次成功结果，避免入口旧副本覆盖新状态。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
// 结果先写入受 trace 管理的 payload，再释放 VM 返回值根，最后 finish 清执行边但保留成功结果。
|
||
task.result = Some(value.clone());
|
||
task.waiting = None;
|
||
// 结果对象先由 Task.trace 保活；局部返回根随后释放，即使存储失败也不跳过释放。
|
||
let stored = vm.replace_payload(reference, task);
|
||
vm.release_value(value)?;
|
||
stored?;
|
||
finish(vm, reference, State::Completed)
|
||
}
|
||
Err(error) => {
|
||
if active {
|
||
// 已终态任务不再改成 Failed，保持自取消/替换结果；实际执行错误仍返回上层。
|
||
fail_callback(vm, reference, &format!("{error:#}"))?;
|
||
}
|
||
Err(error)
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Scheduler::poll/advance_coroutine 的回调错误分支调用，保存有界诊断，再经 finish 统一清理执行资源。
|
||
fn fail_callback(vm: &mut Vm, reference: ObjectRef, message: &str) -> Result<()> {
|
||
// 只更新 error 字段，不自行复制 Task 结果或重新分配一份故障任务对象。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
// 错误预览按字符限长，保证 UTF-8 完整；记录后统一走终态清理而非保留失败协程。
|
||
task.error = Some(message.chars().take(1024).collect::<String>().into());
|
||
vm.replace_payload(reference, task)?;
|
||
finish(vm, reference, State::Failed)
|
||
}
|
||
|
||
/// poll、cancel_all、协程收尾和 fail_callback 共用的终态转换，清执行边并只撤销仍匹配的 owner/key 拥有关系。
|
||
fn finish(vm: &mut Vm, reference: ObjectRef, state: State) -> Result<()> {
|
||
// 先核对当前真实状态，已完成/失败/取消不再被另一个迟到的收尾覆盖。
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
if !matches!(task.state, State::Pending | State::Running) {
|
||
return Ok(());
|
||
}
|
||
// finish 只允许从 pending/running 转终态，重复调用不改变既有 Completed/Cancelled/Failed 的语义。
|
||
task.state = state;
|
||
task.callback = None;
|
||
task.lease = None;
|
||
task.coroutine = None;
|
||
task.waiting = None;
|
||
if state == State::Cancelled {
|
||
// 取消不保存成功结果；Completed/Failed 的诊断/结果由保留 Task token 的源代码继续观察。
|
||
task.text = None;
|
||
task.result = None;
|
||
}
|
||
vm.replace_payload(reference, task.clone())?;
|
||
if let Some(owner) = vm.upgrade(task.owner)? {
|
||
// owner 可能已经回收，无 owner 时无需删表；升级成功则先保留清理结果再 release。
|
||
let result = (|| -> Result<()> {
|
||
// 只有 key 仍指向本任务才删 owner 边，避免旧任务结束时误删同 key 新任务。
|
||
if current(vm.object_store(), owner, reference, &task)?
|
||
&& let Some(scope) = scope(vm.object_store(), owner)?
|
||
{
|
||
vm.dict_delete(scope, &Value::Text(task.key))?;
|
||
}
|
||
Ok(())
|
||
})();
|
||
vm.release(owner)?;
|
||
result?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// ui_bridge::bootstrap 在 Task 类型建立后调用，注册五种创建 API 和 Task 实例方法，返回同域 Scheduler。
|
||
pub fn install(
|
||
vm: &mut Vm,
|
||
roots: &mut Vec<ObjectRef>,
|
||
ui: ObjectRef,
|
||
desktop: ObjectRef,
|
||
) -> Result<Scheduler> {
|
||
// 每个运行时一份弱索引、可运行/退出标记和 provider mailbox，UI API 与 Scheduler 共享这些原生状态。
|
||
let index = Arc::new(Mutex::new(Vec::new()));
|
||
// 初始化没有可推进协程，spawn 及轮末 prune 才会更新这个唤醒提示。
|
||
let runnable = Arc::new(AtomicBool::new(false));
|
||
// 运行时退出单向门闩共享给所有 After callable，不能只在 Scheduler 自己保留一个无效本地副本。
|
||
let quiescing = Arc::new(AtomicBool::new(false));
|
||
// 所有 API 返回同一 Task 名义类型，不能为 after/sleep 各建一个不兼容 Task 类。
|
||
let definition = desktop_type(vm.object_store(), "Task", desktop)?;
|
||
// 每运行时独立完成邮箱/租约额度，工作线程池由 Provider 内部全进程共用。
|
||
let provider = crate::io_tasks::Provider::default();
|
||
for (name, operation) in [
|
||
("after", Operation::Timer),
|
||
("sleep", Operation::Sleep),
|
||
("readText", Operation::ReadText),
|
||
("fetchText", Operation::FetchText),
|
||
("spawn", Operation::Spawn),
|
||
] {
|
||
// I/O 第三参数为字符串并声明 External；其余初始模板用毫秒 Number/Mutating。
|
||
let external = matches!(operation, Operation::ReadText | Operation::FetchText);
|
||
// After 共享索引/门闩/Provider，只有 operation 与签名不同，仍保留各源 API 独立名字。
|
||
let function = vm.allocate(ContextNativeFunction::new(
|
||
name,
|
||
After {
|
||
operation,
|
||
provider: provider.clone(),
|
||
signature: {
|
||
// 基础四参模板为 owner、key、delay/path、callback，随后针对无回调/零延时 API 收窄。
|
||
let mut signature = endor_core::CallSignature::exact(
|
||
vec![
|
||
endor_core::AnnotationType::Any,
|
||
endor_core::AnnotationType::String,
|
||
if external {
|
||
// readText/fetchText 使用路径/URL，不能让 API 签名把它误认作计时延迟。
|
||
endor_core::AnnotationType::String
|
||
} else {
|
||
// Timer/Sleep 使用整数毫秒，整数性与范围由 After::invoke 再查。
|
||
endor_core::AnnotationType::Number
|
||
},
|
||
endor_core::AnnotationType::Function,
|
||
],
|
||
endor_core::AnnotationType::Nominal(definition),
|
||
if external {
|
||
endor_core::CallEffect::External
|
||
} else {
|
||
endor_core::CallEffect::Mutating
|
||
},
|
||
);
|
||
// I/O 可带可选回调；spawn 去掉延时参数，sleep 去掉回调参数，保留各自正确 arity。
|
||
if external {
|
||
// I/O 允许不传回调并直接 await 返回 Task，但 owner/key/path 三参始终必填。
|
||
signature.arity = endor_core::CallArity::range(3, 4);
|
||
}
|
||
if operation == Operation::Spawn {
|
||
// 第三位改为源函数，去掉延时注解而非只把 required 数量改成 3。
|
||
signature.parameters.remove(2);
|
||
signature.arity = endor_core::CallArity::exact(3);
|
||
}
|
||
if operation == Operation::Sleep {
|
||
// sleep 无回调，不能把三参 API 的注解末尾遗留成任意函数尾参。
|
||
signature.parameters.pop();
|
||
signature.arity = endor_core::CallArity::exact(3);
|
||
}
|
||
signature
|
||
},
|
||
index: index.clone(),
|
||
runnable: runnable.clone(),
|
||
quiescing: quiescing.clone(),
|
||
desktop,
|
||
},
|
||
vec![desktop],
|
||
))?;
|
||
roots.push(function);
|
||
vm.set_attr(ui, name, Value::Object(function))?;
|
||
}
|
||
// Task 类型安装 cancel/status/active/result/error：只有 cancel 声明 Mutating，其余只读查询。
|
||
for (name, query) in [
|
||
("cancel", Query::Cancel),
|
||
("status", Query::Status),
|
||
("active", Query::Active),
|
||
("result", Query::Result),
|
||
("error", Query::Error),
|
||
] {
|
||
// 方法存在 Task 类型表上，receiver 由 VM 绑定；返回值契约按 query 区分。
|
||
let method = vm.allocate(ContextNativeFunction::new(
|
||
name,
|
||
Method {
|
||
query,
|
||
desktop,
|
||
signature: endor_core::CallSignature::exact(
|
||
vec![endor_core::AnnotationType::Nominal(definition)],
|
||
match query {
|
||
// status/error 返回文本，result 保留任意源值，cancel/active 则返回布尔。
|
||
Query::Status | Query::Error => endor_core::AnnotationType::String,
|
||
Query::Result => endor_core::AnnotationType::Any,
|
||
_ => endor_core::AnnotationType::Bool,
|
||
},
|
||
if matches!(query, Query::Cancel) {
|
||
// 查询和取消的效果不同，不能让 cancel 在只读布局里改变任务资格。
|
||
endor_core::CallEffect::Mutating
|
||
} else {
|
||
endor_core::CallEffect::ReadOnly
|
||
},
|
||
),
|
||
},
|
||
vec![desktop],
|
||
))?;
|
||
roots.push(method);
|
||
vm.define_method(definition, name, method)?;
|
||
}
|
||
// 返回与全部 After callable 相同的共享状态，UiBridge 后续通过它轮询/退出本 VM 的任务。
|
||
Ok(Scheduler {
|
||
index,
|
||
runnable,
|
||
quiescing,
|
||
desktop,
|
||
provider,
|
||
})
|
||
}
|