diff --git a/src/endor_host.rs b/src/endor_host.rs index b7ccfe2..ba77708 100644 --- a/src/endor_host.rs +++ b/src/endor_host.rs @@ -1346,7 +1346,7 @@ impl Architecture { ExprKind::Number(value) => (value.clone(), "number", vec![]), ExprKind::String(value) => (value.chars().take(40).collect(), "string", vec![]), ExprKind::Symbol(name) => (name.text.clone(), "symbol", vec![]), - ExprKind::Unary { operand, .. } => { + ExprKind::Unary { operand, .. } | ExprKind::Await { operand } => { ("negate".into(), "unary", vec![self.expression(operand)]) } ExprKind::Binary { left, right, .. } => ( diff --git a/src/endor_script.rs b/src/endor_script.rs index af0bc4e..1a6f1c3 100644 --- a/src/endor_script.rs +++ b/src/endor_script.rs @@ -162,7 +162,9 @@ fn shift_expression(expression: &mut Expr, offset: usize) { match &mut expression.kind { ExprKind::Name(_) | ExprKind::Number(_) | ExprKind::String(_) => {} ExprKind::Symbol(name) => shift_span(&mut name.span, offset), - ExprKind::Unary { operand, .. } => shift_expression(operand, offset), + ExprKind::Unary { operand, .. } | ExprKind::Await { operand } => { + shift_expression(operand, offset) + } ExprKind::Binary { left, right, .. } => { shift_expression(left, offset); shift_expression(right, offset); diff --git a/src/tasks.rs b/src/tasks.rs index 6b3513f..e82006a 100644 --- a/src/tasks.rs +++ b/src/tasks.rs @@ -56,6 +56,8 @@ struct Task { lease: Option>, cooperative: bool, coroutine: Option, + waiting: Option, + result: Option, } impl ObjectPayload for Task { fn as_any(&self) -> &dyn std::any::Any { @@ -67,6 +69,23 @@ impl ObjectPayload for Task { fn reference_assignment_only(&self) -> bool { true } + fn poll_await(&self) -> Result { + 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); @@ -74,6 +93,12 @@ impl ObjectPayload for Task { 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 { @@ -197,7 +222,9 @@ pub fn cancel_owner(cx: &mut NativeContext<'_>, owner: ObjectRef) -> Result<(), 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)?; @@ -208,6 +235,7 @@ pub fn cancel_owner(cx: &mut NativeContext<'_>, owner: ObjectRef) -> Result<(), #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Operation { Timer, + Sleep, ReadText, FetchText, Spawn, @@ -240,6 +268,7 @@ impl NativeCallable for After { [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) => @@ -293,7 +322,7 @@ impl NativeCallable for After { } crate::io_tasks::Request::FetchText(path.clone()) } - Operation::Timer | Operation::Spawn => unreachable!(), + Operation::Timer | Operation::Sleep | Operation::Spawn => unreachable!(), }; (0, Some(request)) } else { @@ -352,12 +381,17 @@ impl NativeCallable for After { .ok_or(invalid("task deadline overflow"))?, callback, state: State::Pending, - ready: matches!(self.operation, Operation::Timer | Operation::Spawn), + 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 { @@ -402,7 +436,9 @@ impl NativeCallable for After { old.callback = None; old.lease = None; old.coroutine = None; + old.waiting = None; old.text = None; + old.result = None; cx.replace_payload(previous, old)?; } } @@ -477,10 +513,12 @@ impl NativeCallable for Method { "task result is not available; inspect status and error", )); } - task.text - .as_ref() - .map(|text| Value::Text(text.to_string())) - .unwrap_or(Value::Nil) + 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 { @@ -501,7 +539,9 @@ impl NativeCallable for Method { 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)? @@ -599,6 +639,16 @@ impl Scheduler { 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(()); } @@ -617,6 +667,7 @@ impl Scheduler { // 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)?; @@ -704,9 +755,20 @@ impl Scheduler { if let Some(task) = vm.upgrade(weak)? { let pending = vm.payload::(task, "Task").map(|task| { let pending = matches!(task.state, State::Pending | State::Running); - runnable |= pending && task.cooperative; + runnable |= pending && task.cooperative && task.waiting.is_none(); pending }); + if pending.as_ref().is_ok_and(|pending| *pending) { + let value = vm.payload::(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); @@ -718,6 +780,22 @@ impl Scheduler { Ok(()) } } +fn cyclic_wait(vm: &Vm, source: ObjectRef, mut target: ObjectRef) -> Result { + 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::() 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::(reference, "Task")?.clone(); let coroutine = if let Some(coroutine) = task.coroutine { @@ -767,17 +845,38 @@ fn advance_coroutine(vm: &mut Vm, reference: ObjectRef) -> Result<()> { 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)?; + Ok(endor_core::CoroutineProgress::Waiting { awaitable, .. }) => { if !active { return Ok(()); } - if !unit { - fail_callback(vm, reference, "spawn function must return Unit")?; - bail!("spawn function must return Unit"); + if cyclic_wait(vm, reference, awaitable)? { + fail_callback(vm, reference, "cyclic task await")?; + bail!("cyclic task await"); } + let mut task = vm.payload::(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::(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::(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) => { @@ -805,8 +904,10 @@ fn finish(vm: &mut Vm, reference: ObjectRef, state: State) -> Result<()> { 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)? { @@ -836,6 +937,7 @@ pub fn install( 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), @@ -872,6 +974,10 @@ pub fn install( 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(), diff --git a/vendor/aster b/vendor/aster index b33d322..ae75c54 160000 --- a/vendor/aster +++ b/vendor/aster @@ -1 +1 @@ -Subproject commit b33d322690f2f3f18440081d28dad094d695a1a0 +Subproject commit ae75c54707eb2917d97a4117de9eff5a9f025938