LocalSet (#1892)
## Motivation There's currently an issue in `task::LocalSet` where dropping the local set can result in an infinite loop if a task running in the local set is notified from outside the local set (e.g. by a timer). This was reported in issue #1885. This issue exists because the `Drop` impl for `task::local::Scheduler` does not drain the queue of tasks notified externally, the way the basic scheduler does. Instead, only the local queue is drained, leaving some tasks in place. Since these tasks are never removed, the loop that continues trying to cancel tasks until the owned task list is totally empty continues infinitely. I think this issue was due to the `Drop` impl being written before a remote queue was added to the local scheduler, and the need to close the remote queue as well was overlooked. ## Solution This branch solves the problem by clearing the local scheduler's remote queue as well as the local one. I've added a test that reproduces the behavior. The test fails on master and passes after this change. In addition, this branch factors out the common task queue logic in the basic scheduler runtime and the `LocalSet` struct in `tokio::task`. This is because as more work was done on the `LocalSet`, it has gotten closer and closer to the basic scheduler in behavior, and factoring out the shared code reduces the risk of errors caused by `LocalSet` not doing something that the basic scheduler does. The queues are now encapsulated by a `MpscQueues` struct in `tokio::task::queue` (crate-public). As a follow-up, I'd also like to look into changing this type to use the same remote queue type as the threadpool (a linked list). In particular, I noticed the basic scheduler has a flag that indicates the remote queue has been closed, which is set when dropping the scheduler. This prevents tasks from being added after the scheduler has started shutting down, stopping a potential task leak. Rather than duplicating this code in `LocalSet`, I thought it was probably better to factor it out into a shared type. There are a few cases where there are small differences in behavior, though, so there is still a need for separate types implemented _using_ the new `MpscQueues` struct. However, it should cover most of the identical code. Note that this diff is rather large, due to the refactoring. However, the actual fix for the infinite loop is very simple. It can be reviewed on its own by looking at commit4f46ac6. The refactor is in a separate commit, with the SHA90b5b1f. Fixes #1885 Signed-off-by: Eliza Weisman <[email protected]>
Tokio
A runtime for writing reliable, asynchronous, and slim applications with the Rust programming language. It is:
-
Fast: Tokio's zero-cost abstractions give you bare-metal performance.
-
Reliable: Tokio leverages Rust's ownership, type system, and concurrency model to reduce bugs and ensure thread safety.
-
Scalable: Tokio has a minimal footprint, and handles backpressure and cancellation naturally.
Website | Guides | API Docs | Chat
Overview
Tokio is an event-driven, non-blocking I/O platform for writing asynchronous applications with the Rust programming language. At a high level, it provides a few major components:
- A multithreaded, work-stealing based task scheduler.
- A reactor backed by the operating system's event queue (epoll, kqueue, IOCP, etc...).
- Asynchronous TCP and UDP sockets.
These components provide the runtime components necessary for building an asynchronous application.
Example
A basic TCP echo server with Tokio:
use tokio::net::TcpListener;
use tokio::prelude::*;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut listener = TcpListener::bind("127.0.0.1:8080").await?;
loop {
let (mut socket, _) = listener.accept().await?;
tokio::spawn(async move {
let mut buf = [0; 1024];
// In a loop, read data from the socket and write the data back.
loop {
let n = match socket.read(&mut buf).await {
// socket closed
Ok(n) if n == 0 => return,
Ok(n) => n,
Err(e) => {
eprintln!("failed to read from socket; err = {:?}", e);
return;
}
};
// Write the data back
if let Err(e) = socket.write_all(&buf[0..n]).await {
eprintln!("failed to write to socket; err = {:?}", e);
return;
}
}
});
}
}
More examples can be found here. Note that the master branch
is currently being updated to use async / await. The examples are
not fully ported. Examples for stable Tokio can be found
here.
Getting Help
First, see if the answer to your question can be found in the Guides or the API documentation. If the answer is not there, there is an active community in the Tokio Discord server. We would be happy to try to answer your question. Last, if that doesn't work, try opening an issue with the question.
Contributing
🎈 Thanks for your help improving the project! We are so happy to have you! We have a contributing guide to help you get involved in the Tokio project.
Related Projects
In addition to the crates in this repository, the Tokio project also maintains several other libraries, including:
-
tracing(formerlytokio-trace): A framework for application-level tracing and async-aware diagnostics. -
mio: A low-level, cross-platform abstraction over OS I/O APIs that powerstokio. -
bytes: Utilities for working with bytes, including efficient byte buffers.
Supported Rust Versions
Tokio is built against the latest stable, nightly, and beta Rust releases. The minimum version supported is the stable release from three months before the current stable release version. For example, if the latest stable Rust is 1.29, the minimum version supported is 1.26. The current Tokio version is not guaranteed to build on Rust versions earlier than the minimum supported version.
License
This project is licensed under the MIT license.
Contribution
Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in Tokio by you, shall be licensed as MIT, without any additional terms or conditions.