onion_vm/lambda/scheduler/
async_scheduler.rs

1use arc_gc::gc::GC;
2
3use crate::{
4    lambda::runnable::{Runnable, RuntimeError, StepResult},
5    types::{
6        object::{OnionObject, OnionObjectCell},
7        pair::OnionPair,
8    },
9    unwrap_step_result,
10};
11
12pub struct AsyncScheduler {
13    pub(crate) runnables: Vec<Box<dyn Runnable>>,
14}
15
16impl AsyncScheduler {
17    pub fn new(runnables: Vec<Box<dyn Runnable>>) -> Self {
18        AsyncScheduler { runnables }
19    }
20}
21
22impl Runnable for AsyncScheduler {
23    fn step(&mut self, gc: &mut GC<OnionObjectCell>) -> StepResult {
24        if self.runnables.is_empty() {
25            return StepResult::Return(
26                OnionPair::new_static(
27                    &OnionObject::Boolean(true).stabilize(),
28                    &OnionObject::Undefined(Some("All runnables completed".to_string().into()))
29                        .stabilize(),
30                )
31                .into(),
32            );
33        }
34
35        let mut i = 0;
36        while i < self.runnables.len() {
37            match self.runnables[i].step(gc) {
38                StepResult::Continue => {
39                    i += 1; // 继续处理下一个runnable
40                }
41                StepResult::NewRunnable(new_runnable) => {
42                    self.runnables.push(new_runnable);
43                    unwrap_step_result!(self.runnables[i].receive(
44                        &StepResult::Return(
45                            OnionObject::Undefined(Some("Task Launched".to_string().into()))
46                                .stabilize()
47                                .into(),
48                        ),
49                        gc,
50                    ));
51                    i += 1; // 继续处理下一个runnable
52                }
53                StepResult::ReplaceRunnable(new_runnable) => {
54                    // self.runnables.push(new_runnable);
55                    // self.runnables[i].receive(
56                    //     StepResult::Return(
57                    //         OnionObject::Undefined(Some("Task Launched".to_string())).stabilize(),
58                    //     ),
59                    //     gc,
60                    // )?;
61                    // i += 1; // 继续处理下一个runnable
62                    self.runnables[i] = new_runnable;
63                    i += 1; // 继续处理下一个runnable
64                }
65                StepResult::Return(_) => {
66                    // 移除已完成的runnable
67                    self.runnables.remove(i);
68                }
69                e => return e,
70            }
71        }
72
73        // 如果所有runnable都需要继续，返回Continue
74        StepResult::Continue
75    }
76    fn receive(
77        &mut self,
78        _step_result: &StepResult,
79        _gc: &mut GC<OnionObjectCell>,
80    ) -> Result<(), RuntimeError> {
81        Err(RuntimeError::DetailedError(
82            "AsyncScheduler does not support receive".to_string().into(),
83        ))
84    }
85
86    fn copy(&self) -> Box<dyn Runnable> {
87        Box::new(AsyncScheduler {
88            runnables: self.runnables.iter().map(|r| r.copy()).collect(),
89        })
90    }
91
92    fn format_context(&self) -> Result<serde_json::Value, RuntimeError> {
93        let mut tasks_json_array = serde_json::Value::Array(vec![]);
94        for runnable in &self.runnables {
95            let frame_json = runnable.format_context()?;
96            tasks_json_array.as_array_mut().unwrap().push(frame_json);
97        }
98        // {type: "AsyncScheduler", tasks: tasks_json_array}
99        Ok(serde_json::json!({
100            "type": "AsyncScheduler",
101            "tasks": tasks_json_array
102        }))
103    }
104}
onion_vm/lambda/scheduler/async_scheduler.rs

onion_vm/lambda/scheduler/
async_scheduler.rs
