1use std::cell::RefCell;
33use std::collections::HashMap;
34use std::fmt;
35use std::rc::Rc;
36use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
37use std::sync::{Arc, Mutex};
38use std::thread::{self, ThreadId};
39
40thread_local! {
41 static WORK: RefCell<HashMap<u64, Rc<dyn Fn()>>> = RefCell::new(HashMap::new());
43}
44
45static NEXT: AtomicU64 = AtomicU64::new(1);
46
47static ABANDONED: Mutex<Vec<(ThreadId, u64)>> = Mutex::new(Vec::new());
50static ANY_ABANDONED: AtomicBool = AtomicBool::new(false);
51
52struct Slot {
54 id: u64,
55 owner: ThreadId,
56}
57
58impl Drop for Slot {
59 fn drop(&mut self) {
60 if thread::current().id() == self.owner {
61 let removed = WORK.try_with(|work| work.borrow_mut().remove(&self.id));
64 drop(removed);
65 } else if let Ok(mut abandoned) = ABANDONED.lock() {
66 abandoned.push((self.owner, self.id));
67 ANY_ABANDONED.store(true, Ordering::Release);
68 }
69 }
70}
71
72#[inline(never)]
74fn collect_abandoned() {
75 if !ANY_ABANDONED.load(Ordering::Acquire) {
76 return;
77 }
78 let me = thread::current().id();
79 let mine: Vec<u64> = match ABANDONED.lock() {
80 Ok(mut abandoned) => {
81 let (mine, others): (Vec<_>, Vec<_>) =
82 abandoned.drain(..).partition(|(owner, _)| *owner == me);
83 ANY_ABANDONED.store(!others.is_empty(), Ordering::Release);
84 *abandoned = others;
85 mine.into_iter().map(|(_, id)| id).collect()
86 }
87 Err(_) => return,
88 };
89 let removed: Vec<_> = WORK.with(|work| {
90 let mut work = work.borrow_mut();
91 mine.iter().filter_map(|id| work.remove(id)).collect()
92 });
93 drop(removed);
94}
95
96#[derive(Clone)]
101pub struct Callback {
102 slot: Arc<Slot>,
103}
104
105impl Callback {
106 pub fn new(work: impl Fn() + 'static) -> Self {
109 Self::from_work(Rc::new(work))
110 }
111
112 #[inline(never)]
116 fn from_work(work: Rc<dyn Fn()>) -> Self {
117 collect_abandoned();
118 let id = NEXT.fetch_add(1, Ordering::Relaxed);
119 WORK.with(|table| table.borrow_mut().insert(id, work));
120 Self {
121 slot: Arc::new(Slot {
122 id,
123 owner: thread::current().id(),
124 }),
125 }
126 }
127
128 #[inline(never)]
133 pub fn call(&self) {
134 collect_abandoned();
135 let work = WORK.with(|table| table.borrow().get(&self.slot.id).cloned());
136 match work {
137 Some(work) => work(),
138 None => debug_assert!(
139 thread::current().id() == self.slot.owner,
140 "a callback was run on a different thread from the one that made it"
141 ),
142 }
143 }
144
145 pub fn ptr_eq(&self, other: &Callback) -> bool {
148 self.slot.id == other.slot.id
149 }
150}
151
152impl<F: Fn() + 'static> From<F> for Callback {
153 fn from(work: F) -> Self {
154 Callback::new(work)
155 }
156}
157
158impl PartialEq for Callback {
159 fn eq(&self, other: &Self) -> bool {
162 self.ptr_eq(other)
163 }
164}
165
166impl fmt::Debug for Callback {
167 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
168 f.write_str("Callback(..)")
169 }
170}
171
172pub trait IntoMsg<Msg>: Sized {
190 fn defer(work: impl Fn() -> Self + 'static) -> Msg;
192}
193
194impl<Msg> IntoMsg<Msg> for Msg {
195 fn defer(work: impl Fn() -> Msg + 'static) -> Msg {
196 work()
197 }
198}
199
200impl IntoMsg<Callback> for () {
201 fn defer(work: impl Fn() + 'static) -> Callback {
202 Callback::new(work)
203 }
204}
205
206pub(crate) fn handler1<A, R, Msg>(f: impl Fn(A) -> R + 'static) -> impl Fn(A) -> Msg + 'static
208where
209 A: Clone + 'static,
210 R: IntoMsg<Msg>,
211{
212 let f = Rc::new(f);
213 move |a| {
214 let f = f.clone();
215 R::defer(move || f(a.clone()))
216 }
217}
218
219pub(crate) fn handler2<A, B, R, Msg>(
221 f: impl Fn(A, B) -> R + 'static,
222) -> impl Fn(A, B) -> Msg + 'static
223where
224 A: Clone + 'static,
225 B: Clone + 'static,
226 R: IntoMsg<Msg>,
227{
228 let f = Rc::new(f);
229 move |a, b| {
230 let f = f.clone();
231 R::defer(move || f(a.clone(), b.clone()))
232 }
233}
234
235pub(crate) fn handler4<A, B, C, D, R, Msg>(
237 f: impl Fn(A, B, C, D) -> R + 'static,
238) -> impl Fn(A, B, C, D) -> Msg + 'static
239where
240 A: Clone + 'static,
241 B: Clone + 'static,
242 C: Clone + 'static,
243 D: Clone + 'static,
244 R: IntoMsg<Msg>,
245{
246 let f = Rc::new(f);
247 move |a, b, c, d| {
248 let f = f.clone();
249 R::defer(move || f(a.clone(), b.clone(), c.clone(), d.clone()))
250 }
251}
252
253#[cfg(test)]
254mod tests {
255 use super::*;
256 use std::cell::Cell;
257
258 #[test]
259 fn a_callback_runs_its_work_every_time_and_through_every_clone() {
260 let n = Rc::new(Cell::new(0));
261 let inner = n.clone();
262 let cb = Callback::new(move || inner.set(inner.get() + 1));
263 cb.call();
264 cb.clone().call();
265 assert_eq!(n.get(), 2);
266 }
267
268 #[test]
269 fn a_closure_converts_where_a_callback_is_asked_for() {
270 fn takes(cb: impl Into<Callback>) -> Callback {
271 cb.into()
272 }
273 let hit = Rc::new(Cell::new(false));
274 let flag = hit.clone();
275 takes(move || flag.set(true)).call();
276 assert!(hit.get());
277 let cb = Callback::new(|| {});
279 assert!(takes(cb.clone()).ptr_eq(&cb));
280 }
281
282 #[test]
283 fn equality_is_identity_not_lookalike() {
284 let a = Callback::new(|| {});
285 assert_eq!(a, a.clone());
286 assert_ne!(a, Callback::new(|| {}));
287 }
288
289 #[test]
290 fn a_callback_is_a_message_that_may_cross_threads() {
291 fn message<T: Send + Sync + Clone + 'static>() {}
292 message::<Callback>();
293 }
294
295 #[test]
296 fn the_closure_lives_until_the_last_callback_to_it_goes() {
297 struct Flag(Rc<Cell<bool>>);
298 impl Drop for Flag {
299 fn drop(&mut self) {
300 self.0.set(true);
301 }
302 }
303 let dropped = Rc::new(Cell::new(false));
304 let held = Flag(dropped.clone());
305 let cb = Callback::new(move || {
306 let _ = &held;
307 });
308 let clone = cb.clone();
309 drop(cb);
310 assert!(!dropped.get(), "a clone still holds it");
311 clone.call();
312 drop(clone);
313 assert!(dropped.get(), "and with the last one, it is let go");
314 }
315
316 #[test]
317 fn a_callback_dropped_on_another_thread_is_let_go_by_its_owner() {
318 struct Flag(Rc<Cell<bool>>);
319 impl Drop for Flag {
320 fn drop(&mut self) {
321 self.0.set(true);
322 }
323 }
324 let dropped = Rc::new(Cell::new(false));
325 let held = Flag(dropped.clone());
326 let cb = Callback::new(move || {
327 let _ = &held;
328 });
329 thread::spawn(move || drop(cb)).join().unwrap();
330 assert!(!dropped.get(), "the closure cannot be dropped from there");
331 let _ = Callback::new(|| {});
333 assert!(dropped.get());
334 }
335
336 #[test]
337 #[cfg(debug_assertions)]
338 #[should_panic(expected = "different thread")]
339 fn running_a_callback_elsewhere_is_reported_in_a_debug_build() {
340 let cb = Callback::new(|| {});
341 thread::spawn(move || cb.call())
342 .join()
343 .unwrap_or_else(|e| std::panic::resume_unwind(e));
344 }
345
346 #[test]
347 fn a_callback_that_owns_callbacks_can_be_dropped() {
348 let inner = Callback::new(|| {});
349 let outer = Callback::new(move || inner.call());
350 outer.call();
351 drop(outer);
352 }
353}