actix_rt/
lib.rs

1//! Tokio-based single-threaded async runtime for the Actix ecosystem.
2//!
3//! In most parts of the the Actix ecosystem, it has been chosen to use !Send futures. For this
4//! reason, a single-threaded runtime is appropriate since it is guaranteed that futures will not
5//! be moved between threads. This can result in small performance improvements over cases where
6//! atomics would otherwise be needed.
7//!
8//! To achieve similar performance to multi-threaded, work-stealing runtimes, applications
9//! using `actix-rt` will create multiple, mostly disconnected, single-threaded runtimes.
10//! This approach has good performance characteristics for workloads where the majority of tasks
11//! have similar runtime expense.
12//!
13//! The disadvantage is that idle threads will not steal work from very busy, stuck or otherwise
14//! backlogged threads. Tasks that are disproportionately expensive should be offloaded to the
15//! blocking task thread-pool using [`task::spawn_blocking`].
16//!
17//! # Examples
18//! ```no_run
19//! use std::sync::mpsc;
20//! use actix_rt::{Arbiter, System};
21//!
22//! let _ = System::new();
23//!
24//! let (tx, rx) = mpsc::channel::<u32>();
25//!
26//! let arbiter = Arbiter::new();
27//! arbiter.spawn_fn(move || tx.send(42).unwrap());
28//!
29//! let num = rx.recv().unwrap();
30//! assert_eq!(num, 42);
31//!
32//! arbiter.stop();
33//! arbiter.join().unwrap();
34//! ```
35//!
36//! # `io-uring` Support
37//!
38//! There is experimental support for using io-uring with this crate by enabling the
39//! `io-uring` feature. For now, it is semver exempt.
40//!
41//! Note that there are currently some unimplemented parts of using `actix-rt` with `io-uring`.
42//! In particular, when running a `System`, only `System::block_on` is supported.
43
44#![deny(rust_2018_idioms, nonstandard_style)]
45#![warn(future_incompatible, missing_docs)]
46#![allow(clippy::type_complexity)]
47#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
48#![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
49
50#[cfg(all(not(target_os = "linux"), feature = "io-uring"))]
51compile_error!("io_uring is a linux only feature.");
52
53use std::future::Future;
54
55// Cannot define a main macro when compiled into test harness.
56// Workaround for https://github.com/rust-lang/rust/issues/62127.
57#[cfg(all(feature = "macros", not(test)))]
58pub use actix_macros::main;
59#[cfg(feature = "macros")]
60pub use actix_macros::test;
61
62mod arbiter;
63mod runtime;
64mod system;
65
66pub use tokio::pin;
67use tokio::task::JoinHandle;
68
69pub use self::{
70    arbiter::{Arbiter, ArbiterHandle},
71    runtime::Runtime,
72    system::{System, SystemRunner},
73};
74
75pub mod signal {
76    //! Asynchronous signal handling (Tokio re-exports).
77
78    #[cfg(unix)]
79    pub mod unix {
80        //! Unix specific signals (Tokio re-exports).
81        pub use tokio::signal::unix::*;
82    }
83    pub use tokio::signal::ctrl_c;
84}
85
86pub mod net {
87    //! TCP/UDP/Unix bindings (mostly Tokio re-exports).
88
89    use std::{
90        future::Future,
91        io,
92        task::{Context, Poll},
93    };
94
95    use tokio::io::{AsyncRead, AsyncWrite, Interest};
96    #[cfg(unix)]
97    pub use tokio::net::{UnixDatagram, UnixListener, UnixStream};
98    pub use tokio::{
99        io::Ready,
100        net::{TcpListener, TcpSocket, TcpStream, UdpSocket},
101    };
102
103    /// Extension trait over async read+write types that can also signal readiness.
104    #[doc(hidden)]
105    pub trait ActixStream: AsyncRead + AsyncWrite + Unpin {
106        /// Poll stream and check read readiness of Self.
107        ///
108        /// See [tokio::net::TcpStream::poll_read_ready] for detail on intended use.
109        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>>;
110
111        /// Poll stream and check write readiness of Self.
112        ///
113        /// See [tokio::net::TcpStream::poll_write_ready] for detail on intended use.
114        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>>;
115    }
116
117    impl ActixStream for TcpStream {
118        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
119            let ready = self.ready(Interest::READABLE);
120            tokio::pin!(ready);
121            ready.poll(cx)
122        }
123
124        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
125            let ready = self.ready(Interest::WRITABLE);
126            tokio::pin!(ready);
127            ready.poll(cx)
128        }
129    }
130
131    #[cfg(unix)]
132    impl ActixStream for UnixStream {
133        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
134            let ready = self.ready(Interest::READABLE);
135            tokio::pin!(ready);
136            ready.poll(cx)
137        }
138
139        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
140            let ready = self.ready(Interest::WRITABLE);
141            tokio::pin!(ready);
142            ready.poll(cx)
143        }
144    }
145
146    impl<Io: ActixStream + ?Sized> ActixStream for Box<Io> {
147        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
148            (**self).poll_read_ready(cx)
149        }
150
151        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
152            (**self).poll_write_ready(cx)
153        }
154    }
155}
156
157pub mod time {
158    //! Utilities for tracking time (Tokio re-exports).
159
160    pub use tokio::time::{
161        interval, interval_at, sleep, sleep_until, timeout, Instant, Interval, Sleep, Timeout,
162    };
163}
164
165pub mod task {
166    //! Task management (Tokio re-exports).
167
168    pub use tokio::task::{spawn_blocking, yield_now, JoinError, JoinHandle};
169}
170
171/// Spawns a future on the current thread as a new task.
172///
173/// If not immediately awaited, the task can be cancelled using [`JoinHandle::abort`].
174///
175/// The provided future is spawned as a new task; therefore, panics are caught.
176///
177/// # Panics
178/// Panics if Actix system is not running.
179///
180/// # Examples
181/// ```
182/// # use std::time::Duration;
183/// # actix_rt::Runtime::new().unwrap().block_on(async {
184/// // task resolves successfully
185/// assert_eq!(actix_rt::spawn(async { 1 }).await.unwrap(), 1);
186///
187/// // task panics
188/// assert!(actix_rt::spawn(async {
189///     panic!("panic is caught at task boundary");
190/// })
191/// .await
192/// .unwrap_err()
193/// .is_panic());
194///
195/// // task is cancelled before completion
196/// let handle = actix_rt::spawn(actix_rt::time::sleep(Duration::from_secs(100)));
197/// handle.abort();
198/// assert!(handle.await.unwrap_err().is_cancelled());
199/// # });
200/// ```
201#[track_caller]
202#[inline]
203pub fn spawn<Fut>(f: Fut) -> JoinHandle<Fut::Output>
204where
205    Fut: Future + 'static,
206    Fut::Output: 'static,
207{
208    tokio::task::spawn_local(f)
209}
actix_rt/lib.rs

actix_rt/
lib.rs
