onion_vm/lambda/scheduler/
map_scheduler.rs

1//! Map 调度器实现：用于对容器（如元组）中的每个元素应用映射器函数，并收集结果。
2//!
3//! 该模块实现了 Mapping 结构体及其 Runnable trait，用于在 Onion 虚拟机中高效地进行 map 操作。
4//! 支持惰性调度、逐步执行和结果收集，适用于异步/协作式执行环境。
5use arc_gc::gc::GC;
6
7use crate::{
8    lambda::runnable::{Runnable, RuntimeError, StepResult},
9    types::{
10        lambda::launcher::OnionLambdaRunnableLauncher,
11        object::{OnionObject, OnionObjectCell, OnionStaticObject},
12        tuple::OnionTuple,
13    },
14    utils::format_object_summary,
15};
16
17/// 表示一次 map 操作的调度状态。
18///
19/// - `container`：待映射的容器对象（通常为元组）
20/// - `mapper`：映射器函数对象（lambda 或其他）
21/// - `collected`：已收集的映射结果
22/// - `current_index`：当前处理到的元素索引
23#[derive(Clone)]
24pub struct Mapping {
25    /// 待映射的容器对象（如元组）
26    container: OnionStaticObject,
27    /// 映射器函数对象
28    mapper: OnionStaticObject,
29    /// 已收集的结果
30    collected: Vec<OnionStaticObject>,
31    /// 当前处理的元素索引
32    current_index: usize,
33}
34
35impl Mapping {
36    /// 创建一个新的 Mapping 实例。
37    ///
38    /// # 参数
39    /// * `container` - 待映射的容器对象（通常为元组）
40    /// * `mapper` - 映射器函数对象（lambda 或其他）
41    ///
42    /// # 返回值
43    /// 返回新创建的 Mapping 实例，初始状态下未处理任何元素。
44    pub fn new(container: &OnionStaticObject, mapper: &OnionStaticObject) -> Self {
45        Mapping {
46            container: container.clone(),
47            mapper: mapper.clone(),
48            collected: vec![],
49            current_index: 0,
50        }
51    }
52}
53
54impl Runnable for Mapping {
55    /// 接收子任务的执行结果，并收集到结果列表。
56    ///
57    /// # 参数
58    /// * `step_result` - 子任务的执行结果，期望为 StepResult::Return
59    /// * `_gc` - 垃圾收集器引用（未使用）
60    ///
61    /// # 返回值
62    /// * Ok(()) - 成功收集结果并推进索引
63    /// * Err(RuntimeError) - 收到非预期结果类型
64    fn receive(
65        &mut self,
66        step_result: &StepResult,
67        _gc: &mut GC<OnionObjectCell>,
68    ) -> Result<(), RuntimeError> {
69        match step_result {
70            StepResult::Return(result) => {
71                self.collected.push(result.as_ref().clone());
72                self.current_index += 1; // 移动到下一个元素
73                Ok(())
74            }
75            _ => Err(RuntimeError::DetailedError(
76                "Unexpected step result in mapping".into(),
77            )),
78        }
79    }
80
81    /// 执行 map 操作的一个调度步骤。
82    ///
83    /// - 若当前元素存在，尝试用 mapper 处理之：
84    ///   - 若 mapper 是 lambda，则生成新 Runnable 进行调用
85    ///   - 若 mapper 为 false，跳过当前元素
86    ///   - 其他情况直接收集元素
87    /// - 若所有元素处理完毕，返回收集到的新元组
88    ///
89    /// # 返回值
90    /// * StepResult::NewRunnable - 需要调度新任务
91    /// * StepResult::Continue - 继续下一个元素
92    /// * StepResult::Return - 所有元素处理完毕，返回结果
93    /// * StepResult::Error - 类型错误或其他异常
94    fn step(&mut self, _gc: &mut GC<OnionObjectCell>) -> StepResult {
95        self.container
96            .weak()
97            .with_data(|container| match container {
98                OnionObject::Tuple(tuple) => {
99                    // 使用索引获取当前元素
100                    if let Some(element) = tuple.get_elements().get(self.current_index) {
101                        let element_clone = element.clone();
102                        self.mapper.weak().with_data(|mapper_obj| match mapper_obj {
103                            OnionObject::Lambda(_) => {
104                                let runnable = Box::new(OnionLambdaRunnableLauncher::new(
105                                    mapper_obj,
106                                    element.stabilize(),
107                                    &|r| Ok(r),
108                                )?);
109                                Ok(StepResult::NewRunnable(runnable))
110                            }
111                            OnionObject::Boolean(false) => Ok(StepResult::Continue),
112                            _ => {
113                                self.collected.push(element_clone.stabilize());
114                                Ok(StepResult::Continue)
115                            }
116                        })
117                    } else {
118                        // 所有元素都处理完了
119                        Ok(StepResult::Return(
120                            OnionTuple::new_static_no_ref(&self.collected).into(),
121                        ))
122                    }
123                }
124                _ => Err(RuntimeError::InvalidType(
125                    "Container must be a tuple".into(),
126                )),
127            })
128            .unwrap_or_else(|e| StepResult::Error(e))
129    }
130
131    /// 格式化 map 调度器的当前上下文信息。
132    ///
133    /// 输出当前映射器、容器、进度和已收集结果数量，便于调试。
134    ///
135    /// # 返回值
136    /// 返回格式化的多行字符串，展示 map 操作的状态。
137    fn format_context(&self) -> String {
138        // 使用 format! 宏来构建一个多行的字符串
139        format!(
140            "-> In 'map' operation:\n   - Mapper: {}\n   - Container: {}\n   - Progress: Processing element {} / {}\n   - Collected Items: {}",
141            // 1. 映射器信息
142            // 使用对象的简略表示（比如 debug 格式）
143            format_object_summary(self.mapper.weak()),
144            // 2. 容器信息
145            format_object_summary(self.container.weak()),
146            // 3. 进度信息
147            self.current_index,
148            self.container
149                .weak()
150                .with_data(|c| {
151                    Ok(if let OnionObject::Tuple(t) = c {
152                        t.get_elements().len()
153                    } else {
154                        0 // Or some other placeholder like "N/A"
155                    })
156                })
157                .unwrap_or(0), // Provide a default if weak link is dead
158            // 4. 已收集结果的数量
159            self.collected.len()
160        )
161    }
162}
onion_vm/lambda/scheduler/map_scheduler.rs

onion_vm/lambda/scheduler/
map_scheduler.rs
