Zayden-Jung/endorphin-tessera-gui
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: schedule owner-scoped source tasks with managed coroutine quanta

Browse Source
This commit is contained in:
Codex Bot
2026-10-23 20:02:25 +08:00
parent eb48ec3396
commit 485c35c641
2 changed files with 166 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tasks.rs
+165 -46
View File
@@ -51,6 +51,8 @@ struct Task {
text: Option<Arc<str>>,
error: Option<Arc<str>>,
lease: Option<Arc<crate::io_tasks::Lease>>,
cooperative: bool,
coroutine: Option<ObjectRef>,
}
impl ObjectPayload for Task {
fn as_any(&self) -> &dyn std::any::Any {
@@ -66,6 +68,9 @@ impl ObjectPayload for Task {
if let Some(callback) = self.callback {
visit(callback);
}
if let Some(coroutine) = self.coroutine {
visit(coroutine);
}
}
}
fn invalid(message: &'static str) -> VmError {
@@ -184,10 +189,11 @@ pub fn cancel_owner(cx: &mut NativeContext<'_>, owner: ObjectRef) -> Result<(),
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 {
if matches!(task.state, State::Pending | State::Running) {
task.state = State::Cancelled;
task.callback = None;
task.lease = None;
task.coroutine = None;
task.text = None;
cx.replace_payload(reference, task)?;
}
@@ -201,6 +207,7 @@ enum Operation {
Timer,
ReadText,
FetchText,
Spawn,
}
#[derive(Debug)]
struct After {
@@ -224,9 +231,17 @@ impl NativeCallable for After {
"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, callback] => (owner, key, delay, Some(callback)),
[owner, key, path] if self.operation != Operation::Timer => (owner, key, path, None),
[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",
@@ -254,42 +269,43 @@ impl NativeCallable for After {
if key.is_empty() || key.len() > 128 || key.chars().any(char::is_control) {
return Err(invalid("invalid task key"));
}
let (milliseconds, request) = if self.operation != Operation::Timer {
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"));
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())
}
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",
));
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())
}
crate::io_tasks::Request::FetchText(path.clone())
}
Operation::Timer => unreachable!(),
Operation::Timer | 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)
};
(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"));
}
@@ -307,7 +323,10 @@ impl NativeCallable for After {
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
&& matches!(
cx.store().payload::<Task>(task, "Task")?.state,
State::Pending | State::Running
)
{
live.push(weak);
}
@@ -329,18 +348,21 @@ impl NativeCallable for After {
.ok_or(invalid("task deadline overflow"))?,
callback,
state: State::Pending,
ready: self.operation == Operation::Timer,
ready: matches!(self.operation, Operation::Timer | Operation::Spawn),
text: None,
error: None,
lease: None,
cooperative: self.operation == Operation::Spawn,
coroutine: 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| task.state == State::Pending && task.lease.is_some())
cx.store().payload::<Task>(previous, "Task").map(|task| {
matches!(task.state, State::Pending | State::Running)
&& task.lease.is_some()
})
})
.transpose()?
.unwrap_or(false);
@@ -371,10 +393,11 @@ impl NativeCallable for After {
// 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 {
if matches!(old.state, State::Pending | State::Running) {
old.state = State::Cancelled;
old.callback = None;
old.lease = None;
old.coroutine = None;
old.text = None;
cx.replace_payload(previous, old)?;
}
@@ -416,7 +439,7 @@ impl NativeCallable for Method {
};
let reference = object(value)?;
let mut task = cx.store().payload::<Task>(reference, "Task")?.clone();
let pending = task.state == State::Pending;
let pending = matches!(task.state, State::Pending | State::Running);
let owner = if pending {
cx.upgrade(task.owner)?
} else {
@@ -439,7 +462,9 @@ impl NativeCallable for Method {
),
Query::Active => Value::Bool(active),
Query::Result => {
if !matches!(task.state, State::Running | State::Completed) || task.error.is_some()
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",
@@ -468,6 +493,7 @@ impl NativeCallable for Method {
task.state = State::Cancelled;
task.callback = None;
task.lease = None;
task.coroutine = None;
task.text = None;
cx.replace_payload(reference, task.clone())?;
if let Some(owner) = owner
@@ -537,6 +563,7 @@ impl Scheduler {
.clone();
let now = Instant::now();
let mut work = 0;
let mut advanced = Vec::new();
for weak in snapshot {
if work >= limit {
break;
@@ -546,7 +573,7 @@ impl Scheduler {
};
let result = (|| -> Result<()> {
let mut task = vm.payload::<Task>(reference, "Task")?.clone();
if task.state != State::Pending {
if !matches!(task.state, State::Pending | State::Running) {
return Ok(());
}
let owner = vm.upgrade(task.owner)?;
@@ -565,6 +592,10 @@ impl Scheduler {
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);
}
@@ -574,6 +605,7 @@ impl Scheduler {
// 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;
let mut running = task.clone();
running.state = State::Running;
vm.replace_payload(reference, running)?;
@@ -615,6 +647,16 @@ impl Scheduler {
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)
}
@@ -649,7 +691,7 @@ impl Scheduler {
if let Some(task) = vm.upgrade(weak)? {
let pending = vm
.payload::<Task>(task, "Task")
.map(|task| task.state == State::Pending);
.map(|task| matches!(task.state, State::Pending | State::Running));
vm.release(task)?;
if pending? {
live.push(weak);
@@ -660,6 +702,77 @@ impl Scheduler {
Ok(())
}
}
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::Pending { .. }) => Ok(()),
Ok(endor_core::CoroutineProgress::Complete { value, .. }) => {
let unit = matches!(value, Value::Nil);
vm.release_value(value)?;
if !active {
return Ok(());
}
if !unit {
fail_callback(vm, reference, "spawn function must return Unit")?;
bail!("spawn function must return Unit");
}
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());
@@ -675,6 +788,7 @@ fn finish(vm: &mut Vm, reference: ObjectRef, state: State) -> Result<()> {
task.state = state;
task.callback = None;
task.lease = None;
task.coroutine = None;
if state == State::Cancelled {
task.text = None;
}
@@ -707,8 +821,9 @@ pub fn install(
("after", Operation::Timer),
("readText", Operation::ReadText),
("fetchText", Operation::FetchText),
("spawn", Operation::Spawn),
] {
let external = operation != Operation::Timer;
let external = matches!(operation, Operation::ReadText | Operation::FetchText);
let function = vm.allocate(ContextNativeFunction::new(
name,
After {
@@ -736,6 +851,10 @@ pub fn install(
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);
}
signature
},
index: index.clone(),
vendor/aster
Vendored
+1 -1
Submodule vendor/aster updated: 1c729b4128...b33d322690