onion_vm/lambda/scheduler/
map_scheduler.rs

1use arc_gc::gc::GC;
2
3use crate::{
4    lambda::runnable::{Runnable, RuntimeError, StepResult},
5    types::{
6        lambda::launcher::OnionLambdaRunnableLauncher,
7        object::{OnionObject, OnionObjectCell, OnionStaticObject},
8        tuple::OnionTuple,
9    },
10    utils::format_object_summary,
11};
12
13#[derive(Clone)]
14pub struct Mapping {
15    pub(crate) container: OnionStaticObject,
16    pub(crate) mapper: OnionStaticObject,
17    pub(crate) collected: Vec<OnionStaticObject>,
18    pub(crate) current_index: usize,
19}
20
21impl Runnable for Mapping {
22    fn receive(
23        &mut self,
24        step_result: &StepResult,
25        _gc: &mut GC<OnionObjectCell>,
26    ) -> Result<(), RuntimeError> {
27        match step_result {
28            StepResult::Return(result) => {
29                self.collected.push(result.as_ref().clone());
30                self.current_index += 1; // 移动到下一个元素
31                Ok(())
32            }
33            _ => Err(RuntimeError::DetailedError(
34                "Unexpected step result in mapping".into(),
35            )),
36        }
37    }
38
39    fn step(&mut self, _gc: &mut GC<OnionObjectCell>) -> StepResult {
40        self.container
41            .weak()
42            .with_data(|container| match container {
43                OnionObject::Tuple(tuple) => {
44                    // 使用索引获取当前元素
45                    if let Some(element) = tuple.get_elements().get(self.current_index) {
46                        let element_clone = element.clone();
47                        self.mapper.weak().with_data(|mapper_obj| match mapper_obj {
48                            OnionObject::Lambda(_) => {
49                                let runnable = Box::new(OnionLambdaRunnableLauncher::new_static(
50                                    mapper_obj,
51                                    element.stabilize(),
52                                    &|r| Ok(r),
53                                )?);
54                                Ok(StepResult::NewRunnable(runnable))
55                            }
56                            OnionObject::Boolean(false) => Ok(StepResult::Continue),
57                            _ => {
58                                self.collected.push(element_clone.stabilize());
59                                Ok(StepResult::Continue)
60                            }
61                        })
62                    } else {
63                        // 所有元素都处理完了
64                        Ok(StepResult::Return(
65                            OnionTuple::new_static_no_ref(&self.collected).into(),
66                        ))
67                    }
68                }
69                _ => Err(RuntimeError::InvalidType(
70                    "Container must be a tuple".into(),
71                )),
72            })
73            .unwrap_or_else(|e| StepResult::Error(e))
74    }
75    fn format_context(&self) -> String {
76        // 使用 format! 宏来构建一个多行的字符串
77        format!(
78            "-> In 'map' operation:\n   - Mapper: {}\n   - Container: {}\n   - Progress: Processing element {} / {}\n   - Collected Items: {}",
79            // 1. 映射器信息
80            // 使用对象的简略表示（比如 debug 格式）
81            format_object_summary(self.mapper.weak()),
82            // 2. 容器信息
83            format_object_summary(self.container.weak()),
84            // 3. 进度信息
85            self.current_index,
86            self.container
87                .weak()
88                .with_data(|c| {
89                    Ok(if let OnionObject::Tuple(t) = c {
90                        t.get_elements().len()
91                    } else {
92                        0 // Or some other placeholder like "N/A"
93                    })
94                })
95                .unwrap_or(0), // Provide a default if weak link is dead
96            // 4. 已收集结果的数量
97            self.collected.len()
98        )
99    }
100}
onion_vm/lambda/scheduler/map_scheduler.rs

onion_vm/lambda/scheduler/
map_scheduler.rs
