onion_vm/types/
thread_handle.rs

1use std::{
2    any::Any,
3    collections::VecDeque,
4    fmt::{Debug, Formatter},
5    sync::Mutex,
6    thread::JoinHandle,
7};
8
9use arc_gc::{
10    arc::{GCArc, GCArcWeak},
11    gc::GC,
12    traceable::GCTraceable,
13};
14
15use crate::{
16    lambda::runnable::RuntimeError,
17    types::object::{GCArcStorage, OnionStaticObject},
18};
19
20use super::object::{OnionObject, OnionObjectCell, OnionObjectExt};
21
22/// A wrapper around JoinHandle to make it work with OnionObject
23pub struct OnionThreadHandle {
24    inner: Mutex<(
25        Option<JoinHandle<Result<Box<OnionStaticObject>, RuntimeError>>>,
26        bool,
27        GCArcWeak<OnionObjectCell>,
28    )>,
29}
30
31impl OnionThreadHandle {
32    pub fn new(
33        handle: JoinHandle<Result<Box<OnionStaticObject>, RuntimeError>>,
34        gc: &mut GC<OnionObjectCell>,
35    ) -> (Self, GCArcStorage) {
36        let tmp = gc.create(OnionObjectCell::from(OnionObject::Undefined(None)));
37        (
38            Self {
39                inner: Mutex::new((Some(handle), false, tmp.as_weak())),
40            },
41            GCArcStorage::Single(tmp),
42        )
43    }
44    /// Check if the thread has finished without blocking
45    pub fn is_finished(&self) -> bool {
46        let guard = self.inner.lock().unwrap();
47        guard.1 || guard.0.is_none()
48    }
49}
50
51impl Debug for OnionThreadHandle {
52    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
53        let guard = self.inner.lock().unwrap();
54        write!(
55            f,
56            "OnionThreadHandle(finished: {}, has_handle: {})",
57            guard.1 || guard.0.is_none(),
58            guard.0.is_some()
59        )
60    }
61}
62
63impl GCTraceable<OnionObjectCell> for OnionThreadHandle {
64    fn collect(&self, queue: &mut VecDeque<GCArcWeak<OnionObjectCell>>) {
65        let guard = self.inner.lock().unwrap();
66        queue.push_back(guard.2.clone());
67    }
68}
69
70impl OnionObjectExt for OnionThreadHandle {
71    fn as_any(&self) -> &dyn Any {
72        self
73    }
74
75    fn repr(&self, _ptrs: &Vec<*const OnionObject>) -> Result<String, RuntimeError> {
76        Ok(format!("ThreadHandle(finished: {})", self.is_finished()))
77    }
78
79    fn equals(&self, _other: &OnionObject) -> Result<bool, RuntimeError> {
80        Ok(false)
81    }
82
83    fn upgrade(&self, collected: &mut Vec<GCArc<OnionObjectCell>>) {
84        let guard = self.inner.lock().unwrap();
85        if let Some(strong) = guard.2.upgrade() {
86            collected.push(strong);
87        }
88    }
89    fn is_same(&self, other: &OnionObject) -> Result<bool, RuntimeError> {
90        if let OnionObject::Custom(other_custom) = other {
91            if let Some(other_handle) = other_custom.as_any().downcast_ref::<OnionThreadHandle>() {
92                // Use pointer equality to check if it's the same Arc
93                Ok(std::ptr::eq(self, other_handle))
94            } else {
95                Ok(false)
96            }
97        } else {
98            Ok(false)
99        }
100    }
101
102    fn to_boolean(&self) -> Result<bool, RuntimeError> {
103        // A thread handle is "truthy" if it hasn't finished yet
104        Ok(!self.is_finished())
105    }
106
107    fn type_of(&self) -> Result<String, RuntimeError> {
108        Ok("ThreadHandle".to_string())
109    }
110
111    fn value_of(&self) -> Result<OnionStaticObject, RuntimeError> {
112        let mut guard = self.inner.lock().unwrap();
113        // 句柄已经被取走，意味着线程已经结束
114        if guard.0.is_none() {
115            return if let Some(strong_ref) = guard.2.upgrade() {
116                strong_ref.as_ref().with_data(|data| Ok(data.stabilize()))
117            } else {
118                Err(RuntimeError::BrokenReference)
119            };
120        }
121        // 非阻塞地检查线程是否结束，通过handle来检查
122        if let Some(handle) = guard.0.as_ref() {
123            if !handle.is_finished() {
124                return Err(RuntimeError::Pending);
125            }
126        }
127
128        // 线程已经完成，可以安全地 join 并获取结果
129        if let Some(handle) = guard.0.take() {
130            guard.1 = true;
131            drop(guard); // 释放锁
132            match handle.join() {
133                Ok(result) => match result {
134                    Ok(obj) => {
135                        // 由于Handle是抽象的Mut容器，因此应当将结果写入GCArcWeak中
136
137                        // 步骤 1: 克隆结果，不持有任何目标锁。
138                        // 我们从 obj 中提取出最终要写入的值。
139                        // 这里的 clone() 是浅拷贝，非常快。
140                        // 这一步之后，new_value 就与 obj 的锁解耦了。
141                        let new_value = obj.weak().clone();
142
143                        // 步骤 2: 获取目标锁并写入。
144                        match self.inner.lock().unwrap().2.upgrade() {
145                            Some(arc) => {
146                                // 现在我们可以安全地获取写锁，因为我们不再需要访问 obj。
147                                arc.as_ref().with_data_mut(|to| {
148                                    *to = new_value;
149                                    Ok(())
150                                })?;
151                            }
152                            None => return Err(RuntimeError::BrokenReference),
153                        };
154
155                        // 注意：这里返回的 obj 是 join 的原始结果，
156                        // 而不是我们刚刚写入的值，这在逻辑上是正确的。
157                        // Ok(*obj) 表示线程成功返回了这个 OnionStaticObject。
158                        return Ok(*obj);
159                    }
160                    Err(err) => Err(err),
161                },
162                Err(_) => Err(RuntimeError::DetailedError(
163                    "Thread join failed".into(),
164                )),
165            }
166        } else {
167            Err(RuntimeError::DetailedError(
168                "Failed to take thread handle after finished check"
169                    .to_string()
170                    .into(),
171            ))
172        }
173    }
174
175    fn with_attribute(
176        &self,
177        key: &OnionObject,
178        f: &mut dyn FnMut(&OnionObject) -> Result<(), RuntimeError>,
179    ) -> Result<(), RuntimeError> {
180        match key {
181            OnionObject::String(s) => match s.as_ref() {
182                "is_finished" => f(&OnionObject::Boolean(self.is_finished())),
183                "has_handle" => {
184                    let guard = self.inner.lock().unwrap();
185                    let has_handle = guard.0.is_some();
186                    f(&OnionObject::Boolean(has_handle))
187                }
188                "has_result" => {
189                    let guard = self.inner.lock().unwrap();
190                    let has_result = guard.2.upgrade().is_some();
191                    f(&OnionObject::Boolean(has_result))
192                }
193                _ => Err(RuntimeError::InvalidOperation(
194                    format!("Attribute '{}' not found in ThreadHandle", s).into(),
195                )),
196            },
197            _ => Err(RuntimeError::InvalidOperation(
198                format!("Attribute {:?} not found in ThreadHandle", key).into(),
199            )),
200        }
201    }
202}
onion_vm/types/thread_handle.rs

onion_vm/types/
thread_handle.rs
