614 lines
21 KiB
Rust
614 lines
21 KiB
Rust
//! 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},
|
|
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};
|
|
|
|
const SCOPE: &str = "\0tessera.tasks";
|
|
const MAX_TASKS: usize = 1024;
|
|
const MAX_OWNER_TASKS: usize = 64;
|
|
const MAX_READY: usize = 64;
|
|
|
|
type Index = Arc<Mutex<Vec<WeakObjectRef>>>;
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
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)]
|
|
struct Task {
|
|
owner: WeakObjectRef,
|
|
key: String,
|
|
due: Instant,
|
|
callback: Option<ObjectRef>,
|
|
state: State,
|
|
}
|
|
impl ObjectPayload for Task {
|
|
fn as_any(&self) -> &dyn std::any::Any {
|
|
self
|
|
}
|
|
fn kind_name(&self) -> &'static str {
|
|
"tessera-task"
|
|
}
|
|
fn reference_assignment_only(&self) -> bool {
|
|
true
|
|
}
|
|
fn trace(&self, visit: &mut dyn FnMut(ObjectRef)) {
|
|
if let Some(callback) = self.callback {
|
|
visit(callback);
|
|
}
|
|
}
|
|
}
|
|
fn invalid(message: &'static str) -> VmError {
|
|
VmError::InvalidArgument(message)
|
|
}
|
|
fn object(value: &Value) -> Result<ObjectRef, VmError> {
|
|
match value {
|
|
Value::Object(object) => Ok(*object),
|
|
_ => Err(invalid("task owner and callback must be managed objects")),
|
|
}
|
|
}
|
|
fn scope(store: &ObjectStore, owner: ObjectRef) -> Result<Option<ObjectRef>, VmError> {
|
|
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> {
|
|
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))
|
|
}
|
|
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);
|
|
}
|
|
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);
|
|
}
|
|
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)
|
|
}
|
|
fn current(
|
|
store: &ObjectStore,
|
|
owner: ObjectRef,
|
|
reference: ObjectRef,
|
|
task: &Task,
|
|
) -> Result<bool, VmError> {
|
|
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.
|
|
pub fn cancel_owner(cx: &mut NativeContext<'_>, owner: ObjectRef) -> Result<(), VmError> {
|
|
let Some(scope) = scope(cx.store(), owner)? else {
|
|
return Ok(());
|
|
};
|
|
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 task.state == State::Pending {
|
|
task.state = State::Cancelled;
|
|
task.callback = None;
|
|
cx.replace_payload(reference, task)?;
|
|
}
|
|
cx.dict_delete(scope, &key)?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
struct After {
|
|
signature: endor_core::CallSignature,
|
|
index: Index,
|
|
desktop: ObjectRef,
|
|
}
|
|
impl NativeCallable for After {
|
|
fn signature(&self) -> Option<endor_core::CallSignature> {
|
|
Some(self.signature.clone())
|
|
}
|
|
fn arity(&self) -> Option<endor_core::CallArity> {
|
|
Some(endor_core::CallArity::exact(4))
|
|
}
|
|
|
|
fn invoke(&self, cx: &mut NativeContext<'_>, args: &[Value]) -> Result<NativeResult, VmError> {
|
|
if cx.store().is_read_only() {
|
|
return Err(invalid(
|
|
"asynchronous work is forbidden during read-only rendering",
|
|
));
|
|
}
|
|
let [owner, key, delay, callback] = args else {
|
|
return Err(invalid(
|
|
"ui.after requires owner, key, milliseconds and callback",
|
|
));
|
|
};
|
|
let owner = object(owner)?;
|
|
let callback = object(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"));
|
|
};
|
|
if key.is_empty() || key.len() > 128 || key.chars().any(char::is_control) {
|
|
return Err(invalid("invalid task key"));
|
|
}
|
|
let Value::Number(delay) = delay else {
|
|
return Err(invalid(
|
|
"task delay must be an integer number of milliseconds",
|
|
));
|
|
};
|
|
let milliseconds = delay
|
|
.as_i64()
|
|
.filter(|delay| (0..=86_400_000).contains(delay))
|
|
.ok_or(invalid(
|
|
"task delay must be 0..86400000 integer milliseconds",
|
|
))?;
|
|
if !eligible(cx.store(), owner, self.desktop)? {
|
|
return Err(invalid("task owner is closed or detached"));
|
|
}
|
|
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"))?;
|
|
let mut live = Vec::new();
|
|
for weak in index.iter().copied() {
|
|
if let Some(task) = cx.upgrade(weak)?
|
|
&& cx.store().payload::<Task>(task, "Task")?.state == State::Pending
|
|
{
|
|
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"));
|
|
}
|
|
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: Some(callback),
|
|
state: State::Pending,
|
|
})?;
|
|
cx.attach_type(task, desktop_type(cx.store(), "Task", self.desktop)?)?;
|
|
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()?;
|
|
if let Some(weak) = previous_weak {
|
|
cx.upgrade(weak)?;
|
|
}
|
|
cx.dict_set(task_scope, Value::Text(key.clone()), Value::Object(task))?;
|
|
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 old.state == State::Pending {
|
|
old.state = State::Cancelled;
|
|
old.callback = None;
|
|
cx.replace_payload(previous, old)?;
|
|
}
|
|
}
|
|
}
|
|
let weak = cx.store().downgrade(task)?;
|
|
live.retain(|old| Some(*old) != previous_weak);
|
|
live.push(weak);
|
|
*index = live;
|
|
Ok(NativeResult::Return(Value::Object(task)))
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy)]
|
|
enum Query {
|
|
Cancel,
|
|
Status,
|
|
Active,
|
|
}
|
|
#[derive(Debug)]
|
|
struct Method {
|
|
signature: endor_core::CallSignature,
|
|
query: Query,
|
|
desktop: ObjectRef,
|
|
}
|
|
impl NativeCallable for Method {
|
|
fn signature(&self) -> Option<endor_core::CallSignature> {
|
|
Some(self.signature.clone())
|
|
}
|
|
fn arity(&self) -> Option<endor_core::CallArity> {
|
|
Some(endor_core::CallArity::exact(1))
|
|
}
|
|
|
|
fn invoke(&self, cx: &mut NativeContext<'_>, args: &[Value]) -> Result<NativeResult, VmError> {
|
|
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 = task.state == State::Pending;
|
|
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 {
|
|
Query::Status => Value::Text(
|
|
if pending && !active {
|
|
State::Cancelled.name()
|
|
} else {
|
|
task.state.name()
|
|
}
|
|
.into(),
|
|
),
|
|
Query::Active => Value::Bool(active),
|
|
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;
|
|
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))?;
|
|
}
|
|
}
|
|
Value::Bool(pending && active)
|
|
}
|
|
};
|
|
Ok(NativeResult::Return(value))
|
|
}
|
|
}
|
|
|
|
pub struct Scheduler {
|
|
index: Index,
|
|
desktop: ObjectRef,
|
|
}
|
|
impl Scheduler {
|
|
/// Mutation boundaries collect unreachable owner/callback cycles before
|
|
/// this weak lookup. Idle timer polling does not rescan the entire heap.
|
|
pub fn poll(&self, vm: &mut Vm, limit: usize) -> Result<usize> {
|
|
anyhow::ensure!((1..=MAX_READY).contains(&limit), "invalid task poll limit");
|
|
let snapshot = self
|
|
.index
|
|
.lock()
|
|
.map_err(|_| anyhow::anyhow!("task index poisoned"))?
|
|
.clone();
|
|
let now = Instant::now();
|
|
let mut work = 0;
|
|
for weak in snapshot {
|
|
if work >= limit {
|
|
break;
|
|
}
|
|
let Some(reference) = vm.upgrade(weak)? else {
|
|
continue;
|
|
};
|
|
let result = (|| -> Result<()> {
|
|
let task = vm.payload::<Task>(reference, "Task")?.clone();
|
|
if task.state != State::Pending {
|
|
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
|
|
};
|
|
if !active {
|
|
work += 1;
|
|
return finish(vm, reference, State::Cancelled);
|
|
}
|
|
if task.due > now {
|
|
return Ok(());
|
|
}
|
|
work += 1;
|
|
let callback = task
|
|
.callback
|
|
.ok_or_else(|| anyhow::anyhow!("pending task has no callback"))?;
|
|
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) => {
|
|
let unit = matches!(value, Value::Nil);
|
|
vm.release_value(value)?;
|
|
finish(
|
|
vm,
|
|
reference,
|
|
if unit {
|
|
State::Completed
|
|
} else {
|
|
State::Failed
|
|
},
|
|
)?;
|
|
if !unit {
|
|
bail!("asynchronous callback must return Unit");
|
|
}
|
|
Ok(())
|
|
}
|
|
Err(error) => {
|
|
finish(vm, reference, State::Failed)?;
|
|
Err(error)
|
|
}
|
|
}
|
|
})();
|
|
if let Some(owner) = owner {
|
|
vm.release(owner)?;
|
|
}
|
|
result
|
|
})();
|
|
vm.release(reference)?;
|
|
result?;
|
|
}
|
|
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"))?;
|
|
let count = index.len();
|
|
index.clear();
|
|
Ok(count)
|
|
}
|
|
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();
|
|
for weak in index.iter().copied() {
|
|
if let Some(task) = vm.upgrade(weak)? {
|
|
let pending = vm
|
|
.payload::<Task>(task, "Task")
|
|
.map(|task| task.state == State::Pending);
|
|
vm.release(task)?;
|
|
if pending? {
|
|
live.push(weak);
|
|
}
|
|
}
|
|
}
|
|
*index = live;
|
|
Ok(())
|
|
}
|
|
}
|
|
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(());
|
|
}
|
|
task.state = state;
|
|
task.callback = None;
|
|
vm.replace_payload(reference, task.clone())?;
|
|
if let Some(owner) = vm.upgrade(task.owner)? {
|
|
let result = (|| -> Result<()> {
|
|
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> {
|
|
let index = Arc::new(Mutex::new(Vec::new()));
|
|
let definition = desktop_type(vm.object_store(), "Task", desktop)?;
|
|
let after = vm.allocate(ContextNativeFunction::new(
|
|
"after",
|
|
After {
|
|
signature: endor_core::CallSignature::exact(
|
|
vec![
|
|
endor_core::AnnotationType::Any,
|
|
endor_core::AnnotationType::String,
|
|
endor_core::AnnotationType::Number,
|
|
endor_core::AnnotationType::Function,
|
|
],
|
|
endor_core::AnnotationType::Nominal(definition),
|
|
endor_core::CallEffect::Mutating,
|
|
),
|
|
index: index.clone(),
|
|
desktop,
|
|
},
|
|
vec![desktop],
|
|
))?;
|
|
roots.push(after);
|
|
vm.set_attr(ui, "after", Value::Object(after))?;
|
|
for (name, query) in [
|
|
("cancel", Query::Cancel),
|
|
("status", Query::Status),
|
|
("active", Query::Active),
|
|
] {
|
|
let method = vm.allocate(ContextNativeFunction::new(
|
|
name,
|
|
Method {
|
|
query,
|
|
desktop,
|
|
signature: endor_core::CallSignature::exact(
|
|
vec![endor_core::AnnotationType::Nominal(definition)],
|
|
if matches!(query, Query::Status) {
|
|
endor_core::AnnotationType::String
|
|
} else {
|
|
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, desktop })
|
|
}
|