lifecycle.rs - source
Description: Source of the Rust file `library/std/src/thread/lifecycle.rs`.

#
std/thread/
lifecycle.rs

    1//! The inner logic for thread spawning and joining.
    2
    3use super::current::set_current;
    4use super::id::ThreadId;
    5use super::scoped::ScopeData;
    6use super::thread::Thread;
    7use super::{Result, spawnhook};
    8use crate::cell::UnsafeCell;
    9use crate::marker::PhantomData;
    10use crate::mem::{ManuallyDrop, MaybeUninit};
    11use crate::sync::Arc;
    12use crate::sync::atomic::{Atomic, AtomicUsize, Ordering};
    13use crate::sys::{AsInner, IntoInner, thread as imp};
    14use crate::{env, io, panic};
    15
    16#[cfg_attr(miri, track_caller)] // even without panics, this helps for Miri backtraces
    17pub(super) unsafe fn spawn_unchecked<'scope, F, T>(
    18    name: Option<String>,
    19    stack_size: Option<usize>,
    20    no_hooks: bool,
    21    scope_data: Option<Arc<ScopeData>>,
    22    f: F,
    23) -> io::Result<JoinInner<'scope, T>>
    24where
    25    F: FnOnce() -> T,
    26    F: Send,
    27    T: Send,
    28{
    29    let stack_size = stack_size.unwrap_or_else(|| {
    30        static MIN: Atomic<usize> = AtomicUsize::new(0);
    31
    32        match MIN.load(Ordering::Relaxed) {
    33            0 => {}
    34            n => return n - 1,
    35        }
    36
    37        let amt = env::var_os("RUST_MIN_STACK")
    38            .and_then(|s| s.to_str().and_then(|s| s.parse().ok()))
    39            .unwrap_or(imp::DEFAULT_MIN_STACK_SIZE);
    40
    41        // 0 is our sentinel value, so ensure that we'll never see 0 after
    42        // initialization has run
    43        MIN.store(amt + 1, Ordering::Relaxed);
    44        amt
    45    });
    46
    47    let id = ThreadId::new();
    48    let thread = Thread::new(id, name);
    49
    50    let hooks = if no_hooks {
    51        spawnhook::ChildSpawnHooks::default()
    52    } else {
    53        spawnhook::run_spawn_hooks(&thread)
    54    };
    55
    56    let my_packet: Arc<Packet<'scope, T>> =
    57        Arc::new(Packet { scope: scope_data, result: UnsafeCell::new(None), _marker: PhantomData });
    58    let their_packet = my_packet.clone();
    59
    60    // Pass `f` in `MaybeUninit` because actually that closure might *run longer than the lifetime of `F`*.
    61    // See <https://github.com/rust-lang/rust/issues/101983> for more details.
    62    // To prevent leaks we use a wrapper that drops its contents.
    63    #[repr(transparent)]
    64    struct MaybeDangling<T>(MaybeUninit<T>);
    65    impl<T> MaybeDangling<T> {
    66        fn new(x: T) -> Self {
    67            MaybeDangling(MaybeUninit::new(x))
    68        }
    69        fn into_inner(self) -> T {
    70            // Make sure we don't drop.
    71            let this = ManuallyDrop::new(self);
    72            // SAFETY: we are always initialized.
    73            unsafe { this.0.assume_init_read() }
    74        }
    75    }
    76    impl<T> Drop for MaybeDangling<T> {
    77        fn drop(&mut self) {
    78            // SAFETY: we are always initialized.
    79            unsafe { self.0.assume_init_drop() };
    80        }
    81    }
    82
    83    let f = MaybeDangling::new(f);
    84
    85    // The entrypoint of the Rust thread, after platform-specific thread
    86    // initialization is done.
    87    let rust_start = move || {
    88        let f = f.into_inner();
    89        let try_result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    90            crate::sys::backtrace::__rust_begin_short_backtrace(|| hooks.run());
    91            crate::sys::backtrace::__rust_begin_short_backtrace(f)
    92        }));
    93        // SAFETY: `their_packet` as been built just above and moved by the
    94        // closure (it is an Arc<...>) and `my_packet` will be stored in the
    95        // same `JoinInner` as this closure meaning the mutation will be
    96        // safe (not modify it and affect a value far away).
    97        unsafe { *their_packet.result.get() = Some(try_result) };
    98        // Here `their_packet` gets dropped, and if this is the last `Arc` for that packet that
    99        // will call `decrement_num_running_threads` and therefore signal that this thread is
    100        // done.
    101        drop(their_packet);
    102        // Here, the lifetime `'scope` can end. `main` keeps running for a bit
    103        // after that before returning itself.
    104    };
    105
    106    if let Some(scope_data) = &my_packet.scope {
    107        scope_data.increment_num_running_threads();
    108    }
    109
    110    // SAFETY: dynamic size and alignment of the Box remain the same. See below for why the
    111    // lifetime change is justified.
    112    let rust_start = unsafe {
    113        let ptr = Box::into_raw(Box::new(rust_start));
    114        let ptr = crate::mem::transmute::<
    115            *mut (dyn FnOnce() + Send + '_),
    116            *mut (dyn FnOnce() + Send + 'static),
    117        >(ptr);
    118        Box::from_raw(ptr)
    119    };
    120
    121    let init = Box::new(ThreadInit { handle: thread.clone(), rust_start });
    122
    123    Ok(JoinInner {
    124        // SAFETY:
    125        //
    126        // `imp::Thread::new` takes a closure with a `'static` lifetime, since it's passed
    127        // through FFI or otherwise used with low-level threading primitives that have no
    128        // notion of or way to enforce lifetimes.
    129        //
    130        // As mentioned in the `Safety` section of this function's documentation, the caller of
    131        // this function needs to guarantee that the passed-in lifetime is sufficiently long
    132        // for the lifetime of the thread.
    133        //
    134        // Similarly, the `sys` implementation must guarantee that no references to the closure
    135        // exist after the thread has terminated, which is signaled by `Thread::join`
    136        // returning.
    137        native: unsafe { imp::Thread::new(stack_size, init)? },
    138        thread,
    139        packet: my_packet,
    140    })
    141}
    142
    143/// The data passed to the spawned thread for thread initialization. Any thread
    144/// implementation should start a new thread by calling .init() on this before
    145/// doing anything else to ensure the current thread is properly initialized and
    146/// the global allocator works.
    147pub(crate) struct ThreadInit {
    148    pub handle: Thread,
    149    pub rust_start: Box<dyn FnOnce() + Send>,
    150}
    151
    152impl ThreadInit {
    153    /// Initialize the 'current thread' mechanism on this thread, returning the
    154    /// Rust entry point.
    155    pub fn init(self: Box<Self>) -> Box<dyn FnOnce() + Send> {
    156        // Set the current thread before any (de)allocations on the global allocator occur,
    157        // so that it may call std::thread::current() in its implementation. This is also
    158        // why we take Box<Self>, to ensure the Box is not destroyed until after this point.
    159        // Cloning the handle does not invoke the global allocator, it is an Arc.
    160        if let Err(_thread) = set_current(self.handle.clone()) {
    161            // The current thread should not have set yet. Use an abort to save binary size (see #123356).
    162            rtabort!("current thread handle already set during thread spawn");
    163        }
    164
    165        if let Some(name) = self.handle.cname() {
    166            imp::set_name(name);
    167        }
    168
    169        self.rust_start
    170    }
    171}
    172
    173// This packet is used to communicate the return value between the spawned
    174// thread and the rest of the program. It is shared through an `Arc` and
    175// there's no need for a mutex here because synchronization happens with `join()`
    176// (the caller will never read this packet until the thread has exited).
    177//
    178// An Arc to the packet is stored into a `JoinInner` which in turns is placed
    179// in `JoinHandle`.
    180struct Packet<'scope, T> {
    181    scope: Option<Arc<ScopeData>>,
    182    result: UnsafeCell<Option<Result<T>>>,
    183    _marker: PhantomData<Option<&'scope ScopeData>>,
    184}
    185
    186// Due to the usage of `UnsafeCell` we need to manually implement Sync.
    187// The type `T` should already always be Send (otherwise the thread could not
    188// have been created) and the Packet is Sync because all access to the
    189// `UnsafeCell` synchronized (by the `join()` boundary), and `ScopeData` is Sync.
    190unsafe impl<'scope, T: Send> Sync for Packet<'scope, T> {}
    191
    192impl<'scope, T> Drop for Packet<'scope, T> {
    193    fn drop(&mut self) {
    194        // If this packet was for a thread that ran in a scope, the thread
    195        // panicked, and nobody consumed the panic payload, we make sure
    196        // the scope function will panic.
    197        let unhandled_panic = matches!(self.result.get_mut(), Some(Err(_)));
    198        // Drop the result without causing unwinding.
    199        // This is only relevant for threads that aren't join()ed, as
    200        // join() will take the `result` and set it to None, such that
    201        // there is nothing left to drop here.
    202        // If this panics, we should handle that, because we're outside the
    203        // outermost `catch_unwind` of our thread.
    204        // We just abort in that case, since there's nothing else we can do.
    205        // (And even if we tried to handle it somehow, we'd also need to handle
    206        // the case where the panic payload we get out of it also panics on
    207        // drop, and so on. See issue #86027.)
    208        if let Err(_) = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    209            *self.result.get_mut() = None;
    210        })) {
    211            rtabort!("thread result panicked on drop");
    212        }
    213        // Book-keeping so the scope knows when it's done.
    214        if let Some(scope) = &self.scope {
    215            // Now that there will be no more user code running on this thread
    216            // that can use 'scope, mark the thread as 'finished'.
    217            // It's important we only do this after the `result` has been dropped,
    218            // since dropping it might still use things it borrowed from 'scope.
    219            scope.decrement_num_running_threads(unhandled_panic);
    220        }
    221    }
    222}
    223
    224/// Inner representation for JoinHandle
    225pub(super) struct JoinInner<'scope, T> {
    226    native: imp::Thread,
    227    thread: Thread,
    228    packet: Arc<Packet<'scope, T>>,
    229}
    230
    231impl<'scope, T> JoinInner<'scope, T> {
    232    pub(super) fn is_finished(&self) -> bool {
    233        Arc::strong_count(&self.packet) == 1
    234    }
    235
    236    pub(super) fn thread(&self) -> &Thread {
    237        &self.thread
    238    }
    239
    240    pub(super) fn join(mut self) -> Result<T> {
    241        self.native.join();
    242        Arc::get_mut(&mut self.packet)
    243            // FIXME(fuzzypixelz): returning an error instead of panicking here
    244            // would require updating the documentation of
    245            // `std::thread::Result`; currently we can return `Err` if and only
    246            // if the thread had panicked.
    247            .expect("threads should not terminate unexpectedly")
    248            .result
    249            .get_mut()
    250            .take()
    251            .unwrap()
    252    }
    253}
    254
    255impl<T> AsInner<imp::Thread> for JoinInner<'static, T> {
    256    fn as_inner(&self) -> &imp::Thread {
    257        &self.native
    258    }
    259}
    260
    261impl<T> IntoInner<imp::Thread> for JoinInner<'static, T> {
    262    fn into_inner(self) -> imp::Thread {
    263        self.native
    264    }
    265}