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#![allow(clippy::type_complexity)]
45#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
46#![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
47
48#[cfg(all(not(target_os = "linux"), feature = "io-uring"))]
49compile_error!("io_uring is a linux only feature.");
50
51use std::future::Future;
52
53// Cannot define a main macro when compiled into test harness.
54// Workaround for https://github.com/rust-lang/rust/issues/62127.
55#[cfg(all(feature = "macros", not(test)))]
56pub use actix_macros::main;
57#[cfg(feature = "macros")]
58pub use actix_macros::test;
59
60mod arbiter;
61mod runtime;
62mod system;
63
64#[deprecated(since = "2.11.0", note = "Prefer `std::pin::pin!`.")]
65pub use tokio::pin;
66use tokio::task::JoinHandle;
67
68pub use self::{
69    arbiter::{Arbiter, ArbiterHandle},
70    runtime::Runtime,
71    system::{System, SystemRunner},
72};
73
74pub mod signal {
75    //! Asynchronous signal handling (Tokio re-exports).
76
77    #[cfg(unix)]
78    pub mod unix {
79        //! Unix specific signals (Tokio re-exports).
80        pub use tokio::signal::unix::*;
81    }
82    pub use tokio::signal::ctrl_c;
83}
84
85pub mod net {
86    //! TCP/UDP/Unix bindings (mostly Tokio re-exports).
87
88    use std::{
89        future::Future,
90        io,
91        pin::pin,
92        task::{Context, Poll},
93    };
94
95    use tokio::io::{AsyncRead, AsyncWrite, BufReader, 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            pin!(ready).poll(cx)
121        }
122
123        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
124            let ready = self.ready(Interest::WRITABLE);
125            pin!(ready).poll(cx)
126        }
127    }
128
129    #[cfg(unix)]
130    impl ActixStream for UnixStream {
131        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
132            let ready = self.ready(Interest::READABLE);
133            pin!(ready).poll(cx)
134        }
135
136        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
137            let ready = self.ready(Interest::WRITABLE);
138            pin!(ready).poll(cx)
139        }
140    }
141
142    impl<Io: ActixStream + ?Sized> ActixStream for Box<Io> {
143        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
144            (**self).poll_read_ready(cx)
145        }
146
147        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
148            (**self).poll_write_ready(cx)
149        }
150    }
151
152    impl<Io: ActixStream> ActixStream for BufReader<Io> {
153        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
154            self.get_ref().poll_read_ready(cx)
155        }
156
157        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
158            self.get_ref().poll_write_ready(cx)
159        }
160    }
161}
162
163pub mod time {
164    //! Utilities for tracking time (Tokio re-exports).
165
166    pub use tokio::time::{
167        interval, interval_at, sleep, sleep_until, timeout, Instant, Interval, Sleep, Timeout,
168    };
169}
170
171pub mod task {
172    //! Task management (Tokio re-exports).
173
174    pub use tokio::task::{spawn_blocking, yield_now, JoinError, JoinHandle};
175}
176
177/// Spawns a future on the current thread as a new task.
178///
179/// If not immediately awaited, the task can be cancelled using [`JoinHandle::abort`].
180///
181/// The provided future is spawned as a new task; therefore, panics are caught.
182///
183/// # Panics
184/// Panics if Actix system is not running.
185///
186/// # Examples
187/// ```
188/// # use std::time::Duration;
189/// # actix_rt::Runtime::new().unwrap().block_on(async {
190/// // task resolves successfully
191/// assert_eq!(actix_rt::spawn(async { 1 }).await.unwrap(), 1);
192///
193/// // task panics
194/// assert!(actix_rt::spawn(async {
195///     panic!("panic is caught at task boundary");
196/// })
197/// .await
198/// .unwrap_err()
199/// .is_panic());
200///
201/// // task is cancelled before completion
202/// let handle = actix_rt::spawn(actix_rt::time::sleep(Duration::from_secs(100)));
203/// handle.abort();
204/// assert!(handle.await.unwrap_err().is_cancelled());
205/// # });
206/// ```
207#[track_caller]
208#[inline]
209pub fn spawn<Fut>(f: Fut) -> JoinHandle<Fut::Output>
210where
211    Fut: Future + 'static,
212    Fut::Output: 'static,
213{
214    tokio::task::spawn_local(f)
215}
actix_rt/lib.rs

actix_rt/
lib.rs
