godot_core/task/
futures.rs

1/*
2 * Copyright (c) godot-rust; Bromeon and contributors.
3 * This Source Code Form is subject to the terms of the Mozilla Public
4 * License, v. 2.0. If a copy of the MPL was not distributed with this
5 * file, You can obtain one at https://mozilla.org/MPL/2.0/.
6 */
7
8use core::panic;
9use std::fmt::Display;
10use std::future::{Future, IntoFuture};
11use std::pin::Pin;
12use std::sync::{Arc, Mutex};
13use std::task::{Context, Poll, Waker};
14use std::thread::ThreadId;
15
16use crate::builtin::{Callable, RustCallable, Signal, Variant};
17use crate::classes::object::ConnectFlags;
18use crate::global::godot_error;
19use crate::meta::InParamTuple;
20use crate::meta::sealed::Sealed;
21use crate::obj::{Gd, GodotClass, WithSignals};
22use crate::signal::TypedSignal;
23use crate::sys;
24
25// ----------------------------------------------------------------------------------------------------------------------------------------------
26// Internal re-exports
27#[rustfmt::skip] // Do not reorder.
28pub(crate) use crate::impl_dynamic_send;
29
30/// The panicking counter part to the [`FallibleSignalFuture`].
31///
32/// This future works in the same way as `FallibleSignalFuture`, but panics when the signal object is freed, instead of resolving to a
33/// [`Result::Err`].
34///
35/// # Panics
36/// - If the signal object is freed before the signal has been emitted.
37/// - If one of the signal arguments is `!Send`, but the signal was emitted on a different thread.
38///
39/// During engine shutdown, this does **not** panic: when the signal object is freed before emission as part of the main loop being torn down,
40/// the awaiting task stays suspended and is dropped silently instead. This keeps "fire-and-forget" tasks from spamming errors on application exit.
41///
42/// # Keeping the signal object alive
43/// If the signal object is freed *concurrently* while the future is being dropped, connection cleanup is skipped and the stale connection is
44/// leaked (Godot removes it once the object dies), rather than aborting the process. This is only reachable by moving a `Gd` across threads,
45/// which requires `unsafe` (`Gd` is `!Send`); in safe, single-threaded code it cannot happen. If you do bypass `!Send`, it is your
46/// responsibility to keep the object alive -- retain a strong `Gd` on the runtime thread, or `join()` the other thread -- until no pending
47/// future refers to it.
48pub struct SignalFuture<R: InParamTuple + IntoDynamicSend>(FallibleSignalFuture<R>);
49
50impl<R: InParamTuple + IntoDynamicSend> SignalFuture<R> {
51    fn new(signal: Signal) -> Self {
52        Self(FallibleSignalFuture::new(signal))
53    }
54}
55
56impl<R: InParamTuple + IntoDynamicSend> Future for SignalFuture<R> {
57    type Output = R;
58
59    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
60        let poll_result = self.get_mut().0.poll(cx);
61
62        match poll_result {
63            Poll::Pending => Poll::Pending,
64            Poll::Ready(Ok(value)) => Poll::Ready(value),
65            // A freed signal object normally means a logic error -> panic. But on engine exit, the object may be freed before the
66            // engine-exiting flag is set; `SignalFutureResolver::drop` then marks the future `Dead` instead of leaving it pending. So we
67            // also check the flag here: if the engine is exiting, park silently (the runtime drops the future in `cleanup()`).
68            Poll::Ready(Err(FallibleSignalFutureError)) if crate::task::is_engine_exiting() => {
69                Poll::Pending
70            }
71            Poll::Ready(Err(FallibleSignalFutureError)) => panic!(
72                "the signal object of a SignalFuture was freed, while the future was still waiting for the signal to be emitted"
73            ),
74        }
75    }
76}
77
78// Not derived, otherwise an extra bound `Output: Default` is required.
79struct SignalFutureData<T> {
80    state: SignalFutureState<T>,
81    waker: Option<Waker>,
82}
83
84impl<T> Default for SignalFutureData<T> {
85    fn default() -> Self {
86        Self {
87            state: Default::default(),
88            waker: None,
89        }
90    }
91}
92
93// Only public for itest.
94pub struct SignalFutureResolver<R: IntoDynamicSend> {
95    data: Arc<Mutex<SignalFutureData<R::Target>>>,
96}
97
98impl<R: IntoDynamicSend> Clone for SignalFutureResolver<R> {
99    fn clone(&self) -> Self {
100        Self {
101            data: self.data.clone(),
102        }
103    }
104}
105
106/// For itest to construct and test a resolver.
107#[cfg(feature = "itest")] #[cfg_attr(published_docs, doc(cfg(feature = "itest")))]
108pub fn create_test_signal_future_resolver<R: IntoDynamicSend>() -> SignalFutureResolver<R> {
109    SignalFutureResolver {
110        data: Arc::new(Mutex::new(SignalFutureData::default())),
111    }
112}
113
114impl<R: IntoDynamicSend> SignalFutureResolver<R> {
115    fn new(data: Arc<Mutex<SignalFutureData<R::Target>>>) -> Self {
116        Self { data }
117    }
118}
119
120impl<R: IntoDynamicSend> std::hash::Hash for SignalFutureResolver<R> {
121    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
122        state.write_usize(Arc::as_ptr(&self.data) as usize);
123    }
124}
125
126impl<R: IntoDynamicSend> PartialEq for SignalFutureResolver<R> {
127    fn eq(&self, other: &Self) -> bool {
128        Arc::ptr_eq(&self.data, &other.data)
129    }
130}
131
132impl<R: InParamTuple + IntoDynamicSend> RustCallable for SignalFutureResolver<R> {
133    fn invoke(&mut self, args: &[&Variant]) -> Variant {
134        let waker = {
135            let mut data = self.data.lock().unwrap();
136            data.state = SignalFutureState::Ready(R::from_variant_array(args).into_dynamic_send());
137
138            // We no longer need the waker after we resolved. If the future is polled again, we'll also get a new waker.
139            data.waker.take()
140        };
141
142        if let Some(waker) = waker {
143            waker.wake();
144        }
145
146        Variant::nil()
147    }
148}
149
150impl<R: IntoDynamicSend> Display for SignalFutureResolver<R> {
151    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
152        write!(f, "SignalFutureResolver::<{}>", std::any::type_name::<R>())
153    }
154}
155
156// This resolver will change the futures state when it's being dropped (i.e. the engine removes all connected signal callables). By marking
157// the future as dead we can resolve it to an error value the next time it gets polled.
158impl<R: IntoDynamicSend> Drop for SignalFutureResolver<R> {
159    fn drop(&mut self) {
160        let mut data = self.data.lock().unwrap();
161
162        if !matches!(data.state, SignalFutureState::Pending) {
163            // The future is no longer pending, so no clean up is required.
164            return;
165        }
166
167        // During teardown, leave the future `Pending` (runtime drops it in `cleanup()`) instead of marking it `Dead` and waking it -> that would
168        // cause "signal object freed" error, i.e. a panic for `SignalFuture`, spamming on every exit. See `async_runtime::is_engine_exiting()`.
169        if crate::task::is_engine_exiting() {
170            return;
171        }
172
173        // We mark the future as dead, so the next time it gets polled we can react to it's inability to resolve.
174        data.state = SignalFutureState::Dead;
175
176        // If we got a waker we trigger it to get the future polled. If there is no waker, then the future has not been polled yet and we
177        // simply wait for the runtime to perform the first poll.
178        if let Some(ref waker) = data.waker {
179            waker.wake_by_ref();
180        }
181    }
182}
183
184#[derive(Default)]
185enum SignalFutureState<T> {
186    #[default]
187    Pending,
188    Ready(T),
189    Dead,
190    Dropped,
191}
192
193impl<T> SignalFutureState<T> {
194    fn take(&mut self) -> Self {
195        let new_value = match self {
196            Self::Pending => Self::Pending,
197            Self::Ready(_) | Self::Dead => Self::Dead,
198            Self::Dropped => Self::Dropped,
199        };
200
201        std::mem::replace(self, new_value)
202    }
203}
204
205/// A future that tries to resolve as soon as the provided Godot signal was emitted.
206///
207/// The future might resolve to an error if the signal object is freed before the signal is emitted.
208///
209/// # Panics
210/// - If one of the signal arguments is `!Send`, but the signal was emitted on a different thread.
211///
212/// For behavior when the signal object is freed while a future is being dropped, see
213/// [_Keeping the signal object alive_](SignalFuture#keeping-the-signal-object-alive).
214pub struct FallibleSignalFuture<R: InParamTuple + IntoDynamicSend> {
215    data: Arc<Mutex<SignalFutureData<R::Target>>>,
216    callable: SignalFutureResolver<R>,
217    signal: Signal,
218}
219
220impl<R: InParamTuple + IntoDynamicSend> FallibleSignalFuture<R> {
221    fn new(signal: Signal) -> Self {
222        sys::strict_assert!(
223            !signal.is_null(),
224            "Failed to create future for invalid signal:\n\
225            Either the signal object was already freed, or it\n\
226            was not registered in the object before being used.",
227        );
228
229        let data = Arc::new(Mutex::new(SignalFutureData::default()));
230
231        // The callable currently requires that the return value is Sync + Send.
232        let callable = SignalFutureResolver::new(data.clone());
233
234        signal.connect_flags(
235            &Callable::from_custom(callable.clone()),
236            ConnectFlags::ONE_SHOT,
237        );
238
239        Self {
240            data,
241            callable,
242            signal,
243        }
244    }
245
246    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<Result<R, FallibleSignalFutureError>> {
247        let mut data = self.data.lock().unwrap();
248
249        data.waker.replace(cx.waker().clone());
250
251        let value = data.state.take();
252
253        // Drop the data mutex lock to prevent the mutext from getting poisoned by the potential later panic.
254        drop(data);
255
256        match value {
257            SignalFutureState::Pending => Poll::Pending,
258            SignalFutureState::Dropped => unreachable!(),
259            SignalFutureState::Dead => Poll::Ready(Err(FallibleSignalFutureError)),
260            SignalFutureState::Ready(value) => {
261                let Some(value) = DynamicSend::extract_if_safe(value) else {
262                    panic!(
263                        "the awaited signal was not emitted on the main-thread, but contained a non Send argument"
264                    );
265                };
266
267                Poll::Ready(Ok(value))
268            }
269        }
270    }
271}
272
273/// Error that might be returned  by the [`FallibleSignalFuture`].
274///
275/// This error is being resolved to when the signal object is freed before the awaited singal is emitted.
276#[derive(Debug)]
277pub struct FallibleSignalFutureError;
278
279impl Display for FallibleSignalFutureError {
280    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
281        write!(
282            f,
283            "The signal object was freed before the awaited signal was emitted"
284        )
285    }
286}
287
288impl std::error::Error for FallibleSignalFutureError {}
289
290impl<R: InParamTuple + IntoDynamicSend> Future for FallibleSignalFuture<R> {
291    type Output = Result<R, FallibleSignalFutureError>;
292
293    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
294        self.get_mut().poll(cx)
295    }
296}
297
298impl<R: InParamTuple + IntoDynamicSend> Drop for FallibleSignalFuture<R> {
299    fn drop(&mut self) {
300        // The callable might alredy be destroyed, this occurs during engine shutdown.
301        if self.signal.is_null() {
302            return;
303        }
304
305        let mut data_lock = self.data.lock().unwrap();
306
307        let prev_state = std::mem::replace(&mut data_lock.state, SignalFutureState::Dropped);
308
309        drop(data_lock);
310
311        // Only the still-`Pending` future has a live connection that needs cleanup. Once the signal has fired (`Ready`) or the resolver was
312        // already dropped (`Dead`), the `ONE_SHOT` connection is gone, so there is nothing to disconnect. Skipping the object access in that
313        // case is also crucial for correctness: the signal's object may be freed concurrently (e.g. emitted from another thread that then
314        // drops the last reference), and touching it here -- inside `Drop` -- would panic and escalate to a fatal non-unwinding abort.
315        if !matches!(prev_state, SignalFutureState::Pending) {
316            return;
317        }
318
319        // We create a new Godot Callable from our RustCallable so we get independent reference counting.
320        let gd_callable = Callable::from_custom(self.callable.clone());
321
322        // The future was dropped before the signal fired, so the ONE_SHOT connection is still live and must be disconnected.
323        // Signal::object() resolves the object over separate FFI calls (validate liveness, then inc-ref), which creates a TOCTOU race:
324        // another thread holding the last Gd -- only reachable via `unsafe` (e.g. cross-thread accessor) -- can free it in between.
325        // The inc-ref would then access freed memory and panic/UB. A panic escaping `Drop` aborts the process, so we contain it:
326        // the stale connection is not disconnected, but Godot later does so when the object dies.
327        // The common case (object alive) disconnects normally.
328        //
329        // Calling is_connected()/disconnect() on the Signal directly would re-resolve the object, reopening the TOCTOU window per call.
330        // Resolving once via Signal::object() avoids that: for RefCounted the handle is a strong reference, so it stays alive while used.
331        //
332        // is_connected() is true while the signal hasn't fired yet.
333        let cleanup = || {
334            if let Some(mut object) = self.signal.object() {
335                let signal_name = self.signal.name();
336
337                if object.is_connected(&signal_name, &gd_callable) {
338                    object.disconnect(&signal_name, &gd_callable);
339                }
340            }
341        };
342
343        let context = || "FallibleSignalFuture::drop: object freed concurrently".to_string();
344        let _ = crate::private::handle_panic(context, cleanup);
345    }
346}
347
348impl Signal {
349    /// Creates a fallible future for this signal.
350    ///
351    /// The future will resolve the next time the signal is emitted.
352    /// See [`FallibleSignalFuture`] for details.
353    ///
354    /// Since the `Signal` type does not contain information on the signal argument types, the future output type has to be inferred from
355    /// the call to this function.
356    pub fn to_fallible_future<R: InParamTuple + IntoDynamicSend>(&self) -> FallibleSignalFuture<R> {
357        FallibleSignalFuture::new(self.clone())
358    }
359
360    /// Creates a future for this signal.
361    ///
362    /// The future will resolve the next time the signal is emitted, but might panic if the signal object is freed.
363    /// See [`SignalFuture`] for details.
364    ///
365    /// Since the `Signal` type does not contain information on the signal argument types, the future output type has to be inferred from
366    /// the call to this function.
367    pub fn to_future<R: InParamTuple + IntoDynamicSend>(&self) -> SignalFuture<R> {
368        SignalFuture::new(self.clone())
369    }
370}
371
372impl<C: WithSignals, R: InParamTuple + IntoDynamicSend> TypedSignal<'_, C, R> {
373    /// Creates a fallible future for this signal.
374    ///
375    /// The future will resolve the next time the signal is emitted.
376    /// See [`FallibleSignalFuture`] for details.
377    pub fn to_fallible_future(&self) -> FallibleSignalFuture<R> {
378        FallibleSignalFuture::new(self.to_untyped())
379    }
380
381    /// Creates a future for this signal.
382    ///
383    /// The future will resolve the next time the signal is emitted, but might panic if the signal object is freed.
384    /// See [`SignalFuture`] for details.
385    pub fn to_future(&self) -> SignalFuture<R> {
386        SignalFuture::new(self.to_untyped())
387    }
388}
389
390impl<C: WithSignals, R: InParamTuple + IntoDynamicSend> IntoFuture for &TypedSignal<'_, C, R> {
391    type Output = R;
392
393    type IntoFuture = SignalFuture<R>;
394
395    fn into_future(self) -> Self::IntoFuture {
396        self.to_future()
397    }
398}
399
400/// Convert a value into a type that is [`Send`] at compile-time while the value might not be.
401///
402/// This allows to turn any implementor into a type that is `Send`, but requires to also implement [`DynamicSend`] as well.
403/// The later trait will verify if a value can actually be sent between threads at runtime.
404pub trait IntoDynamicSend: Sealed + 'static {
405    type Target: DynamicSend<Inner = Self>;
406
407    fn into_dynamic_send(self) -> Self::Target;
408}
409
410/// Runtime-checked `Send` capability.
411///
412/// Implemented for types that need a static `Send` bound, but where it is determined at runtime whether sending a value was
413/// actually safe. Only allows to extract the value if sending across threads is safe, thus fulfilling the `Send` supertrait.
414///
415/// # Safety
416/// The implementor has to guarantee that `extract_if_safe` returns `None`, if the value has been sent between threads while being `!Send`.
417///
418/// To uphold the `Send` supertrait guarantees, no public API apart from `extract_if_safe` must exist that would give access to the inner value from another thread.
419pub unsafe trait DynamicSend: Send + Sealed {
420    type Inner;
421
422    fn extract_if_safe(self) -> Option<Self::Inner>;
423}
424
425/// Value that can be sent across threads, but only accessed on its original thread.
426///
427/// When moved to another thread, the inner value can no longer be accessed and will be leaked when the `ThreadConfined` is dropped.
428pub struct ThreadConfined<T> {
429    value: Option<T>,
430    thread_id: ThreadId,
431}
432
433// SAFETY: This type can always be sent across threads, but the inner value can only be accessed on its original thread.
434unsafe impl<T> Send for ThreadConfined<T> {}
435
436impl<T> ThreadConfined<T> {
437    pub(crate) fn new(value: T) -> Self {
438        Self {
439            value: Some(value),
440            thread_id: std::thread::current().id(),
441        }
442    }
443
444    /// Retrieve the inner value, if the current thread is the one in which the `ThreadConfined` was created.
445    ///
446    /// If this fails, the value will be leaked immediately.
447    pub(crate) fn extract(mut self) -> Option<T> {
448        if self.is_original_thread() {
449            self.value.take()
450        } else {
451            None // causes Drop -> leak.
452        }
453    }
454
455    fn is_original_thread(&self) -> bool {
456        self.thread_id == std::thread::current().id()
457    }
458}
459
460impl<T> Drop for ThreadConfined<T> {
461    fn drop(&mut self) {
462        if !self.is_original_thread() {
463            std::mem::forget(self.value.take());
464
465            // Cannot panic, potentially during unwind already.
466            godot_error!(
467                "Dropped ThreadConfined<T> on a different thread than it was created on. The inner T value will be leaked."
468            );
469        }
470    }
471}
472
473unsafe impl<T: GodotClass> DynamicSend for ThreadConfined<Gd<T>> {
474    type Inner = Gd<T>;
475
476    fn extract_if_safe(self) -> Option<Self::Inner> {
477        self.extract()
478    }
479}
480
481impl<T: GodotClass> Sealed for ThreadConfined<Gd<T>> {}
482
483impl<T: GodotClass> IntoDynamicSend for Gd<T> {
484    type Target = ThreadConfined<Self>;
485
486    fn into_dynamic_send(self) -> Self::Target {
487        ThreadConfined::new(self)
488    }
489}
490
491// ----------------------------------------------------------------------------------------------------------------------------------------------
492// Generated impls
493
494#[macro_export(local_inner_macros)]
495macro_rules! impl_dynamic_send {
496    (Send; $($ty:ty),+) => {
497        $(
498            unsafe impl $crate::task::DynamicSend for $ty {
499                type Inner = Self;
500
501                fn extract_if_safe(self) -> Option<Self::Inner> {
502                    Some(self)
503                }
504            }
505
506            impl $crate::task::IntoDynamicSend for $ty {
507                type Target = Self;
508                fn into_dynamic_send(self) -> Self::Target {
509                    self
510                }
511            }
512        )+
513    };
514
515    (tuple; $($arg:ident: $ty:ident),*) => {
516        unsafe impl<$($ty: $crate::task::DynamicSend ),*> $crate::task::DynamicSend for ($($ty,)*) {
517            type Inner = ($($ty::Inner,)*);
518
519            fn extract_if_safe(self) -> Option<Self::Inner> {
520                #[allow(non_snake_case)]
521                let ($($arg,)*) = self;
522
523                #[allow(clippy::unused_unit)]
524                match ($($arg.extract_if_safe(),)*) {
525                    ($(Some($arg),)*) => Some(($($arg,)*)),
526
527                    #[allow(unreachable_patterns)]
528                    _ => None,
529                }
530            }
531        }
532
533        impl<$($ty: $crate::task::IntoDynamicSend),*> $crate::task::IntoDynamicSend for ($($ty,)*) {
534            type Target = ($($ty::Target,)*);
535
536            fn into_dynamic_send(self) -> Self::Target {
537                #[allow(non_snake_case)]
538                let ($($arg,)*) = self;
539
540                #[allow(clippy::unused_unit)]
541                ($($arg.into_dynamic_send(),)*)
542            }
543        }
544    };
545
546    (!Send; $($ty:ident),+) => {
547        $(
548            impl $crate::meta::sealed::Sealed for $crate::task::ThreadConfined<$crate::builtin::$ty> {}
549
550            unsafe impl $crate::task::DynamicSend for $crate::task::ThreadConfined<$crate::builtin::$ty> {
551                type Inner = $crate::builtin::$ty;
552
553                fn extract_if_safe(self) -> Option<Self::Inner> {
554                    self.extract()
555                }
556            }
557
558            impl $crate::task::IntoDynamicSend for $crate::builtin::$ty {
559                type Target = $crate::task::ThreadConfined<$crate::builtin::$ty>;
560
561                fn into_dynamic_send(self) -> Self::Target {
562                    $crate::task::ThreadConfined::new(self)
563                }
564            }
565        )+
566    };
567}
568
569#[cfg(test)] #[cfg_attr(published_docs, doc(cfg(test)))]
570mod tests {
571    use std::sync::Arc;
572    use std::sync::atomic::{AtomicUsize, Ordering};
573    use std::thread;
574
575    use super::{SignalFutureResolver, ThreadConfined};
576    use crate::classes::Object;
577    use crate::obj::Gd;
578    use crate::sys;
579
580    /// Test that the hash of a cloned future resolver is equal to its original version. With this equality in place, we can create new
581    /// Callables that are equal to their original version but have separate reference counting.
582    #[test]
583    fn future_resolver_cloned_hash() {
584        let resolver_a = SignalFutureResolver::<(Gd<Object>, i64)>::new(Arc::default());
585        let resolver_b = resolver_a.clone();
586
587        let hash_a = sys::hash_value(&resolver_a);
588        let hash_b = sys::hash_value(&resolver_b);
589
590        assert_eq!(hash_a, hash_b);
591    }
592
593    // Test that dropping ThreadConfined<T> on another thread leaks the inner value.
594    #[test]
595    #[cfg_attr(
596        all(target_family = "wasm", not(target_feature = "atomics")),
597        ignore = "Threading not available"
598    )]
599    fn thread_confined_extract() {
600        let confined = ThreadConfined::new(772);
601        assert_eq!(confined.extract(), Some(772));
602
603        let confined = ThreadConfined::new(772);
604
605        let handle = thread::spawn(move || {
606            assert!(confined.extract().is_none());
607        });
608        handle.join().unwrap();
609    }
610
611    #[test]
612    #[cfg_attr(
613        all(target_family = "wasm", not(target_feature = "atomics")),
614        ignore = "Threading not available"
615    )]
616    fn thread_confined_leak_on_other_thread() {
617        static COUNTER: AtomicUsize = AtomicUsize::new(0);
618
619        struct DropCounter;
620        impl Drop for DropCounter {
621            fn drop(&mut self) {
622                COUNTER.fetch_add(1, Ordering::SeqCst);
623            }
624        }
625
626        let drop_counter = DropCounter;
627        let confined = ThreadConfined::new(drop_counter);
628
629        let handle = thread::spawn(move || drop(confined));
630        handle.join().unwrap();
631
632        // The counter should still be 0, meaning Drop was not called (leaked).
633        assert_eq!(COUNTER.load(Ordering::SeqCst), 0);
634    }
635}
godot_core/task/futures.rs

godot_core/task/
futures.rs
