tokio 1.52.1

An event-driven, non-blocking I/O platform for writing asynchronous I/O backed applications.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

use crate::runtime::task::{Header, RawTask, Schedule};

use std::marker::PhantomData;
use std::mem::ManuallyDrop;
use std::ops;
use std::ptr::NonNull;
use std::task::{RawWaker, RawWakerVTable, Waker};

pub(super) struct WakerRef<'a, S: 'static> {
    waker: ManuallyDrop<Waker>,
    _p: PhantomData<(&'a Header, S)>,
}

/// Returns a `WakerRef` which avoids having to preemptively increase the
/// refcount if there is no need to do so.
pub(super) fn waker_ref<S>(header: &NonNull<Header>) -> WakerRef<'_, S>
where
    S: Schedule,
{
    // `Waker::will_wake` uses the VTABLE pointer as part of the check. This
    // means that `will_wake` will always return false when using the current
    // task's waker. (discussion at rust-lang/rust#66281).
    //
    // To fix this, we use a single vtable. Since we pass in a reference at this
    // point and not an *owned* waker, we must ensure that `drop` is never
    // called on this waker instance. This is done by wrapping it with
    // `ManuallyDrop` and then never calling drop.
    let waker = unsafe { ManuallyDrop::new(Waker::from_raw(raw_waker(*header))) };

    WakerRef {
        waker,
        _p: PhantomData,
    }
}

impl<S> ops::Deref for WakerRef<'_, S> {
    type Target = Waker;

    fn deref(&self) -> &Waker {
        &self.waker
    }
}

cfg_trace! {
    /// # Safety
    ///
    /// `$header` must be a valid pointer to a [`Header`].
    macro_rules! trace {
        ($header:expr, $op:expr) => {
            if let Some(id) = Header::get_tracing_id(&$header) {
                tracing::trace!(
                    target: "tokio::task::waker",
                    op = $op,
                    task.id = id.into_u64(),
                );
            }
        }
    }
}

cfg_not_trace! {
    macro_rules! trace {
        ($header:expr, $op:expr) => {
            // noop
            let _ = &$header;
        }
    }
}

unsafe fn clone_waker(ptr: *const ()) -> RawWaker {
    // Safety: `ptr` was created from a `Header` pointer in function `waker_ref`.
    let header = unsafe { NonNull::new_unchecked(ptr as *mut Header) };
    #[cfg_attr(not(all(tokio_unstable, feature = "tracing")), allow(unused_unsafe))]
    unsafe {
        trace!(header, "waker.clone");
    }
    unsafe { header.as_ref() }.state.ref_inc();
    raw_waker(header)
}

unsafe fn drop_waker(ptr: *const ()) {
    // Safety: `ptr` was created from a `Header` pointer in function `waker_ref`.
    let ptr = unsafe { NonNull::new_unchecked(ptr as *mut Header) };
    // TODO; replace to #[expect(unused_unsafe)] after bumping MSRV to 1.81.0.
    #[cfg_attr(not(all(tokio_unstable, feature = "tracing")), allow(unused_unsafe))]
    unsafe {
        trace!(ptr, "waker.drop");
    }
    let raw = unsafe { RawTask::from_raw(ptr) };
    raw.drop_reference();
}

unsafe fn wake_by_val(ptr: *const ()) {
    // Safety: `ptr` was created from a `Header` pointer in function `waker_ref`.
    let ptr = unsafe { NonNull::new_unchecked(ptr as *mut Header) };
    // TODO; replace to #[expect(unused_unsafe)] after bumping MSRV to 1.81.0.
    #[cfg_attr(not(all(tokio_unstable, feature = "tracing")), allow(unused_unsafe))]
    unsafe {
        trace!(ptr, "waker.wake");
    }
    let raw = unsafe { RawTask::from_raw(ptr) };
    raw.wake_by_val();
}

// Wake without consuming the waker
unsafe fn wake_by_ref(ptr: *const ()) {
    // Safety: `ptr` was created from a `Header` pointer in function `waker_ref`.
    let ptr = unsafe { NonNull::new_unchecked(ptr as *mut Header) };
    // TODO; replace to #[expect(unused_unsafe)] after bumping MSRV to 1.81.0.
    #[cfg_attr(not(all(tokio_unstable, feature = "tracing")), allow(unused_unsafe))]
    unsafe {
        trace!(ptr, "waker.wake_by_ref");
    }
    let raw = unsafe { RawTask::from_raw(ptr) };
    raw.wake_by_ref();
}

static WAKER_VTABLE: RawWakerVTable =
    RawWakerVTable::new(clone_waker, wake_by_val, wake_by_ref, drop_waker);

fn raw_waker(header: NonNull<Header>) -> RawWaker {
    let ptr = header.as_ptr() as *const ();
    RawWaker::new(ptr, &WAKER_VTABLE)
}