diff --git a/src/tasks.rs b/src/tasks.rs index bfe9094..ce4e28c 100644 --- a/src/tasks.rs +++ b/src/tasks.rs @@ -51,6 +51,8 @@ struct Task { text: Option>, error: Option>, lease: Option>, + cooperative: bool, + coroutine: Option, } 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::(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")?.state == State::Pending + && matches!( + cx.store().payload::(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::(previous, "Task") - .map(|task| task.state == State::Pending && task.lease.is_some()) + cx.store().payload::(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::(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::(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::(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") - .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::(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::(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::(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::(reference, "Task")?.clone(); task.error = Some(message.chars().take(1024).collect::().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(), diff --git a/vendor/aster b/vendor/aster index 1c729b4..b33d322 160000 --- a/vendor/aster +++ b/vendor/aster @@ -1 +1 @@ -Subproject commit 1c729b412871c6c68c9afeb81550e1b6b3eaa3f5 +Subproject commit b33d322690f2f3f18440081d28dad094d695a1a0