Compare commits

...
Author SHA1 Message Date
Alice Ryhl c505a2f81a chore: prepare tokio-macros 1.3.0 (#3931) 2021-07-07 10:58:02 +02:00
Alice Ryhl e2589a0e40 runtime: add OwnedTasks (#3909) 2021-07-07 10:53:57 +02:00
Alice Ryhl 51fad066e2 bench: update spawn benchmarks (#3927) 2021-07-07 10:52:53 +02:00
Alice Ryhl ae233c1e9f Merge branch 'upstream/merge-1.8.1' into 'master' 2021-07-07 10:22:13 +02:00
Alice Ryhl be26ca7625 sync: wrap state in helper struct (#3922) 2021-07-07 10:17:47 +02:00
Carl Lerche 2d41f0303e Merge branch 'tokio-1.8.x' into merge-1.8.1 2021-07-06 16:20:57 -07:00
Carl Lerche e0c527f383 chore: prepare Tokio v1.8.1 release 2021-07-06 15:37:17 -07:00
Carl Lerche 1160e8864c Merge branch 'tokio-1.7.x' into merge-1.7.2 2021-07-06 15:36:24 -07:00
Carl Lerche 998a125717 chore: prepare Tokio 1.7.2 release 2021-07-06 14:35:36 -07:00
Carl Lerche cbfdc9d69e Merge branch 'tokio-1.6.x' into merge-1.6.3 2021-07-06 14:34:34 -07:00
Carl Lerche 08337c5f79 chore: prepare Tokio v1.6.3 2021-07-06 13:45:47 -07:00
Carl Lerche 87f4969fbc Merge branch 'tokio-1.5.x' into merge-1.5.x 2021-07-06 13:44:11 -07:00
Alice Ryhl e385108920 rt: remotely abort tasks on JoinHandle::abort (#3934) 2021-07-06 13:41:25 -07:00
Blas Rodriguez Irizar 4818c2ed05 fs: document performance considerations (#3920) 2021-07-06 16:25:13 +02:00
Blas Rodriguez Irizar 80f0801e19 tokio-macros: compat with clippy::unwrap_used (#3926) 2021-07-06 15:01:13 +02:00
David CARLIER 8148b2107c net: make ucred return pid field for OpenBSD (#3919) 2021-07-06 11:15:15 +02:00
oiovoyo 2e7de1ae1d stream: modify HashMap to StreamMap in example. (#3925) 2021-07-05 16:10:55 +02:00
Alice Ryhl 37e60fc7f9 chore: fix new clippy complaint (#3915) 2021-07-05 09:58:49 +02:00
Alice Ryhl 8170e2787c sync: fix watch wrapper (#3914) 2021-07-02 23:24:02 +02:00
Alice Ryhl 677107d8d9 chore: prepare Tokio v1.8.0 (#3912) 2021-07-02 22:21:10 +02:00
Alice Ryhl c8ecfc894d sync: fix watch borrow_and_update (#3913) 2021-07-02 10:14:34 -07:00
Taiki Endo 08ed41f339 chore: fix typos (#3907) 2021-07-01 02:06:56 +09:00
Alice Ryhl 90e1935c48 test: test with tracing (#3906) 2021-06-30 09:40:10 -07:00
Alice Ryhl b877629cb1 net: handle HUP event with UnixStream (#3898)
Fixes: #3879
2021-06-30 09:39:06 -07:00
Mikhail Zabaluev 0531549b6e io: efficient implementation of vectored writes for BufWriter (#3163) 2021-06-29 20:26:47 +02:00
Alice Ryhl 38204f5fba time: fix Timeout size_hint (#3902) 2021-06-29 20:06:10 +02:00
Jacob Rothstein 8fa29cb00a rt: add tokio::task::Builder (#3881)
Adds a builder API for spawning tasks. Initially, this enables the caller to name the spawned
task in order to provide better visibility into all tasks in the system.
2021-06-29 10:47:30 -07:00
Alice Ryhl b521cc2689 doc: document cancellation safety (#3900)
This patch documents cancellation safety. It also moves the "Avoid racy if preconditions"
section in the select! documentation since otherwise the first code block on the page
shows how not to use it, which seems counterintuitive.
2021-06-29 10:35:42 -07:00
Jake Shadle 57c90c9750 net: add read/try_read etc methods to NamedPipeServer (#3899) 2021-06-29 10:05:20 +02:00
Mikail Bagishov d35ff7064f sync: add drop guard for cancellation token (#3839) 2021-06-28 12:34:58 +02:00
Milo ab0791b817 time: add wait alias to sleep (#3897) 2021-06-28 10:48:07 +02:00
Jake Shadle 959c5c997f net: add ready/try methods to NamedPipeClient (#3866) 2021-06-28 10:43:57 +02:00
sb64 845626410a sync: implement From<T> for OnceCell<T> (#3877) 2021-06-22 15:11:01 +02:00
Carl Lerche 3207479fa7 Merge branch 'tokio-1.7.x' into merge-1.7.1 2021-06-18 16:37:21 -07:00
Carl Lerche 7601dc6d2a rt: avoid early task shutdown (#3870)
Tokio 1.7.0 introduced a change intended to eagerly shutdown newly
spawned tasks if the runtime is in the process of shutting down.
However, it introduced a bug where already spawned tasks could be
shutdown too early, resulting in the potential introduction of deadlocks
if tasks acquired mutexes in drop handlers.

Fixes #3869
2021-06-18 16:32:53 -07:00
Alice Ryhl d1aa2df80e sync: add Receiver::borrow_and_update (#3813) 2021-06-16 18:42:05 +02:00
sb64 e979ad7f2d time: allow users to specify Interval behavior when delayed (#3721) 2021-06-16 13:51:27 +02:00
sb64 5ad3dd3378 time: document auto-advancing behavior of runtime (#3763) 2021-06-16 13:51:12 +02:00
teor 60bd40d529 time: fix typo in Instant::saturating_duration_since docs (#3864) 2021-06-16 12:50:30 +02:00
brain0 4a93af4d25 io: add get_{ref,mut} methods to AsyncFdReadyGuard and AsyncFdReadyMutGuard. (#3807) 2021-06-16 12:45:05 +02:00
Alice Ryhl 34c6a26c01 chore: prepare Tokio v1.7.0 (#3863) 2021-06-15 19:39:33 +02:00
John-John TedroandAlice Ryhl 97e7830364 net: provide NamedPipe{Client, Server} types and builders (#3760)
This builds on https://github.com/tokio-rs/mio/pull/1351 and introduces the
tokio::net::windows::named_pipe module which provides low level types for
building and communicating asynchronously over windows named pipes.

Named pipes require the `net` feature flag to be enabled on Windows.

Co-authored-by: Alice Ryhl <[email protected]>
2021-06-15 08:13:31 +02:00
Carl Lerche 606206ecad Merge branch 'tokio-1.6.x' into merge-1.6.x 2021-06-14 17:51:30 -07:00
Carl Lerche dfe4013ff2 chore: prepare Tokio 1.6.2 release (#3859) 2021-06-14 17:50:34 -07:00
Carl Lerche 18779aa2e2 time: fix time::advance() with sub-ms durations (#3852)
Instead of using sleep in time::advance, this fixes the root of the issue. When futures passed
to `Runtime::block_on` are woken, it bypassed all the machinery around advancing time. By
intercepting wakes in the time driver, we know when the block_on task is woken and skip
advancing time in that case.

Fixes #3837
2021-06-14 16:36:04 -07:00
Alan Somers 2c24a028f6 chore: re-enable test_socket_pair on FreeBSD (#3844)
It was disabled due to an OS bug that has been fixed on all currently
supported releases of FreeBSD
2021-06-14 13:14:35 -07:00
Sean McArthur f55b77aadd tracing: emit waker op as str instead as Debug (#3853) 2021-06-14 10:37:24 -07:00
Yotam Ofek 21de476ae7 sync: export sync::notify::Notified future publicly (#3840) 2021-06-14 10:05:34 +02:00
Alice Ryhl cb147a2b3f sync: deprecate mpsc::RecvError (#3833) 2021-06-11 11:01:34 +02:00
Bart Pelle f759240254 chore: update version in README.md (#3851) 2021-06-11 11:01:22 +02:00
AtulJatia 2e44cd29df net: TcpSocket from std::net::TcpStream (#3838) 2021-06-11 10:58:58 +02:00
Oğuz Bilgener 2ab1fb00a9 sync: add examples to Semaphore (#3808) 2021-06-11 10:46:16 +02:00
Evan Mesterhazy d16e50639a doc: clarify EOF condition for AsyncReadExt::read_buf (#3850) 2021-06-09 15:00:32 +02:00
Sean McArthur e7d74b3119 tracing: instrument task wakers (#3836)
## Motivation

In support of tokio-rs/console#37, we want to understand when a specific task's waker has been interacted with, such as when it is awoken, or if it's forgotten (not cloned), etc.

## Solution

When the tracing feature is enabled, a super trait of Future (InstrumentedFuture) is implemented for Instrumented<F> that allows grabbing the task's ID (well, its span ID), and stores that in the raw task trailer. The waker vtable then emits events and includes that ID.
2021-06-08 14:50:40 -07:00
Tony Han d101feac50 runtime: fix typo in task state doc (#3835) 2021-06-04 22:31:04 +02:00
Taiki Endo 9d8b37d51a macros: suppress clippy::default_numeric_fallback lint in generated code (#3831) 2021-06-02 18:16:23 +09:00
James Hallowell d4c89758fc sync: fix spelling mistake in mpsc::Sender docs (#3828) 2021-06-01 09:56:34 +02:00
Taiki Endo 932be12481 ci: ignore private crates in minimal-versions check (#3827) 2021-05-31 23:39:06 +09:00
Edward Shen bdd6765016 doc: clarify limits on return values of AsyncWrite::poll_write (#3820) 2021-05-31 19:55:55 +09:00
Carl Lerche 8f6d8b25bf chore: add FUNDING.yml (#3824)
Reference Tokio sponsorship
2021-05-28 13:55:16 -07:00
Carl Lerche eb7aee980c Merge branch 'tokio-1.6.x' into merge-1.6.1 2021-05-28 10:04:08 -07:00
Alice Ryhl 1baea398c4 chore: prepare Tokio v1.6.1 2021-05-28 09:59:28 -07:00
Alice Ryhl 3fbcf1ba50 Revert "fs: try doing a non-blocking read before punting to the threadpool (#3518)"
This reverts commit 39706b198c.
2021-05-28 09:59:28 -07:00
j-vanderstoep 21264f1d33 tcp: assign an unused port (#3817) 2021-05-28 09:03:27 +02:00
Rafael Bachmann 81ee3d202a doc: fix minor doc typo (stray backtick) (#3814) 2021-05-26 14:23:39 +02:00
Alice Ryhl a39e6c2439 runtime: immediately drop new task when runtime is shut down (#3752) 2021-05-26 10:24:56 +02:00
Petros Angelatos 4abeca7bc5 io: impl AsyncSeek for BufStream (#3810)
Signed-off-by: Petros Angelatos <[email protected]>
2021-05-26 07:11:47 +09:00
GITSRC 5ed84e1cd8 doc: update README.md (#3806) 2021-05-21 09:55:53 +02:00
Lucio Franco 44a070d1b4 doc: add missing backtick (#3799) 2021-05-19 10:14:08 +02:00
Aaron Taner ce9ca45c92 doc: fix invalid #[doc(inline)] warnings on latest nightly. (#3788)
This commit fixed issue #3787 by removing [doc(inline)] from
macro `cfg_macros` and added proper #[doc(inline)] attributes
to `pub use` items inside `cfg_macros` calls.

It's probably not `cfg_macros`s responsibility to inlining public
macros, though it's conveninent to do so. Notice that in lib.rs:

cfg_macros! {
    /// Implementation detail of the `select!` macro. This macro is **not**
    /// intended to be used as part of the public API and is permitted to
    /// change.
    #[doc(hidden)]
    pub use tokio_macros::select_priv_declare_output_enum;

    ...
}

`#[doc(hidden)]` and `#[doc(inline)]` are conflict with each other
in the sense of correctness.

Fixes: #3787
2021-05-18 01:28:17 +09:00
Folyd f3ed064a26 chore: update *::{max, min}_value() to const *::MAX and *::MIN respectively (#3794) 2021-05-17 12:31:58 +02:00
Sören Meier 652f0ae728 sync: add receiver_count to watch::Sender (#3729) 2021-05-15 11:56:24 +02:00
160 changed files with 6428 additions and 1174 deletions
+1
View File
@@ -0,0 +1 @@
msrv = "1.45"
+3
View File
@@ -0,0 +1,3 @@
# These are supported funding model platforms
github: [tokio-rs]
+3 -3
View File
@@ -122,7 +122,7 @@ jobs:
# Run `tokio` with "unstable" cfg flag.
- name: test tokio full --cfg unstable
run: cargo test --features full
run: cargo test --all-features
working-directory: tokio
env:
RUSTFLAGS: --cfg tokio_unstable -Dwarnings
@@ -238,7 +238,7 @@ jobs:
cargo hack --remove-dev-deps --workspace
# Update Cargo.lock to minimal version dependencies.
cargo update -Z minimal-versions
cargo check --all-features
cargo hack check --all-features --ignore-private
fmt:
name: fmt
@@ -265,7 +265,7 @@ jobs:
steps:
- uses: actions/checkout@v2
- name: Install Rust
run: rustup update ${{ env.minrust }} && rustup default ${{ env.minrust }}
run: rustup update 1.52.1 && rustup default 1.52.1
- name: Install clippy
run: rustup component add clippy
+2 -2
View File
@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
```toml
[dependencies]
tokio = { version = "1.5.0", features = ["full"] }
tokio = { version = "1.8.0", features = ["full"] }
```
Then, on your main.rs:
@@ -66,7 +66,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut listener = TcpListener::bind("127.0.0.1:8080").await?;
let listener = TcpListener::bind("127.0.0.1:8080").await?;
loop {
let (mut socket, _) = listener.accept().await?;
+1 -1
View File
@@ -1,4 +1,4 @@
//! Benchmark implementation details of the theaded scheduler. These benches are
//! Benchmark implementation details of the threaded scheduler. These benches are
//! intended to be used as a form of regression testing and not as a general
//! purpose benchmark demonstrating real-world performance.
+57 -37
View File
@@ -2,63 +2,83 @@
//! This essentially measure the time to enqueue a task in the local and remote
//! case.
#[macro_use]
extern crate bencher;
use bencher::{black_box, Bencher};
async fn work() -> usize {
let val = 1 + 1;
tokio::task::yield_now().await;
black_box(val)
}
fn basic_scheduler_local_spawn(bench: &mut Bencher) {
fn basic_scheduler_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
runtime.block_on(async {
bench.iter(|| {
let h = tokio::spawn(work());
black_box(h);
})
});
}
fn threaded_scheduler_local_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
runtime.block_on(async {
bench.iter(|| {
let h = tokio::spawn(work());
black_box(h);
})
});
}
fn basic_scheduler_remote_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
bench.iter(|| {
let h = runtime.spawn(work());
black_box(h);
runtime.block_on(async {
let h = tokio::spawn(work());
assert_eq!(h.await.unwrap(), 2);
});
});
}
fn threaded_scheduler_remote_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
fn basic_scheduler_spawn10(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
bench.iter(|| {
let h = runtime.spawn(work());
black_box(h);
runtime.block_on(async {
let mut handles = Vec::with_capacity(10);
for _ in 0..10 {
handles.push(tokio::spawn(work()));
}
for handle in handles {
assert_eq!(handle.await.unwrap(), 2);
}
});
});
}
fn threaded_scheduler_spawn(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.build()
.unwrap();
bench.iter(|| {
runtime.block_on(async {
let h = tokio::spawn(work());
assert_eq!(h.await.unwrap(), 2);
});
});
}
fn threaded_scheduler_spawn10(bench: &mut Bencher) {
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.build()
.unwrap();
bench.iter(|| {
runtime.block_on(async {
let mut handles = Vec::with_capacity(10);
for _ in 0..10 {
handles.push(tokio::spawn(work()));
}
for handle in handles {
assert_eq!(handle.await.unwrap(), 2);
}
});
});
}
bencher::benchmark_group!(
spawn,
basic_scheduler_local_spawn,
threaded_scheduler_local_spawn,
basic_scheduler_remote_spawn,
threaded_scheduler_remote_spawn
basic_scheduler_spawn,
basic_scheduler_spawn10,
threaded_scheduler_spawn,
threaded_scheduler_spawn10,
);
bencher::benchmark_main!(spawn);
+15
View File
@@ -22,7 +22,10 @@ serde_json = "1.0"
httparse = "1.0"
time = "0.1"
once_cell = "1.5.2"
rand = "0.8.3"
[target.'cfg(windows)'.dev-dependencies.winapi]
version = "0.3.8"
[[example]]
name = "chat"
@@ -76,3 +79,15 @@ path = "custom-executor.rs"
[[example]]
name = "custom-executor-tokio-context"
path = "custom-executor-tokio-context.rs"
[[example]]
name = "named-pipe"
path = "named-pipe.rs"
[[example]]
name = "named-pipe-ready"
path = "named-pipe-ready.rs"
[[example]]
name = "named-pipe-multi-client"
path = "named-pipe-multi-client.rs"
+98
View File
@@ -0,0 +1,98 @@
use std::io;
#[cfg(windows)]
async fn windows_main() -> io::Result<()> {
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
use tokio::time;
use winapi::shared::winerror;
const PIPE_NAME: &str = r"\\.\pipe\named-pipe-multi-client";
const N: usize = 10;
// The first server needs to be constructed early so that clients can
// be correctly connected. Otherwise a waiting client will error.
//
// Here we also make use of `first_pipe_instance`, which will ensure
// that there are no other servers up and running already.
let mut server = ServerOptions::new()
.first_pipe_instance(true)
.create(PIPE_NAME)?;
let server = tokio::spawn(async move {
// Artificial workload.
time::sleep(Duration::from_secs(1)).await;
for _ in 0..N {
// Wait for client to connect.
server.connect().await?;
let mut inner = server;
// Construct the next server to be connected before sending the one
// we already have of onto a task. This ensures that the server
// isn't closed (after it's done in the task) before a new one is
// available. Otherwise the client might error with
// `io::ErrorKind::NotFound`.
server = ServerOptions::new().create(PIPE_NAME)?;
let _ = tokio::spawn(async move {
let mut buf = vec![0u8; 4];
inner.read_exact(&mut buf).await?;
inner.write_all(b"pong").await?;
Ok::<_, io::Error>(())
});
}
Ok::<_, io::Error>(())
});
let mut clients = Vec::new();
for _ in 0..N {
clients.push(tokio::spawn(async move {
// This showcases a generic connect loop.
//
// We immediately try to create a client, if it's not found or
// the pipe is busy we use the specialized wait function on the
// client builder.
let mut client = loop {
match ClientOptions::new().open(PIPE_NAME) {
Ok(client) => break client,
Err(e) if e.raw_os_error() == Some(winerror::ERROR_PIPE_BUSY as i32) => (),
Err(e) => return Err(e),
}
time::sleep(Duration::from_millis(5)).await;
};
let mut buf = [0u8; 4];
client.write_all(b"ping").await?;
client.read_exact(&mut buf).await?;
Ok::<_, io::Error>(buf)
}));
}
for client in clients {
let result = client.await?;
assert_eq!(&result?[..], b"pong");
}
server.await??;
Ok(())
}
#[tokio::main]
async fn main() -> io::Result<()> {
#[cfg(windows)]
{
windows_main().await?;
}
#[cfg(not(windows))]
{
println!("Named pipes are only supported on Windows!");
}
Ok(())
}
+161
View File
@@ -0,0 +1,161 @@
use std::io;
#[cfg(windows)]
async fn windows_main() -> io::Result<()> {
use tokio::io::Interest;
use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
const PIPE_NAME: &str = r"\\.\pipe\named-pipe-single-client";
let server = ServerOptions::new().create(PIPE_NAME)?;
let server = tokio::spawn(async move {
// Note: we wait for a client to connect.
server.connect().await?;
let buf = {
let mut read_buf = [0u8; 5];
let mut read_buf_cursor = 0;
loop {
server.readable().await?;
let buf = &mut read_buf[read_buf_cursor..];
match server.try_read(buf) {
Ok(n) => {
read_buf_cursor += n;
if read_buf_cursor == read_buf.len() {
break;
}
}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
continue;
}
Err(e) => {
return Err(e);
}
}
}
read_buf
};
{
let write_buf = b"pong\n";
let mut write_buf_cursor = 0;
loop {
let buf = &write_buf[write_buf_cursor..];
if buf.is_empty() {
break;
}
server.writable().await?;
match server.try_write(buf) {
Ok(n) => {
write_buf_cursor += n;
}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
continue;
}
Err(e) => {
return Err(e);
}
}
}
}
Ok::<_, io::Error>(buf)
});
let client = tokio::spawn(async move {
// There's no need to use a connect loop here, since we know that the
// server is already up - `open` was called before spawning any of the
// tasks.
let client = ClientOptions::new().open(PIPE_NAME)?;
let mut read_buf = [0u8; 5];
let mut read_buf_cursor = 0;
let write_buf = b"ping\n";
let mut write_buf_cursor = 0;
loop {
let mut interest = Interest::READABLE;
if write_buf_cursor < write_buf.len() {
interest |= Interest::WRITABLE;
}
let ready = client.ready(interest).await?;
if ready.is_readable() {
let buf = &mut read_buf[read_buf_cursor..];
match client.try_read(buf) {
Ok(n) => {
read_buf_cursor += n;
if read_buf_cursor == read_buf.len() {
break;
}
}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
continue;
}
Err(e) => {
return Err(e);
}
}
}
if ready.is_writable() {
let buf = &write_buf[write_buf_cursor..];
if buf.is_empty() {
continue;
}
match client.try_write(buf) {
Ok(n) => {
write_buf_cursor += n;
}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
continue;
}
Err(e) => {
return Err(e);
}
}
}
}
let buf = String::from_utf8_lossy(&read_buf).into_owned();
Ok::<_, io::Error>(buf)
});
let (server, client) = tokio::try_join!(server, client)?;
assert_eq!(server?, *b"ping\n");
assert_eq!(client?, "pong\n");
Ok(())
}
#[tokio::main]
async fn main() -> io::Result<()> {
#[cfg(windows)]
{
windows_main().await?;
}
#[cfg(not(windows))]
{
println!("Named pipes are only supported on Windows!");
}
Ok(())
}
+60
View File
@@ -0,0 +1,60 @@
use std::io;
#[cfg(windows)]
async fn windows_main() -> io::Result<()> {
use tokio::io::AsyncWriteExt;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
const PIPE_NAME: &str = r"\\.\pipe\named-pipe-single-client";
let server = ServerOptions::new().create(PIPE_NAME)?;
let server = tokio::spawn(async move {
// Note: we wait for a client to connect.
server.connect().await?;
let mut server = BufReader::new(server);
let mut buf = String::new();
server.read_line(&mut buf).await?;
server.write_all(b"pong\n").await?;
Ok::<_, io::Error>(buf)
});
let client = tokio::spawn(async move {
// There's no need to use a connect loop here, since we know that the
// server is already up - `open` was called before spawning any of the
// tasks.
let client = ClientOptions::new().open(PIPE_NAME)?;
let mut client = BufReader::new(client);
let mut buf = String::new();
client.write_all(b"ping\n").await?;
client.read_line(&mut buf).await?;
Ok::<_, io::Error>(buf)
});
let (server, client) = tokio::try_join!(server, client)?;
assert_eq!(server?, "ping\n");
assert_eq!(client?, "pong\n");
Ok(())
}
#[tokio::main]
async fn main() -> io::Result<()> {
#[cfg(windows)]
{
windows_main().await?;
}
#[cfg(not(windows))]
{
println!("Named pipes are only supported on Windows!");
}
Ok(())
}
+6
View File
@@ -1,3 +1,9 @@
# 1.3.0 (July 7, 2021)
- macros: don't trigger `clippy::unwrap_used` ([#3926])
[#3926]: https://github.com/tokio-rs/tokio/pull/3926
# 1.2.0 (May 14, 2021)
- macros: forward input arguments in `#[tokio::test]` ([#3691])
+2 -2
View File
@@ -6,13 +6,13 @@ name = "tokio-macros"
# - Cargo.toml
# - Update CHANGELOG.md.
# - Create "tokio-macros-1.0.x" git tag.
version = "1.2.0"
version = "1.3.0"
edition = "2018"
authors = ["Tokio Contributors <[email protected]>"]
license = "MIT"
repository = "https://github.com/tokio-rs/tokio"
homepage = "https://tokio.rs"
documentation = "https://docs.rs/tokio-macros/1.2.0/tokio_macros"
documentation = "https://docs.rs/tokio-macros/1.3.0/tokio_macros"
description = """
Tokio's proc macros.
"""
+16 -14
View File
@@ -201,12 +201,15 @@ fn parse_knobs(
for arg in args {
match arg {
syn::NestedMeta::Meta(syn::Meta::NameValue(namevalue)) => {
let ident = namevalue.path.get_ident();
if ident.is_none() {
let msg = "Must have specified ident";
return Err(syn::Error::new_spanned(namevalue, msg));
}
match ident.unwrap().to_string().to_lowercase().as_str() {
let ident = namevalue
.path
.get_ident()
.ok_or_else(|| {
syn::Error::new_spanned(&namevalue, "Must have specified ident")
})?
.to_string()
.to_lowercase();
match ident.as_str() {
"worker_threads" => {
config.set_worker_threads(
namevalue.lit.clone(),
@@ -239,12 +242,11 @@ fn parse_knobs(
}
}
syn::NestedMeta::Meta(syn::Meta::Path(path)) => {
let ident = path.get_ident();
if ident.is_none() {
let msg = "Must have specified ident";
return Err(syn::Error::new_spanned(path, msg));
}
let name = ident.unwrap().to_string().to_lowercase();
let name = path
.get_ident()
.ok_or_else(|| syn::Error::new_spanned(&path, "Must have specified ident"))?
.to_string()
.to_lowercase();
let msg = match name.as_str() {
"threaded_scheduler" | "multi_thread" => {
format!(
@@ -326,11 +328,11 @@ fn parse_knobs(
#rt
.enable_all()
.build()
.unwrap()
.expect("Failed building the Runtime")
.block_on(async #body)
}
})
.unwrap();
.expect("Parsing failure");
input.block.brace_token = brace_token;
let result = quote! {
+1 -1
View File
@@ -5,7 +5,7 @@
rust_2018_idioms,
unreachable_pub
)]
#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
#![doc(test(
no_crate_inject,
attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
+1 -1
View File
@@ -30,7 +30,7 @@ signal = ["tokio/signal"]
[dependencies]
futures-core = { version = "0.3.0" }
pin-project-lite = "0.2.0"
tokio = { version = "1.2.0", path = "../tokio", features = ["sync"] }
tokio = { version = "1.8.0", path = "../tokio", features = ["sync"] }
tokio-util = { version = "0.6.3", path = "../tokio-util", optional = true }
[dev-dependencies]
+1 -1
View File
@@ -10,7 +10,7 @@
unreachable_pub
)]
#![cfg_attr(docsrs, feature(doc_cfg))]
#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
#![doc(test(
no_crate_inject,
attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
+1 -1
View File
@@ -515,7 +515,7 @@ pub trait StreamExt: Stream {
/// Skip elements from the underlying stream while the provided predicate
/// resolves to `true`.
///
/// This function, like [`Iterator::skip_while`], will ignore elemets from the
/// This function, like [`Iterator::skip_while`], will ignore elements from the
/// stream until the predicate `f` resolves to `false`. Once one element
/// returns false, the rest of the elements will be yielded.
///
+2 -2
View File
@@ -113,7 +113,7 @@ impl<T: AsRef<str>> sealed::FromStreamPriv<T> for String {
}
fn finalize(_: sealed::Internal, collection: &mut String) -> String {
mem::replace(collection, String::new())
mem::take(collection)
}
}
@@ -132,7 +132,7 @@ impl<T> sealed::FromStreamPriv<T> for Vec<T> {
}
fn finalize(_: sealed::Internal, collection: &mut Vec<T>) -> Vec<T> {
mem::replace(collection, vec![])
mem::take(collection)
}
}
+12 -1
View File
@@ -69,7 +69,18 @@ impl<S: Stream> Stream for Timeout<S> {
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.stream.size_hint()
let (lower, upper) = self.stream.size_hint();
// The timeout stream may insert an error before and after each message
// from the underlying stream, but no more than one error between each
// message. Hence the upper bound is computed as 2x+1.
// Using a helper function to enable use of question mark operator.
fn twice_plus_one(value: Option<usize>) -> Option<usize> {
value?.checked_mul(2)?.checked_add(1)
}
(lower, twice_plus_one(upper))
}
}
+3 -3
View File
@@ -364,11 +364,11 @@ impl<K, V> StreamMap<K, V> {
/// # Examples
///
/// ```
/// use std::collections::HashMap;
/// use tokio_stream::{StreamMap, pending};
///
/// let mut a = HashMap::new();
/// let mut a = StreamMap::new();
/// assert!(a.is_empty());
/// a.insert(1, "a");
/// a.insert(1, pending::<i32>());
/// assert!(!a.is_empty());
/// ```
pub fn is_empty(&self) -> bool {
+3 -3
View File
@@ -11,7 +11,7 @@ use tokio::sync::watch::error::RecvError;
/// A wrapper around [`tokio::sync::watch::Receiver`] that implements [`Stream`].
///
/// This stream will always start by yielding the current value when the WatchStream is polled,
/// regardles of whether it was the initial value or sent afterwards.
/// regardless of whether it was the initial value or sent afterwards.
///
/// # Examples
///
@@ -72,10 +72,10 @@ impl<T: Clone + 'static + Send + Sync> Stream for WatchStream<T> {
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let (result, rx) = ready!(self.inner.poll(cx));
let (result, mut rx) = ready!(self.inner.poll(cx));
match result {
Ok(_) => {
let received = (*rx.borrow()).clone();
let received = (*rx.borrow_and_update()).clone();
self.inner.set(make_future(rx));
Poll::Ready(Some(received))
}
+2
View File
@@ -1,3 +1,5 @@
#![allow(clippy::diverging_sub_expression)]
use std::rc::Rc;
#[allow(dead_code)]
+2 -2
View File
@@ -89,12 +89,12 @@ fn size_overflow() {
}
fn size_hint(&self) -> (usize, Option<usize>) {
(usize::max_value(), Some(usize::max_value()))
(usize::MAX, Some(usize::MAX))
}
}
let m1 = Monster;
let m2 = Monster;
let m = m1.chain(m2);
assert_eq!(m.size_hint(), (usize::max_value(), None));
assert_eq!(m.size_hint(), (usize::MAX, None));
}
+2 -2
View File
@@ -72,12 +72,12 @@ fn size_overflow() {
}
fn size_hint(&self) -> (usize, Option<usize>) {
(usize::max_value(), Some(usize::max_value()))
(usize::MAX, Some(usize::MAX))
}
}
let m1 = Monster;
let m2 = Monster;
let m = m1.merge(m2);
assert_eq!(m.size_hint(), (usize::max_value(), None));
assert_eq!(m.size_hint(), (usize::MAX, None));
}
+27
View File
@@ -0,0 +1,27 @@
use tokio::sync::watch;
use tokio_stream::wrappers::WatchStream;
use tokio_stream::StreamExt;
#[tokio::test]
async fn message_not_twice() {
let (tx, rx) = watch::channel("hello");
let mut counter = 0;
let mut stream = WatchStream::new(rx).map(move |payload| {
println!("{}", payload);
if payload == "goodbye" {
counter += 1;
}
if counter >= 2 {
panic!("too many goodbyes");
}
});
let task = tokio::spawn(async move { while stream.next().await.is_some() {} });
// Send goodbye just once
tx.send("goodbye").unwrap();
drop(tx);
task.await.unwrap();
}
+2 -2
View File
@@ -4,13 +4,13 @@
rust_2018_idioms,
unreachable_pub
)]
#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
#![doc(test(
no_crate_inject,
attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
))]
//! Tokio and Futures based testing utilites
//! Tokio and Futures based testing utilities
pub mod io;
+1 -1
View File
@@ -180,7 +180,7 @@ impl ThreadWaker {
}
}
/// Clears any previously received wakes, avoiding potential spurrious
/// Clears any previously received wakes, avoiding potential spurious
/// wake notifications. This should only be called immediately before running the
/// task.
fn clear(&self) {
+1 -1
View File
@@ -234,7 +234,7 @@ impl Default for AnyDelimiterCodec {
}
}
/// An error occured while encoding or decoding a chunk.
/// An error occurred while encoding or decoding a chunk.
#[derive(Debug)]
pub enum AnyDelimiterCodecError {
/// The maximum chunk length was exceeded.
+1 -1
View File
@@ -28,7 +28,7 @@ use std::io;
/// It is up to the Decoder to keep track of a restart after an EOF,
/// and to decide how to handle such an event by, for example,
/// allowing frames to cross EOF boundaries, re-emitting opening frames, or
/// reseting the entire internal state.
/// resetting the entire internal state.
///
/// [`Framed`]: crate::codec::Framed
/// [`FramedRead`]: crate::codec::FramedRead
+3 -3
View File
@@ -124,7 +124,7 @@ where
// to a combination of the `is_readable` and `eof` flags. States persist across
// loop entries and most state transitions occur with a return.
//
// The intitial state is `reading`.
// The initial state is `reading`.
//
// | state | eof | is_readable |
// |---------|-------|-------------|
@@ -155,10 +155,10 @@ where
// Both signal that there is no such data by returning `None`.
//
// If `decode` couldn't read a frame and the upstream source has returned eof,
// `decode_eof` will attemp to decode the remaining bytes as closing frames.
// `decode_eof` will attempt to decode the remaining bytes as closing frames.
//
// If the underlying AsyncRead is resumable, we may continue after an EOF,
// but must finish emmiting all of it's associated `decode_eof` frames.
// but must finish emitting all of it's associated `decode_eof` frames.
// Furthermore, we don't want to emit any `decode_eof` frames on retried
// reads after an EOF unless we've actually read more data.
if state.is_readable {
+1 -1
View File
@@ -486,7 +486,7 @@ impl LengthDelimitedCodec {
// Skip the required bytes
src.advance(self.builder.length_field_offset);
// match endianess
// match endianness
let n = if self.builder.length_field_is_big_endian {
src.get_uint(field_len)
} else {
+2 -2
View File
@@ -203,12 +203,12 @@ impl Default for LinesCodec {
}
}
/// An error occured while encoding or decoding a line.
/// An error occurred while encoding or decoding a line.
#[derive(Debug)]
pub enum LinesCodecError {
/// The maximum line length was exceeded.
MaxLineLengthExceeded,
/// An IO error occured.
/// An IO error occurred.
Io(io::Error),
}
+1 -1
View File
@@ -67,7 +67,7 @@ pub enum Either<L, R> {
}
/// A small helper macro which reduces amount of boilerplate in the actual trait method implementation.
/// It takes an invokation of method as an argument (e.g. `self.poll(cx)`), and redirects it to either
/// It takes an invocation of method as an argument (e.g. `self.poll(cx)`), and redirects it to either
/// enum variant held in `self`.
macro_rules! delegate_call {
($self:ident.$method:ident($($args:ident),+)) => {
+1 -1
View File
@@ -5,7 +5,7 @@
rust_2018_idioms,
unreachable_pub
)]
#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
#![doc(test(
no_crate_inject,
attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
+13 -2
View File
@@ -1,5 +1,6 @@
//! An asynchronously awaitable `CancellationToken`.
//! The token allows to signal a cancellation request to one or more tasks.
pub(crate) mod guard;
use crate::loom::sync::atomic::AtomicUsize;
use crate::loom::sync::Mutex;
@@ -11,6 +12,8 @@ use core::ptr::NonNull;
use core::sync::atomic::Ordering;
use core::task::{Context, Poll, Waker};
use guard::DropGuard;
/// A token which can be used to signal a cancellation request to one or more
/// tasks.
///
@@ -275,6 +278,14 @@ impl CancellationToken {
}
}
/// Creates a `DropGuard` for this token.
///
/// Returned guard will cancel this token (and all its children) on drop
/// unless disarmed.
pub fn drop_guard(self) -> DropGuard {
DropGuard { inner: Some(self) }
}
unsafe fn register(
&self,
wait_node: &mut ListNode<WaitQueueEntry>,
@@ -764,8 +775,8 @@ impl CancellationTokenState {
return Poll::Ready(());
}
// So far the token is not cancelled. However it could be cancelld before
// we get the chance to store the `Waker`. Therfore we need to check
// So far the token is not cancelled. However it could be cancelled before
// we get the chance to store the `Waker`. Therefore we need to check
// for cancellation again inside the mutex.
let mut guard = self.synchronized.lock().unwrap();
if guard.is_cancelled {
@@ -0,0 +1,27 @@
use crate::sync::CancellationToken;
/// A wrapper for cancellation token which automatically cancels
/// it on drop. It is created using `drop_guard` method on the `CancellationToken`.
#[derive(Debug)]
pub struct DropGuard {
pub(super) inner: Option<CancellationToken>,
}
impl DropGuard {
/// Returns stored cancellation token and removes this drop guard instance
/// (i.e. it will no longer cancel token). Other guards for this token
/// are not affected.
pub fn disarm(mut self) -> CancellationToken {
self.inner
.take()
.expect("`inner` can be only None in a destructor")
}
}
impl Drop for DropGuard {
fn drop(&mut self) {
if let Some(inner) = &self.inner {
inner.cancel();
}
}
}
@@ -222,7 +222,7 @@ impl<T> LinkedList<T> {
}
}
/// Returns whether the linked list doesn not contain any node
/// Returns whether the linked list does not contain any node
pub fn is_empty(&self) -> bool {
if self.head.is_some() {
return false;
+1 -1
View File
@@ -1,7 +1,7 @@
//! Synchronization primitives
mod cancellation_token;
pub use cancellation_token::{CancellationToken, WaitForCancellationFuture};
pub use cancellation_token::{guard::DropGuard, CancellationToken, WaitForCancellationFuture};
mod intrusive_double_linked_list;
+1 -1
View File
@@ -12,9 +12,9 @@ use std::time::Duration;
mod wheel;
#[doc(inline)]
pub mod delay_queue;
#[doc(inline)]
pub use delay_queue::DelayQueue;
// ===== Internal utils =====
+3 -3
View File
@@ -6,9 +6,9 @@ use tokio_util::sync::PollSemaphore;
type SemRet = Option<OwnedSemaphorePermit>;
fn semaphore_poll<'a>(
sem: &'a mut PollSemaphore,
) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + 'a> {
fn semaphore_poll(
sem: &mut PollSemaphore,
) -> tokio_test::task::Spawn<impl Future<Output = SemRet> + '_> {
let fut = futures::future::poll_fn(move |cx| sem.poll_acquire(cx));
tokio_test::task::spawn(fut)
}
+134
View File
@@ -1,3 +1,129 @@
# 1.8.1 (July 6, 2021)
Forward ports 1.5.1 fixes.
### Fixed
- runtime: remotely abort tasks on `JoinHandle::abort` ([#3934])
[#3934]: https://github.com/tokio-rs/tokio/pull/3934
# 1.8.0 (July 2, 2021)
### Added
- io: add `get_{ref,mut}` methods to `AsyncFdReadyGuard` and `AsyncFdReadyMutGuard` ([#3807])
- io: efficient implementation of vectored writes for `BufWriter` ([#3163])
- net: add ready/try methods to `NamedPipe{Client,Server}` ([#3866], [#3899])
- sync: add `watch::Receiver::borrow_and_update` ([#3813])
- sync: implement `From<T>` for `OnceCell<T>` ([#3877])
- time: allow users to specify Interval behaviour when delayed ([#3721])
### Added (unstable)
- rt: add `tokio::task::Builder` ([#3881])
### Fixed
- net: handle HUP event with `UnixStream` ([#3898])
### Documented
- doc: document cancellation safety ([#3900])
- time: add wait alias to sleep ([#3897])
- time: document auto-advancing behaviour of runtime ([#3763])
[#3163]: https://github.com/tokio-rs/tokio/pull/3163
[#3721]: https://github.com/tokio-rs/tokio/pull/3721
[#3763]: https://github.com/tokio-rs/tokio/pull/3763
[#3807]: https://github.com/tokio-rs/tokio/pull/3807
[#3813]: https://github.com/tokio-rs/tokio/pull/3813
[#3866]: https://github.com/tokio-rs/tokio/pull/3866
[#3877]: https://github.com/tokio-rs/tokio/pull/3877
[#3881]: https://github.com/tokio-rs/tokio/pull/3881
[#3897]: https://github.com/tokio-rs/tokio/pull/3897
[#3898]: https://github.com/tokio-rs/tokio/pull/3898
[#3899]: https://github.com/tokio-rs/tokio/pull/3899
[#3900]: https://github.com/tokio-rs/tokio/pull/3900
# 1.7.2 (July 6, 2021)
Forward ports 1.5.1 fixes.
### Fixed
- runtime: remotely abort tasks on `JoinHandle::abort` ([#3934])
[#3934]: https://github.com/tokio-rs/tokio/pull/3934
# 1.7.1 (June 18, 2021)
### Fixed
- runtime: fix early task shutdown during runtime shutdown ([#3870])
[#3870]: https://github.com/tokio-rs/tokio/pull/3870
# 1.7.0 (June 15, 2021)
### Added
- net: add named pipes on windows ([#3760])
- net: add `TcpSocket` from `std::net::TcpStream` conversion ([#3838])
- sync: add `receiver_count` to `watch::Sender` ([#3729])
- sync: export `sync::notify::Notified` future publicly ([#3840])
- tracing: instrument task wakers ([#3836])
### Fixed
- macros: suppress `clippy::default_numeric_fallback` lint in generated code ([#3831])
- runtime: immediately drop new tasks when runtime is shut down ([#3752])
- sync: deprecate unused `mpsc::RecvError` type ([#3833])
### Documented
- io: clarify EOF condition for `AsyncReadExt::read_buf` ([#3850])
- io: clarify limits on return values of `AsyncWrite::poll_write` ([#3820])
- sync: add examples to Semaphore ([#3808])
[#3729]: https://github.com/tokio-rs/tokio/pull/3729
[#3752]: https://github.com/tokio-rs/tokio/pull/3752
[#3760]: https://github.com/tokio-rs/tokio/pull/3760
[#3808]: https://github.com/tokio-rs/tokio/pull/3808
[#3820]: https://github.com/tokio-rs/tokio/pull/3820
[#3831]: https://github.com/tokio-rs/tokio/pull/3831
[#3833]: https://github.com/tokio-rs/tokio/pull/3833
[#3836]: https://github.com/tokio-rs/tokio/pull/3836
[#3838]: https://github.com/tokio-rs/tokio/pull/3838
[#3840]: https://github.com/tokio-rs/tokio/pull/3840
[#3850]: https://github.com/tokio-rs/tokio/pull/3850
# 1.6.3 (July 6, 2021)
Forward ports 1.5.1 fixes.
### Fixed
- runtime: remotely abort tasks on `JoinHandle::abort` ([#3934])
[#3934]: https://github.com/tokio-rs/tokio/pull/3934
# 1.6.2 (June 14, 2021)
### Fixes
- test: sub-ms `time:advance` regression introduced in 1.6 ([#3852])
[#3852]: https://github.com/tokio-rs/tokio/pull/3852
# 1.6.1 (May 28, 2021)
This release reverts [#3518] because it doesn't work on some kernels due to
a kernel bug. ([#3803])
[#3518]: https://github.com/tokio-rs/tokio/issues/3518
[#3803]: https://github.com/tokio-rs/tokio/issues/3803
# 1.6.0 (May 14, 2021)
### Added
@@ -44,6 +170,14 @@
[#3775]: https://github.com/tokio-rs/tokio/pull/3775
[#3780]: https://github.com/tokio-rs/tokio/pull/3780
# 1.5.1 (July 6, 2021)
### Fixed
- runtime: remotely abort tasks on `JoinHandle::abort` ([#3934])
[#3934]: https://github.com/tokio-rs/tokio/pull/3934
# 1.5.0 (April 12, 2021)
### Added
+10 -5
View File
@@ -7,12 +7,12 @@ name = "tokio"
# - README.md
# - Update CHANGELOG.md.
# - Create "v1.0.x" git tag.
version = "1.6.0"
version = "1.8.1"
edition = "2018"
authors = ["Tokio Contributors <[email protected]>"]
license = "MIT"
readme = "README.md"
documentation = "https://docs.rs/tokio/1.6.0/tokio/"
documentation = "https://docs.rs/tokio/1.8.1/tokio/"
repository = "https://github.com/tokio-rs/tokio"
homepage = "https://tokio.rs"
description = """
@@ -42,7 +42,7 @@ full = [
"time",
]
fs = ["libc"]
fs = []
io-util = ["memchr", "bytes"]
# stdin, stdout, stderr
io-std = []
@@ -54,6 +54,7 @@ net = [
"mio/tcp",
"mio/udp",
"mio/uds",
"winapi/namedpipeapi",
]
process = [
"bytes",
@@ -103,11 +104,11 @@ parking_lot = { version = "0.11.0", optional = true }
tracing = { version = "0.1.21", default-features = false, features = ["std"], optional = true } # Not in full
[target.'cfg(unix)'.dependencies]
libc = { version = "0.2.87", optional = true }
libc = { version = "0.2.42", optional = true }
signal-hook-registry = { version = "1.1.1", optional = true }
[target.'cfg(unix)'.dev-dependencies]
libc = { version = "0.2.87" }
libc = { version = "0.2.42" }
nix = { version = "0.19.0" }
[target.'cfg(windows)'.dependencies.winapi]
@@ -115,6 +116,9 @@ version = "0.3.8"
default-features = false
optional = true
[target.'cfg(windows)'.dev-dependencies.ntapi]
version = "0.3.6"
[dev-dependencies]
tokio-test = { version = "0.4.0", path = "../tokio-test" }
tokio-stream = { version = "0.1", path = "../tokio-stream" }
@@ -123,6 +127,7 @@ proptest = "1"
rand = "0.8.0"
tempfile = "3.1.0"
async-stream = "0.3"
socket2 = "0.4"
[target.'cfg(loom)'.dev-dependencies]
loom = { version = "0.5", features = ["futures", "checkpoint"] }
+3 -3
View File
@@ -228,7 +228,7 @@ It is only possible to implement `AsyncRead` and `AsyncWrite` for resource types
themselves and not for `&Resource`. Implementing the traits for `&Resource`
would permit concurrent operations to the resource. Because only a single waker
is stored per direction, any concurrent usage would result in deadlocks. An
alterate implementation would call for a `Vec<Waker>` but this would result in
alternate implementation would call for a `Vec<Waker>` but this would result in
memory leaks.
## Enabling reads and writes for `&TcpStream`
@@ -268,9 +268,9 @@ select! {
}
```
It is also possible to sotre a `TcpStream` in an `Arc`.
It is also possible to store a `TcpStream` in an `Arc`.
```rust
let arc_stream = Arc::new(my_tcp_stream);
let n = arc_stream.by_ref().read(buf).await?;
```
```
+23
View File
@@ -0,0 +1,23 @@
//! Types which are documented locally in the Tokio crate, but does not actually
//! live here.
//!
//! **Note** this module is only visible on docs.rs, you cannot use it directly
//! in your own code.
/// The name of a type which is not defined here.
///
/// This is typically used as an alias for another type, like so:
///
/// ```rust,ignore
/// /// See [some::other::location](https://example.com).
/// type DEFINED_ELSEWHERE = crate::doc::NotDefinedHere;
/// ```
///
/// This type is uninhabitable like the [`never` type] to ensure that no one
/// will ever accidentally use it.
///
/// [`never` type]: https://doc.rust-lang.org/std/primitive.never.html
pub enum NotDefinedHere {}
pub mod os;
pub mod winapi;
+26
View File
@@ -0,0 +1,26 @@
//! See [std::os](https://doc.rust-lang.org/std/os/index.html).
/// Platform-specific extensions to `std` for Windows.
///
/// See [std::os::windows](https://doc.rust-lang.org/std/os/windows/index.html).
pub mod windows {
/// Windows-specific extensions to general I/O primitives.
///
/// See [std::os::windows::io](https://doc.rust-lang.org/std/os/windows/io/index.html).
pub mod io {
/// See [std::os::windows::io::RawHandle](https://doc.rust-lang.org/std/os/windows/io/type.RawHandle.html)
pub type RawHandle = crate::doc::NotDefinedHere;
/// See [std::os::windows::io::AsRawHandle](https://doc.rust-lang.org/std/os/windows/io/trait.AsRawHandle.html)
pub trait AsRawHandle {
/// See [std::os::windows::io::FromRawHandle::from_raw_handle](https://doc.rust-lang.org/std/os/windows/io/trait.AsRawHandle.html#tymethod.as_raw_handle)
fn as_raw_handle(&self) -> RawHandle;
}
/// See [std::os::windows::io::FromRawHandle](https://doc.rust-lang.org/std/os/windows/io/trait.FromRawHandle.html)
pub trait FromRawHandle {
/// See [std::os::windows::io::FromRawHandle::from_raw_handle](https://doc.rust-lang.org/std/os/windows/io/trait.FromRawHandle.html#tymethod.from_raw_handle)
unsafe fn from_raw_handle(handle: RawHandle) -> Self;
}
}
}
+66
View File
@@ -0,0 +1,66 @@
//! See [winapi].
//!
//! [winapi]: https://docs.rs/winapi
/// See [winapi::shared](https://docs.rs/winapi/*/winapi/shared/index.html).
pub mod shared {
/// See [winapi::shared::winerror](https://docs.rs/winapi/*/winapi/shared/winerror/index.html).
#[allow(non_camel_case_types)]
pub mod winerror {
/// See [winapi::shared::winerror::ERROR_ACCESS_DENIED][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/shared/winerror/constant.ERROR_ACCESS_DENIED.html
pub type ERROR_ACCESS_DENIED = crate::doc::NotDefinedHere;
/// See [winapi::shared::winerror::ERROR_PIPE_BUSY][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/shared/winerror/constant.ERROR_PIPE_BUSY.html
pub type ERROR_PIPE_BUSY = crate::doc::NotDefinedHere;
/// See [winapi::shared::winerror::ERROR_MORE_DATA][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/shared/winerror/constant.ERROR_MORE_DATA.html
pub type ERROR_MORE_DATA = crate::doc::NotDefinedHere;
}
}
/// See [winapi::um](https://docs.rs/winapi/*/winapi/um/index.html).
pub mod um {
/// See [winapi::um::winbase](https://docs.rs/winapi/*/winapi/um/winbase/index.html).
#[allow(non_camel_case_types)]
pub mod winbase {
/// See [winapi::um::winbase::PIPE_TYPE_MESSAGE][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_TYPE_MESSAGE.html
pub type PIPE_TYPE_MESSAGE = crate::doc::NotDefinedHere;
/// See [winapi::um::winbase::PIPE_TYPE_BYTE][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_TYPE_BYTE.html
pub type PIPE_TYPE_BYTE = crate::doc::NotDefinedHere;
/// See [winapi::um::winbase::PIPE_CLIENT_END][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_CLIENT_END.html
pub type PIPE_CLIENT_END = crate::doc::NotDefinedHere;
/// See [winapi::um::winbase::PIPE_SERVER_END][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.PIPE_SERVER_END.html
pub type PIPE_SERVER_END = crate::doc::NotDefinedHere;
/// See [winapi::um::winbase::SECURITY_IDENTIFICATION][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/um/winbase/constant.SECURITY_IDENTIFICATION.html
pub type SECURITY_IDENTIFICATION = crate::doc::NotDefinedHere;
}
/// See [winapi::um::minwinbase](https://docs.rs/winapi/*/winapi/um/minwinbase/index.html).
#[allow(non_camel_case_types)]
pub mod minwinbase {
/// See [winapi::um::minwinbase::SECURITY_ATTRIBUTES][winapi]
///
/// [winapi]: https://docs.rs/winapi/*/winapi/um/minwinbase/constant.SECURITY_ATTRIBUTES.html
pub type SECURITY_ATTRIBUTES = crate::doc::NotDefinedHere;
}
}
+2 -189
View File
@@ -491,18 +491,14 @@ impl AsyncRead for File {
loop {
match inner.state {
Idle(ref mut buf_cell) => {
let buf = buf_cell.as_mut().unwrap();
let mut buf = buf_cell.take().unwrap();
if !buf.is_empty() {
buf.copy_to(dst);
*buf_cell = Some(buf);
return Ready(Ok(()));
}
if let Some(x) = try_nonblocking_read(me.std.as_ref(), dst) {
return Ready(x);
}
let mut buf = buf_cell.take().unwrap();
buf.ensure_capacity_for(dst);
let std = me.std.clone();
@@ -760,186 +756,3 @@ impl Inner {
}
}
}
#[cfg(all(target_os = "linux", not(test)))]
pub(crate) fn try_nonblocking_read(
file: &crate::fs::sys::File,
dst: &mut ReadBuf<'_>,
) -> Option<std::io::Result<()>> {
use std::sync::atomic::{AtomicBool, Ordering};
static NONBLOCKING_READ_SUPPORTED: AtomicBool = AtomicBool::new(true);
if !NONBLOCKING_READ_SUPPORTED.load(Ordering::Relaxed) {
return None;
}
let out = preadv2::preadv2_safe(file, dst, -1, preadv2::RWF_NOWAIT);
if let Err(err) = &out {
match err.raw_os_error() {
Some(libc::ENOSYS) => {
NONBLOCKING_READ_SUPPORTED.store(false, Ordering::Relaxed);
return None;
}
Some(libc::ENOTSUP) | Some(libc::EAGAIN) => return None,
_ => {}
}
}
Some(out)
}
#[cfg(any(not(target_os = "linux"), test))]
pub(crate) fn try_nonblocking_read(
_file: &crate::fs::sys::File,
_dst: &mut ReadBuf<'_>,
) -> Option<std::io::Result<()>> {
None
}
#[cfg(target_os = "linux")]
mod preadv2 {
use libc::{c_int, c_long, c_void, iovec, off_t, ssize_t};
use std::os::unix::prelude::AsRawFd;
use crate::io::ReadBuf;
pub(crate) fn preadv2_safe(
file: &std::fs::File,
dst: &mut ReadBuf<'_>,
offset: off_t,
flags: c_int,
) -> std::io::Result<()> {
unsafe {
/* We have to defend against buffer overflows manually here. The slice API makes
* this fairly straightforward. */
let unfilled = dst.unfilled_mut();
let mut iov = iovec {
iov_base: unfilled.as_mut_ptr() as *mut c_void,
iov_len: unfilled.len(),
};
/* We take a File object rather than an fd as reading from a sensitive fd may confuse
* other unsafe code that assumes that only they have access to that fd. */
let bytes_read = preadv2(
file.as_raw_fd(),
&mut iov as *mut iovec as *const iovec,
1,
offset,
flags,
);
if bytes_read < 0 {
Err(std::io::Error::last_os_error())
} else {
/* preadv2 returns the number of bytes read, e.g. the number of bytes that have
* written into `unfilled`. So it's safe to assume that the data is now
* initialised */
dst.assume_init(bytes_read as usize);
dst.advance(bytes_read as usize);
Ok(())
}
}
}
#[cfg(test)]
mod test {
use super::*;
#[test]
fn test_preadv2_safe() {
use std::io::{Seek, Write};
use std::mem::MaybeUninit;
use tempfile::tempdir;
let tmp = tempdir().unwrap();
let filename = tmp.path().join("file");
const MESSAGE: &[u8] = b"Hello this is a test";
{
let mut f = std::fs::File::create(&filename).unwrap();
f.write_all(MESSAGE).unwrap();
}
let f = std::fs::File::open(&filename).unwrap();
let mut buf = [MaybeUninit::<u8>::new(0); 50];
let mut br = ReadBuf::uninit(&mut buf);
// Basic use:
preadv2_safe(&f, &mut br, 0, 0).unwrap();
assert_eq!(br.initialized().len(), MESSAGE.len());
assert_eq!(br.filled(), MESSAGE);
// Here we check that offset works, but also that appending to a non-empty buffer
// behaves correctly WRT initialisation.
preadv2_safe(&f, &mut br, 5, 0).unwrap();
assert_eq!(br.initialized().len(), MESSAGE.len() * 2 - 5);
assert_eq!(br.filled(), b"Hello this is a test this is a test".as_ref());
// offset of -1 means use the current cursor. This has not been advanced by the
// previous reads because we specified an offset there.
preadv2_safe(&f, &mut br, -1, 0).unwrap();
assert_eq!(br.remaining(), 0);
assert_eq!(
br.filled(),
b"Hello this is a test this is a testHello this is a".as_ref()
);
// but the offset should have been advanced by that read
br.clear();
preadv2_safe(&f, &mut br, -1, 0).unwrap();
assert_eq!(br.filled(), b" test");
// This should be in cache, so RWF_NOWAIT should work, but it not being in cache
// (EAGAIN) or not supported by the underlying filesystem (ENOTSUP) is fine too.
br.clear();
match preadv2_safe(&f, &mut br, 0, RWF_NOWAIT) {
Ok(()) => assert_eq!(br.filled(), MESSAGE),
Err(e) => assert!(matches!(
e.raw_os_error(),
Some(libc::ENOTSUP) | Some(libc::EAGAIN)
)),
}
// Test handling large offsets
{
// I hope the underlying filesystem supports sparse files
let mut w = std::fs::OpenOptions::new()
.write(true)
.open(&filename)
.unwrap();
w.set_len(0x1_0000_0000).unwrap();
w.seek(std::io::SeekFrom::Start(0x1_0000_0000)).unwrap();
w.write_all(b"This is a Large File").unwrap();
}
br.clear();
preadv2_safe(&f, &mut br, 0x1_0000_0008, 0).unwrap();
assert_eq!(br.filled(), b"a Large File");
}
}
fn pos_to_lohi(offset: off_t) -> (c_long, c_long) {
// 64-bit offset is split over high and low 32-bits on 32-bit architectures.
// 64-bit architectures still have high and low arguments, but only the low
// one is inspected. See pos_from_hilo in linux/fs/read_write.c.
const HALF_LONG_BITS: usize = core::mem::size_of::<c_long>() * 8 / 2;
(
offset as c_long,
// We want to shift this off_t value by size_of::<c_long>(). We can't do
// it in one shift because if they're both 64-bits we'd be doing u64 >> 64
// which is implementation defined. Instead do it in two halves:
((offset >> HALF_LONG_BITS) >> HALF_LONG_BITS) as c_long,
)
}
pub(crate) const RWF_NOWAIT: c_int = 0x00000008;
unsafe fn preadv2(
fd: c_int,
iov: *const iovec,
iovcnt: c_int,
offset: off_t,
flags: c_int,
) -> ssize_t {
// Call via libc::syscall rather than libc::preadv2. preadv2 is only supported by glibc
// and only since v2.26. By using syscall we don't need to worry about compatiblity with
// old glibc versions and it will work on Android and musl too. The downside is that you
// can't use `LD_PRELOAD` tricks any more to intercept these calls.
let (lo, hi) = pos_to_lohi(offset);
libc::syscall(libc::SYS_preadv2, fd, iov, iovcnt, lo, hi, flags) as ssize_t
}
}
+4
View File
@@ -13,8 +13,12 @@ use std::{io, path::Path};
/// buffer based on the file size when available, so it is generally faster than
/// reading into a vector created with `Vec::new()`.
///
/// This operation is implemented by running the equivalent blocking operation
/// on a separate thread pool using [`spawn_blocking`].
///
/// [`File::open`]: super::File::open
/// [`read_to_end`]: crate::io::AsyncReadExt::read_to_end
/// [`spawn_blocking`]: crate::task::spawn_blocking
///
/// # Errors
///
+5
View File
@@ -13,6 +13,11 @@ use std::task::Poll;
/// Returns a stream over the entries within a directory.
///
/// This is an async version of [`std::fs::read_dir`](std::fs::read_dir)
///
/// This operation is implemented by running the equivalent blocking
/// operation on a separate thread pool using [`spawn_blocking`].
///
/// [`spawn_blocking`]: crate::task::spawn_blocking
pub async fn read_dir(path: impl AsRef<Path>) -> io::Result<ReadDir> {
let path = path.as_ref().to_owned();
let std = asyncify(|| std::fs::read_dir(path)).await?;
+4
View File
@@ -7,6 +7,10 @@ use std::{io, path::Path};
///
/// This is the async equivalent of [`std::fs::read_to_string`][std].
///
/// This operation is implemented by running the equivalent blocking operation
/// on a separate thread pool using [`spawn_blocking`].
///
/// [`spawn_blocking`]: crate::task::spawn_blocking
/// [std]: fn@std::fs::read_to_string
///
/// # Examples
+4
View File
@@ -7,6 +7,10 @@ use std::{io, path::Path};
///
/// This is the async equivalent of [`std::fs::write`][std].
///
/// This operation is implemented by running the equivalent blocking operation
/// on a separate thread pool using [`spawn_blocking`].
///
/// [`spawn_blocking`]: crate::task::spawn_blocking
/// [std]: fn@std::fs::write
///
/// # Examples
+11
View File
@@ -22,3 +22,14 @@ cfg_sync! {
mod block_on;
pub(crate) use block_on::block_on;
}
cfg_trace! {
mod trace;
pub(crate) use trace::InstrumentedFuture as Future;
}
cfg_not_trace! {
cfg_rt! {
pub(crate) use std::future::Future;
}
}
+11
View File
@@ -0,0 +1,11 @@
use std::future::Future;
pub(crate) trait InstrumentedFuture: Future {
fn id(&self) -> Option<tracing::Id>;
}
impl<F: Future> InstrumentedFuture for tracing::instrument::Instrumented<F> {
fn id(&self) -> Option<tracing::Id> {
self.span().id()
}
}
+30
View File
@@ -540,6 +540,16 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyGuard<'a, Inner> {
result => Ok(result),
}
}
/// Returns a shared reference to the inner [`AsyncFd`].
pub fn get_ref(&self) -> &AsyncFd<Inner> {
self.async_fd
}
/// Returns a shared reference to the backing object of the inner [`AsyncFd`].
pub fn get_inner(&self) -> &Inner {
self.get_ref().get_ref()
}
}
impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> {
@@ -601,6 +611,26 @@ impl<'a, Inner: AsRawFd> AsyncFdReadyMutGuard<'a, Inner> {
result => Ok(result),
}
}
/// Returns a shared reference to the inner [`AsyncFd`].
pub fn get_ref(&self) -> &AsyncFd<Inner> {
self.async_fd
}
/// Returns a mutable reference to the inner [`AsyncFd`].
pub fn get_mut(&mut self) -> &mut AsyncFd<Inner> {
self.async_fd
}
/// Returns a shared reference to the backing object of the inner [`AsyncFd`].
pub fn get_inner(&self) -> &Inner {
self.get_ref().get_ref()
}
/// Returns a mutable reference to the backing object of the inner [`AsyncFd`].
pub fn get_inner_mut(&mut self) -> &mut Inner {
self.get_mut().get_mut()
}
}
impl<'a, T: std::fmt::Debug + AsRawFd> std::fmt::Debug for AsyncFdReadyGuard<'a, T> {
+5 -1
View File
@@ -45,7 +45,11 @@ use std::task::{Context, Poll};
pub trait AsyncWrite {
/// Attempt to write bytes from `buf` into the object.
///
/// On success, returns `Poll::Ready(Ok(num_bytes_written))`.
/// On success, returns `Poll::Ready(Ok(num_bytes_written))`. If successful,
/// then it must be guaranteed that `n <= buf.len()`. A return value of `0`
/// typically means that the underlying object is no longer able to accept
/// bytes and will likely not be able to in the future as well, or that the
/// buffer provided is empty.
///
/// If the object is not ready for writing, the method returns
/// `Poll::Pending` and arranges for the current task (via
+1 -1
View File
@@ -58,7 +58,7 @@ impl Interest {
self.0.is_writable()
}
/// Add together two `Interst` values.
/// Add together two `Interest` values.
///
/// This function works from a `const` context.
///
+1 -1
View File
@@ -96,7 +96,7 @@ const ADDRESS: bit::Pack = bit::Pack::least_significant(24);
//
// The generation prevents a race condition where a slab slot is reused for a
// new socket while the I/O driver is about to apply a readiness event. The
// generaton value is checked when setting new readiness. If the generation do
// generation value is checked when setting new readiness. If the generation do
// not match, then the readiness event is discarded.
const GENERATION: bit::Pack = ADDRESS.then(7);
+3 -3
View File
@@ -84,9 +84,9 @@ cfg_io_readiness! {
// The `ScheduledIo::readiness` (`AtomicUsize`) is packed full of goodness.
//
// | reserved | generation | driver tick | readinesss |
// |----------+------------+--------------+------------|
// | 1 bit | 7 bits + 8 bits + 16 bits |
// | reserved | generation | driver tick | readiness |
// |----------+------------+--------------+-----------|
// | 1 bit | 7 bits + 8 bits + 16 bits |
const READINESS: bit::Pack = bit::Pack::least_significant(16);
+3 -3
View File
@@ -45,7 +45,7 @@ impl<'a> ReadBuf<'a> {
/// Creates a new `ReadBuf` from a fully uninitialized buffer.
///
/// Use `assume_init` if part of the buffer is known to be already inintialized.
/// Use `assume_init` if part of the buffer is known to be already initialized.
#[inline]
pub fn uninit(buf: &'a mut [MaybeUninit<u8>]) -> ReadBuf<'a> {
ReadBuf {
@@ -85,7 +85,7 @@ impl<'a> ReadBuf<'a> {
#[inline]
pub fn take(&mut self, n: usize) -> ReadBuf<'_> {
let max = std::cmp::min(self.remaining(), n);
// Saftey: We don't set any of the `unfilled_mut` with `MaybeUninit::uninit`.
// Safety: We don't set any of the `unfilled_mut` with `MaybeUninit::uninit`.
unsafe { ReadBuf::uninit(&mut self.unfilled_mut()[..max]) }
}
@@ -217,7 +217,7 @@ impl<'a> ReadBuf<'a> {
///
/// # Panics
///
/// Panics if the filled region of the buffer would become larger than the intialized region.
/// Panics if the filled region of the buffer would become larger than the initialized region.
#[inline]
pub fn set_filled(&mut self, n: usize) {
assert!(
+2 -2
View File
@@ -52,10 +52,10 @@ where
buf = &buf[..crate::io::blocking::MAX_BUF];
// Now there are two possibilites.
// Now there are two possibilities.
// If caller gave is binary buffer, we **should not** shrink it
// anymore, because excessive shrinking hits performance.
// If caller gave as binary buffer, we **must** additionaly
// If caller gave as binary buffer, we **must** additionally
// shrink it to strip incomplete char at the end of buffer.
// that's why check we will perform now is allowed to have
// false-positive.
+35 -9
View File
@@ -105,8 +105,10 @@ cfg_io_util! {
/// async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize>;
/// ```
///
/// This function does not provide any guarantees about whether it
/// completes immediately or asynchronously
/// This method does not provide any guarantees about whether it
/// completes immediately or asynchronously.
///
/// # Return
///
/// If the return value of this method is `Ok(n)`, then it must be
/// guaranteed that `0 <= n <= buf.len()`. A nonzero `n` value indicates
@@ -136,6 +138,12 @@ cfg_io_util! {
/// variant will be returned. If an error is returned then it must be
/// guaranteed that no bytes were read.
///
/// # Cancel safety
///
/// This method is cancel safe. If you use it as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that no data was read.
///
/// # Examples
///
/// [`File`][crate::fs::File]s implement `Read`:
@@ -175,14 +183,19 @@ cfg_io_util! {
/// Usually, only a single `read` syscall is issued, even if there is
/// more space in the supplied buffer.
///
/// This function does not provide any guarantees about whether it
/// completes immediately or asynchronously
/// This method does not provide any guarantees about whether it
/// completes immediately or asynchronously.
///
/// # Return
///
/// On a successful read, the number of read bytes is returned. If the
/// supplied buffer is not empty and the function returns `Ok(0)` then
/// the source has reached an "end-of-file" event.
/// A nonzero `n` value indicates that the buffer `buf` has been filled
/// in with `n` bytes of data from this source. If `n` is `0`, then it
/// can indicate one of two scenarios:
///
/// 1. This reader has reached its "end of file" and will likely no longer
/// be able to produce bytes. Note that this does not mean that the
/// reader will *always* no longer be able to produce bytes.
/// 2. The buffer specified had a remaining capacity of zero.
///
/// # Errors
///
@@ -190,6 +203,12 @@ cfg_io_util! {
/// variant will be returned. If an error is returned then it must be
/// guaranteed that no bytes were read.
///
/// # Cancel safety
///
/// This method is cancel safe. If you use it as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that no data was read.
///
/// # Examples
///
/// [`File`] implements `Read` and [`BytesMut`] implements [`BufMut`]:
@@ -254,6 +273,13 @@ cfg_io_util! {
/// it has read, but it will never read more than would be necessary to
/// completely fill the buffer.
///
/// # Cancel safety
///
/// This method is not cancellation safe. If the method is used as the
/// event in a [`tokio::select!`](crate::select) statement and some
/// other branch completes first, then some data may already have been
/// read into `buf`.
///
/// # Examples
///
/// [`File`][crate::fs::File]s implement `Read`:
@@ -579,7 +605,7 @@ cfg_io_util! {
/// async fn main() -> io::Result<()> {
/// let mut reader = Cursor::new(vec![0x80, 0, 0, 0, 0, 0, 0, 0]);
///
/// assert_eq!(i64::min_value(), reader.read_i64().await?);
/// assert_eq!(i64::MIN, reader.read_i64().await?);
/// Ok(())
/// }
/// ```
@@ -659,7 +685,7 @@ cfg_io_util! {
/// 0, 0, 0, 0, 0, 0, 0, 0
/// ]);
///
/// assert_eq!(i128::min_value(), reader.read_i128().await?);
/// assert_eq!(i128::MIN, reader.read_i128().await?);
/// Ok(())
/// }
/// ```
+46 -6
View File
@@ -97,6 +97,13 @@ cfg_io_util! {
/// It is **not** considered an error if the entire buffer could not be
/// written to this writer.
///
/// # Cancel safety
///
/// This method is cancellation safe in the sense that if it is used as
/// the event in a [`tokio::select!`](crate::select) statement and some
/// other branch completes first, then it is guaranteed that no data was
/// written to this `AsyncWrite`.
///
/// # Examples
///
/// ```no_run
@@ -129,6 +136,13 @@ cfg_io_util! {
///
/// See [`AsyncWrite::poll_write_vectored`] for more details.
///
/// # Cancel safety
///
/// This method is cancellation safe in the sense that if it is used as
/// the event in a [`tokio::select!`](crate::select) statement and some
/// other branch completes first, then it is guaranteed that no data was
/// written to this `AsyncWrite`.
///
/// # Examples
///
/// ```no_run
@@ -195,6 +209,13 @@ cfg_io_util! {
/// It is **not** considered an error if the entire buffer could not be
/// written to this writer.
///
/// # Cancel safety
///
/// This method is cancellation safe in the sense that if it is used as
/// the event in a [`tokio::select!`](crate::select) statement and some
/// other branch completes first, then it is guaranteed that no data was
/// written to this `AsyncWrite`.
///
/// # Examples
///
/// [`File`] implements [`AsyncWrite`] and [`Cursor`]`<&[u8]>` implements [`Buf`]:
@@ -243,6 +264,7 @@ cfg_io_util! {
/// while buf.has_remaining() {
/// self.write_buf(&mut buf).await?;
/// }
/// Ok(())
/// }
/// ```
///
@@ -254,6 +276,15 @@ cfg_io_util! {
/// The buffer is advanced after each chunk is successfully written. After failure,
/// `src.chunk()` will return the chunk that failed to write.
///
/// # Cancel safety
///
/// If `write_all_buf` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then the data in the provided buffer may have been
/// partially written. However, it is guaranteed that the provided
/// buffer has been [advanced] by the amount of bytes that have been
/// partially written.
///
/// # Examples
///
/// [`File`] implements [`AsyncWrite`] and [`Cursor`]`<&[u8]>` implements [`Buf`]:
@@ -261,6 +292,7 @@ cfg_io_util! {
/// [`File`]: crate::fs::File
/// [`Buf`]: bytes::Buf
/// [`Cursor`]: std::io::Cursor
/// [advanced]: bytes::Buf::advance
///
/// ```no_run
/// use tokio::io::{self, AsyncWriteExt};
@@ -300,6 +332,14 @@ cfg_io_util! {
/// has been successfully written or such an error occurs. The first
/// error generated from this method will be returned.
///
/// # Cancel safety
///
/// This method is not cancellation safe. If it is used as the event
/// in a [`tokio::select!`](crate::select) statement and some other
/// branch completes first, then the provided buffer may have been
/// partially written, but future calls to `write_all` will start over
/// from the beginning of the buffer.
///
/// # Errors
///
/// This function will return the first error that [`write`] returns.
@@ -621,8 +661,8 @@ cfg_io_util! {
/// async fn main() -> io::Result<()> {
/// let mut writer = Vec::new();
///
/// writer.write_i64(i64::min_value()).await?;
/// writer.write_i64(i64::max_value()).await?;
/// writer.write_i64(i64::MIN).await?;
/// writer.write_i64(i64::MAX).await?;
///
/// assert_eq!(writer, b"\x80\x00\x00\x00\x00\x00\x00\x00\x7f\xff\xff\xff\xff\xff\xff\xff");
/// Ok(())
@@ -699,7 +739,7 @@ cfg_io_util! {
/// async fn main() -> io::Result<()> {
/// let mut writer = Vec::new();
///
/// writer.write_i128(i128::min_value()).await?;
/// writer.write_i128(i128::MIN).await?;
///
/// assert_eq!(writer, vec![
/// 0x80, 0, 0, 0, 0, 0, 0, 0,
@@ -930,8 +970,8 @@ cfg_io_util! {
/// async fn main() -> io::Result<()> {
/// let mut writer = Vec::new();
///
/// writer.write_i64_le(i64::min_value()).await?;
/// writer.write_i64_le(i64::max_value()).await?;
/// writer.write_i64_le(i64::MIN).await?;
/// writer.write_i64_le(i64::MAX).await?;
///
/// assert_eq!(writer, b"\x00\x00\x00\x00\x00\x00\x00\x80\xff\xff\xff\xff\xff\xff\xff\x7f");
/// Ok(())
@@ -1008,7 +1048,7 @@ cfg_io_util! {
/// async fn main() -> io::Result<()> {
/// let mut writer = Vec::new();
///
/// writer.write_i128_le(i128::min_value()).await?;
/// writer.write_i128_le(i128::MIN).await?;
///
/// assert_eq!(writer, vec![
/// 0, 0, 0, 0, 0, 0, 0,
+14 -2
View File
@@ -2,7 +2,7 @@ use crate::io::util::DEFAULT_BUF_SIZE;
use crate::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
use pin_project_lite::pin_project;
use std::io::{self, SeekFrom};
use std::io::{self, IoSlice, SeekFrom};
use std::pin::Pin;
use std::task::{Context, Poll};
use std::{cmp, fmt, mem};
@@ -198,7 +198,7 @@ impl<R: AsyncRead + AsyncSeek> AsyncSeek for BufReader<R> {
// it should be safe to assume that remainder fits within an i64 as the alternative
// means we managed to allocate 8 exbibytes and that's absurd.
// But it's not out of the realm of possibility for some weird underlying reader to
// support seeking by i64::min_value() so we need to handle underflow when subtracting
// support seeking by i64::MIN so we need to handle underflow when subtracting
// remainder.
if let Some(offset) = n.checked_sub(remainder) {
self.as_mut()
@@ -268,6 +268,18 @@ impl<R: AsyncRead + AsyncWrite> AsyncWrite for BufReader<R> {
self.get_pin_mut().poll_write(cx, buf)
}
fn poll_write_vectored(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &[IoSlice<'_>],
) -> Poll<io::Result<usize>> {
self.get_pin_mut().poll_write_vectored(cx, bufs)
}
fn is_write_vectored(&self) -> bool {
self.get_ref().is_write_vectored()
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.get_pin_mut().poll_flush(cx)
}
+42 -2
View File
@@ -1,8 +1,8 @@
use crate::io::util::{BufReader, BufWriter};
use crate::io::{AsyncBufRead, AsyncRead, AsyncWrite, ReadBuf};
use crate::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
use pin_project_lite::pin_project;
use std::io;
use std::io::{self, IoSlice, SeekFrom};
use std::pin::Pin;
use std::task::{Context, Poll};
@@ -127,6 +127,18 @@ impl<RW: AsyncRead + AsyncWrite> AsyncWrite for BufStream<RW> {
self.project().inner.poll_write(cx, buf)
}
fn poll_write_vectored(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &[IoSlice<'_>],
) -> Poll<io::Result<usize>> {
self.project().inner.poll_write_vectored(cx, bufs)
}
fn is_write_vectored(&self) -> bool {
self.inner.is_write_vectored()
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.project().inner.poll_flush(cx)
}
@@ -146,6 +158,34 @@ impl<RW: AsyncRead + AsyncWrite> AsyncRead for BufStream<RW> {
}
}
/// Seek to an offset, in bytes, in the underlying stream.
///
/// The position used for seeking with `SeekFrom::Current(_)` is the
/// position the underlying stream would be at if the `BufStream` had no
/// internal buffer.
///
/// Seeking always discards the internal buffer, even if the seek position
/// would otherwise fall within it. This guarantees that calling
/// `.into_inner()` immediately after a seek yields the underlying reader
/// at the same position.
///
/// See [`AsyncSeek`] for more details.
///
/// Note: In the edge case where you're seeking with `SeekFrom::Current(n)`
/// where `n` minus the internal buffer length overflows an `i64`, two
/// seeks will be performed instead of one. If the second seek returns
/// `Err`, the underlying reader will be left at the same position it would
/// have if you called `seek` with `SeekFrom::Current(0)`.
impl<RW: AsyncRead + AsyncWrite + AsyncSeek> AsyncSeek for BufStream<RW> {
fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> io::Result<()> {
self.project().inner.start_seek(position)
}
fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
self.project().inner.poll_complete(cx)
}
}
impl<RW: AsyncRead + AsyncWrite> AsyncBufRead for BufStream<RW> {
fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
self.project().inner.poll_fill_buf(cx)
+67 -1
View File
@@ -3,7 +3,7 @@ use crate::io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite, ReadBuf};
use pin_project_lite::pin_project;
use std::fmt;
use std::io::{self, SeekFrom, Write};
use std::io::{self, IoSlice, SeekFrom, Write};
use std::pin::Pin;
use std::task::{Context, Poll};
@@ -133,6 +133,72 @@ impl<W: AsyncWrite> AsyncWrite for BufWriter<W> {
}
}
fn poll_write_vectored(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
mut bufs: &[IoSlice<'_>],
) -> Poll<io::Result<usize>> {
if self.inner.is_write_vectored() {
let total_len = bufs
.iter()
.fold(0usize, |acc, b| acc.saturating_add(b.len()));
if total_len > self.buf.capacity() - self.buf.len() {
ready!(self.as_mut().flush_buf(cx))?;
}
let me = self.as_mut().project();
if total_len >= me.buf.capacity() {
// It's more efficient to pass the slices directly to the
// underlying writer than to buffer them.
// The case when the total_len calculation saturates at
// usize::MAX is also handled here.
me.inner.poll_write_vectored(cx, bufs)
} else {
bufs.iter().for_each(|b| me.buf.extend_from_slice(b));
Poll::Ready(Ok(total_len))
}
} else {
// Remove empty buffers at the beginning of bufs.
while bufs.first().map(|buf| buf.len()) == Some(0) {
bufs = &bufs[1..];
}
if bufs.is_empty() {
return Poll::Ready(Ok(0));
}
// Flush if the first buffer doesn't fit.
let first_len = bufs[0].len();
if first_len > self.buf.capacity() - self.buf.len() {
ready!(self.as_mut().flush_buf(cx))?;
debug_assert!(self.buf.is_empty());
}
let me = self.as_mut().project();
if first_len >= me.buf.capacity() {
// The slice is at least as large as the buffering capacity,
// so it's better to write it directly, bypassing the buffer.
debug_assert!(me.buf.is_empty());
return me.inner.poll_write(cx, &bufs[0]);
} else {
me.buf.extend_from_slice(&bufs[0]);
bufs = &bufs[1..];
}
let mut total_written = first_len;
debug_assert!(total_written != 0);
// Append the buffers that fit in the internal buffer.
for buf in bufs {
if buf.len() > me.buf.capacity() - me.buf.len() {
break;
} else {
me.buf.extend_from_slice(buf);
total_written += buf.len();
}
}
Poll::Ready(Ok(total_written))
}
}
fn is_write_vectored(&self) -> bool {
true
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
ready!(self.as_mut().flush_buf(cx))?;
self.get_pin_mut().poll_flush(cx)
+7 -7
View File
@@ -10,12 +10,12 @@ use std::task::{Context, Poll};
pin_project! {
/// Future for the [`read_until`](crate::io::AsyncBufReadExt::read_until) method.
/// The delimeter is included in the resulting vector.
/// The delimiter is included in the resulting vector.
#[derive(Debug)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct ReadUntil<'a, R: ?Sized> {
reader: &'a mut R,
delimeter: u8,
delimiter: u8,
buf: &'a mut Vec<u8>,
// The number of bytes appended to buf. This can be less than buf.len() if
// the buffer was not empty when the operation was started.
@@ -28,7 +28,7 @@ pin_project! {
pub(crate) fn read_until<'a, R>(
reader: &'a mut R,
delimeter: u8,
delimiter: u8,
buf: &'a mut Vec<u8>,
) -> ReadUntil<'a, R>
where
@@ -36,7 +36,7 @@ where
{
ReadUntil {
reader,
delimeter,
delimiter,
buf,
read: 0,
_pin: PhantomPinned,
@@ -46,14 +46,14 @@ where
pub(super) fn read_until_internal<R: AsyncBufRead + ?Sized>(
mut reader: Pin<&mut R>,
cx: &mut Context<'_>,
delimeter: u8,
delimiter: u8,
buf: &mut Vec<u8>,
read: &mut usize,
) -> Poll<io::Result<usize>> {
loop {
let (done, used) = {
let available = ready!(reader.as_mut().poll_fill_buf(cx))?;
if let Some(i) = memchr::memchr(delimeter, available) {
if let Some(i) = memchr::memchr(delimiter, available) {
buf.extend_from_slice(&available[..=i]);
(true, i + 1)
} else {
@@ -74,6 +74,6 @@ impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadUntil<'_, R> {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let me = self.project();
read_until_internal(Pin::new(*me.reader), cx, *me.delimeter, me.buf, me.read)
read_until_internal(Pin::new(*me.reader), cx, *me.delimiter, me.buf, me.read)
}
}
+1 -1
View File
@@ -95,7 +95,7 @@ where
let n = ready!(read_until_internal(
me.reader, cx, *me.delim, me.buf, me.read,
))?;
// read_until_internal resets me.read to zero once it finds the delimeter
// read_until_internal resets me.read to zero once it finds the delimiter
debug_assert_eq!(*me.read, 0);
if n == 0 && me.buf.is_empty() {
+32 -1
View File
@@ -9,7 +9,8 @@
rust_2018_idioms,
unreachable_pub
)]
#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
#![deny(unused_must_use)]
#![cfg_attr(docsrs, deny(rustdoc::broken_intra_doc_links))]
#![doc(test(
no_crate_inject,
attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
@@ -442,6 +443,28 @@ mod util;
/// ```
pub mod stream {}
// local re-exports of platform specific things, allowing for decent
// documentation to be shimmed in on docs.rs
#[cfg(docsrs)]
pub mod doc;
#[cfg(docsrs)]
#[allow(unused)]
pub(crate) use self::doc::os;
#[cfg(not(docsrs))]
#[allow(unused)]
pub(crate) use std::os;
#[cfg(docsrs)]
#[allow(unused)]
pub(crate) use self::doc::winapi;
#[cfg(all(not(docsrs), windows, feature = "net"))]
#[allow(unused)]
pub(crate) use ::winapi;
cfg_macros! {
/// Implementation detail of the `select!` macro. This macro is **not**
/// intended to be used as part of the public API and is permitted to
@@ -453,15 +476,20 @@ cfg_macros! {
#[cfg(feature = "rt-multi-thread")]
#[cfg(not(test))] // Work around for rust-lang/rust#62127
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
#[doc(inline)]
pub use tokio_macros::main;
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
#[doc(inline)]
pub use tokio_macros::test;
cfg_not_rt_multi_thread! {
#[cfg(not(test))] // Work around for rust-lang/rust#62127
#[doc(inline)]
pub use tokio_macros::main_rt as main;
#[doc(inline)]
pub use tokio_macros::test_rt as test;
}
}
@@ -469,7 +497,10 @@ cfg_macros! {
// Always fail if rt is not enabled.
cfg_not_rt! {
#[cfg(not(test))]
#[doc(inline)]
pub use tokio_macros::main_fail as main;
#[doc(inline)]
pub use tokio_macros::test_fail as test;
}
}
+10 -1
View File
@@ -157,7 +157,6 @@ macro_rules! cfg_macros {
$(
#[cfg(feature = "macros")]
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
#[doc(inline)]
$item
)*
}
@@ -183,6 +182,16 @@ macro_rules! cfg_net_unix {
}
}
macro_rules! cfg_net_windows {
($($item:item)*) => {
$(
#[cfg(all(any(docsrs, windows), feature = "net"))]
#[cfg_attr(docsrs, doc(cfg(all(windows, feature = "net"))))]
$item
)*
}
}
macro_rules! cfg_process {
($($item:item)*) => {
$(
+136 -86
View File
@@ -23,10 +23,10 @@
/// returns the result of evaluating the completed branch's `<handler>`
/// expression.
///
/// Additionally, each branch may include an optional `if` precondition. This
/// precondition is evaluated **before** the `<async expression>`. If the
/// precondition returns `false`, the branch is entirely disabled. This
/// capability is useful when using `select!` within a loop.
/// Additionally, each branch may include an optional `if` precondition. If the
/// precondition returns `false`, then the branch is disabled. The provided
/// `<async expression>` is still evaluated but the resulting future is never
/// polled. This capability is useful when using `select!` within a loop.
///
/// The complete lifecycle of a `select!` expression is as follows:
///
@@ -42,12 +42,10 @@
/// to the provided `<pattern>`, if the pattern matches, evaluate `<handler>`
/// and return. If the pattern **does not** match, disable the current branch
/// and for the remainder of the current call to `select!`. Continue from step 3.
/// 5. If **all** branches are disabled, evaluate the `else` expression. If none
/// is provided, panic.
/// 5. If **all** branches are disabled, evaluate the `else` expression. If no
/// else branch is provided, panic.
///
/// # Notes
///
/// ### Runtime characteristics
/// # Runtime characteristics
///
/// By running all async expressions on the current task, the expressions are
/// able to run **concurrently** but not in **parallel**. This means all
@@ -58,76 +56,7 @@
///
/// [`tokio::spawn`]: crate::spawn
///
/// ### Avoid racy `if` preconditions
///
/// Given that `if` preconditions are used to disable `select!` branches, some
/// caution must be used to avoid missing values.
///
/// For example, here is **incorrect** usage of `sleep` with `if`. The objective
/// is to repeatedly run an asynchronous task for up to 50 milliseconds.
/// However, there is a potential for the `sleep` completion to be missed.
///
/// ```no_run
/// use tokio::time::{self, Duration};
///
/// async fn some_async_work() {
/// // do work
/// }
///
/// #[tokio::main]
/// async fn main() {
/// let sleep = time::sleep(Duration::from_millis(50));
/// tokio::pin!(sleep);
///
/// while !sleep.is_elapsed() {
/// tokio::select! {
/// _ = &mut sleep, if !sleep.is_elapsed() => {
/// println!("operation timed out");
/// }
/// _ = some_async_work() => {
/// println!("operation completed");
/// }
/// }
/// }
/// }
/// ```
///
/// In the above example, `sleep.is_elapsed()` may return `true` even if
/// `sleep.poll()` never returned `Ready`. This opens up a potential race
/// condition where `sleep` expires between the `while !sleep.is_elapsed()`
/// check and the call to `select!` resulting in the `some_async_work()` call to
/// run uninterrupted despite the sleep having elapsed.
///
/// One way to write the above example without the race would be:
///
/// ```
/// use tokio::time::{self, Duration};
///
/// async fn some_async_work() {
/// # time::sleep(Duration::from_millis(10)).await;
/// // do work
/// }
///
/// #[tokio::main]
/// async fn main() {
/// let sleep = time::sleep(Duration::from_millis(50));
/// tokio::pin!(sleep);
///
/// loop {
/// tokio::select! {
/// _ = &mut sleep => {
/// println!("operation timed out");
/// break;
/// }
/// _ = some_async_work() => {
/// println!("operation completed");
/// }
/// }
/// }
/// }
/// ```
///
/// ### Fairness
/// # Fairness
///
/// By default, `select!` randomly picks a branch to check first. This provides
/// some level of fairness when calling `select!` in a loop with branches that
@@ -151,10 +80,60 @@
///
/// # Panics
///
/// `select!` panics if all branches are disabled **and** there is no provided
/// `else` branch. A branch is disabled when the provided `if` precondition
/// returns `false` **or** when the pattern does not match the result of `<async
/// expression>`.
/// The `select!` macro panics if all branches are disabled **and** there is no
/// provided `else` branch. A branch is disabled when the provided `if`
/// precondition returns `false` **or** when the pattern does not match the
/// result of `<async expression>`.
///
/// # Cancellation safety
///
/// When using `select!` in a loop to receive messages from multiple sources,
/// you should make sure that the receive call is cancellation safe to avoid
/// losing messages. This section goes through various common methods and
/// describes whether they are cancel safe. The lists in this section are not
/// exhaustive.
///
/// The following methods are cancellation safe:
///
/// * [`tokio::sync::mpsc::Receiver::recv`](crate::sync::mpsc::Receiver::recv)
/// * [`tokio::sync::mpsc::UnboundedReceiver::recv`](crate::sync::mpsc::UnboundedReceiver::recv)
/// * [`tokio::sync::broadcast::Receiver::recv`](crate::sync::broadcast::Receiver::recv)
/// * [`tokio::sync::watch::Receiver::changed`](crate::sync::watch::Receiver::changed)
/// * [`tokio::net::TcpListener::accept`](crate::net::TcpListener::accept)
/// * [`tokio::net::UnixListener::accept`](crate::net::UnixListener::accept)
/// * [`tokio::io::AsyncReadExt::read`](crate::io::AsyncReadExt::read) on any `AsyncRead`
/// * [`tokio::io::AsyncReadExt::read_buf`](crate::io::AsyncReadExt::read_buf) on any `AsyncRead`
/// * [`tokio::io::AsyncWriteExt::write`](crate::io::AsyncWriteExt::write) on any `AsyncWrite`
/// * [`tokio::io::AsyncWriteExt::write_buf`](crate::io::AsyncWriteExt::write_buf) on any `AsyncWrite`
/// * [`tokio_stream::StreamExt::next`](https://docs.rs/tokio-stream/0.1/tokio_stream/trait.StreamExt.html#method.next) on any `Stream`
/// * [`futures::stream::StreamExt::next`](https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.next) on any `Stream`
///
/// The following methods are not cancellation safe and can lead to loss of data:
///
/// * [`tokio::io::AsyncReadExt::read_exact`](crate::io::AsyncReadExt::read_exact)
/// * [`tokio::io::AsyncReadExt::read_to_end`](crate::io::AsyncReadExt::read_to_end)
/// * [`tokio::io::AsyncReadExt::read_to_string`](crate::io::AsyncReadExt::read_to_string)
/// * [`tokio::io::AsyncWriteExt::write_all`](crate::io::AsyncWriteExt::write_all)
///
/// The following methods are not cancellation safe because they use a queue for
/// fairness and cancellation makes you lose your place in the queue:
///
/// * [`tokio::sync::Mutex::lock`](crate::sync::Mutex::lock)
/// * [`tokio::sync::RwLock::read`](crate::sync::RwLock::read)
/// * [`tokio::sync::RwLock::write`](crate::sync::RwLock::write)
/// * [`tokio::sync::Semaphore::acquire`](crate::sync::Semaphore::acquire)
/// * [`tokio::sync::Notify::notified`](crate::sync::Notify::notified)
///
/// To determine whether your own methods are cancellation safe, look for the
/// location of uses of `.await`. This is because when an asynchronous method is
/// cancelled, that always happens at an `.await`. If your function behaves
/// correctly even if it is restarted while waiting at an `.await`, then it is
/// cancellation safe.
///
/// Be aware that cancelling something that is not cancellation safe is not
/// necessarily wrong. For example, if you are cancelling a task because the
/// application is shutting down, then you probably don't care that partially
/// read data is lost.
///
/// # Examples
///
@@ -310,7 +289,7 @@
/// loop {
/// tokio::select! {
/// // If you run this example without `biased;`, the polling order is
/// // psuedo-random, and the assertions on the value of count will
/// // pseudo-random, and the assertions on the value of count will
/// // (probably) fail.
/// biased;
///
@@ -338,6 +317,77 @@
/// }
/// }
/// ```
///
/// ## Avoid racy `if` preconditions
///
/// Given that `if` preconditions are used to disable `select!` branches, some
/// caution must be used to avoid missing values.
///
/// For example, here is **incorrect** usage of `sleep` with `if`. The objective
/// is to repeatedly run an asynchronous task for up to 50 milliseconds.
/// However, there is a potential for the `sleep` completion to be missed.
///
/// ```no_run,should_panic
/// use tokio::time::{self, Duration};
///
/// async fn some_async_work() {
/// // do work
/// }
///
/// #[tokio::main]
/// async fn main() {
/// let sleep = time::sleep(Duration::from_millis(50));
/// tokio::pin!(sleep);
///
/// while !sleep.is_elapsed() {
/// tokio::select! {
/// _ = &mut sleep, if !sleep.is_elapsed() => {
/// println!("operation timed out");
/// }
/// _ = some_async_work() => {
/// println!("operation completed");
/// }
/// }
/// }
///
/// panic!("This example shows how not to do it!");
/// }
/// ```
///
/// In the above example, `sleep.is_elapsed()` may return `true` even if
/// `sleep.poll()` never returned `Ready`. This opens up a potential race
/// condition where `sleep` expires between the `while !sleep.is_elapsed()`
/// check and the call to `select!` resulting in the `some_async_work()` call to
/// run uninterrupted despite the sleep having elapsed.
///
/// One way to write the above example without the race would be:
///
/// ```
/// use tokio::time::{self, Duration};
///
/// async fn some_async_work() {
/// # time::sleep(Duration::from_millis(10)).await;
/// // do work
/// }
///
/// #[tokio::main]
/// async fn main() {
/// let sleep = time::sleep(Duration::from_millis(50));
/// tokio::pin!(sleep);
///
/// loop {
/// tokio::select! {
/// _ = &mut sleep => {
/// println!("operation timed out");
/// break;
/// }
/// _ = some_async_work() => {
/// println!("operation completed");
/// }
/// }
/// }
/// }
/// ```
#[macro_export]
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
macro_rules! select {
@@ -398,7 +448,7 @@ macro_rules! select {
// set the appropriate bit in `disabled`.
$(
if !$c {
let mask = 1 << $crate::count!( $($skip)* );
let mask: util::Mask = 1 << $crate::count!( $($skip)* );
disabled |= mask;
}
)*
@@ -417,7 +467,7 @@ macro_rules! select {
let mut is_pending = false;
// Choose a starting index to begin polling the futures at. In
// practice, this will either be a psuedo-randomly generrated
// practice, this will either be a pseudo-randomly generated
// number by default, or the constant 0 if `biased;` is
// supplied.
let start = $start;
+4
View File
@@ -46,3 +46,7 @@ cfg_net_unix! {
pub use unix::listener::UnixListener;
pub use unix::stream::UnixStream;
}
cfg_net_windows! {
pub mod windows;
}
+7
View File
@@ -125,6 +125,13 @@ impl TcpListener {
/// established, the corresponding [`TcpStream`] and the remote peer's
/// address will be returned.
///
/// # Cancel safety
///
/// This method is cancel safe. If the method is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that no new connections were
/// accepted by this method.
///
/// [`TcpStream`]: struct@crate::net::TcpStream
///
/// # Examples
+42
View File
@@ -482,6 +482,48 @@ impl TcpSocket {
let mio = self.inner.listen(backlog)?;
TcpListener::new(mio)
}
/// Converts a [`std::net::TcpStream`] into a `TcpSocket`. The provided
/// socket must not have been connected prior to calling this function. This
/// function is typically used together with crates such as [`socket2`] to
/// configure socket options that are not available on `TcpSocket`.
///
/// [`std::net::TcpStream`]: struct@std::net::TcpStream
/// [`socket2`]: https://docs.rs/socket2/
///
/// # Examples
///
/// ```
/// use tokio::net::TcpSocket;
/// use socket2::{Domain, Socket, Type};
///
/// #[tokio::main]
/// async fn main() -> std::io::Result<()> {
///
/// let socket2_socket = Socket::new(Domain::IPV4, Type::STREAM, None)?;
///
/// let socket = TcpSocket::from_std_stream(socket2_socket.into());
///
/// Ok(())
/// }
/// ```
pub fn from_std_stream(std_stream: std::net::TcpStream) -> TcpSocket {
#[cfg(unix)]
{
use std::os::unix::io::{FromRawFd, IntoRawFd};
let raw_fd = std_stream.into_raw_fd();
unsafe { TcpSocket::from_raw_fd(raw_fd) }
}
#[cfg(windows)]
{
use std::os::windows::io::{FromRawSocket, IntoRawSocket};
let raw_socket = std_stream.into_raw_socket();
unsafe { TcpSocket::from_raw_socket(raw_socket) }
}
}
}
impl fmt::Debug for TcpSocket {
+3 -3
View File
@@ -30,7 +30,7 @@ pub struct ReadHalf<'a>(&'a TcpStream);
/// Borrowed write half of a [`TcpStream`], created by [`split`].
///
/// Note that in the [`AsyncWrite`] implemenation of this type, [`poll_shutdown`] will
/// Note that in the [`AsyncWrite`] implementation of this type, [`poll_shutdown`] will
/// shut down the TCP stream in the write direction.
///
/// Writing to an `WriteHalf` is usually done using the convenience methods found
@@ -57,7 +57,7 @@ impl ReadHalf<'_> {
/// `Waker` from the `Context` passed to the most recent call is scheduled
/// to receive a wakeup.
///
/// See the [`TcpStream::poll_peek`] level documenation for more details.
/// See the [`TcpStream::poll_peek`] level documentation for more details.
///
/// # Examples
///
@@ -95,7 +95,7 @@ impl ReadHalf<'_> {
/// connected, without removing that data from the queue. On success,
/// returns the number of bytes peeked.
///
/// See the [`TcpStream::peek`] level documenation for more details.
/// See the [`TcpStream::peek`] level documentation for more details.
///
/// [`TcpStream::peek`]: TcpStream::peek
///
+2 -2
View File
@@ -112,7 +112,7 @@ impl OwnedReadHalf {
/// `Waker` from the `Context` passed to the most recent call is scheduled
/// to receive a wakeup.
///
/// See the [`TcpStream::poll_peek`] level documenation for more details.
/// See the [`TcpStream::poll_peek`] level documentation for more details.
///
/// # Examples
///
@@ -150,7 +150,7 @@ impl OwnedReadHalf {
/// connected, without removing that data from the queue. On success,
/// returns the number of bytes peeked.
///
/// See the [`TcpStream::peek`] level documenation for more details.
/// See the [`TcpStream::peek`] level documentation for more details.
///
/// [`TcpStream::peek`]: TcpStream::peek
///
+27
View File
@@ -356,6 +356,13 @@ impl TcpStream {
/// can be used to concurrently read / write to the same socket on a single
/// task without splitting the socket.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read or write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// Concurrently read and write to the stream on the same task without
@@ -420,6 +427,13 @@ impl TcpStream {
/// This function is equivalent to `ready(Interest::READABLE)` and is usually
/// paired with `try_read()`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -725,6 +739,13 @@ impl TcpStream {
/// This function is equivalent to `ready(Interest::WRITABLE)` and is usually
/// paired with `try_write()`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -1152,6 +1173,12 @@ impl TcpStream {
split_owned(self)
}
// == Poll IO functions that takes `&self` ==
//
// To read or write without mutable access to the `UnixStream`, combine the
// `poll_read_ready` or `poll_write_ready` methods with the `try_read` or
// `try_write` methods.
pub(crate) fn poll_read_priv(
&self,
cx: &mut Context<'_>,
+51 -4
View File
@@ -327,6 +327,13 @@ impl UdpSocket {
/// false-positive and attempting an operation will return with
/// `io::ErrorKind::WouldBlock`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read or write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// Concurrently receive from and send to the socket on the same task
@@ -390,6 +397,13 @@ impl UdpSocket {
/// false-positive and attempting a `try_send()` will return with
/// `io::ErrorKind::WouldBlock`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -442,6 +456,12 @@ impl UdpSocket {
/// On success, the number of bytes sent is returned, otherwise, the
/// encountered error is returned.
///
/// # Cancel safety
///
/// This method is cancel safe. If `send` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that the message was not sent.
///
/// # Examples
///
/// ```no_run
@@ -559,6 +579,13 @@ impl UdpSocket {
/// false-positive and attempting a `try_recv()` will return with
/// `io::ErrorKind::WouldBlock`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -613,6 +640,13 @@ impl UdpSocket {
/// The [`connect`] method will connect this socket to a remote address.
/// This method will fail if the socket is not connected.
///
/// # Cancel safety
///
/// This method is cancel safe. If `recv_from` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, it is guaranteed that no messages were received on this
/// socket.
///
/// [`connect`]: method@Self::connect
///
/// ```no_run
@@ -665,7 +699,7 @@ impl UdpSocket {
/// [`connect`]: method@Self::connect
pub fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
let n = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
// Safety: will not read the maybe uninitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
@@ -882,6 +916,12 @@ impl UdpSocket {
///
/// [`ToSocketAddrs`]: crate::net::ToSocketAddrs
///
/// # Cancel safety
///
/// This method is cancel safe. If `send_to` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that the message was not sent.
///
/// # Example
///
/// ```no_run
@@ -945,7 +985,7 @@ impl UdpSocket {
///
/// # Returns
///
/// If successfull, returns the number of bytes sent
/// If successful, returns the number of bytes sent
///
/// Users should ensure that when the remote cannot receive, the
/// [`ErrorKind::WouldBlock`] is properly handled. An error can also occur
@@ -1005,6 +1045,13 @@ impl UdpSocket {
/// size to hold the message bytes. If a message is too long to fit in the
/// supplied buffer, excess bytes may be discarded.
///
/// # Cancel safety
///
/// This method is cancel safe. If `recv_from` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, it is guaranteed that no messages were received on this
/// socket.
///
/// # Example
///
/// ```no_run
@@ -1053,7 +1100,7 @@ impl UdpSocket {
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<SocketAddr>> {
let (n, addr) = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
// Safety: will not read the maybe uninitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
@@ -1192,7 +1239,7 @@ impl UdpSocket {
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<SocketAddr>> {
let (n, addr) = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
// Safety: will not read the maybe uninitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
+49 -2
View File
@@ -106,6 +106,13 @@ impl UnixDatagram {
/// false-positive and attempting an operation will return with
/// `io::ErrorKind::WouldBlock`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read or write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// Concurrently receive from and send to the socket on the same task
@@ -171,6 +178,13 @@ impl UnixDatagram {
/// false-positive and attempting a `try_send()` will return with
/// `io::ErrorKind::WouldBlock`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -221,6 +235,13 @@ impl UnixDatagram {
/// false-positive and attempting a `try_recv()` will return with
/// `io::ErrorKind::WouldBlock`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -490,6 +511,12 @@ impl UnixDatagram {
/// Sends data on the socket to the socket's peer.
///
/// # Cancel safety
///
/// This method is cancel safe. If `send` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that the message was not sent.
///
/// # Examples
/// ```
/// # use std::error::Error;
@@ -613,6 +640,13 @@ impl UnixDatagram {
/// Receives data from the socket.
///
/// # Cancel safety
///
/// This method is cancel safe. If `recv` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, it is guaranteed that no messages were received on this
/// socket.
///
/// # Examples
/// ```
/// # use std::error::Error;
@@ -820,6 +854,12 @@ impl UnixDatagram {
/// Sends data on the socket to the specified address.
///
/// # Cancel safety
///
/// This method is cancel safe. If `send_to` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that the message was not sent.
///
/// # Examples
/// ```
/// # use std::error::Error;
@@ -863,6 +903,13 @@ impl UnixDatagram {
/// Receives data from the socket.
///
/// # Cancel safety
///
/// This method is cancel safe. If `recv_from` is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, it is guaranteed that no messages were received on this
/// socket.
///
/// # Examples
/// ```
/// # use std::error::Error;
@@ -927,7 +974,7 @@ impl UnixDatagram {
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<SocketAddr>> {
let (n, addr) = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
// Safety: will not read the maybe uninitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
@@ -1028,7 +1075,7 @@ impl UnixDatagram {
/// [`connect`]: method@Self::connect
pub fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
let n = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
// Safety: will not read the maybe uninitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
+7
View File
@@ -128,6 +128,13 @@ impl UnixListener {
}
/// Accepts a new incoming connection to this listener.
///
/// # Cancel safety
///
/// This method is cancel safe. If the method is used as the event in a
/// [`tokio::select!`](crate::select) statement and some other branch
/// completes first, then it is guaranteed that no new connections were
/// accepted by this method.
pub async fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
let (mio, addr) = self
.io
+1 -1
View File
@@ -29,7 +29,7 @@ pub struct ReadHalf<'a>(&'a UnixStream);
/// Borrowed write half of a [`UnixStream`], created by [`split`].
///
/// Note that in the [`AsyncWrite`] implemenation of this type, [`poll_shutdown`] will
/// Note that in the [`AsyncWrite`] implementation of this type, [`poll_shutdown`] will
/// shut down the UnixStream stream in the write direction.
///
/// Writing to an `WriteHalf` is usually done using the convenience methods found
+29 -8
View File
@@ -51,6 +51,11 @@ impl UnixStream {
let stream = UnixStream::new(stream)?;
poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;
if let Some(e) = stream.io.take_error()? {
return Err(e);
}
Ok(stream)
}
@@ -60,6 +65,13 @@ impl UnixStream {
/// can be used to concurrently read / write to the same socket on a single
/// task without splitting the socket.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read or write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// Concurrently read and write to the stream on the same task without
@@ -126,6 +138,13 @@ impl UnixStream {
/// This function is equivalent to `ready(Interest::READABLE)` and is usually
/// paired with `try_read()`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to read that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -435,6 +454,13 @@ impl UnixStream {
/// This function is equivalent to `ready(Interest::WRITABLE)` and is usually
/// paired with `try_write()`.
///
/// # Cancel safety
///
/// This method is cancel safe. Once a readiness event occurs, the method
/// will continue to return immediately until the readiness event is
/// consumed by an attempt to write that fails with `WouldBlock` or
/// `Poll::Pending`.
///
/// # Examples
///
/// ```no_run
@@ -826,14 +852,9 @@ impl AsyncWrite for UnixStream {
impl UnixStream {
// == Poll IO functions that takes `&self` ==
//
// They are not public because (taken from the doc of `PollEvented`):
//
// While `PollEvented` is `Sync` (if the underlying I/O type is `Sync`), the
// caller must ensure that there are at most two tasks that use a
// `PollEvented` instance concurrently. One for reading and one for writing.
// While violating this requirement is "safe" from a Rust memory model point
// of view, it will result in unexpected behavior in the form of lost
// notifications and tasks hanging.
// To read or write without mutable access to the `UnixStream`, combine the
// `poll_read_ready` or `poll_write_ready` methods with the `try_read` or
// `try_write` methods.
pub(crate) fn poll_read_priv(
&self,
+54 -16
View File
@@ -25,21 +25,19 @@ impl UCred {
/// Gets PID (process ID) of the process.
///
/// This is only implemented under Linux, Android, iOS, macOS, Solaris and
/// Illumos. On other plaforms this will always return `None`.
/// Illumos. On other platforms this will always return `None`.
pub fn pid(&self) -> Option<pid_t> {
self.pid
}
}
#[cfg(any(target_os = "linux", target_os = "android"))]
#[cfg(any(target_os = "linux", target_os = "android", target_os = "openbsd"))]
pub(crate) use self::impl_linux::get_peer_cred;
#[cfg(any(
target_os = "dragonfly",
target_os = "freebsd",
target_os = "netbsd",
target_os = "openbsd"
))]
#[cfg(any(target_os = "netbsd"))]
pub(crate) use self::impl_netbsd::get_peer_cred;
#[cfg(any(target_os = "dragonfly", target_os = "freebsd"))]
pub(crate) use self::impl_bsd::get_peer_cred;
#[cfg(any(target_os = "macos", target_os = "ios"))]
@@ -48,13 +46,16 @@ pub(crate) use self::impl_macos::get_peer_cred;
#[cfg(any(target_os = "solaris", target_os = "illumos"))]
pub(crate) use self::impl_solaris::get_peer_cred;
#[cfg(any(target_os = "linux", target_os = "android"))]
#[cfg(any(target_os = "linux", target_os = "android", target_os = "openbsd"))]
pub(crate) mod impl_linux {
use crate::net::unix::UnixStream;
use libc::{c_void, getsockopt, socklen_t, SOL_SOCKET, SO_PEERCRED};
use std::{io, mem};
#[cfg(target_os = "openbsd")]
use libc::sockpeercred as ucred;
#[cfg(any(target_os = "linux", target_os = "android"))]
use libc::ucred;
pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
@@ -73,7 +74,7 @@ pub(crate) mod impl_linux {
// These paranoid checks should be optimized-out
assert!(mem::size_of::<u32>() <= mem::size_of::<usize>());
assert!(ucred_size <= u32::max_value() as usize);
assert!(ucred_size <= u32::MAX as usize);
let mut ucred_size = ucred_size as socklen_t;
@@ -97,12 +98,49 @@ pub(crate) mod impl_linux {
}
}
#[cfg(any(
target_os = "dragonfly",
target_os = "freebsd",
target_os = "netbsd",
target_os = "openbsd"
))]
#[cfg(any(target_os = "netbsd"))]
pub(crate) mod impl_netbsd {
use crate::net::unix::UnixStream;
use libc::{c_void, getsockopt, socklen_t, unpcbid, LOCAL_PEEREID, SOL_SOCKET};
use std::io;
use std::mem::size_of;
use std::os::unix::io::AsRawFd;
pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
unsafe {
let raw_fd = sock.as_raw_fd();
let mut unpcbid = unpcbid {
unp_pid: 0,
unp_euid: 0,
unp_egid: 0,
};
let unpcbid_size = size_of::<unpcbid>();
let mut unpcbid_size = unpcbid_size as socklen_t;
let ret = getsockopt(
raw_fd,
SOL_SOCKET,
LOCAL_PEEREID,
&mut unpcbid as *mut unpcbid as *mut c_void,
&mut unpcbid_size,
);
if ret == 0 && unpcbid_size as usize == size_of::<unpcbid>() {
Ok(super::UCred {
uid: unpcbid.unp_euid,
gid: unpcbid.unp_egid,
pid: Some(unpcbid.unp_pid),
})
} else {
Err(io::Error::last_os_error())
}
}
}
}
#[cfg(any(target_os = "dragonfly", target_os = "freebsd"))]
pub(crate) mod impl_bsd {
use crate::net::unix::UnixStream;
+3
View File
@@ -0,0 +1,3 @@
//! Windows specific network types.
pub mod named_pipe;
File diff suppressed because it is too large Load Diff
+53 -78
View File
@@ -2,16 +2,14 @@ use crate::future::poll_fn;
use crate::loom::sync::atomic::AtomicBool;
use crate::loom::sync::Mutex;
use crate::park::{Park, Unpark};
use crate::runtime::task::{self, JoinHandle, Schedule, Task};
use crate::runtime::task::{self, JoinHandle, OwnedTasks, Schedule, Task};
use crate::sync::notify::Notify;
use crate::util::linked_list::{Link, LinkedList};
use crate::util::{waker_ref, Wake, WakerRef};
use std::cell::RefCell;
use std::collections::VecDeque;
use std::fmt;
use std::future::Future;
use std::ptr::NonNull;
use std::sync::atomic::Ordering::{AcqRel, Acquire, Release};
use std::sync::Arc;
use std::task::Poll::{Pending, Ready};
@@ -57,9 +55,6 @@ pub(crate) struct Spawner {
}
struct Tasks {
/// Collection of all active tasks spawned onto this executor.
owned: LinkedList<Task<Arc<Shared>>, <Task<Arc<Shared>> as Link>::Target>,
/// Local run queue.
///
/// Tasks notified from the current thread are pushed into this queue.
@@ -69,28 +64,28 @@ struct Tasks {
/// A remote scheduler entry.
///
/// These are filled in by remote threads sending instructions to the scheduler.
enum Entry {
enum RemoteMsg {
/// A remote thread wants to spawn a task.
Schedule(task::Notified<Arc<Shared>>),
/// A remote thread wants a task to be released by the scheduler. We only
/// have access to its header.
Release(NonNull<task::Header>),
}
// Safety: Used correctly, the task header is "thread safe". Ultimately the task
// is owned by the current thread executor, for which this instruction is being
// sent.
unsafe impl Send for Entry {}
unsafe impl Send for RemoteMsg {}
/// Scheduler state shared between threads.
struct Shared {
/// Remote run queue
queue: Mutex<VecDeque<Entry>>,
/// Remote run queue. None if the `Runtime` has been dropped.
queue: Mutex<Option<VecDeque<RemoteMsg>>>,
/// Unpark the blocked thread
/// Collection of all active tasks spawned onto this executor.
owned: OwnedTasks<Arc<Shared>>,
/// Unpark the blocked thread.
unpark: Box<dyn Unpark>,
// indicates whether the blocked on thread was woken
/// Indicates whether the blocked on thread was woken.
woken: AtomicBool,
}
@@ -124,7 +119,8 @@ impl<P: Park> BasicScheduler<P> {
let spawner = Spawner {
shared: Arc::new(Shared {
queue: Mutex::new(VecDeque::with_capacity(INITIAL_CAPACITY)),
queue: Mutex::new(Some(VecDeque::with_capacity(INITIAL_CAPACITY))),
owned: OwnedTasks::new(),
unpark: unpark as Box<dyn Unpark>,
woken: AtomicBool::new(false),
}),
@@ -132,7 +128,6 @@ impl<P: Park> BasicScheduler<P> {
let inner = Mutex::new(Some(Inner {
tasks: Some(Tasks {
owned: LinkedList::new(),
queue: VecDeque::with_capacity(INITIAL_CAPACITY),
}),
spawner: spawner.clone(),
@@ -227,7 +222,7 @@ impl<P: Park> Inner<P> {
.borrow_mut()
.queue
.pop_front()
.map(Entry::Schedule)
.map(RemoteMsg::Schedule)
})
} else {
context
@@ -235,7 +230,7 @@ impl<P: Park> Inner<P> {
.borrow_mut()
.queue
.pop_front()
.map(Entry::Schedule)
.map(RemoteMsg::Schedule)
.or_else(|| scheduler.spawner.pop())
};
@@ -251,26 +246,7 @@ impl<P: Park> Inner<P> {
};
match entry {
Entry::Schedule(task) => crate::coop::budget(|| task.run()),
Entry::Release(ptr) => {
// Safety: the task header is only legally provided
// internally in the header, so we know that it is a
// valid (or in particular *allocated*) header that
// is part of the linked list.
unsafe {
let removed = context.tasks.borrow_mut().owned.remove(ptr);
// TODO: This seems like it should hold, because
// there doesn't seem to be an avenue for anyone
// else to fiddle with the owned tasks
// collection *after* a remote thread has marked
// it as released, and at that point, the only
// location at which it can be removed is here
// or in the Drop implementation of the
// scheduler.
debug_assert!(removed.is_some());
}
}
RemoteMsg::Schedule(task) => crate::coop::budget(|| task.run()),
}
}
@@ -335,14 +311,7 @@ impl<P: Park> Drop for BasicScheduler<P> {
};
enter(&mut inner, |scheduler, context| {
// Loop required here to ensure borrow is dropped between iterations
#[allow(clippy::while_let_loop)]
loop {
let task = match context.tasks.borrow_mut().owned.pop_back() {
Some(task) => task,
None => break,
};
while let Some(task) = context.shared.owned.pop_back() {
task.shutdown();
}
@@ -351,20 +320,27 @@ impl<P: Park> Drop for BasicScheduler<P> {
task.shutdown();
}
// Drain remote queue
for entry in scheduler.spawner.shared.queue.lock().drain(..) {
match entry {
Entry::Schedule(task) => {
task.shutdown();
}
Entry::Release(..) => {
// Do nothing, each entry in the linked list was *just*
// dropped by the scheduler above.
// Drain remote queue and set it to None
let mut remote_queue = scheduler.spawner.shared.queue.lock();
// Using `Option::take` to replace the shared queue with `None`.
if let Some(remote_queue) = remote_queue.take() {
for entry in remote_queue {
match entry {
RemoteMsg::Schedule(task) => {
task.shutdown();
}
}
}
}
// By dropping the mutex lock after the full duration of the above loop,
// any thread that sees the queue in the `None` state is guaranteed that
// the runtime has fully shut down.
//
// The assert below is unrelated to this mutex.
drop(remote_queue);
assert!(context.tasks.borrow().owned.is_empty());
assert!(context.shared.owned.is_empty());
});
}
}
@@ -381,7 +357,7 @@ impl Spawner {
/// Spawns a future onto the thread pool
pub(crate) fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F: crate::future::Future + Send + 'static,
F::Output: Send + 'static,
{
let (task, handle) = task::joinable(future);
@@ -389,8 +365,11 @@ impl Spawner {
handle
}
fn pop(&self) -> Option<Entry> {
self.shared.queue.lock().pop_front()
fn pop(&self) -> Option<RemoteMsg> {
match self.shared.queue.lock().as_mut() {
Some(queue) => queue.pop_front(),
None => None,
}
}
fn waker_ref(&self) -> WakerRef<'_> {
@@ -416,27 +395,14 @@ impl Schedule for Arc<Shared> {
fn bind(task: Task<Self>) -> Arc<Shared> {
CURRENT.with(|maybe_cx| {
let cx = maybe_cx.expect("scheduler context missing");
cx.tasks.borrow_mut().owned.push_front(task);
cx.shared.owned.push_front(task);
cx.shared.clone()
})
}
fn release(&self, task: &Task<Self>) -> Option<Task<Self>> {
CURRENT.with(|maybe_cx| {
let ptr = NonNull::from(task.header());
if let Some(cx) = maybe_cx {
// safety: the task is inserted in the list in `bind`.
unsafe { cx.tasks.borrow_mut().owned.remove(ptr) }
} else {
self.queue.lock().push_back(Entry::Release(ptr));
self.unpark.unpark();
// Returning `None` here prevents the task plumbing from being
// freed. It is then up to the scheduler through the queue we
// just added to, or its Drop impl to free the task.
None
}
})
// SAFETY: Inserted into the list in bind above.
unsafe { self.owned.remove(task) }
}
fn schedule(&self, task: task::Notified<Self>) {
@@ -445,8 +411,17 @@ impl Schedule for Arc<Shared> {
cx.tasks.borrow_mut().queue.push_back(task);
}
_ => {
self.queue.lock().push_back(Entry::Schedule(task));
self.unpark.unpark();
let mut guard = self.queue.lock();
if let Some(queue) = guard.as_mut() {
queue.push_back(RemoteMsg::Schedule(task));
drop(guard);
self.unpark.unpark();
} else {
// The runtime has shut down. We drop the new task
// immediately.
drop(guard);
task.shutdown();
}
}
});
}
-13
View File
@@ -4,7 +4,6 @@ use crate::loom::sync::{Arc, Condvar, Mutex};
use crate::loom::thread;
use crate::runtime::blocking::schedule::NoopSchedule;
use crate::runtime::blocking::shutdown;
use crate::runtime::blocking::task::BlockingTask;
use crate::runtime::builder::ThreadNameFn;
use crate::runtime::context;
use crate::runtime::task::{self, JoinHandle};
@@ -86,18 +85,6 @@ where
rt.spawn_blocking(func)
}
#[allow(dead_code)]
pub(crate) fn try_spawn_blocking<F, R>(func: F) -> Result<(), ()>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let rt = context::current().expect(CONTEXT_MISSING_ERROR);
let (task, _handle) = task::joinable(BlockingTask::new(func));
rt.blocking_spawner.spawn(task, &rt)
}
// ===== impl BlockingPool =====
impl BlockingPool {
+1 -1
View File
@@ -413,7 +413,7 @@ impl Builder {
/// Sets a custom timeout for a thread in the blocking pool.
///
/// By default, the timeout for a thread is set to 10 seconds. This can
/// be overriden using .thread_keep_alive().
/// be overridden using .thread_keep_alive().
///
/// # Example
///
+1 -1
View File
@@ -64,7 +64,7 @@ cfg_rt! {
// # Warning
//
// This is hidden for a reason. Do not use without fully understanding
// executors. Misuing can easily cause your program to deadlock.
// executors. Misusing can easily cause your program to deadlock.
cfg_rt_multi_thread! {
pub(crate) fn exit<F: FnOnce() -> R, R>(f: F) -> R {
// Reset in case the closure panics
+22 -7
View File
@@ -145,7 +145,7 @@ impl Handle {
F::Output: Send + 'static,
{
#[cfg(all(tokio_unstable, feature = "tracing"))]
let future = crate::util::trace::task(future, "task");
let future = crate::util::trace::task(future, "task", None);
self.spawner.spawn(future)
}
@@ -174,8 +174,20 @@ impl Handle {
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
self.spawn_blocking_inner(func, None)
}
#[cfg_attr(tokio_track_caller, track_caller)]
pub(crate) fn spawn_blocking_inner<F, R>(&self, func: F, name: Option<&str>) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let fut = BlockingTask::new(func);
#[cfg(all(tokio_unstable, feature = "tracing"))]
let func = {
let fut = {
use tracing::Instrument;
#[cfg(tokio_track_caller)]
let location = std::panic::Location::caller();
#[cfg(tokio_track_caller)]
@@ -184,6 +196,7 @@ impl Handle {
"task",
kind = %"blocking",
function = %std::any::type_name::<F>(),
task.name = %name.unwrap_or_default(),
spawn.location = %format_args!("{}:{}:{}", location.file(), location.line(), location.column()),
);
#[cfg(not(tokio_track_caller))]
@@ -191,14 +204,16 @@ impl Handle {
target: "tokio::task",
"task",
kind = %"blocking",
task.name = %name.unwrap_or_default(),
function = %std::any::type_name::<F>(),
);
move || {
let _g = span.enter();
func()
}
fut.instrument(span)
};
let (task, handle) = task::joinable(BlockingTask::new(func));
#[cfg(not(all(tokio_unstable, feature = "tracing")))]
let _ = name;
let (task, handle) = task::joinable(fut);
let _ = self.blocking_spawner.spawn(task, &self);
handle
}
+24 -12
View File
@@ -109,7 +109,10 @@ impl<T> Local<T> {
}
/// Pushes a task to the back of the local queue, skipping the LIFO slot.
pub(super) fn push_back(&mut self, mut task: task::Notified<T>, inject: &Inject<T>) {
pub(super) fn push_back(&mut self, mut task: task::Notified<T>, inject: &Inject<T>)
where
T: crate::runtime::task::Schedule,
{
let tail = loop {
let head = self.inner.head.load(Acquire);
let (steal, real) = unpack(head);
@@ -121,9 +124,14 @@ impl<T> Local<T> {
// There is capacity for the task
break tail;
} else if steal != real {
// Concurrently stealing, this will free up capacity, so
// only push the new task onto the inject queue
inject.push(task);
// Concurrently stealing, this will free up capacity, so only
// push the new task onto the inject queue
//
// If the task fails to be pushed on the injection queue, there
// is nothing to be done at this point as the task cannot be a
// newly spawned task. Shutting down this task is handled by the
// worker shutdown process.
let _ = inject.push(task);
return;
} else {
// Push the current task and half of the queue into the
@@ -504,16 +512,19 @@ impl<T: 'static> Inject<T> {
}
/// Pushes a value into the queue.
pub(super) fn push(&self, task: task::Notified<T>) {
///
/// Returns `Err(task)` if pushing fails due to the queue being shutdown.
/// The caller is expected to call `shutdown()` on the task **if and only
/// if** it is a newly spawned task.
pub(super) fn push(&self, task: task::Notified<T>) -> Result<(), task::Notified<T>>
where
T: crate::runtime::task::Schedule,
{
// Acquire queue lock
let mut p = self.pointers.lock();
if p.is_closed {
// Drop the mutex to avoid a potential deadlock when
// re-entering.
drop(p);
drop(task);
return;
return Err(task);
}
// safety: only mutated with the lock held
@@ -532,6 +543,7 @@ impl<T: 'static> Inject<T> {
p.tail = Some(task);
self.len.store(len + 1, Release);
Ok(())
}
pub(super) fn push_batch(
@@ -617,7 +629,7 @@ fn set_next(header: NonNull<task::Header>, val: Option<NonNull<task::Header>>) {
/// Split the head value into the real head and the index a stealer is working
/// on.
fn unpack(n: u32) -> (u16, u16) {
let real = n & u16::max_value() as u32;
let real = n & u16::MAX as u32;
let steal = n >> 16;
(steal as u16, real as u16)
@@ -630,5 +642,5 @@ fn pack(steal: u16, real: u16) -> u32 {
#[test]
fn test_local_queue_capacity() {
assert!(LOCAL_QUEUE_CAPACITY - 1 <= u8::max_value() as usize);
assert!(LOCAL_QUEUE_CAPACITY - 1 <= u8::MAX as usize);
}
+1 -2
View File
@@ -1,8 +1,7 @@
cfg_rt! {
use crate::future::Future;
use crate::runtime::basic_scheduler;
use crate::task::JoinHandle;
use std::future::Future;
}
cfg_rt_multi_thread! {
+12 -15
View File
@@ -9,13 +9,13 @@
//! Make sure to consult the relevant safety section of each function before
//! use.
use crate::future::Future;
use crate::loom::cell::UnsafeCell;
use crate::runtime::task::raw::{self, Vtable};
use crate::runtime::task::state::State;
use crate::runtime::task::{Notified, Schedule, Task};
use crate::util::linked_list;
use std::future::Future;
use std::pin::Pin;
use std::ptr::NonNull;
use std::task::{Context, Poll, Waker};
@@ -66,11 +66,12 @@ pub(crate) struct Header {
/// Pointer to next task, used with the injection queue
pub(crate) queue_next: UnsafeCell<Option<NonNull<Header>>>,
/// Pointer to the next task in the transfer stack
pub(super) stack_next: UnsafeCell<Option<NonNull<Header>>>,
/// Table of function pointers for executing actions on the task.
pub(super) vtable: &'static Vtable,
/// The tracing ID for this instrumented task.
#[cfg(all(tokio_unstable, feature = "tracing"))]
pub(super) id: Option<tracing::Id>,
}
unsafe impl Send for Header {}
@@ -93,13 +94,16 @@ impl<T: Future, S: Schedule> Cell<T, S> {
/// Allocates a new task cell, containing the header, trailer, and core
/// structures.
pub(super) fn new(future: T, state: State) -> Box<Cell<T, S>> {
#[cfg(all(tokio_unstable, feature = "tracing"))]
let id = future.id();
Box::new(Cell {
header: Header {
state,
owned: UnsafeCell::new(linked_list::Pointers::new()),
queue_next: UnsafeCell::new(None),
stack_next: UnsafeCell::new(None),
vtable: raw::vtable::<T, S>(),
#[cfg(all(tokio_unstable, feature = "tracing"))]
id,
},
core: Core {
scheduler: Scheduler {
@@ -123,7 +127,7 @@ impl<S: Schedule> Scheduler<S> {
/// Bind a scheduler to the task.
///
/// This only happens on the first poll and must be preceeded by a call to
/// This only happens on the first poll and must be preceded by a call to
/// `is_bound` to determine if binding is appropriate or not.
///
/// # Safety
@@ -212,7 +216,7 @@ impl<T: Future> CoreStage<T> {
/// # Safety
///
/// The caller must ensure it is safe to mutate the `state` field. This
/// requires ensuring mutal exclusion between any concurrent thread that
/// requires ensuring mutual exclusion between any concurrent thread that
/// might modify the future or output field.
///
/// The mutual exclusion is implemented by `Harness` and the `Lifecycle`
@@ -276,7 +280,7 @@ impl<T: Future> CoreStage<T> {
use std::mem;
self.stage.with_mut(|ptr| {
// Safety:: the caller ensures mutal exclusion to the field.
// Safety:: the caller ensures mutual exclusion to the field.
match mem::replace(unsafe { &mut *ptr }, Stage::Consumed) {
Stage::Finished(output) => output,
_ => panic!("JoinHandle polled after completion"),
@@ -291,13 +295,6 @@ impl<T: Future> CoreStage<T> {
cfg_rt_multi_thread! {
impl Header {
pub(crate) fn shutdown(&self) {
use crate::runtime::task::RawTask;
let task = unsafe { RawTask::from_raw(self.into()) };
task.shutdown();
}
pub(crate) unsafe fn set_next(&self, next: Option<NonNull<Header>>) {
self.queue_next.with_mut(|ptr| *ptr = next);
}
+18 -2
View File
@@ -1,9 +1,9 @@
use crate::future::Future;
use crate::runtime::task::core::{Cell, Core, CoreStage, Header, Scheduler, Trailer};
use crate::runtime::task::state::Snapshot;
use crate::runtime::task::waker::waker_ref;
use crate::runtime::task::{JoinError, Notified, Schedule, Task};
use std::future::Future;
use std::mem;
use std::panic;
use std::ptr::NonNull;
@@ -146,6 +146,11 @@ where
}
}
#[cfg(all(tokio_unstable, feature = "tracing"))]
pub(super) fn id(&self) -> Option<&tracing::Id> {
self.header().id.as_ref()
}
/// Forcibly shutdown the task
///
/// Attempt to transition to `Running` in order to forcibly shutdown the
@@ -158,12 +163,23 @@ where
return;
}
// By transitioning the lifcycle to `Running`, we have permission to
// By transitioning the lifecycle to `Running`, we have permission to
// drop the future.
let err = cancel_task(&self.core().stage);
self.complete(Err(err), true)
}
/// Remotely abort the task
///
/// This is similar to `shutdown` except that it asks the runtime to perform
/// the shutdown. This is necessary to avoid the shutdown happening in the
/// wrong thread for non-Send tasks.
pub(super) fn remote_abort(self) {
if self.header().state.transition_to_notified_and_cancel() {
self.core().scheduler.schedule(Notified(self.to_task()));
}
}
// ====== internal ======
fn complete(self, output: super::Result<T::Output>, is_join_interested: bool) {
+1 -1
View File
@@ -192,7 +192,7 @@ impl<T> JoinHandle<T> {
/// ```
pub fn abort(&self) {
if let Some(raw) = self.raw {
raw.shutdown();
raw.remote_abort();
}
}
}
+33
View File
@@ -0,0 +1,33 @@
use crate::loom::sync::Mutex;
use crate::runtime::task::Task;
use crate::util::linked_list::{Link, LinkedList};
pub(crate) struct OwnedTasks<S: 'static> {
list: Mutex<LinkedList<Task<S>, <Task<S> as Link>::Target>>,
}
impl<S: 'static> OwnedTasks<S> {
pub(crate) fn new() -> Self {
Self {
list: Mutex::new(LinkedList::new()),
}
}
pub(crate) fn push_front(&self, task: Task<S>) {
self.list.lock().push_front(task);
}
pub(crate) fn pop_back(&self) -> Option<Task<S>> {
self.list.lock().pop_back()
}
/// The caller must ensure that if the provided task is stored in a
/// linked list, then it is in this linked list.
pub(crate) unsafe fn remove(&self, task: &Task<S>) -> Option<Task<S>> {
self.list.lock().remove(task.header().into())
}
pub(crate) fn is_empty(&self) -> bool {
self.list.lock().is_empty()
}
}
+7 -10
View File
@@ -13,6 +13,9 @@ mod join;
#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
pub use self::join::JoinHandle;
mod list;
pub(super) use self::list::OwnedTasks;
mod raw;
use self::raw::RawTask;
@@ -21,14 +24,9 @@ use self::state::State;
mod waker;
cfg_rt_multi_thread! {
mod stack;
pub(crate) use self::stack::TransferStack;
}
use crate::future::Future;
use crate::util::linked_list;
use std::future::Future;
use std::marker::PhantomData;
use std::ptr::NonNull;
use std::{fmt, mem};
@@ -62,11 +60,10 @@ pub(crate) trait Schedule: Sync + Sized + 'static {
fn bind(task: Task<Self>) -> Self;
/// The task has completed work and is ready to be released. The scheduler
/// is free to drop it whenever.
/// should release it immediately and return it. The task module will batch
/// the ref-dec with setting other options.
///
/// If the scheduler can immediately release the task, it should return
/// it as part of the function. This enables the task module to batch
/// the ref-dec with other options.
/// If the scheduler has already released the task, then None is returned.
fn release(&self, task: &Task<Self>) -> Option<Task<Self>>;
/// Schedule the task
+15 -1
View File
@@ -1,6 +1,6 @@
use crate::future::Future;
use crate::runtime::task::{Cell, Harness, Header, Schedule, State};
use std::future::Future;
use std::ptr::NonNull;
use std::task::{Poll, Waker};
@@ -22,6 +22,9 @@ pub(super) struct Vtable {
/// The join handle has been dropped
pub(super) drop_join_handle_slow: unsafe fn(NonNull<Header>),
/// The task is remotely aborted
pub(super) remote_abort: unsafe fn(NonNull<Header>),
/// Scheduler is being shutdown
pub(super) shutdown: unsafe fn(NonNull<Header>),
}
@@ -33,6 +36,7 @@ pub(super) fn vtable<T: Future, S: Schedule>() -> &'static Vtable {
dealloc: dealloc::<T, S>,
try_read_output: try_read_output::<T, S>,
drop_join_handle_slow: drop_join_handle_slow::<T, S>,
remote_abort: remote_abort::<T, S>,
shutdown: shutdown::<T, S>,
}
}
@@ -89,6 +93,11 @@ impl RawTask {
let vtable = self.header().vtable;
unsafe { (vtable.shutdown)(self.ptr) }
}
pub(super) fn remote_abort(self) {
let vtable = self.header().vtable;
unsafe { (vtable.remote_abort)(self.ptr) }
}
}
impl Clone for RawTask {
@@ -125,6 +134,11 @@ unsafe fn drop_join_handle_slow<T: Future, S: Schedule>(ptr: NonNull<Header>) {
harness.drop_join_handle_slow()
}
unsafe fn remote_abort<T: Future, S: Schedule>(ptr: NonNull<Header>) {
let harness = Harness::<T, S>::from_raw(ptr);
harness.remote_abort()
}
unsafe fn shutdown<T: Future, S: Schedule>(ptr: NonNull<Header>) {
let harness = Harness::<T, S>::from_raw(ptr);
harness.shutdown()
-83
View File
@@ -1,83 +0,0 @@
use crate::loom::sync::atomic::AtomicPtr;
use crate::runtime::task::{Header, Task};
use std::marker::PhantomData;
use std::ptr::{self, NonNull};
use std::sync::atomic::Ordering::{Acquire, Relaxed, Release};
/// Concurrent stack of tasks, used to pass ownership of a task from one worker
/// to another.
pub(crate) struct TransferStack<T: 'static> {
head: AtomicPtr<Header>,
_p: PhantomData<T>,
}
impl<T: 'static> TransferStack<T> {
pub(crate) fn new() -> TransferStack<T> {
TransferStack {
head: AtomicPtr::new(ptr::null_mut()),
_p: PhantomData,
}
}
pub(crate) fn push(&self, task: Task<T>) {
let task = task.into_raw();
// We don't care about any memory associated w/ setting the `head`
// field, just the current value.
//
// The compare-exchange creates a release sequence.
let mut curr = self.head.load(Relaxed);
loop {
unsafe {
task.as_ref()
.stack_next
.with_mut(|ptr| *ptr = NonNull::new(curr))
};
let res = self
.head
.compare_exchange(curr, task.as_ptr() as *mut _, Release, Relaxed);
match res {
Ok(_) => return,
Err(actual) => {
curr = actual;
}
}
}
}
pub(crate) fn drain(&self) -> impl Iterator<Item = Task<T>> {
struct Iter<T: 'static>(Option<NonNull<Header>>, PhantomData<T>);
impl<T: 'static> Iterator for Iter<T> {
type Item = Task<T>;
fn next(&mut self) -> Option<Task<T>> {
let task = self.0?;
// Move the cursor forward
self.0 = unsafe { task.as_ref().stack_next.with(|ptr| *ptr) };
// Return the task
unsafe { Some(Task::from_raw(task)) }
}
}
impl<T: 'static> Drop for Iter<T> {
fn drop(&mut self) {
use std::process;
if self.0.is_some() {
// we have bugs
process::abort();
}
}
}
let ptr = self.head.swap(ptr::null_mut(), Acquire);
Iter(NonNull::new(ptr), PhantomData)
}
}
+16 -4
View File
@@ -29,12 +29,15 @@ const LIFECYCLE_MASK: usize = 0b11;
const NOTIFIED: usize = 0b100;
/// The join handle is still around
#[allow(clippy::unusual_byte_groupings)] // https://github.com/rust-lang/rust-clippy/issues/6556
const JOIN_INTEREST: usize = 0b1_000;
/// A join handle waker has been set
#[allow(clippy::unusual_byte_groupings)] // https://github.com/rust-lang/rust-clippy/issues/6556
const JOIN_WAKER: usize = 0b10_000;
/// The task has been forcibly cancelled.
#[allow(clippy::unusual_byte_groupings)] // https://github.com/rust-lang/rust-clippy/issues/6556
const CANCELLED: usize = 0b100_000;
/// All bits
@@ -52,7 +55,7 @@ const REF_ONE: usize = 1 << REF_COUNT_SHIFT;
/// State a task is initialized with
///
/// A task is initialized with two references: one for the scheduler and one for
/// the `JoinHandle`. As the task starts with a `JoinHandle`, `JOIN_INTERST` is
/// the `JoinHandle`. As the task starts with a `JoinHandle`, `JOIN_INTEREST` is
/// set. A new task is immediately pushed into the run queue for execution and
/// starts with the `NOTIFIED` flag set.
const INITIAL_STATE: usize = (REF_ONE * 2) | JOIN_INTEREST | NOTIFIED;
@@ -64,7 +67,7 @@ impl State {
pub(super) fn new() -> State {
// A task is initialized with three references: one for the scheduler,
// one for the `JoinHandle`, one for the task handle made available in
// release. As the task starts with a `JoinHandle`, `JOIN_INTERST` is
// release. As the task starts with a `JoinHandle`, `JOIN_INTEREST` is
// set. A new task is immediately pushed into the run queue for
// execution and starts with the `NOTIFIED` flag set.
State {
@@ -177,6 +180,15 @@ impl State {
prev.will_need_queueing()
}
/// Set the cancelled bit and transition the state to `NOTIFIED`.
///
/// Returns `true` if the task needs to be submitted to the pool for
/// execution
pub(super) fn transition_to_notified_and_cancel(&self) -> bool {
let prev = Snapshot(self.val.fetch_or(NOTIFIED | CANCELLED, AcqRel));
prev.will_need_queueing()
}
/// Set the `CANCELLED` bit and attempt to transition to `Running`.
///
/// Returns `true` if the transition to `Running` succeeded.
@@ -306,7 +318,7 @@ impl State {
let prev = self.val.fetch_add(REF_ONE, Relaxed);
// If the reference count overflowed, abort.
if prev > isize::max_value() as usize {
if prev > isize::MAX as usize {
process::abort();
}
}
@@ -410,7 +422,7 @@ impl Snapshot {
}
fn ref_inc(&mut self) {
assert!(self.0 <= isize::max_value() as usize);
assert!(self.0 <= isize::MAX as usize);
self.0 += REF_ONE;
}
+30 -1
View File
@@ -1,7 +1,7 @@
use crate::future::Future;
use crate::runtime::task::harness::Harness;
use crate::runtime::task::{Header, Schedule};
use std::future::Future;
use std::marker::PhantomData;
use std::mem::ManuallyDrop;
use std::ops;
@@ -44,12 +44,38 @@ impl<S> ops::Deref for WakerRef<'_, S> {
}
}
cfg_trace! {
macro_rules! trace {
($harness:expr, $op:expr) => {
if let Some(id) = $harness.id() {
tracing::trace!(
target: "tokio::task::waker",
op = $op,
task.id = id.into_u64(),
);
}
}
}
}
cfg_not_trace! {
macro_rules! trace {
($harness:expr, $op:expr) => {
// noop
let _ = &$harness;
}
}
}
unsafe fn clone_waker<T, S>(ptr: *const ()) -> RawWaker
where
T: Future,
S: Schedule,
{
let header = ptr as *const Header;
let ptr = NonNull::new_unchecked(ptr as *mut Header);
let harness = Harness::<T, S>::from_raw(ptr);
trace!(harness, "waker.clone");
(*header).state.ref_inc();
raw_waker::<T, S>(header)
}
@@ -61,6 +87,7 @@ where
{
let ptr = NonNull::new_unchecked(ptr as *mut Header);
let harness = Harness::<T, S>::from_raw(ptr);
trace!(harness, "waker.drop");
harness.drop_reference();
}
@@ -71,6 +98,7 @@ where
{
let ptr = NonNull::new_unchecked(ptr as *mut Header);
let harness = Harness::<T, S>::from_raw(ptr);
trace!(harness, "waker.wake");
harness.wake_by_val();
}
@@ -82,6 +110,7 @@ where
{
let ptr = NonNull::new_unchecked(ptr as *mut Header);
let harness = Harness::<T, S>::from_raw(ptr);
trace!(harness, "waker.wake_by_ref");
harness.wake_by_ref();
}

Some files were not shown because too many files have changed in this diff Show More