feat: suspend managed calls on generic awaitable operands
This commit is contained in:
+53
-3
@@ -88,7 +88,7 @@ impl Vm {
|
||||
}
|
||||
let mut frames = vec![frame];
|
||||
let result = self
|
||||
.eval_frames(&mut frames, limits, &mut executed, usize::MAX)
|
||||
.eval_frames(&mut frames, limits, &mut executed, usize::MAX, false)
|
||||
.map(|value| value.expect("unbounded slice completes or fails"));
|
||||
if result.is_err() {
|
||||
self.last_traceback = frames
|
||||
@@ -189,7 +189,13 @@ impl Vm {
|
||||
.suspended_call
|
||||
.take()
|
||||
.ok_or(VmError::InvalidArgument("no suspended call"))?;
|
||||
let result = self.eval_frames(&mut call.frames, call.limits, &mut call.executed, slice_end);
|
||||
let result = self.eval_frames(
|
||||
&mut call.frames,
|
||||
call.limits,
|
||||
&mut call.executed,
|
||||
slice_end,
|
||||
false,
|
||||
);
|
||||
if let Ok(None) = result {
|
||||
let instructions = call.executed;
|
||||
self.suspended_call = Some(call);
|
||||
@@ -376,8 +382,14 @@ impl Vm {
|
||||
_ => unreachable!(),
|
||||
};
|
||||
let slice_end = executed.saturating_add(slice);
|
||||
let result = self.eval_frames(&mut frames, limits, &mut executed, slice_end);
|
||||
let result = self.eval_frames(&mut frames, limits, &mut executed, slice_end, true);
|
||||
if let Ok(None) = result {
|
||||
let waiting = frames.last().filter(|frame| frame.waiting).map(|frame| {
|
||||
match Self::top_value(frame).expect("await operand remains on stack") {
|
||||
Value::Object(reference) => reference,
|
||||
_ => unreachable!(),
|
||||
}
|
||||
});
|
||||
let frames = std::sync::Arc::new(frames);
|
||||
let mut roots = Vec::new();
|
||||
for frame in frames.iter() {
|
||||
@@ -395,6 +407,12 @@ impl Vm {
|
||||
self.release(root)?;
|
||||
}
|
||||
published?;
|
||||
if let Some(awaitable) = waiting {
|
||||
return Ok(CoroutineProgress::Waiting {
|
||||
awaitable,
|
||||
instructions: executed,
|
||||
});
|
||||
}
|
||||
return Ok(CoroutineProgress::Pending {
|
||||
instructions: executed,
|
||||
});
|
||||
@@ -596,6 +614,7 @@ impl Vm {
|
||||
name: data.name().into(),
|
||||
instruction_offset: 0,
|
||||
current_offset: 0,
|
||||
waiting: false,
|
||||
localsplus: vec![None; executable.slot_count()],
|
||||
stack: Vec::with_capacity(executable.stack_size()),
|
||||
executable,
|
||||
@@ -981,6 +1000,7 @@ impl Vm {
|
||||
limits: ExecutionLimits,
|
||||
executed: &mut usize,
|
||||
slice_end: usize,
|
||||
allow_wait: bool,
|
||||
) -> Result<Option<Value>, VmError> {
|
||||
loop {
|
||||
if *executed >= limits.max_instructions {
|
||||
@@ -997,6 +1017,36 @@ impl Vm {
|
||||
let opcode = decoded.instruction.opcode;
|
||||
let argument = decoded.instruction.argument as usize;
|
||||
match opcode {
|
||||
Opcode::Await => {
|
||||
if self.objects.is_read_only() {
|
||||
return Err(VmError::UnsupportedOperation(
|
||||
"await during read-only execution",
|
||||
));
|
||||
}
|
||||
let Value::Object(awaitable) = Self::top_value(frame)? else {
|
||||
return Err(VmError::ObjectKindMismatch("awaitable object"));
|
||||
};
|
||||
match self.object(awaitable)?.payload().poll_await()? {
|
||||
crate::AwaitPoll::Pending => {
|
||||
if !allow_wait {
|
||||
return Err(VmError::UnsupportedOperation(
|
||||
"pending await outside a coroutine",
|
||||
));
|
||||
}
|
||||
// Keep the evaluated operand, never replay its call
|
||||
// or the statements before this await instruction.
|
||||
frame.waiting = true;
|
||||
frame.instruction_offset = decoded.offset;
|
||||
return Ok(None);
|
||||
}
|
||||
crate::AwaitPoll::Ready(value) => {
|
||||
self.retain_value(&value)?;
|
||||
self.discard(frame, 1);
|
||||
frame.stack.push(StackEntry::Value(value));
|
||||
frame.waiting = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
Opcode::Nop => {}
|
||||
Opcode::LoadConst => {
|
||||
let value = frame.executable.constants()[argument].clone();
|
||||
|
||||
@@ -38,6 +38,7 @@ pub struct EndorFrame {
|
||||
pub(crate) name: String,
|
||||
pub(crate) instruction_offset: usize,
|
||||
pub(crate) current_offset: usize,
|
||||
pub(crate) waiting: bool,
|
||||
pub(crate) localsplus: Vec<Option<Value>>,
|
||||
pub(crate) stack: Vec<StackEntry>,
|
||||
}
|
||||
@@ -118,8 +119,17 @@ pub(crate) struct SuspendedCall {
|
||||
/// not keep a VM-wide heap scope active between quanta.
|
||||
#[derive(Debug)]
|
||||
pub enum CoroutineProgress {
|
||||
Pending { instructions: usize },
|
||||
Complete { value: Value, instructions: usize },
|
||||
Waiting {
|
||||
awaitable: ObjectRef,
|
||||
instructions: usize,
|
||||
},
|
||||
Pending {
|
||||
instructions: usize,
|
||||
},
|
||||
Complete {
|
||||
value: Value,
|
||||
instructions: usize,
|
||||
},
|
||||
}
|
||||
#[derive(Debug)]
|
||||
pub(crate) enum CoroutineState {
|
||||
|
||||
@@ -48,8 +48,8 @@ pub use frame::{
|
||||
// Preserve existing embedding imports; objects owns these definitions.
|
||||
pub use objects::ObjectStore;
|
||||
pub use objects::{
|
||||
AnnotationType, ClassField, EndorTraitObject, EndorTypeObject, MethodRequirement, RootTypes,
|
||||
TypeConstructor, TypedMethodRequirement,
|
||||
AnnotationType, AwaitPoll, ClassField, EndorTraitObject, EndorTypeObject, MethodRequirement,
|
||||
RootTypes, TypeConstructor, TypedMethodRequirement,
|
||||
};
|
||||
pub use objects::{
|
||||
CallArity, CallEffect, CallSignature, CallType, ChangeBatch, ChangeKey, CodeLayout, CompareOp,
|
||||
|
||||
@@ -394,7 +394,7 @@ fn verify(
|
||||
| Opcode::StoreDeref
|
||||
| Opcode::PopJumpIfFalse
|
||||
| Opcode::PopJumpIfTrue => (1, 1, 0),
|
||||
Opcode::ToBool | Opcode::UnaryNot | Opcode::LoadAttr => (1, 1, 1),
|
||||
Opcode::ToBool | Opcode::UnaryNot | Opcode::LoadAttr | Opcode::Await => (1, 1, 1),
|
||||
Opcode::CompareOp => (2, 2, 1),
|
||||
Opcode::StoreAttr => (2, 2, 0),
|
||||
Opcode::Copy => (argument, 0, 1),
|
||||
|
||||
@@ -123,6 +123,7 @@ impl ObjectStore {
|
||||
let argument = decoded.instruction.argument as usize;
|
||||
let mut stack = states[index].clone().expect("reachable effect state");
|
||||
match opcode {
|
||||
Opcode::Await => return Ok(None),
|
||||
Opcode::LoadConst => stack.push(Fact::value(code.constants()[argument].clone())),
|
||||
Opcode::LoadFast => stack.push(if argument == 0 {
|
||||
receiver.map(Fact::Object).unwrap_or(Fact::Unknown)
|
||||
|
||||
@@ -4,6 +4,8 @@ use std::fmt;
|
||||
/// retained for embedders; defining it here avoids a dependency on the VM.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum VmError {
|
||||
AwaitFailed(String),
|
||||
AwaitCancelled,
|
||||
ForeignHandle,
|
||||
InvalidHandle,
|
||||
IdentityExhausted,
|
||||
@@ -49,6 +51,8 @@ pub enum VmError {
|
||||
impl fmt::Display for VmError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::AwaitFailed(message) => write!(f, "await failed: {message}"),
|
||||
Self::AwaitCancelled => f.write_str("await cancelled"),
|
||||
Self::ForeignHandle => f.write_str("handle belongs to another VM"),
|
||||
Self::InvalidHandle => f.write_str("handle does not identify a live object"),
|
||||
Self::IdentityExhausted => f.write_str("process-local VM identities exhausted"),
|
||||
|
||||
@@ -35,6 +35,7 @@ pub enum Opcode {
|
||||
BuildMap = 27,
|
||||
MakeFunction = 28,
|
||||
BuildClass = 29,
|
||||
Await = 30,
|
||||
ExtendedArg = 255,
|
||||
}
|
||||
|
||||
@@ -71,6 +72,7 @@ impl Opcode {
|
||||
27 => Self::BuildMap,
|
||||
28 => Self::MakeFunction,
|
||||
29 => Self::BuildClass,
|
||||
30 => Self::Await,
|
||||
255 => Self::ExtendedArg,
|
||||
_ => return None,
|
||||
})
|
||||
@@ -85,6 +87,7 @@ impl Opcode {
|
||||
| Self::ReturnValue
|
||||
| Self::ToBool
|
||||
| Self::UnaryNot
|
||||
| Self::Await
|
||||
| Self::BuildClass
|
||||
)
|
||||
}
|
||||
|
||||
@@ -43,8 +43,8 @@ pub use number::{
|
||||
EndorDecimal, EndorInteger, EndorNumber, EndorNumberObject, EndorRational, EndorReal,
|
||||
};
|
||||
pub use object::{
|
||||
CompareOp, EndorNativeCall, EndorNativeFunctionObject, EndorObject, NativeCall, NativeFunction,
|
||||
ObjectHeader, ObjectPayload, Record, Value,
|
||||
AwaitPoll, CompareOp, EndorNativeCall, EndorNativeFunctionObject, EndorObject, NativeCall,
|
||||
NativeFunction, ObjectHeader, ObjectPayload, Record, Value,
|
||||
};
|
||||
pub use store::ObjectStore;
|
||||
pub use types::{
|
||||
|
||||
@@ -42,6 +42,14 @@ impl CompareOp {
|
||||
}
|
||||
}
|
||||
|
||||
/// Non-blocking awaitable observation. Ready object values are borrowed from
|
||||
/// the managed graph and must be retained by the consumer before dropping it.
|
||||
#[derive(Debug)]
|
||||
pub enum AwaitPoll {
|
||||
Pending,
|
||||
Ready(Value),
|
||||
}
|
||||
|
||||
/// Rust-side operation interface, not a language type or metaclass.
|
||||
///
|
||||
/// Implementations must report every payload-owned object reference in trace(),
|
||||
@@ -51,6 +59,10 @@ impl CompareOp {
|
||||
/// Rust Drop releases host resources when the payload is destroyed.
|
||||
/// Equal payloads must provide compatible hashes, or reject hashing explicitly.
|
||||
pub trait ObjectPayload: Any + fmt::Debug + Send + Sync {
|
||||
fn poll_await(&self) -> Result<AwaitPoll, VmError> {
|
||||
Err(VmError::UnsupportedOperation("await"))
|
||||
}
|
||||
|
||||
/// Explicit function category; callable Type objects are not functions.
|
||||
fn is_function(&self) -> bool {
|
||||
false
|
||||
|
||||
Reference in New Issue
Block a user