1141 lines
49 KiB
Rust
1141 lines
49 KiB
Rust
//! owner/key 管理的异步任务与协作式源函数调度。弱索引用于查找，回调、等待对象和协程帧是托管图中的真实边。
|
||
//! 任务可能处于等待 I/O、定时器、可执行协程或等待另一任务的状态；取消 owner/key 不能让晚到结果再次执行。
|
||
//! 调度使用有界量子和公平推进；await 挂起后保留操作数与帧，不重放此前副作用。失败和依赖环有明确处理路径。
|
||
//!
|
||
//! Owner-managed asynchronous work. Only weak identities enter the scheduler;
|
||
//! callbacks and scopes are ordinary traced edges in the Aster object graph.
|
||
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";
|
||
const MAX_TASKS: usize = 1024;
|
||
const MAX_OWNER_TASKS: usize = 64;
|
||
const MAX_READY: usize = 64;
|
||
|
||
// 调度器只保存弱任务身份，不因排队把任务、回调或 owner 隐式保活。
|
||
type Index = Arc<Mutex<Vec<WeakObjectRef>>>;
|
||
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
/// 任务对源代码可见的生命周期状态，完成/取消/失败各有不同 await 语义。
|
||
enum State {
|
||
Pending,
|
||
Running,
|
||
Completed,
|
||
Cancelled,
|
||
Failed,
|
||
}
|
||
impl State {
|
||
/// 返回内部枚举或对象对应的稳定源名称。
|
||
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,
|
||
key: String,
|
||
// 单调时钟截止点；ready 与到期是两个独立条件，I/O 完成不受系统日期调整影响。
|
||
due: Instant,
|
||
callback: Option<ObjectRef>,
|
||
state: State,
|
||
// 定时/协作任务创建即 ready，但仍等 due；I/O 任务等 provider 结果写回后才 ready。
|
||
ready: bool,
|
||
text: Option<Arc<str>>,
|
||
error: Option<Arc<str>>,
|
||
// 仅任务 payload 持有 I/O 租约；取消/终态清除此边即可撤销后台交付并归还额度。
|
||
lease: Option<Arc<crate::io_tasks::Lease>>,
|
||
cooperative: bool,
|
||
coroutine: Option<ObjectRef>,
|
||
// 协程 await 的对象是真实托管强边；等待期间由它保活被等待值，不以弱调度索引代替。
|
||
waiting: Option<ObjectRef>,
|
||
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
|
||
}
|
||
/// 把 Task 状态投影成 await 的就绪、等待、取消或失败结果，不在等待期间主动轮询执行。
|
||
fn poll_await(&self) -> Result<endor_core::AwaitPoll, VmError> {
|
||
match self.state {
|
||
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(|| {
|
||
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)) {
|
||
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);
|
||
}
|
||
}
|
||
}
|
||
/// 统一构造本模块的参数错误，让调用方停止处理无效输入。
|
||
fn invalid(message: &'static str) -> VmError {
|
||
VmError::InvalidArgument(message)
|
||
}
|
||
/// 从源值中提取并检查托管对象身份，错误类型沿 VM 错误返回。
|
||
fn object(value: &Value) -> Result<ObjectRef, VmError> {
|
||
match value {
|
||
Value::Object(object) => Ok(*object),
|
||
_ => Err(invalid("task owner and callback must be managed objects")),
|
||
}
|
||
}
|
||
/// 取得 owner 关联的托管任务作用域，供 key 替换和取消使用。
|
||
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>() {
|
||
return Ok(value.tasks);
|
||
}
|
||
if let Some(value) = payload.as_any().downcast_ref::<Mount>() {
|
||
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)) => {
|
||
// Validate even the private host field before using it as an owner edge.
|
||
store.dict_items(scope)?;
|
||
Ok(Some(scope))
|
||
}
|
||
Err(VmError::AttributeNotFound(_)) => Ok(None),
|
||
Ok(_) => Err(invalid("invalid task scope")),
|
||
Err(error) => Err(error),
|
||
}
|
||
}
|
||
/// 为可拥有任务的对象安装作用域，真实引用仍由托管图追踪。
|
||
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>() {
|
||
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>() {
|
||
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>() {
|
||
let mut value = value.clone();
|
||
value.tasks = Some(scope);
|
||
return cx.replace_payload(owner, value);
|
||
}
|
||
cx.set_attr(owner, SCOPE, Value::Object(scope))
|
||
}
|
||
/// 检查任务的 owner 仍有效且任务具备交付条件，避免晚到结果复活已关闭页面。
|
||
fn eligible(store: &ObjectStore, owner: ObjectRef, desktop: ObjectRef) -> Result<bool, VmError> {
|
||
let payload = store.object(owner)?.payload();
|
||
if let Some(mount) = payload.as_any().downcast_ref::<Mount>() {
|
||
if !mount.mounted {
|
||
return Ok(false);
|
||
}
|
||
// Mount 仅 mounted 不足以交付，所属 Page 还必须在某个活动窗口页面注册表里，关闭再打开不能恢复旧任务。
|
||
if let Some(page) = mount.page {
|
||
// No retained root is needed for this synchronous borrowed lookup.
|
||
let windows = object(&store.get_attr(desktop, "windows")?)?;
|
||
for (_, value) in store.dict_items(windows)? {
|
||
for candidate in crate::window::pages(store, object(value)?)? {
|
||
if store.downgrade(*candidate)? == page {
|
||
return Ok(true);
|
||
}
|
||
}
|
||
}
|
||
return Ok(false);
|
||
}
|
||
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)
|
||
{
|
||
let windows = object(&store.get_attr(desktop, "windows")?)?;
|
||
for (_, value) in store.dict_items(windows)? {
|
||
let window = object(value)?;
|
||
if crate::window::pages(store, window)?.contains(&owner) {
|
||
return Ok(true);
|
||
}
|
||
}
|
||
return Ok(false);
|
||
}
|
||
Ok(true)
|
||
}
|
||
/// 检查弱索引对应的任务是否仍是 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) => {
|
||
store.dict_get(scope, &Value::Text(task.key.clone()))? == Some(Value::Object(reference))
|
||
}
|
||
None => false,
|
||
})
|
||
}
|
||
|
||
/// Called at the actual close/detach boundary, including close-then-reopen in
|
||
/// one language input. A later appearance never revives these identities.
|
||
/// 取消某个 owner 的全部任务，防止页面/窗口生命周期结束后继续执行回调。
|
||
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() {
|
||
let reference = object(&value)?;
|
||
let mut task = cx.store().payload::<Task>(reference, "Task")?.clone();
|
||
if matches!(task.state, State::Pending | State::Running) {
|
||
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 {
|
||
operation: Operation,
|
||
provider: crate::io_tasks::Provider,
|
||
signature: endor_core::CallSignature,
|
||
index: Index,
|
||
runnable: Arc<AtomicBool>,
|
||
quiescing: Arc<AtomicBool>,
|
||
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)
|
||
}
|
||
|
||
/// 先拒绝退出或只读域的新工作，再解码 owner/key/提供方参数，取消同键旧任务并建立受作用域管理的新 Task。
|
||
fn invoke(&self, cx: &mut NativeContext<'_>, args: &[Value]) -> Result<NativeResult, VmError> {
|
||
// 退出域先拒绝新任务；只读渲染也禁止异步提交，即使某构造操作只分配新的任务对象。
|
||
if self.quiescing.load(Ordering::Acquire) {
|
||
return Err(invalid("runtime is exiting; new tasks are forbidden"));
|
||
}
|
||
if cx.store().is_read_only() {
|
||
return Err(invalid(
|
||
"asynchronous work is forbidden during read-only rendering",
|
||
));
|
||
}
|
||
// 统一解码不同 API 参数：spawn 用零延时与回调，sleep 无回调，I/O 第三个位置是路径/URL。
|
||
let zero = Value::Number(0_i64.into());
|
||
let (owner, key, delay, callback) = match args {
|
||
[owner, key, callback] if self.operation == Operation::Spawn => {
|
||
(owner, key, &zero, Some(callback))
|
||
}
|
||
[owner, key, delay] if self.operation == Operation::Sleep => (owner, key, delay, None),
|
||
[owner, key, delay, callback] => (owner, key, delay, Some(callback)),
|
||
[owner, key, path]
|
||
if matches!(self.operation, Operation::ReadText | Operation::FetchText) =>
|
||
{
|
||
(owner, key, path, None)
|
||
}
|
||
_ => {
|
||
return Err(invalid(
|
||
"asynchronous request requires owner, key, delay/path and optional I/O callback",
|
||
));
|
||
}
|
||
};
|
||
let owner = object(owner)?;
|
||
let callback = callback.map(object).transpose()?;
|
||
// 所有回调都按零显式参数执行，必须是 Function 且参数元数据确认接受零参，不能等执行时再碰运气。
|
||
if let Some(callback) = callback {
|
||
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"));
|
||
}
|
||
}
|
||
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"));
|
||
}
|
||
let (milliseconds, request) =
|
||
if matches!(self.operation, Operation::ReadText | Operation::FetchText) {
|
||
let Value::Text(path) = delay else {
|
||
return Err(invalid("I/O input must be a String"));
|
||
};
|
||
let request = match self.operation {
|
||
Operation::ReadText => {
|
||
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 => {
|
||
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 {
|
||
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)? {
|
||
return Err(invalid("task owner is closed or detached"));
|
||
}
|
||
// 在分配新任务前找同 key 旧身份，额度检查允许替换但不允许借其他 key 继续增加总量。
|
||
let existing_scope = scope(cx.store(), owner)?;
|
||
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
|
||
)
|
||
{
|
||
live.push(weak);
|
||
}
|
||
}
|
||
if live.len() >= MAX_TASKS && previous.is_none() {
|
||
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()
|
||
{
|
||
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| {
|
||
cx.store().payload::<Task>(previous, "Task").map(|task| {
|
||
matches!(task.state, State::Pending | State::Running)
|
||
&& task.lease.is_some()
|
||
})
|
||
})
|
||
.transpose()?
|
||
.unwrap_or(false);
|
||
let lease = self
|
||
.provider
|
||
.submit(cx.store().downgrade(task)?, request, replacing)?;
|
||
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 {
|
||
scope
|
||
} else {
|
||
let scope = cx.new_dict()?;
|
||
install_scope(cx, owner, scope)?;
|
||
scope
|
||
};
|
||
// Publish the new identity before invalidating the old result. Nothing
|
||
// external is dispatched until the worker accepts this VM generation.
|
||
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 {
|
||
// The replacement above can release the old Task's last owner edge.
|
||
// Preserve it through the replacement even without a saved token.
|
||
if cx.store().object(previous).is_ok() {
|
||
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)?;
|
||
}
|
||
}
|
||
}
|
||
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 {
|
||
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)]
|
||
/// 保存 owner、签名或操作枚举的原生方法适配器，统一由 VM 调用。
|
||
struct Method {
|
||
signature: endor_core::CallSignature,
|
||
query: Query,
|
||
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))
|
||
}
|
||
|
||
/// 验证 Task 接收者并按当前状态处理 status/active/result/error/cancel；仍活跃的任务需重新确认 owner/key 身份。
|
||
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"));
|
||
};
|
||
let reference = object(value)?;
|
||
let mut task = cx.store().payload::<Task>(reference, "Task")?.clone();
|
||
let pending = matches!(task.state, State::Pending | State::Running);
|
||
let owner = if pending {
|
||
cx.upgrade(task.owner)?
|
||
} else {
|
||
None
|
||
};
|
||
let active = if let Some(owner) = owner {
|
||
eligible(cx.store(), owner, self.desktop)?
|
||
&& current(cx.store(), owner, reference, &task)?
|
||
} else {
|
||
false
|
||
};
|
||
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)
|
||
{
|
||
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 仍留在任务内，可由回调查询。
|
||
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)?
|
||
{
|
||
cx.dict_delete(scope, &Value::Text(task.key))?;
|
||
}
|
||
}
|
||
// 返回本次是否取消了仍有效的待执行任务；重复 cancel 或已失去资格的任务返回 false。
|
||
Value::Bool(pending && active)
|
||
}
|
||
};
|
||
Ok(NativeResult::Return(value))
|
||
}
|
||
}
|
||
|
||
/// 弱任务索引、公平调度位置和 provider，索引不会单独把任务保活。
|
||
pub struct Scheduler {
|
||
index: Index,
|
||
runnable: Arc<AtomicBool>,
|
||
quiescing: Arc<AtomicBool>,
|
||
desktop: ObjectRef,
|
||
provider: crate::io_tasks::Provider,
|
||
}
|
||
impl Scheduler {
|
||
/// 停止接收该域的新任务工作，为关闭/替换运行时做准备。
|
||
pub fn quiesce(&self, vm: &mut Vm) -> Result<usize> {
|
||
// 先设置不可逆的退出门闩再取消全部任务，防止清理回调间隙继续接收新工作。
|
||
self.quiescing.store(true, Ordering::Release);
|
||
self.cancel_all(vm)
|
||
}
|
||
/// 检查是否有可推进的协程，而不是仅仅存在正在等待外部结果的任务。
|
||
pub fn has_cooperative_work(&self) -> bool {
|
||
self.runnable.load(Ordering::Acquire)
|
||
}
|
||
/// Mutation boundaries collect unreachable owner/callback cycles before
|
||
/// this weak lookup. Idle timer polling does not rescan the entire heap.
|
||
/// 在当前 VM 中有界交付就绪任务、执行回调或推进协程；开始布局后由外层刷新流程继续。
|
||
pub fn poll(&self, vm: &mut Vm, limit: usize) -> Result<usize> {
|
||
anyhow::ensure!((1..=MAX_READY).contains(&limit), "invalid task poll limit");
|
||
// Provider threads only fill the mailbox. Only this VM worker updates
|
||
// managed payloads, and delivery below rechecks owner/key generation.
|
||
// provider 只产纯数据，VM worker 此处重新核对任务 pending、租约仍在、owner 有效且 key 仍当前。
|
||
for (weak, completion) in self.provider.take_ready()? {
|
||
if let Some(reference) = vm.upgrade(weak)? {
|
||
let result = (|| -> Result<()> {
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
if task.state != State::Pending || task.lease.is_none() {
|
||
return Ok(());
|
||
}
|
||
let owner = vm.upgrade(task.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 {
|
||
Ok(text) => task.text = Some(text.into()),
|
||
Err(error) => task.error = Some(error.into()),
|
||
}
|
||
vm.replace_payload(reference, task)?;
|
||
Ok(())
|
||
})();
|
||
vm.release(reference)?;
|
||
result?;
|
||
}
|
||
}
|
||
let snapshot = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?
|
||
.clone();
|
||
// 本轮使用同一 now 判断到期，避免长回调执行后在同轮无限吸收后来才到期的定时器。
|
||
let now = Instant::now();
|
||
let mut work = 0;
|
||
let mut advanced = Vec::new();
|
||
for weak in snapshot {
|
||
if work >= limit {
|
||
break;
|
||
}
|
||
let Some(reference) = vm.upgrade(weak)? else {
|
||
continue;
|
||
};
|
||
let result = (|| -> Result<()> {
|
||
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
||
if !matches!(task.state, State::Pending | State::Running) {
|
||
return Ok(());
|
||
}
|
||
let owner = vm.upgrade(task.owner)?;
|
||
let result = (|| -> Result<()> {
|
||
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 {
|
||
match vm.object(waiting)?.payload().poll_await() {
|
||
Ok(endor_core::AwaitPoll::Pending) => return Ok(()),
|
||
Err(VmError::AwaitCancelled) => {
|
||
work += 1;
|
||
return finish(vm, reference, State::Cancelled);
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
if task.due > now || !task.ready {
|
||
return Ok(());
|
||
}
|
||
work += 1;
|
||
// 记录实际推进的任务，轮末移到索引尾部，保证一回调空闲预算下仍能轮转多个协程。
|
||
advanced.push(weak);
|
||
// spawn 使用可恢复协程而非普通 call；其余 I/O/定时器回调作为一次完整有界调用执行。
|
||
if task.cooperative {
|
||
return advance_coroutine(vm, reference);
|
||
}
|
||
if task.error.is_some() {
|
||
return finish(vm, reference, State::Failed);
|
||
}
|
||
let Some(callback) = task.callback else {
|
||
return finish(vm, reference, State::Completed);
|
||
};
|
||
// The provider is finished. Keep the result in the Task,
|
||
// but release its I/O slot before running source callbacks.
|
||
task.lease = None;
|
||
task.coroutine = None;
|
||
task.waiting = None;
|
||
let mut running = task.clone();
|
||
running.state = State::Running;
|
||
vm.replace_payload(reference, running)?;
|
||
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 {
|
||
fail_callback(
|
||
vm,
|
||
reference,
|
||
"asynchronous callback must return Unit",
|
||
)?;
|
||
bail!("asynchronous callback must return Unit");
|
||
}
|
||
finish(vm, reference, State::Completed)
|
||
}
|
||
Err(error) => {
|
||
fail_callback(vm, reference, &format!("{error:#}"))?;
|
||
Err(error)
|
||
}
|
||
}
|
||
})();
|
||
if let Some(owner) = owner {
|
||
vm.release(owner)?;
|
||
}
|
||
result
|
||
})();
|
||
vm.release(reference)?;
|
||
result?;
|
||
}
|
||
// Round-robin resumable work so a long call cannot monopolize the
|
||
// one-callback idle budget and starve other runnable Tasks.
|
||
if !advanced.is_empty() {
|
||
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)
|
||
}
|
||
/// 取消当前域全部任务，切断回调、协程和等待交付资格。
|
||
pub fn cancel_all(&self, vm: &mut Vm) -> Result<usize> {
|
||
let snapshot = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?
|
||
.clone();
|
||
for weak in snapshot {
|
||
if let Some(task) = vm.upgrade(weak)? {
|
||
let result = finish(vm, task, State::Cancelled);
|
||
vm.release(task)?;
|
||
result?;
|
||
}
|
||
}
|
||
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)
|
||
}
|
||
/// 移除已释放的弱任务索引，索引本身不让任务或其 owner 存活。
|
||
fn prune(&self, vm: &mut Vm) -> Result<()> {
|
||
let mut index = self
|
||
.index
|
||
.lock()
|
||
.map_err(|_| anyhow::anyhow!("task index poisoned"))?;
|
||
let mut live = Vec::new();
|
||
let mut runnable = false;
|
||
for weak in index.iter().copied() {
|
||
if let Some(task) = vm.upgrade(weak)? {
|
||
let pending = vm.payload::<Task>(task, "Task").map(|task| {
|
||
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) {
|
||
let value = vm.payload::<Task>(task, "Task")?;
|
||
if value.cooperative
|
||
&& let Some(waiting) = value.waiting
|
||
{
|
||
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(())
|
||
}
|
||
}
|
||
/// 沿等待链检查任务依赖环，防止 await 永远无法被满足。
|
||
fn cyclic_wait(vm: &Vm, source: ObjectRef, mut target: ObjectRef) -> Result<bool> {
|
||
let mut seen = std::collections::HashSet::new();
|
||
loop {
|
||
// 等待链到达 source 或重复已有目标都表示环；不是 Task 或没有 waiting 边时链终止，不判环。
|
||
if target == source || !seen.insert(target) {
|
||
return Ok(true);
|
||
}
|
||
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;
|
||
}
|
||
}
|
||
|
||
/// 恢复已保存的源函数帧，处理完成、软量子挂起或 await 等待，保留已发生的副作用。
|
||
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 {
|
||
coroutine
|
||
} else {
|
||
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);
|
||
}
|
||
};
|
||
// Native preparation can itself cancel or replace this Task.
|
||
// 准备 coroutine 期间原生代码也可能取消/替换任务，必须重读当前状态，不能把旧 task 副本覆盖回去。
|
||
let current = vm.payload::<Task>(reference, "Task")?.state;
|
||
if !matches!(current, State::Pending | State::Running) {
|
||
vm.release(coroutine)?;
|
||
return Ok(());
|
||
}
|
||
task.state = State::Running;
|
||
task.callback = None;
|
||
task.coroutine = Some(coroutine);
|
||
// 先由 payload 建立协程拥有边，再释放创建根；安装失败也尝试释放局部根。
|
||
let installed = vm.replace_payload(reference, task);
|
||
vm.release(coroutine)?;
|
||
installed?;
|
||
coroutine
|
||
};
|
||
let result = vm
|
||
// 每次最多执行 256 指令软量子，达到量子时仅挂起；总 100000 指令上限仍由协程本身累计。
|
||
.resume_coroutine(coroutine, 256)
|
||
.map_err(|error| crate::endor_host::runtime_error(vm, error));
|
||
// Cancellation/replacement from the running function must win. A completed
|
||
// value still owns a root and must be released even when delivery is stale.
|
||
// 源代码运行期间可取消自身，取消/替换优先于本轮完成；完成值即使不交付也必须释放其拥有根。
|
||
let active = matches!(
|
||
vm.payload::<Task>(reference, "Task")?.state,
|
||
State::Pending | State::Running
|
||
);
|
||
match result {
|
||
Ok(endor_core::CoroutineProgress::Waiting { awaitable, .. }) => {
|
||
if !active {
|
||
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 {
|
||
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.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;
|
||
let stored = vm.replace_payload(reference, task);
|
||
vm.release_value(value)?;
|
||
stored?;
|
||
finish(vm, reference, State::Completed)
|
||
}
|
||
Err(error) => {
|
||
if active {
|
||
fail_callback(vm, reference, &format!("{error:#}"))?;
|
||
}
|
||
Err(error)
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 记录回调执行失败并沿运行时故障规则停止后续不安全的交付。
|
||
fn fail_callback(vm: &mut Vm, reference: ObjectRef, message: &str) -> Result<()> {
|
||
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)
|
||
}
|
||
|
||
/// 将任务转入终态并释放回调、provider 租约、协程和等待引用；取消还会清除保存的结果。
|
||
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 {
|
||
task.text = None;
|
||
task.result = None;
|
||
}
|
||
vm.replace_payload(reference, task.clone())?;
|
||
if let Some(owner) = vm.upgrade(task.owner)? {
|
||
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(())
|
||
}
|
||
|
||
/// 把本模块的类型、方法或能力注册到所属运行时上下文。
|
||
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()));
|
||
let runnable = Arc::new(AtomicBool::new(false));
|
||
let quiescing = Arc::new(AtomicBool::new(false));
|
||
let definition = desktop_type(vm.object_store(), "Task", desktop)?;
|
||
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),
|
||
] {
|
||
let external = matches!(operation, Operation::ReadText | Operation::FetchText);
|
||
let function = vm.allocate(ContextNativeFunction::new(
|
||
name,
|
||
After {
|
||
operation,
|
||
provider: provider.clone(),
|
||
signature: {
|
||
let mut signature = endor_core::CallSignature::exact(
|
||
vec![
|
||
endor_core::AnnotationType::Any,
|
||
endor_core::AnnotationType::String,
|
||
if external {
|
||
endor_core::AnnotationType::String
|
||
} else {
|
||
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 {
|
||
signature.arity = endor_core::CallArity::range(3, 4);
|
||
}
|
||
if operation == Operation::Spawn {
|
||
signature.parameters.remove(2);
|
||
signature.arity = endor_core::CallArity::exact(3);
|
||
}
|
||
if operation == Operation::Sleep {
|
||
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),
|
||
] {
|
||
let method = vm.allocate(ContextNativeFunction::new(
|
||
name,
|
||
Method {
|
||
query,
|
||
desktop,
|
||
signature: endor_core::CallSignature::exact(
|
||
vec![endor_core::AnnotationType::Nominal(definition)],
|
||
match query {
|
||
Query::Status | Query::Error => endor_core::AnnotationType::String,
|
||
Query::Result => endor_core::AnnotationType::Any,
|
||
_ => endor_core::AnnotationType::Bool,
|
||
},
|
||
if matches!(query, Query::Cancel) {
|
||
endor_core::CallEffect::Mutating
|
||
} else {
|
||
endor_core::CallEffect::ReadOnly
|
||
},
|
||
),
|
||
},
|
||
vec![desktop],
|
||
))?;
|
||
roots.push(method);
|
||
vm.define_method(definition, name, method)?;
|
||
}
|
||
Ok(Scheduler {
|
||
index,
|
||
runnable,
|
||
quiescing,
|
||
desktop,
|
||
provider,
|
||
})
|
||
}
|