onion_vm/types/
async_handle.rs

1use std::{
2    any::Any,
3    collections::VecDeque,
4    fmt::{Debug, Formatter},
5    sync::{Arc, Mutex},
6};
7
8use arc_gc::{
9    arc::{GCArc, GCArcWeak},
10    gc::GC,
11    traceable::GCTraceable,
12};
13
14use crate::{
15    lambda::runnable::RuntimeError,
16    types::object::{GCArcStorage, OnionStaticObject},
17};
18
19use super::object::{OnionObject, OnionObjectCell, OnionObjectExt};
20
21pub struct OnionAsyncHandle {
22    inner: Mutex<(GCArcWeak<OnionObjectCell>, bool)>,
23}
24
25impl OnionAsyncHandle {
26    // 创建一个新的异步句柄，此时没有缓存结果
27    pub fn new(gc: &mut GC<OnionObjectCell>) -> (Arc<Self>, GCArcStorage) {
28        let tmp = gc.create(OnionObjectCell::from(OnionObject::Undefined(None)));
29        (
30            Arc::new(Self {
31                inner: Mutex::new((tmp.as_weak(), false)),
32            }),
33            GCArcStorage::Single(tmp),
34        )
35    }
36
37    /// 检查 handle 是否完成（供调度器使用）
38    pub fn is_finished(&self) -> bool {
39        self.inner.lock().unwrap().1
40    }
41
42    /// 设置任务结果（供调度器使用）
43    /// 为了保证幂等律，对result使用with_data
44    pub fn set_result(&self, result: &OnionObject) -> Result<(), RuntimeError> {
45        let mut guard = self.inner.lock().unwrap();
46        guard.0.upgrade().map_or_else(
47            || Err(RuntimeError::BrokenReference),
48            |strong_ref| {
49                result.with_data(|data| {
50                    strong_ref.as_ref().with_data_mut(|strong_data| {
51                        *strong_data = data.clone();
52                        Ok(())
53                    })
54                })
55            },
56        )?;
57        guard.1 = true; // 设置为完成状态
58        Ok(())
59    }
60
61    /// 设置任务为完成状态但无结果（供调度器使用）
62    pub fn set_finished(&self) {
63        let mut guard = self.inner.lock().unwrap();
64        guard.1 = true;
65    }
66}
67
68impl Debug for OnionAsyncHandle {
69    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
70        // 只打印 is_finished 状态，避免 Debug trait 问题
71        let guard = self.inner.lock().unwrap();
72        write!(f, "OnionAsyncHandle(finished: {})", guard.1)
73    }
74}
75
76impl GCTraceable<OnionObjectCell> for OnionAsyncHandle {
77    fn collect(&self, queue: &mut VecDeque<GCArcWeak<OnionObjectCell>>) {
78        let guard = self.inner.lock().unwrap();
79        queue.push_back(guard.0.clone());
80    }
81}
82
83impl OnionObjectExt for OnionAsyncHandle {
84    fn as_any(&self) -> &dyn Any {
85        self
86    }
87
88    fn repr(&self, _ptrs: &Vec<*const OnionObject>) -> Result<String, RuntimeError> {
89        Ok(format!("AsyncHandle(finished: {})", self.is_finished()))
90    }
91
92    fn equals(&self, _other: &OnionObject) -> Result<bool, RuntimeError> {
93        // Async handles are only equal if they're the exact same handle
94        Ok(false)
95    }
96
97    fn upgrade(&self, collected: &mut Vec<GCArc<OnionObjectCell>>) {
98        let guard = self.inner.lock().unwrap();
99        if let Some(strong_ref) = guard.0.upgrade() {
100            collected.push(strong_ref);
101        }
102    }
103
104    fn is_same(&self, other: &OnionObject) -> Result<bool, RuntimeError> {
105        if let OnionObject::Custom(other_custom) = other {
106            if let Some(other_handle) = other_custom.as_any().downcast_ref::<OnionAsyncHandle>() {
107                // Use pointer equality to check if it's the same Arc
108                Ok(std::ptr::eq(self, other_handle))
109            } else {
110                Ok(false)
111            }
112        } else {
113            Ok(false)
114        }
115    }
116
117    fn to_boolean(&self) -> Result<bool, RuntimeError> {
118        // A async handle is "truthy" if it hasn't finished yet
119        Ok(!self.is_finished())
120    }
121
122    fn type_of(&self) -> Result<String, RuntimeError> {
123        Ok("AsyncHandle".to_string())
124    }
125
126    fn value_of(&self) -> Result<OnionStaticObject, RuntimeError> {
127        let guard = self.inner.lock().unwrap();
128        let (ref weak_result, is_finished) = *guard;
129        if is_finished {
130            if let Some(strong_ref) = weak_result.upgrade() {
131                // 获取对象内容并 stabilize
132                strong_ref.as_ref().with_data(|data| Ok(data.stabilize()))
133            } else {
134                // 这种情况发生在用户尝试跨GC堆传递等情况，考虑到我们没有分布式GC，直接返回 BrokenReference
135                Err(RuntimeError::BrokenReference)
136            }
137        } else {
138            Err(RuntimeError::Pending)
139        }
140    }
141
142    fn to_string(&self, ptrs: &Vec<*const OnionObject>) -> Result<String, RuntimeError> {
143        let finished = self.is_finished();
144        if finished {
145            match self.value_of() {
146                Ok(val) => Ok(format!(
147                    "AsyncHandle(finished: true, result: {})",
148                    val.weak().to_string(ptrs)?
149                )),
150                Err(RuntimeError::BrokenReference) => {
151                    Ok("AsyncHandle(finished: true, result: <broken reference>)".to_string())
152                }
153                Err(_) => Ok("AsyncHandle(finished: true, result: <unknown error>)".to_string()),
154            }
155        } else {
156            let guard = self.inner.lock().unwrap();
157            let (ref weak_result, _) = *guard;
158            match weak_result.upgrade() {
159                Some(v) => Ok(format!(
160                    "AsyncHandle(finished: false, result: {:?})",
161                    v.as_ref().0.read().unwrap().to_string(ptrs)
162                )),
163                None => Ok("AsyncHandle(finished: false, result: <broken reference>)".to_string()),
164            }
165        }
166    }
167
168    fn with_attribute(
169        &self,
170        key: &OnionObject,
171        f: &mut dyn FnMut(&OnionObject) -> Result<(), RuntimeError>,
172    ) -> Result<(), RuntimeError> {
173        match key {
174            OnionObject::String(s) => match s.as_ref() {
175                "is_finished" => f(&OnionObject::Boolean(self.is_finished())),
176                "has_result" => {
177                    let guard = self.inner.lock().unwrap();
178                    let has_result = guard.0.upgrade().is_some();
179                    f(&OnionObject::Boolean(has_result))
180                }
181                _ => Err(RuntimeError::InvalidOperation(
182                    format!("Attribute '{}' not found in AsyncHandle", s).into(),
183                )),
184            },
185            _ => Err(RuntimeError::InvalidOperation(
186                format!("Attribute {:?} not found in AsyncHandle", key).into(),
187            )),
188        }
189    }
190}
onion_vm/types/async_handle.rs

onion_vm/types/
async_handle.rs
