From 16816c38c747ae48ebc5b45f3b3bf66120ff8fac Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Fri, 23 Oct 2026 20:14:15 +0800 Subject: [PATCH] feat: suspend managed calls on generic awaitable operands --- crates/core/src/eval.rs | 56 +++++++++++++++++++++++++++++-- crates/core/src/frame.rs | 14 ++++++-- crates/core/src/lib.rs | 4 +-- crates/objects/src/code.rs | 2 +- crates/objects/src/effects.rs | 1 + crates/objects/src/error.rs | 4 +++ crates/objects/src/instruction.rs | 3 ++ crates/objects/src/lib.rs | 4 +-- crates/objects/src/object.rs | 12 +++++++ 9 files changed, 90 insertions(+), 10 deletions(-) diff --git a/crates/core/src/eval.rs b/crates/core/src/eval.rs index e5791bf..bd5d229 100644 --- a/crates/core/src/eval.rs +++ b/crates/core/src/eval.rs @@ -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, 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(); diff --git a/crates/core/src/frame.rs b/crates/core/src/frame.rs index b75f42c..36875c2 100644 --- a/crates/core/src/frame.rs +++ b/crates/core/src/frame.rs @@ -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>, pub(crate) stack: Vec, } @@ -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 { diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index 3549dd5..06c8911 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -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, diff --git a/crates/objects/src/code.rs b/crates/objects/src/code.rs index 57703fe..577935a 100644 --- a/crates/objects/src/code.rs +++ b/crates/objects/src/code.rs @@ -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), diff --git a/crates/objects/src/effects.rs b/crates/objects/src/effects.rs index 6f4bf5c..2e7516a 100644 --- a/crates/objects/src/effects.rs +++ b/crates/objects/src/effects.rs @@ -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) diff --git a/crates/objects/src/error.rs b/crates/objects/src/error.rs index 20eacd5..93af03e 100644 --- a/crates/objects/src/error.rs +++ b/crates/objects/src/error.rs @@ -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"), diff --git a/crates/objects/src/instruction.rs b/crates/objects/src/instruction.rs index 157bb1c..8084afd 100644 --- a/crates/objects/src/instruction.rs +++ b/crates/objects/src/instruction.rs @@ -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 ) } diff --git a/crates/objects/src/lib.rs b/crates/objects/src/lib.rs index ea07ab2..9535f73 100644 --- a/crates/objects/src/lib.rs +++ b/crates/objects/src/lib.rs @@ -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::{ diff --git a/crates/objects/src/object.rs b/crates/objects/src/object.rs index c7746ec..e1103dc 100644 --- a/crates/objects/src/object.rs +++ b/crates/objects/src/object.rs @@ -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 { + Err(VmError::UnsupportedOperation("await")) + } + /// Explicit function category; callable Type objects are not functions. fn is_function(&self) -> bool { false