nuclei 0.5.0

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

//! 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 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.
    pub fn shutdown_background(self) {
        self.inner.workers.stop_all();
        // Detached: dropping the handles does not wait. `Drop` finds none
        // left to join.
        drop(self.inner.workers.take_threads());
    }
}

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)))
}