rune/runtime/
vm_execution.rs

1use core::fmt;
2use core::future::Future;
3use core::mem::{replace, take};
4use core::pin::{pin, Pin};
5use core::task::{ready, Context, Poll, RawWaker, RawWakerVTable, Waker};
6
7use crate::alloc::prelude::*;
8use crate::async_vm_try;
9use crate::runtime::budget::Budget;
10use crate::runtime::{budget, env, Awaited};
11use crate::shared::AssertSend;
12use crate::sync::Arc;
13
14use super::{
15    Address, GeneratorState, Globals, Handover, Output, RuntimeContext, Unit, Value, Vm,
16    VmDiagnostics, VmError, VmErrorKind, VmHalt, VmHaltInfo,
17};
18
19static COMPLETE_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
20    |_| RawWaker::new(&(), &COMPLETE_WAKER_VTABLE),
21    |_| {},
22    |_| {},
23    |_| {},
24);
25
26// SAFETY: This waker does nothing.
27static COMPLETE_WAKER: Waker =
28    unsafe { Waker::from_raw(RawWaker::new(&(), &COMPLETE_WAKER_VTABLE)) };
29
30/// The state of an execution. We keep track of this because it's important to
31/// correctly interact with functions that yield (like generators and streams)
32/// by initially just calling the function, then by providing a value pushed
33/// onto the stack.
34#[derive(Debug, Clone, Copy, PartialEq)]
35#[non_exhaustive]
36pub(crate) enum ExecutionState {
37    /// The initial state of an execution.
38    Initial,
39    /// execution is waiting.
40    Resumed(Output),
41    /// Suspended execution.
42    Suspended,
43    /// Execution exited.
44    Exited(Option<Address>),
45}
46
47impl fmt::Display for ExecutionState {
48    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
49        match self {
50            ExecutionState::Initial => write!(f, "initial"),
51            ExecutionState::Resumed(out) => write!(f, "resumed({out})"),
52            ExecutionState::Suspended => write!(f, "suspended"),
53            ExecutionState::Exited(..) => write!(f, "exited"),
54        }
55    }
56}
57
58#[derive(TryClone)]
59#[try_clone(crate)]
60pub(crate) struct VmExecutionState {
61    pub(crate) context: Option<Arc<RuntimeContext>>,
62    pub(crate) unit: Option<Arc<Unit>>,
63    pub(crate) globals: Option<Globals>,
64}
65
66/// The execution environment for a virtual machine.
67///
68/// When an execution is dropped, the stack of the stack of the head machine
69/// will be cleared.
70pub struct VmExecution<T> {
71    /// The current head vm which holds the execution.
72    vm: T,
73    /// The state of an execution.
74    state: ExecutionState,
75    /// Indicates the current stack of suspended contexts.
76    states: Vec<VmExecutionState>,
77}
78
79impl<T> VmExecution<T> {
80    /// Construct an execution from a virtual machine.
81    #[inline]
82    pub(crate) fn new(vm: T) -> Self {
83        Self {
84            vm,
85            state: ExecutionState::Initial,
86            states: Vec::new(),
87        }
88    }
89
90    /// Get a reference to the current virtual machine.
91    #[inline]
92    pub fn vm(&self) -> &Vm
93    where
94        T: AsRef<Vm>,
95    {
96        self.vm.as_ref()
97    }
98
99    /// Get a mutable reference the current virtual machine.
100    #[inline]
101    pub fn vm_mut(&mut self) -> &mut Vm
102    where
103        T: AsMut<Vm>,
104    {
105        self.vm.as_mut()
106    }
107}
108
109impl<T> VmExecution<T>
110where
111    T: AsMut<Vm>,
112{
113    /// Coerce the current execution into a generator if appropriate.
114    ///
115    /// ```
116    /// use rune::Vm;
117    /// use rune::sync::Arc;
118    ///
119    /// let mut sources = rune::sources! {
120    ///     entry => {
121    ///         pub fn main() {
122    ///             yield 1;
123    ///             yield 2;
124    ///         }
125    ///     }
126    /// };
127    ///
128    /// let unit = rune::prepare(&mut sources).build()?;
129    /// let unit = Arc::try_new(unit)?;
130    /// let mut vm = Vm::without_runtime(unit)?;
131    ///
132    /// let mut generator = vm.execute(["main"], ())?.into_generator();
133    ///
134    /// let mut n = 1i64;
135    ///
136    /// while let Some(value) = generator.next()? {
137    ///     let value: i64 = rune::from_value(value)?;
138    ///     assert_eq!(value, n);
139    ///     n += 1;
140    /// }
141    /// # Ok::<_, rune::support::Error>(())
142    /// ```
143    pub fn into_generator(self) -> VmGenerator<T> {
144        VmGenerator {
145            execution: Some(self),
146        }
147    }
148
149    /// Coerce the current execution into a stream if appropriate.
150    ///
151    /// ```
152    /// use rune::Vm;
153    /// use rune::sync::Arc;
154    ///
155    /// # futures_executor::block_on(async move {
156    /// let mut sources = rune::sources! {
157    ///     entry => {
158    ///         pub async fn main() {
159    ///             yield 1;
160    ///             yield 2;
161    ///         }
162    ///     }
163    /// };
164    ///
165    /// let unit = rune::prepare(&mut sources).build()?;
166    /// let unit = Arc::try_new(unit)?;
167    /// let mut vm = Vm::without_runtime(unit)?;
168    ///
169    /// let mut stream = vm.execute(["main"], ())?.into_stream();
170    ///
171    /// let mut n = 1i64;
172    ///
173    /// while let Some(value) = stream.next().await? {
174    ///     let value: i64 = rune::from_value(value)?;
175    ///     assert_eq!(value, n);
176    ///     n += 1;
177    /// }
178    /// # Ok::<_, rune::support::Error>(())
179    /// # })?;
180    /// # Ok::<_, rune::support::Error>(())
181    /// ```
182    pub fn into_stream(self) -> VmStream<T> {
183        VmStream {
184            execution: Some(self),
185        }
186    }
187}
188
189impl<T> VmExecution<T>
190where
191    T: AsMut<Vm>,
192{
193    /// Synchronously complete the current execution.
194    ///
195    /// # Errors
196    ///
197    /// If anything except the completion of the execution is encountered, this
198    /// will result in an error.
199    ///
200    /// To handle other outcomes and more configurability see
201    /// [`VmExecution::resume`].
202    pub fn complete(&mut self) -> Result<Value, VmError> {
203        self.resume().complete()?.into_complete()
204    }
205
206    /// Asynchronously complete the current execution.
207    ///
208    /// # Errors
209    ///
210    /// If anything except the completion of the execution is encountered, this
211    /// will result in an error.
212    ///
213    /// To handle other outcomes and more configurability see
214    /// [`VmExecution::resume`].
215    pub async fn async_complete(&mut self) -> Result<Value, VmError> {
216        self.resume().await?.into_complete()
217    }
218
219    /// Resume the current execution.
220    ///
221    /// To complete this operation synchronously, use [`VmResume::complete`].
222    ///
223    /// ## Resume with a value
224    ///
225    /// To resume an execution with a value, use [`VmResume::with_value`]. This
226    /// requires that the execution has yielded first, otherwise an error will
227    /// be produced.
228    ///
229    /// ## Resume with diagnostics
230    ///
231    /// To associated [`VmDiagnostics`] with the execution, use
232    /// [`VmResume::with_diagnostics`].
233    pub fn resume(&mut self) -> VmResume<'_, 'static, T> {
234        VmResume {
235            execution: self,
236            diagnostics: None,
237            awaited: None,
238            init: Some(Value::empty()),
239        }
240    }
241
242    /// End execution and perform debug checks.
243    pub(crate) fn end(&mut self) -> Result<Value, VmError> {
244        let ExecutionState::Exited(addr) = self.state else {
245            return Err(VmError::new(VmErrorKind::ExpectedExitedExecutionState {
246                actual: self.state,
247            }));
248        };
249
250        let value = match addr {
251            Some(addr) => self.vm.as_mut().stack().at(addr).clone(),
252            None => Value::unit(),
253        };
254
255        debug_assert!(self.states.is_empty(), "Execution states should be empty");
256        Ok(value)
257    }
258
259    /// Push a virtual machine state onto the execution.
260    #[tracing::instrument(skip_all)]
261    pub(crate) fn push_state(&mut self, state: VmExecutionState) -> Result<(), VmError> {
262        tracing::trace!("pushing suspended state");
263        let vm = self.vm.as_mut();
264        let context = state.context.map(|c| replace(vm.context_mut(), c));
265        let unit = state.unit.map(|u| replace(vm.unit_mut(), u));
266        let globals = state.globals.map(|g| replace(vm.globals_mut(), g));
267        self.states.try_push(VmExecutionState {
268            context,
269            unit,
270            globals,
271        })?;
272        Ok(())
273    }
274
275    /// Pop a virtual machine state from the execution and transfer the top of
276    /// the stack from the popped machine.
277    #[tracing::instrument(skip_all)]
278    fn pop_state(&mut self) -> Result<(), VmError> {
279        tracing::trace!("popping suspended state");
280
281        let state = self.states.pop().ok_or(VmErrorKind::NoRunningVm)?;
282        let vm = self.vm.as_mut();
283
284        if let Some(context) = state.context {
285            *vm.context_mut() = context;
286        }
287
288        if let Some(unit) = state.unit {
289            *vm.unit_mut() = unit;
290        }
291
292        if let Some(globals) = state.globals {
293            *vm.globals_mut() = globals;
294        }
295
296        Ok(())
297    }
298}
299
300impl VmExecution<&mut Vm> {
301    /// Convert the current execution into one which owns its virtual machine.
302    pub fn into_owned(self) -> VmExecution<Vm> {
303        let stack = take(self.vm.stack_mut());
304        let head = Vm::with_stack(self.vm.context().clone(), self.vm.unit().clone(), stack)
305            .with_globals(self.vm.globals().clone());
306
307        VmExecution {
308            vm: head,
309            states: self.states,
310            state: self.state,
311        }
312    }
313}
314
315/// A wrapper that makes [`VmExecution`] [`Send`].
316///
317/// This is accomplished by preventing any [`Value`] from escaping the [`Vm`].
318/// As long as this is maintained, it is safe to send the execution across,
319/// threads, and therefore schedule the future associated with the execution on
320/// a thread pool like Tokio's through [tokio::spawn].
321///
322/// [tokio::spawn]: https://docs.rs/tokio/0/tokio/runtime/struct.Runtime.html#method.spawn
323pub struct VmSendExecution(pub(crate) VmExecution<Vm>);
324
325// Safety: we wrap all APIs around the [VmExecution], preventing values from
326// escaping from contained virtual machine.
327unsafe impl Send for VmSendExecution {}
328
329impl VmSendExecution {
330    /// Complete the current execution with support for async instructions.
331    ///
332    /// This requires that the result of the Vm is converted into a
333    /// [crate::FromValue] that also implements [Send],  which prevents non-Send
334    /// values from escaping from the virtual machine.
335    pub fn complete(mut self) -> impl Future<Output = Result<Value, VmError>> + Send + 'static {
336        let future = async move { self.0.resume().await.and_then(VmOutcome::into_complete) };
337
338        // Safety: we wrap all APIs around the [VmExecution], preventing values
339        // from escaping from contained virtual machine.
340        unsafe { AssertSend::new(future) }
341    }
342
343    /// Alias for [`VmSendExecution::complete`].
344    #[deprecated = "Use `VmSendExecution::complete`"]
345    pub fn async_complete(self) -> impl Future<Output = Result<Value, VmError>> + Send + 'static {
346        self.complete()
347    }
348
349    /// Complete the current execution with support for async instructions.
350    ///
351    /// This requires that the result of the Vm is converted into a
352    /// [crate::FromValue] that also implements [Send],  which prevents non-Send
353    /// values from escaping from the virtual machine.
354    pub fn complete_with_diagnostics(
355        mut self,
356        diagnostics: &mut dyn VmDiagnostics,
357    ) -> impl Future<Output = Result<Value, VmError>> + Send + '_ {
358        let future = async move {
359            self.0
360                .resume()
361                .with_diagnostics(diagnostics)
362                .await
363                .and_then(VmOutcome::into_complete)
364        };
365
366        // Safety: we wrap all APIs around the [VmExecution], preventing values
367        // from escaping from contained virtual machine.
368        unsafe { AssertSend::new(future) }
369    }
370
371    /// Alias for [`VmSendExecution::complete_with_diagnostics`].
372    #[deprecated = "Use `VmSendExecution::complete_with_diagnostics`"]
373    pub fn async_complete_with_diagnostics(
374        self,
375        diagnostics: &mut dyn VmDiagnostics,
376    ) -> impl Future<Output = Result<Value, VmError>> + Send + '_ {
377        self.complete_with_diagnostics(diagnostics)
378    }
379}
380
381impl<T> TryClone for VmExecution<T>
382where
383    T: AsRef<Vm> + AsMut<Vm> + TryClone,
384{
385    #[inline]
386    fn try_clone(&self) -> Result<Self, rune_alloc::Error> {
387        Ok(Self {
388            vm: self.vm.try_clone()?,
389            state: self.state,
390            states: self.states.try_clone()?,
391        })
392    }
393}
394
395/// The outcome of completing an execution through a [`VmResume`] operation.
396#[non_exhaustive]
397pub enum VmOutcome {
398    /// A value has been produced by the execution returning.
399    Complete(Value),
400    /// A value has been yielded by the execution.
401    Yielded(Value),
402    /// The execution has been limited.
403    Limited,
404}
405
406impl VmOutcome {
407    /// Convert the outcome into a [`GeneratorState`].
408    ///
409    /// # Errors
410    ///
411    /// If the execution is not in a state compatible with producing a generator
412    /// state, such as having been completed or yielded, this will produce an
413    /// error.
414    pub fn into_generator_state(self) -> Result<GeneratorState, VmError> {
415        match self {
416            VmOutcome::Complete(value) => Ok(GeneratorState::Complete(value)),
417            VmOutcome::Yielded(value) => Ok(GeneratorState::Yielded(value)),
418            VmOutcome::Limited => Err(VmError::new(VmErrorKind::Halted {
419                halt: VmHaltInfo::Limited,
420            })),
421        }
422    }
423
424    /// Convert the outcome into a completed value.
425    ///
426    /// # Errors
427    ///
428    /// If the execution hasn't returned, this will produce an error.
429    pub fn into_complete(self) -> Result<Value, VmError> {
430        match self {
431            VmOutcome::Complete(value) => Ok(value),
432            VmOutcome::Yielded(..) => Err(VmError::new(VmErrorKind::Halted {
433                halt: VmHaltInfo::Yielded,
434            })),
435            VmOutcome::Limited => Err(VmError::new(VmErrorKind::Halted {
436                halt: VmHaltInfo::Limited,
437            })),
438        }
439    }
440}
441
442/// An execution that has been resumed.
443///
444/// This can either be completed as a future, which allows the execution to
445/// perform asynchronous operations, or it can be completed by calling
446/// [`VmResume::complete`] which will produce an error in case asynchronous
447/// operations that need to be suspended are encountered.
448pub struct VmResume<'this, 'diag, T> {
449    execution: &'this mut VmExecution<T>,
450    diagnostics: Option<&'diag mut dyn VmDiagnostics>,
451    init: Option<Value>,
452    awaited: Option<Awaited>,
453}
454
455impl<'this, 'diag, T> VmResume<'this, 'diag, T> {
456    /// Associated a budget with the resumed execution.
457    pub fn with_budget(self, budget: usize) -> Budget<Self> {
458        budget::with(budget, self)
459    }
460
461    /// Associate a value with the resumed execution.
462    ///
463    /// This is necessary to provide a value for a generator which has yielded.
464    pub fn with_value(self, value: Value) -> VmResume<'this, 'diag, T> {
465        Self {
466            init: Some(value),
467            ..self
468        }
469    }
470
471    /// Associate diagnostics with the execution.
472    pub fn with_diagnostics<'a>(
473        self,
474        diagnostics: &'a mut dyn VmDiagnostics,
475    ) -> VmResume<'this, 'a, T> {
476        VmResume {
477            execution: self.execution,
478            diagnostics: Some(diagnostics),
479            init: self.init,
480            awaited: self.awaited,
481        }
482    }
483}
484
485impl<'this, 'diag, T> VmResume<'this, 'diag, T>
486where
487    T: AsMut<Vm>,
488{
489    /// Try to synchronously complete the run, returning the generator state it produced.
490    ///
491    /// This will error if the execution is suspended through awaiting.
492    pub fn complete(self) -> Result<VmOutcome, VmError> {
493        let this = pin!(self);
494        let mut cx = Context::from_waker(&COMPLETE_WAKER);
495
496        match this.poll(&mut cx) {
497            Poll::Ready(result) => result,
498            Poll::Pending => Err(VmError::new(VmErrorKind::Halted {
499                halt: VmHaltInfo::Awaited,
500            })),
501        }
502    }
503}
504
505impl<'this, 'diag, T> Future for VmResume<'this, 'diag, T>
506where
507    T: AsMut<Vm>,
508{
509    type Output = Result<VmOutcome, VmError>;
510
511    #[inline]
512    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
513        // SAFETY: We are ensuring that we never move this value or any
514        // projected fields.
515        let this = unsafe { Pin::get_unchecked_mut(self) };
516
517        poll_resume(
518            &mut *this.execution,
519            &mut this.init,
520            &mut this.awaited,
521            this.diagnostics.as_deref_mut(),
522            cx,
523        )
524    }
525}
526
527/// Drive an execution which has been resumed.
528///
529/// The state a resumed execution is driven with is taken apart here so that an
530/// execution which is borrowed - [`VmResume`] - and one which is owned -
531/// [`VmResumeOwned`] - are driven by the same loop.
532fn poll_resume<T>(
533    execution: &mut VmExecution<T>,
534    init: &mut Option<Value>,
535    awaited: &mut Option<Awaited>,
536    mut diagnostics: Option<&mut (dyn VmDiagnostics + '_)>,
537    cx: &mut Context<'_>,
538) -> Poll<Result<VmOutcome, VmError>>
539where
540    T: AsMut<Vm>,
541{
542    // An execution which is driven from inside the native frames of another one
543    // costs a native frame of its own, and a script decides how deeply it nests
544    // them, so the nesting is bounded here rather than left to overflow.
545    let _guard = async_vm_try!(env::enter_execution());
546
547    if let Some(value) = init.take() {
548        let state = replace(&mut execution.state, ExecutionState::Suspended);
549
550        if let ExecutionState::Resumed(out) = state {
551            let vm = execution.vm.as_mut();
552            async_vm_try!(vm.store(out, value));
553        }
554    }
555
556    loop {
557        let vm = execution.vm.as_mut();
558
559        if let Some(value) = &mut *awaited {
560            // SAFETY: The awaited value is never moved for as long as it is
561            // being polled.
562            let value = unsafe { Pin::new_unchecked(value) };
563            async_vm_try!(ready!(value.poll(cx, vm)));
564            *awaited = None;
565        }
566
567        let result = vm.run(match diagnostics {
568            Some(ref mut value) => Some(&mut **value),
569            None => None,
570        });
571
572        match async_vm_try!(VmError::with_vm(result, vm)) {
573            VmHalt::Exited(addr) => {
574                execution.state = ExecutionState::Exited(addr);
575            }
576            VmHalt::Awaited(value) => {
577                *awaited = Some(value);
578                continue;
579            }
580            VmHalt::VmCall(vm_call) => {
581                async_vm_try!(vm_call.into_execution(execution));
582                continue;
583            }
584            VmHalt::Yielded(addr, out) => {
585                let value = match addr {
586                    Some(addr) => vm.stack().at(addr).clone(),
587                    None => Value::unit(),
588                };
589
590                execution.state = ExecutionState::Resumed(out);
591                return Poll::Ready(Ok(VmOutcome::Yielded(value)));
592            }
593            VmHalt::Limited => {
594                return Poll::Ready(Ok(VmOutcome::Limited));
595            }
596        }
597
598        if execution.states.is_empty() {
599            let value = async_vm_try!(execution.end());
600            return Poll::Ready(Ok(VmOutcome::Complete(value)));
601        }
602
603        async_vm_try!(execution.pop_state());
604    }
605}
606
607/// A future which drives an execution it owns.
608///
609/// A future built out of an `async` block owns its execution too, but that
610/// execution is only reachable from inside of the block. This is what an async
611/// call produces instead, so that the values a suspended execution holds can be
612/// handed over rather than dropped in place - see `Future::from_execution`.
613pub(crate) struct VmResumeOwned {
614    execution: VmExecution<Vm>,
615    init: Option<Value>,
616    awaited: Option<Awaited>,
617}
618
619impl VmResumeOwned {
620    /// Drive the given execution to completion.
621    pub(crate) fn new(execution: VmExecution<Vm>) -> Self {
622        Self {
623            execution,
624            init: Some(Value::empty()),
625            awaited: None,
626        }
627    }
628
629    /// Hand over the values the execution is working over.
630    ///
631    /// Nothing borrows the stack in between polls, so this is available even
632    /// while the execution is suspended.
633    /// Swap the next value the execution is working over with `value`.
634    pub(crate) fn dismantle(&mut self, out: &mut Handover<'_>) {
635        out.consume(self.execution.vm_mut().stack_mut());
636    }
637
638    /// Test whether this execution has yet to run a single instruction.
639    ///
640    /// An execution in this state holds nothing but the arguments of the call
641    /// which produced it, so the machine which awaits it can splice it in as an
642    /// ordinary call frame instead of driving it as a nested machine - see
643    /// [`Vm::splice_call`].
644    ///
645    /// [`Vm::splice_call`]: crate::runtime::Vm::splice_call
646    pub(crate) fn is_unstarted(&self) -> bool {
647        self.init.is_some()
648            && self.awaited.is_none()
649            && self.execution.states.is_empty()
650            && matches!(self.execution.state, ExecutionState::Initial)
651            && self.execution.vm.call_frames().is_empty()
652    }
653
654    /// Take the machine out of an unstarted execution, leaving an empty one
655    /// which shares its unit, context and storage behind.
656    pub(crate) fn take_vm(&mut self) -> Vm {
657        let vm = &mut self.execution.vm;
658
659        let replacement = Vm::new(vm.context().clone(), vm.unit().clone());
660        replace(vm, replacement)
661    }
662}
663
664impl Future for VmResumeOwned {
665    type Output = Result<Value, VmError>;
666
667    #[inline]
668    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
669        // SAFETY: We are ensuring that we never move this value or any
670        // projected fields.
671        let this = unsafe { Pin::get_unchecked_mut(self) };
672
673        let outcome = ready!(poll_resume(
674            &mut this.execution,
675            &mut this.init,
676            &mut this.awaited,
677            None,
678            cx
679        ));
680
681        Poll::Ready(outcome.and_then(VmOutcome::into_complete))
682    }
683}
684
685/// A [`VmExecution`] that can be used with a generator api.
686pub struct VmGenerator<T> {
687    execution: Option<VmExecution<T>>,
688}
689
690impl<T> VmGenerator<T>
691where
692    T: AsMut<Vm>,
693{
694    /// Get the next value produced by this generator.
695    ///
696    /// See [`VmExecution::into_generator`].
697    pub fn next(&mut self) -> Result<Option<Value>, VmError> {
698        let Some(execution) = &mut self.execution else {
699            return Ok(None);
700        };
701
702        let outcome = execution.resume().complete()?;
703
704        match outcome {
705            VmOutcome::Complete(_) => {
706                self.execution = None;
707                Ok(None)
708            }
709            VmOutcome::Yielded(value) => Ok(Some(value)),
710            VmOutcome::Limited => Err(VmError::new(VmErrorKind::Halted {
711                halt: VmHaltInfo::Limited,
712            })),
713        }
714    }
715
716    /// Resume the generator with a value and get the next [`GeneratorState`].
717    ///
718    /// See [`VmExecution::into_generator`].
719    pub fn resume(&mut self, value: Value) -> Result<GeneratorState, VmError> {
720        let execution = self
721            .execution
722            .as_mut()
723            .ok_or(VmErrorKind::GeneratorComplete)?;
724
725        let outcome = execution.resume().with_value(value).complete()?;
726
727        match outcome {
728            VmOutcome::Complete(value) => {
729                self.execution = None;
730                Ok(GeneratorState::Complete(value))
731            }
732            VmOutcome::Yielded(value) => Ok(GeneratorState::Yielded(value)),
733            VmOutcome::Limited => Err(VmError::new(VmErrorKind::Halted {
734                halt: VmHaltInfo::Limited,
735            })),
736        }
737    }
738}
739
740/// A [`VmExecution`] that can be used with a stream api.
741pub struct VmStream<T> {
742    execution: Option<VmExecution<T>>,
743}
744
745impl<T> VmStream<T>
746where
747    T: AsMut<Vm>,
748{
749    /// Get the next value produced by this stream.
750    ///
751    /// See [`VmExecution::into_stream`].
752    pub async fn next(&mut self) -> Result<Option<Value>, VmError> {
753        let Some(execution) = &mut self.execution else {
754            return Ok(None);
755        };
756
757        match execution.resume().await? {
758            VmOutcome::Complete(value) => {
759                self.execution = None;
760                Ok(Some(value))
761            }
762            VmOutcome::Yielded(..) => Ok(None),
763            VmOutcome::Limited => Err(VmError::new(VmErrorKind::Halted {
764                halt: VmHaltInfo::Limited,
765            })),
766        }
767    }
768
769    /// Resume the stream with a value and return the next [`GeneratorState`].
770    ///
771    /// See [`VmExecution::into_stream`].
772    pub async fn resume(&mut self, value: Value) -> Result<GeneratorState, VmError> {
773        let execution = self
774            .execution
775            .as_mut()
776            .ok_or(VmErrorKind::GeneratorComplete)?;
777
778        match execution.resume().with_value(value).await? {
779            VmOutcome::Complete(value) => {
780                self.execution = None;
781                Ok(GeneratorState::Complete(value))
782            }
783            VmOutcome::Yielded(value) => Ok(GeneratorState::Yielded(value)),
784            VmOutcome::Limited => Err(VmError::new(VmErrorKind::Halted {
785                halt: VmHaltInfo::Limited,
786            })),
787        }
788    }
789}
rune/runtime/vm_execution.rs

rune/runtime/
vm_execution.rs
