//! Core task module. //! //! # Safety //! //! The functions in this module are private to the `task` module. All of them //! should be considered `unsafe` to use, but are not marked as such since it //! would be too noisy. //! //! Make sure to consult the relevant safety section of each function before //! use.
// It doesn't make sense to enforce `unsafe_op_in_unsafe_fn` for this module because // // * This module is doing the low-level task management that requires tons of unsafe // operations. // * Excessive `unsafe {}` blocks hurt readability significantly. // TODO: replace with `#[expect(unsafe_op_in_unsafe_fn)]` after bumpping // the MSRV to 1.81.0. #![allow(unsafe_op_in_unsafe_fn)]
use std::num::NonZeroU64; #[cfg(tokio_unstable)] use std::panic::Location; use std::pin::Pin; use std::ptr::NonNull; use std::task::{Context, Poll, Waker};
/// The core of the task. /// /// Holds the future or output, depending on the stage of execution. /// /// Any changes to the layout of this struct _must_ also be reflected in the /// `const` fns in raw.rs. #[repr(C)] pub(super) struct Core<T: Future, S> { /// Scheduler used to drive this future. pub(super) scheduler: S,
/// The task's ID, used for populating `JoinError`s. pub(super) task_id: Id,
/// The source code location where the task was spawned. /// /// This is used for populating the `TaskMeta` passed to the task runtime /// hooks. #[cfg(tokio_unstable)] pub(super) spawned_at: &'static Location<'static>,
/// Either the future or the output. pub(super) stage: CoreStage<T>,
}
/// Crate public as this is also needed by the pool. #[repr(C)] pub(crate) struct Header { /// Task state. pub(super) state: State,
/// Pointer to next task, used with the injection queue. pub(super) queue_next: UnsafeCell<Option<NonNull<Header>>>,
/// Table of function pointers for executing actions on the task. pub(super) vtable: &'static Vtable,
/// This integer contains the id of the `OwnedTasks` or `LocalOwnedTasks` /// that this task is stored in. If the task is not in any list, should be /// the id of the list that it was previously in, or `None` if it has never /// been in any list. /// /// Once a task has been bound to a list, it can never be bound to another /// list, even if removed from the first list. /// /// The id is not unset when removed from a list because we want to be able /// to read the id without synchronization, even if it is concurrently being /// removed from the list. pub(super) owner_id: UnsafeCell<Option<NonZeroU64>>,
/// The tracing ID for this instrumented task. #[cfg(all(tokio_unstable, feature = "tracing"))] pub(super) tracing_id: Option<tracing::Id>,
}
unsafeimpl Send for Header {} unsafeimpl Sync for Header {}
/// Cold data is stored after the future. Data is considered cold if it is only /// used during creation or shutdown of the task. pub(super) struct Trailer { /// Pointers for the linked list in the `OwnedTasks` that owns this task. pub(super) owned: linked_list::Pointers<Header>, /// Consumer task waiting on completion of this task. pub(super) waker: UnsafeCell<Option<Waker>>, /// Optional hooks needed in the harness. #[cfg_attr(not(tokio_unstable), allow(dead_code))] //TODO: remove when hooks are stabilized pub(super) hooks: TaskHarnessScheduleHooks,
}
/// Either the future or the output. #[repr(C)] // https://github.com/rust-lang/miri/issues/3780 pub(super) enum Stage<T: Future> {
Running(T),
Finished(super::Result<T::Output>),
Consumed,
}
#[cfg(debug_assertions)]
{ // Using a separate function for this code avoids instantiating it separately for every `T`. unsafefn check<S>(
header: &Header,
trailer: &Trailer,
scheduler: &S,
task_id: &Id, #[cfg(tokio_unstable)] spawn_location: &&'static Location<'static>,
) { let trailer_addr = trailer as *const Trailer as usize; let trailer_ptr = unsafe { Header::get_trailer(NonNull::from(header)) };
assert_eq!(trailer_addr, trailer_ptr.as_ptr() as usize);
let scheduler_addr = scheduler as *const S as usize; let scheduler_ptr = unsafe { Header::get_scheduler::<S>(NonNull::from(header)) };
assert_eq!(scheduler_addr, scheduler_ptr.as_ptr() as usize);
let id_addr = task_id as *const Id as usize; let id_ptr = unsafe { Header::get_id_ptr(NonNull::from(header)) };
assert_eq!(id_addr, id_ptr.as_ptr() as usize);
#[cfg(tokio_unstable)]
{ let spawn_location_addr =
spawn_location as *const &'static Location<'static> as usize; let spawn_location_ptr = unsafe { Header::get_spawn_location_ptr(NonNull::from(header)) };
assert_eq!(spawn_location_addr, spawn_location_ptr.as_ptr() as usize);
}
} unsafe {
check(
&result.header,
&result.trailer,
&result.core.scheduler,
&result.core.task_id, #[cfg(tokio_unstable)]
&result.core.spawned_at,
);
}
}
/// Set and clear the task id in the context when the future is executed or /// dropped, or when the output produced by the future is dropped. pub(crate) struct TaskIdGuard {
parent_task_id: Option<Id>,
}
impl Drop for TaskIdGuard { fn drop(&mutself) {
context::set_current_task_id(self.parent_task_id);
}
}
impl<T: Future, S: Schedule> Core<T, S> { /// Polls the future. /// /// # Safety /// /// The caller must ensure it is safe to mutate the `state` field. This /// requires ensuring mutual exclusion between any concurrent thread that /// might modify the future or output field. /// /// The mutual exclusion is implemented by `Harness` and the `Lifecycle` /// component of the task state. /// /// `self` must also be pinned. This is handled by storing the task on the /// heap. pub(super) fn poll(&self, mut cx: Context<'_>) -> Poll<T::Output> { let res = { self.stage.stage.with_mut(|ptr| { // Safety: The caller ensures mutual exclusion to the field. let future = matchunsafe { &mut *ptr } {
Stage::Running(future) => future,
_ => unreachable!("unexpected stage"),
};
// Safety: The caller ensures the future is pinned. let future = unsafe { Pin::new_unchecked(future) };
let _guard = TaskIdGuard::enter(self.task_id);
future.poll(&mut cx)
})
};
if res.is_ready() { self.drop_future_or_output();
}
res
}
/// Drops the future. /// /// # Safety /// /// The caller must ensure it is safe to mutate the `stage` field. pub(super) fn drop_future_or_output(&self) { // Safety: the caller ensures mutual exclusion to the field. unsafe { self.set_stage(Stage::Consumed);
}
}
/// Stores the task output. /// /// # Safety /// /// The caller must ensure it is safe to mutate the `stage` field. pub(super) fn store_output(&self, output: super::Result<T::Output>) { // Safety: the caller ensures mutual exclusion to the field. unsafe { self.set_stage(Stage::Finished(output));
}
}
/// Takes the task output. /// /// # Safety /// /// The caller must ensure it is safe to mutate the `stage` field. pub(super) fn take_output(&self) -> super::Result<T::Output> { use std::mem;
self.stage.stage.with_mut(|ptr| { // Safety:: the caller ensures mutual exclusion to the field. match mem::replace(unsafe { &mut *ptr }, Stage::Consumed) {
Stage::Finished(output) => output,
_ => panic!("JoinHandle polled after completion"),
}
})
}
// safety: The caller must guarantee exclusive access to this field, and // must ensure that the id is either `None` or the id of the OwnedTasks // containing this task. pub(super) unsafefn set_owner_id(&self, owner: NonZeroU64) { self.owner_id.with_mut(|ptr| *ptr = Some(owner));
}
pub(super) fn get_owner_id(&self) -> Option<NonZeroU64> { // safety: If there are concurrent writes, then that write has violated // the safety requirements on `set_owner_id`. unsafe { self.owner_id.with(|ptr| *ptr) }
}
/// Gets a pointer to the `Trailer` of the task containing this `Header`. /// /// # Safety /// /// The provided raw pointer must point at the header of a task. pub(super) unsafefn get_trailer(me: NonNull<Header>) -> NonNull<Trailer> { let offset = me.as_ref().vtable.trailer_offset; let trailer = me.as_ptr().cast::<u8>().add(offset).cast::<Trailer>();
NonNull::new_unchecked(trailer)
}
/// Gets a pointer to the scheduler of the task containing this `Header`. /// /// # Safety /// /// The provided raw pointer must point at the header of a task. /// /// The generic type S must be set to the correct scheduler type for this /// task. pub(super) unsafefn get_scheduler<S>(me: NonNull<Header>) -> NonNull<S> { let offset = me.as_ref().vtable.scheduler_offset; let scheduler = me.as_ptr().cast::<u8>().add(offset).cast::<S>();
NonNull::new_unchecked(scheduler)
}
/// Gets a pointer to the id of the task containing this `Header`. /// /// # Safety /// /// The provided raw pointer must point at the header of a task. pub(super) unsafefn get_id_ptr(me: NonNull<Header>) -> NonNull<Id> { let offset = me.as_ref().vtable.id_offset; let id = me.as_ptr().cast::<u8>().add(offset).cast::<Id>();
NonNull::new_unchecked(id)
}
/// Gets the id of the task containing this `Header`. /// /// # Safety /// /// The provided raw pointer must point at the header of a task. pub(super) unsafefn get_id(me: NonNull<Header>) -> Id { let ptr = Header::get_id_ptr(me).as_ptr();
*ptr
}
/// Gets a pointer to the source code location where the task containing /// this `Header` was spawned. /// /// # Safety /// /// The provided raw pointer must point at the header of a task. #[cfg(tokio_unstable)] pub(super) unsafefn get_spawn_location_ptr(
me: NonNull<Header>,
) -> NonNull<&'static Location<'static>> { let offset = me.as_ref().vtable.spawn_location_offset; let spawned_at = me
.as_ptr()
.cast::<u8>()
.add(offset)
.cast::<&'static Location<'static>>();
NonNull::new_unchecked(spawned_at)
}
/// Gets the source code location where the task containing /// this `Header` was spawned /// /// # Safety /// /// The provided raw pointer must point at the header of a task. #[cfg(tokio_unstable)] pub(super) unsafefn get_spawn_location(me: NonNull<Header>) -> &'static Location<'static> { let ptr = Header::get_spawn_location_ptr(me).as_ptr();
*ptr
}
/// Gets the tracing id of the task containing this `Header`. /// /// # Safety /// /// The provided raw pointer must point at the header of a task. #[cfg(all(tokio_unstable, feature = "tracing"))] pub(super) unsafefn get_tracing_id(me: &NonNull<Header>) -> Option<&tracing::Id> {
me.as_ref().tracing_id.as_ref()
}
}
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.