1030 lines
38 KiB
Rust
1030 lines
38 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,
|
|
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};
|
|
|
|
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,
|
|
ready: bool,
|
|
text: Option<Arc<str>>,
|
|
error: Option<Arc<str>>,
|
|
lease: Option<Arc<crate::io_tasks::Lease>>,
|
|
cooperative: bool,
|
|
coroutine: Option<ObjectRef>,
|
|
waiting: Option<ObjectRef>,
|
|
result: Option<Value>,
|
|
}
|
|
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 poll_await(&self) -> Result<endor_core::AwaitPoll, VmError> {
|
|
match self.state {
|
|
State::Pending | State::Running => Ok(endor_core::AwaitPoll::Pending),
|
|
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(),
|
|
)),
|
|
}
|
|
}
|
|
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)
|
|
}
|
|
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 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)]
|
|
enum Operation {
|
|
Timer,
|
|
Sleep,
|
|
ReadText,
|
|
FetchText,
|
|
Spawn,
|
|
}
|
|
#[derive(Debug)]
|
|
struct After {
|
|
operation: Operation,
|
|
provider: crate::io_tasks::Provider,
|
|
signature: endor_core::CallSignature,
|
|
index: Index,
|
|
runnable: Arc<AtomicBool>,
|
|
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(self.signature.arity)
|
|
}
|
|
|
|
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 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()?;
|
|
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"));
|
|
};
|
|
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",
|
|
));
|
|
};
|
|
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"));
|
|
}
|
|
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)?
|
|
&& 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"));
|
|
}
|
|
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 {
|
|
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)?;
|
|
}
|
|
|
|
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 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)?;
|
|
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)]
|
|
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 = 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 {
|
|
Query::Status => Value::Text(
|
|
if pending && !active {
|
|
State::Cancelled.name()
|
|
} else {
|
|
task.state.name()
|
|
}
|
|
.into(),
|
|
),
|
|
Query::Active => Value::Bool(active),
|
|
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;
|
|
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))?;
|
|
}
|
|
}
|
|
Value::Bool(pending && active)
|
|
}
|
|
};
|
|
Ok(NativeResult::Return(value))
|
|
}
|
|
}
|
|
|
|
pub struct Scheduler {
|
|
index: Index,
|
|
runnable: Arc<AtomicBool>,
|
|
desktop: ObjectRef,
|
|
provider: crate::io_tasks::Provider,
|
|
}
|
|
impl Scheduler {
|
|
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.
|
|
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.
|
|
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(());
|
|
}
|
|
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();
|
|
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
|
|
};
|
|
if !active {
|
|
work += 1;
|
|
return finish(vm, reference, State::Cancelled);
|
|
}
|
|
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);
|
|
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) => {
|
|
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);
|
|
}
|
|
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();
|
|
self.runnable.store(false, Ordering::Release);
|
|
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();
|
|
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);
|
|
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(())
|
|
}
|
|
}
|
|
fn cyclic_wait(vm: &Vm, source: ObjectRef, mut target: ObjectRef) -> Result<bool> {
|
|
let mut seen = std::collections::HashSet::new();
|
|
loop {
|
|
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;
|
|
}
|
|
}
|
|
|
|
fn advance_coroutine(vm: &mut Vm, reference: ObjectRef) -> Result<()> {
|
|
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
|
|
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.
|
|
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);
|
|
let installed = vm.replace_payload(reference, task);
|
|
vm.release(coroutine)?;
|
|
installed?;
|
|
coroutine
|
|
};
|
|
let result = vm
|
|
.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(());
|
|
}
|
|
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(())
|
|
}
|
|
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();
|
|
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();
|
|
task.error = Some(message.chars().take(1024).collect::<String>().into());
|
|
vm.replace_payload(reference, task)?;
|
|
finish(vm, reference, State::Failed)
|
|
}
|
|
|
|
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;
|
|
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<()> {
|
|
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 runnable = 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
|
|
},
|
|
);
|
|
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(),
|
|
desktop,
|
|
},
|
|
vec![desktop],
|
|
))?;
|
|
roots.push(function);
|
|
vm.set_attr(ui, name, Value::Object(function))?;
|
|
}
|
|
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,
|
|
desktop,
|
|
provider,
|
|
})
|
|
}
|