nuclei 0.5.2

Proactive IO & runtime system
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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! A runtime set on a thread without `block_on` ([`Runtime::enter`]), the
//! handle of the runtime a thread is inside without falling back to the
//! default one ([`RuntimeHandle::try_current`]), runtime ids, the worker a
//! thread is, and a shutdown that does not wait.

use core::fmt;
use core::marker::PhantomData;
use std::sync::Arc;

use super::{Inner, Runtime, RuntimeHandle};
use crate::rt::context;

/// A runtime's id, distinct among the runtimes of the process.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RuntimeId(u64);

impl RuntimeId {
    pub(crate) fn from_raw(id: u64) -> Self {
        Self(id)
    }
}

impl fmt::Display for RuntimeId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

/// The runtime [`Runtime::enter`] set on this thread, until it drops: the
/// free functions ([`spawn`](crate::spawn), ...) use that runtime meanwhile.
/// It stays on the thread that made it.
#[must_use = "the runtime is set only while the guard lives"]
pub struct EnterGuard<'a> {
    _set: context::RuntimeSet,
    _runtime: PhantomData<&'a ()>,
    _not_send: PhantomData<*const ()>,
}

impl fmt::Debug for EnterGuard<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("EnterGuard").finish_non_exhaustive()
    }
}

/// [`RuntimeHandle::try_current`] found no runtime set on this thread.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct TryCurrentError {
    _private: (),
}

impl TryCurrentError {
    /// Whether no runtime was set on the thread (the one reason today).
    pub fn is_missing_context(&self) -> bool {
        true
    }
}

impl fmt::Display for TryCurrentError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(
            "no runtime is set on this thread: call it inside `block_on`, a task, or `enter`",
        )
    }
}

impl std::error::Error for TryCurrentError {}

fn enter(inner: &Arc<Inner>) -> EnterGuard<'_> {
    EnterGuard {
        _set: context::set_runtime(inner),
        _runtime: PhantomData,
        _not_send: PhantomData,
    }
}

impl Runtime {
    /// Sets this runtime on the calling thread until the guard drops, so the
    /// free functions use it outside `block_on`.
    ///
    /// ```
    /// let runtime = nuclei::Runtime::builder().worker_threads(1).build().unwrap();
    /// let _guard = runtime.enter();
    /// let task = nuclei::spawn(async { 2 + 2 });
    /// assert_eq!(runtime.block_on(task), 4);
    /// ```
    pub fn enter(&self) -> EnterGuard<'_> {
        enter(&self.inner)
    }

    /// The runtime's id.
    pub fn id(&self) -> RuntimeId {
        RuntimeId(self.inner.id)
    }

    /// Stops the workers and returns at once: the worker threads drop their
    /// tasks and exit on their own, without being waited for, and what the
    /// handles of those tasks left behind is delivered on threads of its
    /// own.
    pub fn shutdown_background(self) {
        let lingering = self.inner.lingering();
        lingering.close();
        self.inner.workers.stop_all();
        // Detached: dropping the handles does not wait. `Drop` finds none
        // left to join, and no lingering to wait for.
        drop(self.inner.workers.take_threads());
        lingering.end();
    }
}

impl RuntimeHandle {
    /// Sets the handle's runtime on the calling thread until the guard
    /// drops.
    pub fn enter(&self) -> EnterGuard<'_> {
        enter(&self.inner)
    }

    /// The runtime's id.
    pub fn id(&self) -> RuntimeId {
        RuntimeId(self.inner.id)
    }

    /// The handle of the runtime set on this thread (inside its `block_on`,
    /// on one of its workers, or under [`enter`](Self::enter)); unlike
    /// [`current`](Self::current) it never falls back to the default
    /// runtime.
    pub fn try_current() -> Result<Self, TryCurrentError> {
        context::runtime()
            .map(|inner| Self { inner })
            .ok_or(TryCurrentError { _private: () })
    }
}

/// The index of the worker this code runs on (0, 1, ... among the workers
/// of its runtime), `None` off the workers (in `block_on`'s own thread, on
/// the blocking pool, outside any runtime).
pub fn worker_index() -> Option<usize> {
    let runtime = context::runtime()?;
    context::current(|shard| shard.and_then(|shard| shard.worker_index(runtime.id)))
}