mirror/async-std
2
0
Fork 1
mirror of https://github.com/async-rs/async-std.git synced 2025-01-30 09:15:33 +00:00
Code Issues Releases Wiki Activity

Merge branch 'master' into fix_doc_test

Browse source
This commit is contained in:
k-nasa 2020-05-20 13:50:39 +09:00
commit d30603affe
64 changed files with 1591 additions and 1572 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
.github/workflows/ci.yml vendored
View file

@ -29,6 +29,24 @@ jobs:
toolchain: ${{ matrix.rust }}
override: true
- name: Cache cargo registry
uses: actions/cache@v1
with:
path: ~/.cargo/registry
key: ${{ matrix.os }}-${{ matrix.rust }}-cargo-registry-${{ hashFiles('**/Cargo.toml') }}
- name: Cache cargo index
uses: actions/cache@v1
with:
path: ~/.cargo/git
key: ${{ matrix.os }}-${{ matrix.rust }}-cargo-index-${{ hashFiles('**/Cargo.toml') }}
- name: Cache cargo build
uses: actions/cache@v1
with:
path: target
key: ${{ matrix.os }}-${{ matrix.rust }}-cargo-build-target-${{ hashFiles('**/Cargo.toml') }}
- name: check
uses: actions-rs/cargo@v1
with:
@ -40,6 +58,14 @@ jobs:
with:
command: check
args: --features unstable --all --bins --examples --tests
- name: check wasm
uses: actions-rs/cargo@v1
with:
command: check
target: wasm32-unknown-unknown
override: true
args: --features unstable --all --bins --tests
- name: check bench
uses: actions-rs/cargo@v1
@ -66,12 +92,6 @@ jobs:
command: test
args: --all --features "unstable attributes"
- name: documentation test
uses: actions-rs/cargo@v1
with:
command: test
args: --doc --features "unstable attributes"
build__with_no_std:
name: Build with no-std
runs-on: ubuntu-latest
@ -90,6 +110,39 @@ jobs:
command: check
args: --no-default-features --features alloc --target thumbv7m-none-eabi -Z avoid-dev-deps
cross:
name: Cross compile
runs-on: ubuntu-latest
strategy:
matrix:
target:
- i686-unknown-linux-gnu
- powerpc-unknown-linux-gnu
- powerpc64-unknown-linux-gnu
- mips-unknown-linux-gnu
- arm-linux-androideabi
steps:
- uses: actions/checkout@master
- name: Install nightly
uses: actions-rs/toolchain@v1
with:
toolchain: nightly
override: true
- name: Install cross
run: cargo install cross
- name: check
run: cross check --all --target ${{ matrix.target }}
- name: check unstable
run: cross check --all --features unstable --target ${{ matrix.target }}
- name: test
run: cross test --all --features unstable --target ${{ matrix.target }}
check_fmt_and_docs:
name: Checking fmt and docs
runs-on: ubuntu-latest
@ -117,15 +170,3 @@ jobs:
- name: Docs
run: cargo doc --features docs
# clippy_check:
# name: Clippy check
# runs-on: ubuntu-latest
# steps:
# - uses: actions/checkout@v1
# - name: Install rust
# run: rustup update beta && rustup default beta
# - name: Install clippy
# run: rustup component add clippy
# - name: clippy
# run: cargo clippy --all --features unstable

25
CHANGELOG.md
View file

@ -7,6 +7,30 @@ and this project adheres to [Semantic Versioning](https://book.async.rs/overview
## [Unreleased]
## Changed
- For `wasm`, switched underlying `Timer` implementation to [`futures-timer`](https://github.com/async-rs/futures-timer). ([#776](https://github.com/async-rs/async-std/pull/776))
# [1.6.0-beta.1] - 2020-05-07
## Added
- Added `task::spawn_local`. ([#757](https://github.com/async-rs/async-std/pull/757))
- Added out of the box support for `wasm`. ([#757](https://github.com/async-rs/async-std/pull/757))
- Added `JoinHandle::cancel` ([#757](https://github.com/async-rs/async-std/pull/757))
- Added `sync::Condvar` ([#369](https://github.com/async-rs/async-std/pull/369))
- Added `sync::Sender::try_send` and `sync::Receiver::try_recv` ([#585](https://github.com/async-rs/async-std/pull/585))
- Added `no_std` support for `task`, `future` and `stream` ([#680](https://github.com/async-rs/async-std/pull/680))
## Changed
- Switched underlying runtime to [`smol`](https://github.com/stjepang/smol/). ([#757](https://github.com/async-rs/async-std/pull/757))
- Switched implementation of `sync::Barrier` to use `sync::Condvar` like `std` does. ([#581](https://github.com/async-rs/async-std/pull/581))
## Fixed
- Allow compilation on 32 bit targets, by using `AtomicUsize` for `TaskId`. ([#756](https://github.com/async-rs/async-std/pull/756))
# [1.5.0] - 2020-02-03
[API Documentation](https://docs.rs/async-std/1.5.0/async-std)
@ -678,6 +702,7 @@ task::blocking(async {
- Initial beta release
[Unreleased]: https://github.com/async-rs/async-std/compare/v1.5.0...HEAD
[1.6.0-beta.1]: https://github.com/async-rs/async-std/compare/v1.4.0...v1.6.0-beta.1
[1.5.0]: https://github.com/async-rs/async-std/compare/v1.4.0...v1.5.0
[1.4.0]: https://github.com/async-rs/async-std/compare/v1.3.0...v1.4.0
[1.3.0]: https://github.com/async-rs/async-std/compare/v1.2.0...v1.3.0

58
Cargo.toml
View file

@ -1,9 +1,10 @@
[package]
name = "async-std"
version = "1.5.0"
version = "1.6.0-beta.1"
authors = [
"Stjepan Glavina <stjepang@gmail.com>",
"Yoshua Wuyts <yoshuawuyts@gmail.com>",
"Friedel Ziegelmayer <me@dignifiedquire.com>",
"Contributors to async-std",
]
edition = "2018"
@ -24,18 +25,13 @@ rustdoc-args = ["--cfg", "feature=\"docs\""]
default = [
"std",
"async-task",
"crossbeam-channel",
"crossbeam-deque",
"futures-timer",
"kv-log-macro",
"log",
"mio",
"mio-uds",
"num_cpus",
"pin-project-lite",
]
docs = ["attributes", "unstable", "default"]
unstable = ["std", "broadcaster", "futures-timer"]
unstable = ["std"]
attributes = ["async-attributes"]
std = [
"alloc",
@ -46,6 +42,10 @@ std = [
"once_cell",
"pin-utils",
"slab",
"smol",
"futures-timer",
"wasm-bindgen-futures",
"futures-channel",
]
alloc = [
"futures-core/alloc",
@ -54,31 +54,39 @@ alloc = [
[dependencies]
async-attributes = { version = "1.1.1", optional = true }
async-task = { version = "1.2.1", optional = true }
broadcaster = { version = "1.0.0", optional = true }
crossbeam-channel = { version = "0.4.0", optional = true }
crossbeam-deque = { version = "0.7.2", optional = true }
crossbeam-utils = { version = "0.7.0", optional = true }
futures-core = { version = "0.3.1", optional = true, default-features = false }
futures-io = { version = "0.3.1", optional = true }
futures-timer = { version = "2.0.2", optional = true }
async-task = { version = "3.0.0", optional = true }
crossbeam-utils = { version = "0.7.2", optional = true }
futures-core = { version = "0.3.4", optional = true, default-features = false }
futures-io = { version = "0.3.4", optional = true }
kv-log-macro = { version = "1.0.4", optional = true }
log = { version = "0.4.8", features = ["kv_unstable"], optional = true }
memchr = { version = "2.3.0", optional = true }
mio = { version = "0.6.19", optional = true }
mio-uds = { version = "0.6.7", optional = true }
num_cpus = { version = "1.11.1", optional = true }
once_cell = { version = "1.2.0", optional = true }
pin-project-lite = { version = "0.1.2", optional = true }
memchr = { version = "2.3.3", optional = true }
num_cpus = { version = "1.12.0", optional = true }
once_cell = { version = "1.3.1", optional = true }
pin-project-lite = { version = "0.1.4", optional = true }
pin-utils = { version = "0.1.0-alpha.4", optional = true }
slab = { version = "0.4.2", optional = true }
# Devdepencency, but they are not allowed to be optional :/
surf = { version = "1.0.3", optional = true }
[target.'cfg(not(target_os = "unknown"))'.dependencies]
smol = { version = "0.1.1", optional = true }
[target.'cfg(target_arch = "wasm32")'.dependencies]
futures-timer = { version = "3.0.2", optional = true, features = ["wasm-bindgen"] }
wasm-bindgen-futures = { version = "0.4.10", optional = true }
futures-channel = { version = "0.3.4", optional = true }
[target.'cfg(target_arch = "wasm32")'.dev-dependencies]
wasm-bindgen-test = "0.3.10"
[dev-dependencies]
femme = "1.3.0"
rand = "0.7.3"
surf = "1.0.3"
tempdir = "0.3.7"
futures = "0.3.1"
futures = "0.3.4"
rand_xorshift = "0.2.0"
[[test]]
name = "stream"
@ -87,3 +95,7 @@ required-features = ["unstable"]
[[example]]
name = "tcp-ipv4-and-6-echo"
required-features = ["unstable"]
[[example]]
name = "surf-web"
required-features = ["surf"]

7
README.md
View file

@ -8,6 +8,11 @@
<br />
<div align="center">
<!-- CI status -->
<a href="https://github.com/async-rs/async-std/actions">
<img src="https://github.com/async-rs/async-std/workflows/CI/badge.svg"
alt="CI Status" />
</a>
<!-- Crates version -->
<a href="https://crates.io/crates/async-std">
<img src="https://img.shields.io/crates/v/async-std.svg?style=flat-square"
@ -136,6 +141,8 @@ documentation] on how to enable them.
* [Xactor](https://crates.io/crates/xactor) — Xactor is a rust actors framework based on async-std.
* [async-graphql](https://crates.io/crates/async-graphql) — A GraphQL server library implemented in rust, with full support for async/await.
## License
<sup>

16
src/fs/file.rs
View file

@ -12,7 +12,7 @@ use crate::future;
use crate::io::{self, Read, Seek, SeekFrom, Write};
use crate::path::Path;
use crate::prelude::*;
use crate::task::{self, spawn_blocking, Context, Poll, Waker};
use crate::task::{spawn_blocking, Context, Poll, Waker};
use crate::utils::Context as _;
/// An open file on the filesystem.
@ -315,7 +315,7 @@ impl Drop for File {
// non-blocking fashion, but our only other option here is losing data remaining in the
// write cache. Good task schedulers should be resilient to occasional blocking hiccups in
// file destructors so we don't expect this to be a common problem in practice.
let _ = task::block_on(self.flush());
let _ = smol::block_on(self.flush());
}
}
@ -867,3 +867,15 @@ impl LockGuard<State> {
Poll::Ready(Ok(()))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn async_file_drop() {
crate::task::block_on(async move {
File::open(file!()).await.unwrap();
});
}
}

6
src/future/future/delay.rs
View file

@ -2,10 +2,10 @@ use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
use futures_timer::Delay;
use pin_project_lite::pin_project;
use crate::task::{Context, Poll};
use crate::utils::Timer;
pin_project! {
#[doc(hidden)]
@ -14,13 +14,13 @@ pin_project! {
#[pin]
future: F,
#[pin]
delay: Delay,
delay: Timer,
}
}
impl<F> DelayFuture<F> {
pub fn new(future: F, dur: Duration) -> DelayFuture<F> {
let delay = Delay::new(dur);
let delay = Timer::after(dur);
DelayFuture { future, delay }
}

17
src/future/timeout.rs
View file

@ -1,13 +1,13 @@
use std::error::Error;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
use std::future::Future;
use futures_timer::Delay;
use pin_project_lite::pin_project;
use crate::task::{Context, Poll};
use crate::utils::Timer;
/// Awaits a future or times out after a duration of time.
///
@ -33,11 +33,7 @@ pub async fn timeout<F, T>(dur: Duration, f: F) -> Result<T, TimeoutError>
where
F: Future<Output = T>,
{
let f = TimeoutFuture {
future: f,
delay: Delay::new(dur),
};
f.await
TimeoutFuture::new(f, dur).await
}
pin_project! {
@ -46,14 +42,17 @@ pin_project! {
#[pin]
future: F,
#[pin]
delay: Delay,
delay: Timer,
}
}
impl<F> TimeoutFuture<F> {
#[allow(dead_code)]
pub(super) fn new(future: F, dur: Duration) -> TimeoutFuture<F> {
TimeoutFuture { future: future, delay: Delay::new(dur) }
TimeoutFuture {
future,
delay: Timer::after(dur),
}
}
}

11
src/io/mod.rs
View file

@ -307,22 +307,33 @@ cfg_std! {
cfg_default! {
// For use in the print macros.
#[doc(hidden)]
#[cfg(not(target_os = "unknown"))]
pub use stdio::{_eprint, _print};
#[cfg(not(target_os = "unknown"))]
pub use stderr::{stderr, Stderr};
#[cfg(not(target_os = "unknown"))]
pub use stdin::{stdin, Stdin};
#[cfg(not(target_os = "unknown"))]
pub use stdout::{stdout, Stdout};
pub use timeout::timeout;
mod timeout;
#[cfg(not(target_os = "unknown"))]
mod stderr;
#[cfg(not(target_os = "unknown"))]
mod stdin;
#[cfg(not(target_os = "unknown"))]
mod stdio;
#[cfg(not(target_os = "unknown"))]
mod stdout;
}
cfg_unstable_default! {
#[cfg(not(target_os = "unknown"))]
pub use stderr::StderrLock;
#[cfg(not(target_os = "unknown"))]
pub use stdin::StdinLock;
#[cfg(not(target_os = "unknown"))]
pub use stdout::StdoutLock;
}

13
src/io/read/mod.rs
View file

@ -17,9 +17,9 @@ use std::mem;
use crate::io::IoSliceMut;
pub use take::Take;
pub use bytes::Bytes;
pub use chain::Chain;
pub use take::Take;
extension_trait! {
use std::pin::Pin;
@ -477,13 +477,13 @@ unsafe fn initialize<R: futures_io::AsyncRead>(_reader: &R, buf: &mut [u8]) {
std::ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len())
}
#[cfg(test)]
#[cfg(all(test, not(target_os = "unknown")))]
mod tests {
use crate::io;
use crate::prelude::*;
#[test]
fn test_read_by_ref() -> io::Result<()> {
fn test_read_by_ref() {
crate::task::block_on(async {
let mut f = io::Cursor::new(vec![0u8, 1, 2, 3, 4, 5, 6, 7, 8]);
let mut buffer = Vec::new();
@ -493,14 +493,13 @@ mod tests {
let reference = f.by_ref();
// read at most 5 bytes
assert_eq!(reference.take(5).read_to_end(&mut buffer).await?, 5);
assert_eq!(reference.take(5).read_to_end(&mut buffer).await.unwrap(), 5);
assert_eq!(&buffer, &[0, 1, 2, 3, 4])
} // drop our &mut reference so we can use f again
// original file still usable, read the rest
assert_eq!(f.read_to_end(&mut other_buffer).await?, 4);
assert_eq!(f.read_to_end(&mut other_buffer).await.unwrap(), 4);
assert_eq!(&other_buffer, &[5, 6, 7, 8]);
Ok(())
})
});
}
}

2
src/io/read/take.rs
View file

@ -218,7 +218,7 @@ impl<T: BufRead> BufRead for Take<T> {
}
}
#[cfg(test)]
#[cfg(all(test, not(target_os = "unknown")))]
mod tests {
use crate::io;
use crate::prelude::*;

8
src/io/timeout.rs
View file

@ -1,12 +1,12 @@
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use std::future::Future;
use futures_timer::Delay;
use pin_project_lite::pin_project;
use crate::io;
use crate::utils::Timer;
/// Awaits an I/O future or times out after a duration of time.
///
@ -37,7 +37,7 @@ where
F: Future<Output = io::Result<T>>,
{
Timeout {
timeout: Delay::new(dur),
timeout: Timer::after(dur),
future: f,
}
.await
@ -53,7 +53,7 @@ pin_project! {
#[pin]
future: F,
#[pin]
timeout: Delay,
timeout: Timer,
}
}

27
src/lib.rs
View file

@ -197,7 +197,7 @@
//!
//! ```toml
//! [dependencies.async-std]
//! version = "1.0.0"
//! version = "1.6.0-beta.1"
//! features = ["unstable"]
//! ```
//!
@ -210,7 +210,7 @@
//!
//! ```toml
//! [dependencies.async-std]
//! version = "1.0.0"
//! version = "1.6.0-beta.1"
//! features = ["attributes"]
//! ```
//!
@ -219,7 +219,7 @@
//!
//! ```toml
//! [dependencies.async-std]
//! version = "1.0.0"
//! version = "1.6.0-beta.1"
//! default-features = false
//! features = ["std"]
//! ```
@ -229,10 +229,25 @@
//!
//! ```toml
//! [dependencies.async-std]
//! version = "1.5.0"
//! version = "1.6.0-beta.1"
//! default-features = false
//! features = ["alloc"]
//! ```
//!
//! # Runtime configuration
//!
//! Several environment variables are available to tune the async-std
//! runtime:
//!
//! * `ASYNC_STD_THREAD_COUNT`: The number of threads that the
//! async-std runtime will start. By default, this is one per logical
//! cpu as reported by the [num_cpus](num_cpus) crate, which may be
//! different than the number of physical cpus. Async-std _will panic_
//! if this is set to any value other than a positive integer.
//! * `ASYNC_STD_THREAD_NAME`: The name that async-std's runtime
//! threads report to the operating system. The default value is
//! `"async-std/runtime"`.
//!
#![cfg_attr(not(feature = "std"), no_std)]
#![cfg_attr(feature = "docs", feature(doc_cfg))]
@ -270,13 +285,17 @@ cfg_std! {
}
cfg_default! {
#[cfg(not(target_os = "unknown"))]
pub mod fs;
pub mod path;
pub mod net;
#[cfg(not(target_os = "unknown"))]
pub(crate) mod rt;
}
cfg_unstable! {
pub mod pin;
#[cfg(not(target_os = "unknown"))]
pub mod process;
mod unit;

380
src/net/driver/mod.rs
View file

@ -1,380 +0,0 @@
use std::fmt;
use std::sync::{Arc, Mutex};
use mio::{self, Evented};
use once_cell::sync::Lazy;
use slab::Slab;
use crate::io;
use crate::task::{Context, Poll, Waker};
use crate::utils::abort_on_panic;
/// Data associated with a registered I/O handle.
#[derive(Debug)]
struct Entry {
/// A unique identifier.
token: mio::Token,
/// Tasks that are blocked on reading from this I/O handle.
readers: Mutex<Readers>,
/// Thasks that are blocked on writing to this I/O handle.
writers: Mutex<Writers>,
}
/// The set of `Waker`s interested in read readiness.
#[derive(Debug)]
struct Readers {
/// Flag indicating read readiness.
/// (cf. `Watcher::poll_read_ready`)
ready: bool,
/// The `Waker`s blocked on reading.
wakers: Vec<Waker>
}
/// The set of `Waker`s interested in write readiness.
#[derive(Debug)]
struct Writers {
/// Flag indicating write readiness.
/// (cf. `Watcher::poll_write_ready`)
ready: bool,
/// The `Waker`s blocked on writing.
wakers: Vec<Waker>
}
/// The state of a networking driver.
struct Reactor {
/// A mio instance that polls for new events.
poller: mio::Poll,
/// A collection of registered I/O handles.
entries: Mutex<Slab<Arc<Entry>>>,
/// Dummy I/O handle that is only used to wake up the polling thread.
notify_reg: (mio::Registration, mio::SetReadiness),
/// An identifier for the notification handle.
notify_token: mio::Token,
}
impl Reactor {
/// Creates a new reactor for polling I/O events.
fn new() -> io::Result<Reactor> {
let poller = mio::Poll::new()?;
let notify_reg = mio::Registration::new2();
let mut reactor = Reactor {
poller,
entries: Mutex::new(Slab::new()),
notify_reg,
notify_token: mio::Token(0),
};
// Register a dummy I/O handle for waking up the polling thread.
let entry = reactor.register(&reactor.notify_reg.0)?;
reactor.notify_token = entry.token;
Ok(reactor)
}
/// Registers an I/O event source and returns its associated entry.
fn register(&self, source: &dyn Evented) -> io::Result<Arc<Entry>> {
let mut entries = self.entries.lock().unwrap();
// Reserve a vacant spot in the slab and use its key as the token value.
let vacant = entries.vacant_entry();
let token = mio::Token(vacant.key());
// Allocate an entry and insert it into the slab.
let entry = Arc::new(Entry {
token,
readers: Mutex::new(Readers { ready: false, wakers: Vec::new() }),
writers: Mutex::new(Writers { ready: false, wakers: Vec::new() }),
});
vacant.insert(entry.clone());
// Register the I/O event source in the poller.
let interest = mio::Ready::all();
let opts = mio::PollOpt::edge();
self.poller.register(source, token, interest, opts)?;
Ok(entry)
}
/// Deregisters an I/O event source associated with an entry.
fn deregister(&self, source: &dyn Evented, entry: &Entry) -> io::Result<()> {
// Deregister the I/O object from the mio instance.
self.poller.deregister(source)?;
// Remove the entry associated with the I/O object.
self.entries.lock().unwrap().remove(entry.token.0);
Ok(())
}
// fn notify(&self) {
// self.notify_reg
// .1
// .set_readiness(mio::Ready::readable())
// .unwrap();
// }
}
/// The state of the global networking driver.
static REACTOR: Lazy<Reactor> = Lazy::new(|| {
// Spawn a thread that waits on the poller for new events and wakes up tasks blocked on I/O
// handles.
std::thread::Builder::new()
.name("async-std/net".to_string())
.spawn(move || {
// If the driver thread panics, there's not much we can do. It is not a
// recoverable error and there is no place to propagate it into so we just abort.
abort_on_panic(|| {
main_loop().expect("async networking thread has panicked");
})
})
.expect("cannot start a thread driving blocking tasks");
Reactor::new().expect("cannot initialize reactor")
});
/// Waits on the poller for new events and wakes up tasks blocked on I/O handles.
fn main_loop() -> io::Result<()> {
let reactor = &REACTOR;
let mut events = mio::Events::with_capacity(1000);
loop {
// Block on the poller until at least one new event comes in.
reactor.poller.poll(&mut events, None)?;
// Lock the entire entry table while we're processing new events.
let entries = reactor.entries.lock().unwrap();
for event in events.iter() {
let token = event.token();
if token == reactor.notify_token {
// If this is the notification token, we just need the notification state.
reactor.notify_reg.1.set_readiness(mio::Ready::empty())?;
} else {
// Otherwise, look for the entry associated with this token.
if let Some(entry) = entries.get(token.0) {
// Set the readiness flags from this I/O event.
let readiness = event.readiness();
// Wake up reader tasks blocked on this I/O handle.
if !(readiness & reader_interests()).is_empty() {
let mut readers = entry.readers.lock().unwrap();
readers.ready = true;
for w in readers.wakers.drain(..) {
w.wake();
}
}
// Wake up writer tasks blocked on this I/O handle.
if !(readiness & writer_interests()).is_empty() {
let mut writers = entry.writers.lock().unwrap();
writers.ready = true;
for w in writers.wakers.drain(..) {
w.wake();
}
}
}
}
}
}
}
/// An I/O handle powered by the networking driver.
///
/// This handle wraps an I/O event source and exposes a "futurized" interface on top of it,
/// implementing traits `AsyncRead` and `AsyncWrite`.
pub struct Watcher<T: Evented> {
/// Data associated with the I/O handle.
entry: Arc<Entry>,
/// The I/O event source.
source: Option<T>,
}
impl<T: Evented> Watcher<T> {
/// Creates a new I/O handle.
///
/// The provided I/O event source will be kept registered inside the reactor's poller for the
/// lifetime of the returned I/O handle.
pub fn new(source: T) -> Watcher<T> {
Watcher {
entry: REACTOR
.register(&source)
.expect("cannot register an I/O event source"),
source: Some(source),
}
}
/// Returns a reference to the inner I/O event source.
pub fn get_ref(&self) -> &T {
self.source.as_ref().unwrap()
}
/// Polls the inner I/O source for a non-blocking read operation.
///
/// If the operation returns an error of the `io::ErrorKind::WouldBlock` kind, the current task
/// will be registered for wakeup when the I/O source becomes readable.
pub fn poll_read_with<'a, F, R>(&'a self, cx: &mut Context<'_>, mut f: F) -> Poll<io::Result<R>>
where
F: FnMut(&'a T) -> io::Result<R>,
{
// If the operation isn't blocked, return its result.
match f(self.source.as_ref().unwrap()) {
Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
res => return Poll::Ready(res),
}
// Lock the waker list.
let mut readers = self.entry.readers.lock().unwrap();
// Try running the operation again.
match f(self.source.as_ref().unwrap()) {
Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
res => return Poll::Ready(res),
}
// Register the task if it isn't registered already.
if readers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {
readers.wakers.push(cx.waker().clone());
}
readers.ready = false;
Poll::Pending
}
/// Polls the inner I/O source for a non-blocking write operation.
///
/// If the operation returns an error of the `io::ErrorKind::WouldBlock` kind, the current task
/// will be registered for wakeup when the I/O source becomes writable.
pub fn poll_write_with<'a, F, R>(
&'a self,
cx: &mut Context<'_>,
mut f: F,
) -> Poll<io::Result<R>>
where
F: FnMut(&'a T) -> io::Result<R>,
{
// If the operation isn't blocked, return its result.
match f(self.source.as_ref().unwrap()) {
Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
res => return Poll::Ready(res),
}
// Lock the waker list.
let mut writers = self.entry.writers.lock().unwrap();
// Try running the operation again.
match f(self.source.as_ref().unwrap()) {
Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
res => return Poll::Ready(res),
}
// Register the task if it isn't registered already.
if writers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {
writers.wakers.push(cx.waker().clone());
}
writers.ready = false;
Poll::Pending
}
/// Polls the inner I/O source until a non-blocking read can be performed.
///
/// If non-blocking reads are currently not possible, the `Waker`
/// will be saved and notified when it can read non-blocking
/// again.
#[allow(dead_code)]
pub fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<()> {
// Lock the waker list.
let mut readers = self.entry.readers.lock().unwrap();
if readers.ready {
return Poll::Ready(())
}
// Register the task if it isn't registered already.
if readers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {
readers.wakers.push(cx.waker().clone());
}
Poll::Pending
}
/// Polls the inner I/O source until a non-blocking write can be performed.
///
/// If non-blocking writes are currently not possible, the `Waker`
/// will be saved and notified when it can write non-blocking
/// again.
pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<()> {
// Lock the waker list.
let mut writers = self.entry.writers.lock().unwrap();
if writers.ready {
return Poll::Ready(())
}
// Register the task if it isn't registered already.
if writers.wakers.iter().all(|w| !w.will_wake(cx.waker())) {
writers.wakers.push(cx.waker().clone());
}
Poll::Pending
}
/// Deregisters and returns the inner I/O source.
///
/// This method is typically used to convert `Watcher`s to raw file descriptors/handles.
#[allow(dead_code)]
pub fn into_inner(mut self) -> T {
let source = self.source.take().unwrap();
REACTOR
.deregister(&source, &self.entry)
.expect("cannot deregister I/O event source");
source
}
}
impl<T: Evented> Drop for Watcher<T> {
fn drop(&mut self) {
if let Some(ref source) = self.source {
REACTOR
.deregister(source, &self.entry)
.expect("cannot deregister I/O event source");
}
}
}
impl<T: Evented + fmt::Debug> fmt::Debug for Watcher<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Watcher")
.field("entry", &self.entry)
.field("source", &self.source)
.finish()
}
}
/// Returns a mask containing flags that interest tasks reading from I/O handles.
#[inline]
fn reader_interests() -> mio::Ready {
mio::Ready::all() - mio::Ready::writable()
}
/// Returns a mask containing flags that interest tasks writing into I/O handles.
#[inline]
fn writer_interests() -> mio::Ready {
mio::Ready::writable() | hup()
}
/// Returns a flag containing the hangup status.
#[inline]
fn hup() -> mio::Ready {
#[cfg(unix)]
let ready = mio::unix::UnixReady::hup().into();
#[cfg(not(unix))]
let ready = mio::Ready::empty();
ready
}

7
src/net/mod.rs
View file

@ -61,11 +61,16 @@ pub use std::net::Shutdown;
pub use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
pub use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6};
#[cfg(not(target_os = "unknown"))]
pub use addr::ToSocketAddrs;
#[cfg(not(target_os = "unknown"))]
pub use tcp::{Incoming, TcpListener, TcpStream};
#[cfg(not(target_os = "unknown"))]
pub use udp::UdpSocket;
#[cfg(not(target_os = "unknown"))]
mod addr;
pub(crate) mod driver;
#[cfg(not(target_os = "unknown"))]
mod tcp;
#[cfg(not(target_os = "unknown"))]
mod udp;

69
src/net/tcp/listener.rs
View file

@ -1,13 +1,13 @@
use std::future::Future;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use crate::future;
use smol::Async;
use crate::io;
use crate::net::driver::Watcher;
use crate::net::{TcpStream, ToSocketAddrs};
use crate::stream::Stream;
use crate::sync::Arc;
use crate::task::{Context, Poll};
/// A TCP socket server, listening for connections.
@ -49,7 +49,7 @@ use crate::task::{Context, Poll};
/// ```
#[derive(Debug)]
pub struct TcpListener {
watcher: Watcher<mio::net::TcpListener>,
watcher: Async<std::net::TcpListener>,
}
impl TcpListener {
@ -79,11 +79,9 @@ impl TcpListener {
let addrs = addrs.to_socket_addrs().await?;
for addr in addrs {
match mio::net::TcpListener::bind(&addr) {
Ok(mio_listener) => {
return Ok(TcpListener {
watcher: Watcher::new(mio_listener),
});
match Async::<std::net::TcpListener>::bind(&addr) {
Ok(listener) => {
return Ok(TcpListener { watcher: listener });
}
Err(err) => last_err = Some(err),
}
@ -114,13 +112,9 @@ impl TcpListener {
/// # Ok(()) }) }
/// ```
pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
let (io, addr) =
future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.accept_std()))
.await?;
let mio_stream = mio::net::TcpStream::from_stream(io)?;
let (stream, addr) = self.watcher.accept().await?;
let stream = TcpStream {
watcher: Arc::new(Watcher::new(mio_stream)),
watcher: Arc::new(stream),
};
Ok((stream, addr))
}
@ -206,9 +200,8 @@ impl<'a> Stream for Incoming<'a> {
impl From<std::net::TcpListener> for TcpListener {
/// Converts a `std::net::TcpListener` into its asynchronous equivalent.
fn from(listener: std::net::TcpListener) -> TcpListener {
let mio_listener = mio::net::TcpListener::from_std(listener).unwrap();
TcpListener {
watcher: Watcher::new(mio_listener),
watcher: Async::new(listener).expect("TcpListener is known to be good"),
}
}
}
@ -230,29 +223,31 @@ cfg_unix! {
impl IntoRawFd for TcpListener {
fn into_raw_fd(self) -> RawFd {
self.watcher.into_inner().into_raw_fd()
self.watcher.into_raw_fd()
}
}
}
cfg_windows! {
// use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
//
// impl AsRawSocket for TcpListener {
// fn as_raw_socket(&self) -> RawSocket {
// self.raw_socket
// }
// }
//
// impl FromRawSocket for TcpListener {
// unsafe fn from_raw_socket(handle: RawSocket) -> TcpListener {
// net::TcpListener::from_raw_socket(handle).try_into().unwrap()
// }
// }
//
// impl IntoRawSocket for TcpListener {
// fn into_raw_socket(self) -> RawSocket {
// self.raw_socket
// }
// }
use crate::os::windows::io::{
AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket,
};
impl AsRawSocket for TcpListener {
fn as_raw_socket(&self) -> RawSocket {
self.watcher.as_raw_socket()
}
}
impl FromRawSocket for TcpListener {
unsafe fn from_raw_socket(handle: RawSocket) -> TcpListener {
std::net::TcpListener::from_raw_socket(handle).into()
}
}
impl IntoRawSocket for TcpListener {
fn into_raw_socket(self) -> RawSocket {
self.watcher.into_raw_socket()
}
}
}

94
src/net/tcp/stream.rs
View file

@ -1,12 +1,12 @@
use std::io::{IoSlice, IoSliceMut, Read as _, Write as _};
use std::io::{IoSlice, IoSliceMut};
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use crate::future;
use smol::Async;
use crate::io::{self, Read, Write};
use crate::net::driver::Watcher;
use crate::net::ToSocketAddrs;
use crate::sync::Arc;
use crate::task::{Context, Poll};
/// A TCP stream between a local and a remote socket.
@ -47,7 +47,7 @@ use crate::task::{Context, Poll};
/// ```
#[derive(Debug, Clone)]
pub struct TcpStream {
pub(super) watcher: Arc<Watcher<mio::net::TcpStream>>,
pub(super) watcher: Arc<Async<std::net::TcpStream>>,
}
impl TcpStream {
@ -75,28 +75,16 @@ impl TcpStream {
let addrs = addrs.to_socket_addrs().await?;
for addr in addrs {
// mio's TcpStream::connect is non-blocking and may just be in progress
// when it returns with `Ok`. We therefore wait for write readiness to
// be sure the connection has either been established or there was an
// error which we check for afterwards.
let watcher = match mio::net::TcpStream::connect(&addr) {
Ok(s) => Watcher::new(s),
match Async::<std::net::TcpStream>::connect(&addr).await {
Ok(stream) => {
return Ok(TcpStream {
watcher: Arc::new(stream),
});
}
Err(e) => {
last_err = Some(e);
continue;
}
};
future::poll_fn(|cx| watcher.poll_write_ready(cx)).await;
match watcher.get_ref().take_error() {
Ok(None) => {
return Ok(TcpStream {
watcher: Arc::new(watcher),
});
}
Ok(Some(e)) => last_err = Some(e),
Err(e) => last_err = Some(e),
}
}
@ -214,7 +202,7 @@ impl TcpStream {
/// # Ok(()) }) }
/// ```
pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.peek(buf))).await
self.watcher.peek(buf).await
}
/// Gets the value of the `TCP_NODELAY` option on this socket.
@ -317,7 +305,7 @@ impl Read for &TcpStream {
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
self.watcher.poll_read_with(cx, |mut inner| inner.read(buf))
Pin::new(&mut &*self.watcher).poll_read(cx, buf)
}
}
@ -353,26 +341,23 @@ impl Write for &TcpStream {
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
self.watcher
.poll_write_with(cx, |mut inner| inner.write(buf))
Pin::new(&mut &*self.watcher).poll_write(cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.watcher.poll_write_with(cx, |mut inner| inner.flush())
Pin::new(&mut &*self.watcher).poll_flush(cx)
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
self.shutdown(std::net::Shutdown::Write)?;
Poll::Ready(Ok(()))
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut &*self.watcher).poll_close(cx)
}
}
impl From<std::net::TcpStream> for TcpStream {
/// Converts a `std::net::TcpStream` into its asynchronous equivalent.
fn from(stream: std::net::TcpStream) -> TcpStream {
let mio_stream = mio::net::TcpStream::from_stream(stream).unwrap();
TcpStream {
watcher: Arc::new(Watcher::new(mio_stream)),
watcher: Arc::new(Async::new(stream).expect("TcpStream is known to be good")),
}
}
}
@ -403,23 +388,28 @@ cfg_unix! {
}
cfg_windows! {
// use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
//
// impl AsRawSocket for TcpStream {
// fn as_raw_socket(&self) -> RawSocket {
// self.raw_socket
// }
// }
//
// impl FromRawSocket for TcpStream {
// unsafe fn from_raw_socket(handle: RawSocket) -> TcpStream {
// net::TcpStream::from_raw_socket(handle).try_into().unwrap()
// }
// }
//
// impl IntoRawSocket for TcpListener {
// fn into_raw_socket(self) -> RawSocket {
// self.raw_socket
// }
// }
use crate::os::windows::io::{
RawSocket, AsRawSocket, FromRawSocket, IntoRawSocket
};
impl AsRawSocket for TcpStream {
fn as_raw_socket(&self) -> RawSocket {
self.watcher.get_ref().as_raw_socket()
}
}
impl FromRawSocket for TcpStream {
unsafe fn from_raw_socket(handle: RawSocket) -> TcpStream {
std::net::TcpStream::from_raw_socket(handle).into()
}
}
impl IntoRawSocket for TcpStream {
fn into_raw_socket(self) -> RawSocket {
// TODO(stjepang): This does not mean `RawFd` is now the sole owner of the file
// descriptor because it's possible that there are other clones of this `TcpStream`
// using it at the same time. We should probably document that behavior.
self.as_raw_socket()
}
}
}

120
src/net/udp/mod.rs
View file

@ -2,8 +2,8 @@ use std::io;
use std::net::SocketAddr;
use std::net::{Ipv4Addr, Ipv6Addr};
use crate::future;
use crate::net::driver::Watcher;
use smol::Async;
use crate::net::ToSocketAddrs;
use crate::utils::Context as _;
@ -45,7 +45,7 @@ use crate::utils::Context as _;
/// ```
#[derive(Debug)]
pub struct UdpSocket {
watcher: Watcher<mio::net::UdpSocket>,
watcher: Async<std::net::UdpSocket>,
}
impl UdpSocket {
@ -69,16 +69,12 @@ impl UdpSocket {
/// ```
pub async fn bind<A: ToSocketAddrs>(addrs: A) -> io::Result<UdpSocket> {
let mut last_err = None;
let addrs = addrs
.to_socket_addrs()
.await?;
let addrs = addrs.to_socket_addrs().await?;
for addr in addrs {
match mio::net::UdpSocket::bind(&addr) {
Ok(mio_socket) => {
return Ok(UdpSocket {
watcher: Watcher::new(mio_socket),
});
match Async::<std::net::UdpSocket>::bind(&addr) {
Ok(socket) => {
return Ok(UdpSocket { watcher: socket });
}
Err(err) => last_err = Some(err),
}
@ -102,7 +98,7 @@ impl UdpSocket {
/// ```no_run
/// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
/// #
/// use async_std::net::UdpSocket;
/// use async_std::net::UdpSocket;
///
/// let socket = UdpSocket::bind("127.0.0.1:0").await?;
/// let addr = socket.local_addr()?;
@ -153,12 +149,10 @@ impl UdpSocket {
}
};
future::poll_fn(|cx| {
self.watcher
.poll_write_with(cx, |inner| inner.send_to(buf, &addr))
})
.await
.context(|| format!("could not send packet to {}", addr))
self.watcher
.send_to(buf, addr)
.await
.context(|| format!("could not send packet to {}", addr))
}
/// Receives data from the socket.
@ -181,22 +175,7 @@ impl UdpSocket {
/// # Ok(()) }) }
/// ```
pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
future::poll_fn(|cx| {
self.watcher
.poll_read_with(cx, |inner| inner.recv_from(buf))
})
.await
.context(|| {
use std::fmt::Write;
let mut error = String::from("could not receive data on ");
if let Ok(addr) = self.local_addr() {
let _ = write!(&mut error, "{}", addr);
} else {
error.push_str("socket");
}
error
})
self.watcher.recv_from(buf).await
}
/// Connects the UDP socket to a remote address.
@ -267,19 +246,7 @@ impl UdpSocket {
/// # Ok(()) }) }
/// ```
pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
future::poll_fn(|cx| self.watcher.poll_write_with(cx, |inner| inner.send(buf)))
.await
.context(|| {
use std::fmt::Write;
let mut error = String::from("could not send data on ");
if let Ok(addr) = self.local_addr() {
let _ = write!(&mut error, "{}", addr);
} else {
error.push_str("socket");
}
error
})
self.watcher.send(buf).await
}
/// Receives data from the socket.
@ -303,19 +270,7 @@ impl UdpSocket {
/// # Ok(()) }) }
/// ```
pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.recv(buf)))
.await
.context(|| {
use std::fmt::Write;
let mut error = String::from("could not receive data on ");
if let Ok(addr) = self.local_addr() {
let _ = write!(&mut error, "{}", addr);
} else {
error.push_str("socket");
}
error
})
self.watcher.recv(buf).await
}
/// Gets the value of the `SO_BROADCAST` option for this socket.
@ -498,9 +453,8 @@ impl UdpSocket {
impl From<std::net::UdpSocket> for UdpSocket {
/// Converts a `std::net::UdpSocket` into its asynchronous equivalent.
fn from(socket: std::net::UdpSocket) -> UdpSocket {
let mio_socket = mio::net::UdpSocket::from_socket(socket).unwrap();
UdpSocket {
watcher: Watcher::new(mio_socket),
watcher: Async::new(socket).expect("UdpSocket is known to be good"),
}
}
}
@ -522,29 +476,31 @@ cfg_unix! {
impl IntoRawFd for UdpSocket {
fn into_raw_fd(self) -> RawFd {
self.watcher.into_inner().into_raw_fd()
self.watcher.into_raw_fd()
}
}
}
cfg_windows! {
// use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
//
// impl AsRawSocket for UdpSocket {
// fn as_raw_socket(&self) -> RawSocket {
// self.raw_socket
// }
// }
//
// impl FromRawSocket for UdpSocket {
// unsafe fn from_raw_socket(handle: RawSocket) -> UdpSocket {
// net::UdpSocket::from_raw_socket(handle).into()
// }
// }
//
// impl IntoRawSocket for UdpSocket {
// fn into_raw_socket(self) -> RawSocket {
// self.raw_socket
// }
// }
use crate::os::windows::io::{
RawSocket, AsRawSocket, IntoRawSocket, FromRawSocket
};
impl AsRawSocket for UdpSocket {
fn as_raw_socket(&self) -> RawSocket {
self.watcher.get_ref().as_raw_socket()
}
}
impl FromRawSocket for UdpSocket {
unsafe fn from_raw_socket(handle: RawSocket) -> UdpSocket {
std::net::UdpSocket::from_raw_socket(handle).into()
}
}
impl IntoRawSocket for UdpSocket {
fn into_raw_socket(self) -> RawSocket {
self.watcher.into_raw_socket()
}
}
}

52
src/os/unix/net/datagram.rs
View file

@ -2,14 +2,14 @@
use std::fmt;
use std::net::Shutdown;
use std::os::unix::net::UnixDatagram as StdUnixDatagram;
use smol::Async;
use super::SocketAddr;
use crate::future;
use crate::io;
use crate::net::driver::Watcher;
use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
use crate::path::Path;
use crate::task::spawn_blocking;
/// A Unix datagram socket.
///
@ -40,13 +40,13 @@ use crate::task::spawn_blocking;
/// # Ok(()) }) }
/// ```
pub struct UnixDatagram {
watcher: Watcher<mio_uds::UnixDatagram>,
watcher: Async<StdUnixDatagram>,
}
impl UnixDatagram {
fn new(socket: mio_uds::UnixDatagram) -> UnixDatagram {
fn new(socket: StdUnixDatagram) -> UnixDatagram {
UnixDatagram {
watcher: Watcher::new(socket),
watcher: Async::new(socket).expect("UnixDatagram is known to be good"),
}
}
@ -65,8 +65,8 @@ impl UnixDatagram {
/// ```
pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixDatagram> {
let path = path.as_ref().to_owned();
let socket = spawn_blocking(move || mio_uds::UnixDatagram::bind(path)).await?;
Ok(UnixDatagram::new(socket))
let socket = Async::<StdUnixDatagram>::bind(path)?;
Ok(UnixDatagram { watcher: socket })
}
/// Creates a Unix datagram which is not bound to any address.
@ -83,7 +83,7 @@ impl UnixDatagram {
/// # Ok(()) }) }
/// ```
pub fn unbound() -> io::Result<UnixDatagram> {
let socket = mio_uds::UnixDatagram::unbound()?;
let socket = StdUnixDatagram::unbound()?;
Ok(UnixDatagram::new(socket))
}
@ -103,7 +103,7 @@ impl UnixDatagram {
/// # Ok(()) }) }
/// ```
pub fn pair() -> io::Result<(UnixDatagram, UnixDatagram)> {
let (a, b) = mio_uds::UnixDatagram::pair()?;
let (a, b) = StdUnixDatagram::pair()?;
let a = UnixDatagram::new(a);
let b = UnixDatagram::new(b);
Ok((a, b))
@ -195,11 +195,7 @@ impl UnixDatagram {
/// # Ok(()) }) }
/// ```
pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
future::poll_fn(|cx| {
self.watcher
.poll_read_with(cx, |inner| inner.recv_from(buf))
})
.await
self.watcher.recv_from(buf).await
}
/// Receives data from the socket.
@ -220,7 +216,7 @@ impl UnixDatagram {
/// # Ok(()) }) }
/// ```
pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.recv(buf))).await
self.watcher.recv(buf).await
}
/// Sends data on the socket to the specified address.
@ -240,11 +236,7 @@ impl UnixDatagram {
/// # Ok(()) }) }
/// ```
pub async fn send_to<P: AsRef<Path>>(&self, buf: &[u8], path: P) -> io::Result<usize> {
future::poll_fn(|cx| {
self.watcher
.poll_write_with(cx, |inner| inner.send_to(buf, path.as_ref()))
})
.await
self.watcher.send_to(buf, path.as_ref()).await
}
/// Sends data on the socket to the socket's peer.
@ -265,7 +257,7 @@ impl UnixDatagram {
/// # Ok(()) }) }
/// ```
pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
future::poll_fn(|cx| self.watcher.poll_write_with(cx, |inner| inner.send(buf))).await
self.watcher.send(buf).await
}
/// Shut down the read, write, or both halves of this connection.
@ -310,31 +302,31 @@ impl fmt::Debug for UnixDatagram {
}
}
impl From<std::os::unix::net::UnixDatagram> for UnixDatagram {
impl From<StdUnixDatagram> for UnixDatagram {
/// Converts a `std::os::unix::net::UnixDatagram` into its asynchronous equivalent.
fn from(datagram: std::os::unix::net::UnixDatagram) -> UnixDatagram {
let mio_datagram = mio_uds::UnixDatagram::from_datagram(datagram).unwrap();
fn from(datagram: StdUnixDatagram) -> UnixDatagram {
UnixDatagram {
watcher: Watcher::new(mio_datagram),
watcher: Async::new(datagram).expect("UnixDatagram is known to be good"),
}
}
}
impl AsRawFd for UnixDatagram {
fn as_raw_fd(&self) -> RawFd {
self.watcher.get_ref().as_raw_fd()
self.watcher.as_raw_fd()
}
}
impl FromRawFd for UnixDatagram {
unsafe fn from_raw_fd(fd: RawFd) -> UnixDatagram {
let datagram = std::os::unix::net::UnixDatagram::from_raw_fd(fd);
datagram.into()
let raw = StdUnixDatagram::from_raw_fd(fd);
let datagram = Async::<StdUnixDatagram>::new(raw).expect("invalid file descriptor");
UnixDatagram { watcher: datagram }
}
}
impl IntoRawFd for UnixDatagram {
fn into_raw_fd(self) -> RawFd {
self.watcher.into_inner().into_raw_fd()
self.watcher.into_raw_fd()
}
}

53
src/os/unix/net/listener.rs
View file

@ -1,18 +1,20 @@
//! Unix-specific networking extensions.
use std::fmt;
use std::pin::Pin;
use std::future::Future;
use std::os::unix::net::UnixListener as StdUnixListener;
use std::pin::Pin;
use smol::Async;
use super::SocketAddr;
use super::UnixStream;
use crate::future;
use crate::io;
use crate::net::driver::Watcher;
use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
use crate::path::Path;
use crate::stream::Stream;
use crate::task::{spawn_blocking, Context, Poll};
use crate::sync::Arc;
use crate::task::{Context, Poll};
/// A Unix domain socket server, listening for connections.
///
@ -48,7 +50,7 @@ use crate::task::{spawn_blocking, Context, Poll};
/// # Ok(()) }) }
/// ```
pub struct UnixListener {
watcher: Watcher<mio_uds::UnixListener>,
watcher: Async<StdUnixListener>,
}
impl UnixListener {
@ -67,11 +69,9 @@ impl UnixListener {
/// ```
pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixListener> {
let path = path.as_ref().to_owned();
let listener = spawn_blocking(move || mio_uds::UnixListener::bind(path)).await?;
let listener = Async::<StdUnixListener>::bind(path)?;
Ok(UnixListener {
watcher: Watcher::new(listener),
})
Ok(UnixListener { watcher: listener })
}
/// Accepts a new incoming connection to this listener.
@ -91,29 +91,9 @@ impl UnixListener {
/// # Ok(()) }) }
/// ```
pub async fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
future::poll_fn(|cx| {
let res = futures_core::ready!(self.watcher.poll_read_with(cx, |inner| {
match inner.accept_std() {
// Converting to `WouldBlock` so that the watcher will
// add the waker of this task to a list of readers.
Ok(None) => Err(io::ErrorKind::WouldBlock.into()),
res => res,
}
}));
let (stream, addr) = self.watcher.accept().await?;
match res? {
Some((io, addr)) => {
let mio_stream = mio_uds::UnixStream::from_stream(io)?;
let stream = UnixStream {
watcher: Watcher::new(mio_stream),
};
Poll::Ready(Ok((stream, addr)))
}
// This should never happen since `None` is converted to `WouldBlock`
None => unreachable!(),
}
})
.await
Ok((UnixStream { watcher: Arc::new(stream) }, addr))
}
/// Returns a stream of incoming connections.
@ -204,19 +184,18 @@ impl Stream for Incoming<'_> {
}
}
impl From<std::os::unix::net::UnixListener> for UnixListener {
impl From<StdUnixListener> for UnixListener {
/// Converts a `std::os::unix::net::UnixListener` into its asynchronous equivalent.
fn from(listener: std::os::unix::net::UnixListener) -> UnixListener {
let mio_listener = mio_uds::UnixListener::from_listener(listener).unwrap();
fn from(listener: StdUnixListener) -> UnixListener {
UnixListener {
watcher: Watcher::new(mio_listener),
watcher: Async::new(listener).expect("UnixListener is known to be good"),
}
}
}
impl AsRawFd for UnixListener {
fn as_raw_fd(&self) -> RawFd {
self.watcher.get_ref().as_raw_fd()
self.watcher.as_raw_fd()
}
}
@ -229,6 +208,6 @@ impl FromRawFd for UnixListener {
impl IntoRawFd for UnixListener {
fn into_raw_fd(self) -> RawFd {
self.watcher.into_inner().into_raw_fd()
self.watcher.into_raw_fd()
}
}

52
src/os/unix/net/stream.rs
View file

@ -1,16 +1,18 @@
//! Unix-specific networking extensions.
use std::fmt;
use std::io::{Read as _, Write as _};
use std::net::Shutdown;
use std::os::unix::net::UnixStream as StdUnixStream;
use std::pin::Pin;
use smol::Async;
use super::SocketAddr;
use crate::io::{self, Read, Write};
use crate::net::driver::Watcher;
use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
use crate::path::Path;
use crate::task::{spawn_blocking, Context, Poll};
use crate::sync::Arc;
use crate::task::{Context, Poll};
/// A Unix stream socket.
///
@ -35,8 +37,9 @@ use crate::task::{spawn_blocking, Context, Poll};
/// #
/// # Ok(()) }) }
/// ```
#[derive(Clone)]
pub struct UnixStream {
pub(super) watcher: Watcher<mio_uds::UnixStream>,
pub(super) watcher: Arc<Async<StdUnixStream>>,
}
impl UnixStream {
@ -55,15 +58,9 @@ impl UnixStream {
/// ```
pub async fn connect<P: AsRef<Path>>(path: P) -> io::Result<UnixStream> {
let path = path.as_ref().to_owned();
let stream = Arc::new(Async::<StdUnixStream>::connect(path).await?);
spawn_blocking(move || {
let std_stream = std::os::unix::net::UnixStream::connect(path)?;
let mio_stream = mio_uds::UnixStream::from_stream(std_stream)?;
Ok(UnixStream {
watcher: Watcher::new(mio_stream),
})
})
.await
Ok(UnixStream { watcher: stream })
}
/// Creates an unnamed pair of connected sockets.
@ -82,12 +79,12 @@ impl UnixStream {
/// # Ok(()) }) }
/// ```
pub fn pair() -> io::Result<(UnixStream, UnixStream)> {
let (a, b) = mio_uds::UnixStream::pair()?;
let (a, b) = Async::<StdUnixStream>::pair()?;
let a = UnixStream {
watcher: Watcher::new(a),
watcher: Arc::new(a),
};
let b = UnixStream {
watcher: Watcher::new(b),
watcher: Arc::new(b),
};
Ok((a, b))
}
@ -167,7 +164,7 @@ impl Read for &UnixStream {
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
self.watcher.poll_read_with(cx, |mut inner| inner.read(buf))
Pin::new(&mut &*self.watcher).poll_read(cx, buf)
}
}
@ -195,16 +192,15 @@ impl Write for &UnixStream {
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
self.watcher
.poll_write_with(cx, |mut inner| inner.write(buf))
Pin::new(&mut &*self.watcher).poll_write(cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.watcher.poll_write_with(cx, |mut inner| inner.flush())
Pin::new(&mut &*self.watcher).poll_flush(cx)
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut &*self.watcher).poll_close(cx)
}
}
@ -225,19 +221,17 @@ impl fmt::Debug for UnixStream {
}
}
impl From<std::os::unix::net::UnixStream> for UnixStream {
impl From<StdUnixStream> for UnixStream {
/// Converts a `std::os::unix::net::UnixStream` into its asynchronous equivalent.
fn from(stream: std::os::unix::net::UnixStream) -> UnixStream {
let mio_stream = mio_uds::UnixStream::from_stream(stream).unwrap();
UnixStream {
watcher: Watcher::new(mio_stream),
}
fn from(stream: StdUnixStream) -> UnixStream {
let stream = Async::new(stream).expect("UnixStream is known to be good");
UnixStream { watcher: Arc::new(stream) }
}
}
impl AsRawFd for UnixStream {
fn as_raw_fd(&self) -> RawFd {
self.watcher.get_ref().as_raw_fd()
self.watcher.as_raw_fd()
}
}
@ -250,6 +244,6 @@ impl FromRawFd for UnixStream {
impl IntoRawFd for UnixStream {
fn into_raw_fd(self) -> RawFd {
self.watcher.into_inner().into_raw_fd()
self.as_raw_fd()
}
}

32
src/os/windows/io.rs
View file

@ -2,7 +2,8 @@
cfg_not_docs! {
pub use std::os::windows::io::{
AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle, RawSocket,
AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle,
AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket,
};
}
@ -45,4 +46,33 @@ cfg_docs! {
/// it once it's no longer needed.
fn into_raw_handle(self) -> RawHandle;
}
/// Creates I/O objects from raw sockets.
pub trait FromRawSocket {
/// Creates a new I/O object from the given raw socket.
///
/// This function will consume ownership of the socket provided and it will be closed when the returned object goes out of scope.
///
/// This function is also unsafe as the primitives currently returned have the contract that they are the sole owner of the
/// file descriptor they are wrapping. Usage of this function could accidentally allow violating this contract which can cause
/// memory unsafety in code that relies on it being true.
unsafe fn from_raw_socket(sock: RawSocket) -> Self;
}
/// Extracts raw sockets.
pub trait AsRawSocket {
/// Extracts the underlying raw socket from this object.
fn as_raw_socket(&self) -> RawSocket;
}
/// A trait to express the ability to consume an object and acquire ownership of
/// its raw `SOCKET`.
pub trait IntoRawSocket {
/// Consumes this object, returning the raw underlying socket.
///
/// This function **transfers ownership** of the underlying socket to the
/// caller. Callers are then the unique owners of the socket and must close
/// it once it's no longer needed.
fn into_raw_socket(self) -> RawSocket;
}
}

12
src/path/path.rs
View file

@ -4,9 +4,9 @@ use std::ffi::{OsStr, OsString};
use std::rc::Rc;
use std::sync::Arc;
use crate::fs;
use crate::io;
use crate::path::{Ancestors, Components, Display, Iter, PathBuf, StripPrefixError};
#[cfg(not(target_os = "unknown"))]
use crate::{fs, io};
/// A slice of a path.
///
@ -584,6 +584,7 @@ impl Path {
/// #
/// # Ok(()) }) }
/// ```
#[cfg(not(target_os = "unknown"))]
pub async fn metadata(&self) -> io::Result<fs::Metadata> {
fs::metadata(self).await
}
@ -607,6 +608,7 @@ impl Path {
/// #
/// # Ok(()) }) }
/// ```
#[cfg(not(target_os = "unknown"))]
pub async fn symlink_metadata(&self) -> io::Result<fs::Metadata> {
fs::symlink_metadata(self).await
}
@ -632,6 +634,7 @@ impl Path {
/// #
/// # Ok(()) }) }
/// ```
#[cfg(not(target_os = "unknown"))]
pub async fn canonicalize(&self) -> io::Result<PathBuf> {
fs::canonicalize(self).await
}
@ -654,6 +657,7 @@ impl Path {
/// #
/// # Ok(()) }) }
/// ```
#[cfg(not(target_os = "unknown"))]
pub async fn read_link(&self) -> io::Result<PathBuf> {
fs::read_link(self).await
}
@ -688,6 +692,7 @@ impl Path {
/// #
/// # Ok(()) }) }
/// ```
#[cfg(not(target_os = "unknown"))]
pub async fn read_dir(&self) -> io::Result<fs::ReadDir> {
fs::read_dir(self).await
}
@ -717,6 +722,7 @@ impl Path {
/// check errors, call [fs::metadata].
///
/// [fs::metadata]: ../fs/fn.metadata.html
#[cfg(not(target_os = "unknown"))]
pub async fn exists(&self) -> bool {
fs::metadata(self).await.is_ok()
}
@ -749,6 +755,7 @@ impl Path {
///
/// [fs::metadata]: ../fs/fn.metadata.html
/// [fs::Metadata::is_file]: ../fs/struct.Metadata.html#method.is_file
#[cfg(not(target_os = "unknown"))]
pub async fn is_file(&self) -> bool {
fs::metadata(self)
.await
@ -785,6 +792,7 @@ impl Path {
///
/// [fs::metadata]: ../fs/fn.metadata.html
/// [fs::Metadata::is_dir]: ../fs/struct.Metadata.html#method.is_dir
#[cfg(not(target_os = "unknown"))]
pub async fn is_dir(&self) -> bool {
fs::metadata(self)
.await

34
src/rt/mod.rs Normal file
View file

@ -0,0 +1,34 @@
//! The runtime.
use std::env;
use std::thread;
use once_cell::sync::Lazy;
use crate::future;
/// Dummy runtime struct.
pub struct Runtime {}
/// The global runtime.
pub static RUNTIME: Lazy<Runtime> = Lazy::new(|| {
// Create an executor thread pool.
let thread_count = env::var("ASYNC_STD_THREAD_COUNT")
.map(|env| {
env.parse()
.expect("ASYNC_STD_THREAD_COUNT must be a number")
})
.unwrap_or_else(|_| num_cpus::get())
.max(1);
let thread_name = env::var("ASYNC_STD_THREAD_NAME").unwrap_or("async-std/runtime".to_string());
for _ in 0..thread_count {
thread::Builder::new()
.name(thread_name.clone())
.spawn(|| smol::run(future::pending::<()>()))
.expect("cannot start a runtime thread");
}
Runtime {}
});

131
src/stream/interval.rs
View file

@ -1,10 +1,10 @@
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use std::time::Duration;
use crate::future::Future;
use crate::stream::Stream;
use futures_timer::Delay;
use crate::utils::Timer;
/// Creates a new stream that yields at a set interval.
///
@ -45,7 +45,7 @@ use futures_timer::Delay;
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
pub fn interval(dur: Duration) -> Interval {
Interval {
delay: Delay::new(dur),
delay: Timer::after(dur),
interval: dur,
}
}
@ -60,7 +60,7 @@ pub fn interval(dur: Duration) -> Interval {
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
#[derive(Debug)]
pub struct Interval {
delay: Delay,
delay: Timer,
interval: Duration,
}
@ -71,125 +71,8 @@ impl Stream for Interval {
if Pin::new(&mut self.delay).poll(cx).is_pending() {
return Poll::Pending;
}
let when = Instant::now();
let next = next_interval(when, Instant::now(), self.interval);
self.delay.reset(next);
let interval = self.interval;
let _ = std::mem::replace(&mut self.delay, Timer::after(interval));
Poll::Ready(Some(()))
}
}
/// Converts Duration object to raw nanoseconds if possible
///
/// This is useful to divide intervals.
///
/// While technically for large duration it's impossible to represent any
/// duration as nanoseconds, the largest duration we can represent is about
/// 427_000 years. Large enough for any interval we would use or calculate in
/// async-std.
fn duration_to_nanos(dur: Duration) -> Option<u64> {
dur.as_secs()
.checked_mul(1_000_000_000)
.and_then(|v| v.checked_add(u64::from(dur.subsec_nanos())))
}
fn next_interval(prev: Instant, now: Instant, interval: Duration) -> Instant {
let new = prev + interval;
if new > now {
return new;
}
let spent_ns = duration_to_nanos(now.duration_since(prev)).expect("interval should be expired");
let interval_ns =
duration_to_nanos(interval).expect("interval is less that 427 thousand years");
let mult = spent_ns / interval_ns + 1;
assert!(
mult < (1 << 32),
"can't skip more than 4 billion intervals of {:?} \
(trying to skip {})",
interval,
mult
);
prev + interval * (mult as u32)
}
#[cfg(test)]
mod test {
use super::next_interval;
use std::cmp::Ordering;
use std::time::{Duration, Instant};
struct Timeline(Instant);
impl Timeline {
fn new() -> Timeline {
Timeline(Instant::now())
}
fn at(&self, millis: u64) -> Instant {
self.0 + Duration::from_millis(millis)
}
fn at_ns(&self, sec: u64, nanos: u32) -> Instant {
self.0 + Duration::new(sec, nanos)
}
}
fn dur(millis: u64) -> Duration {
Duration::from_millis(millis)
}
// The math around Instant/Duration isn't 100% precise due to rounding
// errors, see #249 for more info
fn almost_eq(a: Instant, b: Instant) -> bool {
match a.cmp(&b) {
Ordering::Equal => true,
Ordering::Greater => a - b < Duration::from_millis(1),
Ordering::Less => b - a < Duration::from_millis(1),
}
}
#[test]
fn norm_next() {
let tm = Timeline::new();
assert!(almost_eq(
next_interval(tm.at(1), tm.at(2), dur(10)),
tm.at(11)
));
assert!(almost_eq(
next_interval(tm.at(7777), tm.at(7788), dur(100)),
tm.at(7877)
));
assert!(almost_eq(
next_interval(tm.at(1), tm.at(1000), dur(2100)),
tm.at(2101)
));
}
#[test]
fn fast_forward() {
let tm = Timeline::new();
assert!(almost_eq(
next_interval(tm.at(1), tm.at(1000), dur(10)),
tm.at(1001)
));
assert!(almost_eq(
next_interval(tm.at(7777), tm.at(8888), dur(100)),
tm.at(8977)
));
assert!(almost_eq(
next_interval(tm.at(1), tm.at(10000), dur(2100)),
tm.at(10501)
));
}
/// TODO: this test actually should be successful, but since we can't
/// multiply Duration on anything larger than u32 easily we decided
/// to allow it to fail for now
#[test]
#[should_panic(expected = "can't skip more than 4 billion intervals")]
fn large_skip() {
let tm = Timeline::new();
assert_eq!(
next_interval(tm.at_ns(0, 1), tm.at_ns(25, 0), Duration::new(0, 2)),
tm.at_ns(25, 1)
);
}
}

5
src/stream/stream/delay.rs
View file

@ -6,6 +6,7 @@ use pin_project_lite::pin_project;
use crate::stream::Stream;
use crate::task::{Context, Poll};
use crate::utils::Timer;
pin_project! {
#[doc(hidden)]
@ -14,7 +15,7 @@ pin_project! {
#[pin]
stream: S,
#[pin]
delay: futures_timer::Delay,
delay: Timer,
delay_done: bool,
}
}
@ -23,7 +24,7 @@ impl<S> Delay<S> {
pub(super) fn new(stream: S, dur: Duration) -> Self {
Delay {
stream,
delay: futures_timer::Delay::new(dur),
delay: Timer::after(dur),
delay_done: false,
}
}

10
src/stream/stream/throttle.rs
View file

@ -1,12 +1,12 @@
use std::future::Future;
use std::pin::Pin;
use std::time::{Duration, Instant};
use std::time::Duration;
use futures_timer::Delay;
use pin_project_lite::pin_project;
use crate::stream::Stream;
use crate::task::{Context, Poll};
use crate::utils::Timer;
pin_project! {
/// A stream that only yields one element once every `duration`.
@ -25,7 +25,7 @@ pin_project! {
#[pin]
blocked: bool,
#[pin]
delay: Delay,
delay: Timer,
}
}
@ -35,7 +35,7 @@ impl<S: Stream> Throttle<S> {
stream,
duration,
blocked: false,
delay: Delay::new(Duration::default()),
delay: Timer::after(Duration::default()),
}
}
}
@ -59,7 +59,7 @@ impl<S: Stream> Stream for Throttle<S> {
Poll::Ready(None) => Poll::Ready(None),
Poll::Ready(Some(v)) => {
*this.blocked = true;
this.delay.reset(Instant::now() + *this.duration);
let _ = std::mem::replace(&mut *this.delay, Timer::after(*this.duration));
Poll::Ready(Some(v))
}
}

6
src/stream/stream/timeout.rs
View file

@ -4,11 +4,11 @@ use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
use futures_timer::Delay;
use pin_project_lite::pin_project;
use crate::stream::Stream;
use crate::task::{Context, Poll};
use crate::utils::Timer;
pin_project! {
/// A stream with timeout time set
@ -17,13 +17,13 @@ pin_project! {
#[pin]
stream: S,
#[pin]
delay: Delay,
delay: Timer,
}
}
impl<S: Stream> Timeout<S> {
pub(crate) fn new(stream: S, dur: Duration) -> Self {
let delay = Delay::new(dur);
let delay = Timer::after(dur);
Self { stream, delay }
}

61
src/sync/barrier.rs
View file

@ -1,6 +1,4 @@
use broadcaster::BroadcastChannel;
use crate::sync::Mutex;
use crate::sync::{Condvar,Mutex};
/// A barrier enables multiple tasks to synchronize the beginning
/// of some computation.
@ -36,14 +34,13 @@ use crate::sync::Mutex;
#[derive(Debug)]
pub struct Barrier {
state: Mutex<BarrierState>,
wait: BroadcastChannel<(usize, usize)>,
n: usize,
cvar: Condvar,
num_tasks: usize,
}
// The inner state of a double barrier
#[derive(Debug)]
struct BarrierState {
waker: BroadcastChannel<(usize, usize)>,
count: usize,
generation_id: usize,
}
@ -81,25 +78,14 @@ impl Barrier {
///
/// let barrier = Barrier::new(10);
/// ```
pub fn new(mut n: usize) -> Barrier {
let waker = BroadcastChannel::new();
let wait = waker.clone();
if n == 0 {
// if n is 0, it's not clear what behavior the user wants.
// in std::sync::Barrier, an n of 0 exhibits the same behavior as n == 1, where every
// .wait() immediately unblocks, so we adopt that here as well.
n = 1;
}
pub fn new(n: usize) -> Barrier {
Barrier {
state: Mutex::new(BarrierState {
waker,
count: 0,
generation_id: 1,
}),
n,
wait,
cvar: Condvar::new(),
num_tasks: n,
}
}
@ -143,35 +129,20 @@ impl Barrier {
/// # });
/// ```
pub async fn wait(&self) -> BarrierWaitResult {
let mut lock = self.state.lock().await;
let local_gen = lock.generation_id;
let mut state = self.state.lock().await;
let local_gen = state.generation_id;
state.count += 1;
lock.count += 1;
if lock.count < self.n {
let mut wait = self.wait.clone();
let mut generation_id = lock.generation_id;
let mut count = lock.count;
drop(lock);
while local_gen == generation_id && count < self.n {
let (g, c) = wait.recv().await.expect("sender has not been closed");
generation_id = g;
count = c;
if state.count < self.num_tasks {
while local_gen == state.generation_id && state.count < self.num_tasks {
state = self.cvar.wait(state).await;
}
BarrierWaitResult(false)
} else {
lock.count = 0;
lock.generation_id = lock.generation_id.wrapping_add(1);
lock.waker
.send(&(lock.generation_id, lock.count))
.await
.expect("there should be at least one receiver");
state.count = 0;
state.generation_id = state.generation_id.wrapping_add(1);
self.cvar.notify_all();
BarrierWaitResult(true)
}
}
@ -202,7 +173,7 @@ impl BarrierWaitResult {
}
}
#[cfg(test)]
#[cfg(all(test, not(target_os = "unknown")))]
mod test {
use futures::channel::mpsc::unbounded;
use futures::sink::SinkExt;

417
src/sync/condvar.rs Normal file
View file

@ -0,0 +1,417 @@
use std::fmt;
use std::pin::Pin;
use std::time::Duration;
use super::mutex::{guard_lock, MutexGuard};
use crate::future::{timeout, Future};
use crate::sync::WakerSet;
use crate::task::{Context, Poll};
#[derive(Debug, PartialEq, Eq, Copy, Clone)]
pub struct WaitTimeoutResult(bool);
/// A type indicating whether a timed wait on a condition variable returned due to a time out or
/// not
impl WaitTimeoutResult {
/// Returns `true` if the wait was known to have timed out.
pub fn timed_out(self) -> bool {
self.0
}
}
/// A Condition Variable
///
/// This type is an async version of [`std::sync::Mutex`].
///
/// [`std::sync::Condvar`]: https://doc.rust-lang.org/std/sync/struct.Condvar.html
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use std::sync::Arc;
///
/// use async_std::sync::{Mutex, Condvar};
/// use async_std::task;
///
/// let pair = Arc::new((Mutex::new(false), Condvar::new()));
/// let pair2 = pair.clone();
///
/// // Inside of our lock, spawn a new thread, and then wait for it to start.
/// task::spawn(async move {
/// let (lock, cvar) = &*pair2;
/// let mut started = lock.lock().await;
/// *started = true;
/// // We notify the condvar that the value has changed.
/// cvar.notify_one();
/// });
///
/// // Wait for the thread to start up.
/// let (lock, cvar) = &*pair;
/// let mut started = lock.lock().await;
/// while !*started {
/// started = cvar.wait(started).await;
/// }
///
/// # })
/// ```
pub struct Condvar {
wakers: WakerSet,
}
unsafe impl Send for Condvar {}
unsafe impl Sync for Condvar {}
impl Default for Condvar {
fn default() -> Self {
Condvar::new()
}
}
impl Condvar {
/// Creates a new condition variable
///
/// # Examples
///
/// ```
/// use async_std::sync::Condvar;
///
/// let cvar = Condvar::new();
/// ```
pub fn new() -> Self {
Condvar {
wakers: WakerSet::new(),
}
}
/// Blocks the current task until this condition variable receives a notification.
///
/// Unlike the std equivalent, this does not check that a single mutex is used at runtime.
/// However, as a best practice avoid using with multiple mutexes.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// use std::sync::Arc;
///
/// use async_std::sync::{Mutex, Condvar};
/// use async_std::task;
///
/// let pair = Arc::new((Mutex::new(false), Condvar::new()));
/// let pair2 = pair.clone();
///
/// task::spawn(async move {
/// let (lock, cvar) = &*pair2;
/// let mut started = lock.lock().await;
/// *started = true;
/// // We notify the condvar that the value has changed.
/// cvar.notify_one();
/// });
///
/// // Wait for the thread to start up.
/// let (lock, cvar) = &*pair;
/// let mut started = lock.lock().await;
/// while !*started {
/// started = cvar.wait(started).await;
/// }
/// # })
/// ```
#[allow(clippy::needless_lifetimes)]
pub async fn wait<'a, T>(&self, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> {
let mutex = guard_lock(&guard);
self.await_notify(guard).await;
mutex.lock().await
}
fn await_notify<'a, T>(&self, guard: MutexGuard<'a, T>) -> AwaitNotify<'_, 'a, T> {
AwaitNotify {
cond: self,
guard: Some(guard),
key: None,
}
}
/// Blocks the current taks until this condition variable receives a notification and the
/// required condition is met. Spurious wakeups are ignored and this function will only
/// return once the condition has been met.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use std::sync::Arc;
///
/// use async_std::sync::{Mutex, Condvar};
/// use async_std::task;
///
/// let pair = Arc::new((Mutex::new(false), Condvar::new()));
/// let pair2 = pair.clone();
///
/// task::spawn(async move {
/// let (lock, cvar) = &*pair2;
/// let mut started = lock.lock().await;
/// *started = true;
/// // We notify the condvar that the value has changed.
/// cvar.notify_one();
/// });
///
/// // Wait for the thread to start up.
/// let (lock, cvar) = &*pair;
/// // As long as the value inside the `Mutex<bool>` is `false`, we wait.
/// let _guard = cvar.wait_until(lock.lock().await, |started| { *started }).await;
/// #
/// # })
/// ```
#[allow(clippy::needless_lifetimes)]
pub async fn wait_until<'a, T, F>(
&self,
mut guard: MutexGuard<'a, T>,
mut condition: F,
) -> MutexGuard<'a, T>
where
F: FnMut(&mut T) -> bool,
{
while !condition(&mut *guard) {
guard = self.wait(guard).await;
}
guard
}
/// Waits on this condition variable for a notification, timing out after a specified duration.
///
/// For these reasons `Condvar::wait_timeout_until` is recommended in most cases.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use std::sync::Arc;
/// use std::time::Duration;
///
/// use async_std::sync::{Mutex, Condvar};
/// use async_std::task;
///
/// let pair = Arc::new((Mutex::new(false), Condvar::new()));
/// let pair2 = pair.clone();
///
/// task::spawn(async move {
/// let (lock, cvar) = &*pair2;
/// let mut started = lock.lock().await;
/// *started = true;
/// // We notify the condvar that the value has changed.
/// cvar.notify_one();
/// });
///
/// // wait for the thread to start up
/// let (lock, cvar) = &*pair;
/// let mut started = lock.lock().await;
/// loop {
/// let result = cvar.wait_timeout(started, Duration::from_millis(10)).await;
/// started = result.0;
/// if *started == true {
/// // We received the notification and the value has been updated, we can leave.
/// break
/// }
/// }
/// #
/// # })
/// ```
#[allow(clippy::needless_lifetimes)]
pub async fn wait_timeout<'a, T>(
&self,
guard: MutexGuard<'a, T>,
dur: Duration,
) -> (MutexGuard<'a, T>, WaitTimeoutResult) {
let mutex = guard_lock(&guard);
match timeout(dur, self.wait(guard)).await {
Ok(guard) => (guard, WaitTimeoutResult(false)),
Err(_) => (mutex.lock().await, WaitTimeoutResult(true)),
}
}
/// Waits on this condition variable for a notification, timing out after a specified duration.
/// Spurious wakes will not cause this function to return.
///
/// # Examples
/// ```
/// # async_std::task::block_on(async {
/// use std::sync::Arc;
/// use std::time::Duration;
///
/// use async_std::sync::{Mutex, Condvar};
/// use async_std::task;
///
/// let pair = Arc::new((Mutex::new(false), Condvar::new()));
/// let pair2 = pair.clone();
///
/// task::spawn(async move {
/// let (lock, cvar) = &*pair2;
/// let mut started = lock.lock().await;
/// *started = true;
/// // We notify the condvar that the value has changed.
/// cvar.notify_one();
/// });
///
/// // wait for the thread to start up
/// let (lock, cvar) = &*pair;
/// let result = cvar.wait_timeout_until(
/// lock.lock().await,
/// Duration::from_millis(100),
/// |&mut started| started,
/// ).await;
/// if result.1.timed_out() {
/// // timed-out without the condition ever evaluating to true.
/// }
/// // access the locked mutex via result.0
/// # });
/// ```
#[allow(clippy::needless_lifetimes)]
pub async fn wait_timeout_until<'a, T, F>(
&self,
guard: MutexGuard<'a, T>,
dur: Duration,
condition: F,
) -> (MutexGuard<'a, T>, WaitTimeoutResult)
where
F: FnMut(&mut T) -> bool,
{
let mutex = guard_lock(&guard);
match timeout(dur, self.wait_until(guard, condition)).await {
Ok(guard) => (guard, WaitTimeoutResult(false)),
Err(_) => (mutex.lock().await, WaitTimeoutResult(true)),
}
}
/// Wakes up one blocked task on this condvar.
///
/// # Examples
///
/// ```
/// # fn main() { async_std::task::block_on(async {
/// use std::sync::Arc;
///
/// use async_std::sync::{Mutex, Condvar};
/// use async_std::task;
///
/// let pair = Arc::new((Mutex::new(false), Condvar::new()));
/// let pair2 = pair.clone();
///
/// task::spawn(async move {
/// let (lock, cvar) = &*pair2;
/// let mut started = lock.lock().await;
/// *started = true;
/// // We notify the condvar that the value has changed.
/// cvar.notify_one();
/// });
///
/// // Wait for the thread to start up.
/// let (lock, cvar) = &*pair;
/// let mut started = lock.lock().await;
/// while !*started {
/// started = cvar.wait(started).await;
/// }
/// # }) }
/// ```
pub fn notify_one(&self) {
self.wakers.notify_one();
}
/// Wakes up all blocked tasks on this condvar.
///
/// # Examples
/// ```
/// # fn main() { async_std::task::block_on(async {
/// #
/// use std::sync::Arc;
///
/// use async_std::sync::{Mutex, Condvar};
/// use async_std::task;
///
/// let pair = Arc::new((Mutex::new(false), Condvar::new()));
/// let pair2 = pair.clone();
///
/// task::spawn(async move {
/// let (lock, cvar) = &*pair2;
/// let mut started = lock.lock().await;
/// *started = true;
/// // We notify the condvar that the value has changed.
/// cvar.notify_all();
/// });
///
/// // Wait for the thread to start up.
/// let (lock, cvar) = &*pair;
/// let mut started = lock.lock().await;
/// // As long as the value inside the `Mutex<bool>` is `false`, we wait.
/// while !*started {
/// started = cvar.wait(started).await;
/// }
/// #
/// # }) }
/// ```
pub fn notify_all(&self) {
self.wakers.notify_all();
}
}
impl fmt::Debug for Condvar {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.pad("Condvar { .. }")
}
}
/// A future that waits for another task to notify the condition variable.
///
/// This is an internal future that `wait` and `wait_until` await on.
struct AwaitNotify<'a, 'b, T> {
/// The condition variable that we are waiting on
cond: &'a Condvar,
/// The lock used with `cond`.
/// This will be released the first time the future is polled,
/// after registering the context to be notified.
guard: Option<MutexGuard<'b, T>>,
/// A key into the conditions variable's `WakerSet`.
/// This is set to the index of the `Waker` for the context each time
/// the future is polled and not completed.
key: Option<usize>,
}
impl<'a, 'b, T> Future for AwaitNotify<'a, 'b, T> {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.guard.take() {
Some(_) => {
self.key = Some(self.cond.wakers.insert(cx));
// the guard is dropped when we return, which frees the lock
Poll::Pending
}
None => {
if let Some(key) = self.key {
if self.cond.wakers.remove_if_notified(key, cx) {
self.key = None;
Poll::Ready(())
} else {
Poll::Pending
}
} else {
// This should only happen if it is polled twice after receiving a notification
Poll::Ready(())
}
}
}
}
}
impl<'a, 'b, T> Drop for AwaitNotify<'a, 'b, T> {
fn drop(&mut self) {
if let Some(key) = self.key {
self.cond.wakers.cancel(key);
}
}
}

2
src/sync/mod.rs
View file

@ -185,8 +185,10 @@ mod rwlock;
cfg_unstable! {
pub use barrier::{Barrier, BarrierWaitResult};
pub use channel::{channel, Sender, Receiver, RecvError, TryRecvError, TrySendError};
pub use condvar::Condvar;
mod barrier;
mod condvar;
mod channel;
}

5
src/sync/mutex.rs
View file

@ -287,3 +287,8 @@ impl<T: ?Sized> DerefMut for MutexGuard<'_, T> {
unsafe { &mut *self.0.value.get() }
}
}
#[cfg(feature = "unstable")]
pub fn guard_lock<'a, T>(guard: &MutexGuard<'a, T>) -> &'a Mutex<T> {
guard.0
}

22
src/sync/waker_set.rs
View file

@ -80,6 +80,28 @@ impl WakerSet {
}
}
/// If the waker for this key is still waiting for a notification, then update
/// the waker for the entry, and return false. If the waker has been notified,
/// treat the entry as completed and return true.
#[cfg(feature = "unstable")]
pub fn remove_if_notified(&self, key: usize, cx: &Context<'_>) -> bool {
let mut inner = self.lock();
match &mut inner.entries[key] {
None => {
inner.entries.remove(key);
true
}
Some(w) => {
// We were never woken, so update instead
if !w.will_wake(cx.waker()) {
*w = cx.waker().clone();
}
false
}
}
}
/// Removes the waker of a cancelled operation.
///
/// Returns `true` if another blocked operation from the set was notified.

111
src/task/block_on.rs
View file

@ -1,14 +1,6 @@
use std::cell::Cell;
use std::future::Future;
use std::mem::{self, ManuallyDrop};
use std::sync::Arc;
use std::task::{RawWaker, RawWakerVTable};
use crossbeam_utils::sync::Parker;
use kv_log_macro::trace;
use log::log_enabled;
use crate::task::{Context, Poll, Task, Waker};
use crate::task::Builder;
/// Spawns a task and blocks the current thread on its result.
///
@ -33,105 +25,20 @@ use crate::task::{Context, Poll, Task, Waker};
/// })
/// }
/// ```
#[cfg(not(target_os = "unknown"))]
pub fn block_on<F, T>(future: F) -> T
where
F: Future<Output = T>,
{
// Create a new task handle.
let task = Task::new(None);
// Log this `block_on` operation.
if log_enabled!(log::Level::Trace) {
trace!("block_on", {
task_id: task.id().0,
parent_task_id: Task::get_current(|t| t.id().0).unwrap_or(0),
});
}
let future = async move {
// Drop task-locals on exit.
defer! {
Task::get_current(|t| unsafe { t.drop_locals() });
}
// Log completion on exit.
defer! {
if log_enabled!(log::Level::Trace) {
Task::get_current(|t| {
trace!("completed", {
task_id: t.id().0,
});
});
}
}
future.await
};
// Run the future as a task.
unsafe { Task::set_current(&task, || run(future)) }
Builder::new().blocking(future)
}
/// Blocks the current thread on a future's result.
fn run<F, T>(future: F) -> T
/// Spawns a task and waits for it to finish.
#[cfg(target_os = "unknown")]
pub fn block_on<F, T>(future: F)
where
F: Future<Output = T>,
F: Future<Output = T> + 'static,
T: 'static,
{
thread_local! {
// May hold a pre-allocated parker that can be reused for efficiency.
//
// Note that each invocation of `block` needs its own parker. In particular, if `block`
// recursively calls itself, we must make sure that each recursive call uses a distinct
// parker instance.
static CACHE: Cell<Option<Arc<Parker>>> = Cell::new(None);
}
// Virtual table for wakers based on `Arc<Parker>`.
static VTABLE: RawWakerVTable = {
unsafe fn clone_raw(ptr: *const ()) -> RawWaker {
let arc = ManuallyDrop::new(Arc::from_raw(ptr as *const Parker));
#[allow(clippy::redundant_clone)]
mem::forget(arc.clone());
RawWaker::new(ptr, &VTABLE)
}
unsafe fn wake_raw(ptr: *const ()) {
let arc = Arc::from_raw(ptr as *const Parker);
arc.unparker().unpark();
}
unsafe fn wake_by_ref_raw(ptr: *const ()) {
let arc = ManuallyDrop::new(Arc::from_raw(ptr as *const Parker));
arc.unparker().unpark();
}
unsafe fn drop_raw(ptr: *const ()) {
drop(Arc::from_raw(ptr as *const Parker))
}
RawWakerVTable::new(clone_raw, wake_raw, wake_by_ref_raw, drop_raw)
};
// Pin the future on the stack.
pin_utils::pin_mut!(future);
CACHE.with(|cache| {
// Reuse a cached parker or create a new one for this invocation of `block`.
let arc_parker: Arc<Parker> = cache.take().unwrap_or_else(|| Arc::new(Parker::new()));
let ptr = (&*arc_parker as *const Parker) as *const ();
// Create a waker and task context.
let waker = unsafe { ManuallyDrop::new(Waker::from_raw(RawWaker::new(ptr, &VTABLE))) };
let cx = &mut Context::from_waker(&waker);
loop {
if let Poll::Ready(t) = future.as_mut().poll(cx) {
// Save the parker for the next invocation of `block`.
cache.set(Some(arc_parker));
return t;
}
arc_parker.park();
}
})
Builder::new().local(future).unwrap();
}

176
src/task/builder.rs
View file

@ -1,11 +1,12 @@
use kv_log_macro::trace;
use log::log_enabled;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use pin_project_lite::pin_project;
use crate::io;
use crate::task::executor;
use crate::task::{JoinHandle, Task};
use crate::utils::abort_on_panic;
use crate::task::{JoinHandle, Task, TaskLocalsWrapper};
/// Task builder that configures the settings of a new task.
#[derive(Debug, Default)]
@ -27,58 +28,151 @@ impl Builder {
self
}
fn build<F, T>(self, future: F) -> SupportTaskLocals<F>
where
F: Future<Output = T>,
{
let name = self.name.map(Arc::new);
// Create a new task handle.
let task = Task::new(name);
#[cfg(not(target_os = "unknown"))]
once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
let tag = TaskLocalsWrapper::new(task.clone());
SupportTaskLocals { tag, future }
}
/// Spawns a task with the configured settings.
#[cfg(not(target_os = "unknown"))]
pub fn spawn<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
where
F: Future<Output = T> + Send + 'static,
T: Send + 'static,
{
// Create a new task handle.
let task = Task::new(self.name);
let wrapped = self.build(future);
// Log this `spawn` operation.
if log_enabled!(log::Level::Trace) {
trace!("spawn", {
task_id: task.id().0,
parent_task_id: Task::get_current(|t| t.id().0).unwrap_or(0),
});
}
kv_log_macro::trace!("spawn", {
task_id: wrapped.tag.id().0,
parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
});
let future = async move {
// Drop task-locals on exit.
defer! {
Task::get_current(|t| unsafe { t.drop_locals() });
}
let task = wrapped.tag.task().clone();
let smol_task = smol::Task::spawn(wrapped).into();
// Log completion on exit.
defer! {
if log_enabled!(log::Level::Trace) {
Task::get_current(|t| {
trace!("completed", {
task_id: t.id().0,
});
});
}
}
Ok(JoinHandle::new(smol_task, task))
}
future.await
};
/// Spawns a task locally with the configured settings.
#[cfg(all(not(target_os = "unknown"), feature = "unstable"))]
pub fn local<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
where
F: Future<Output = T> + 'static,
T: 'static,
{
let wrapped = self.build(future);
let schedule = move |t| executor::schedule(Runnable(t));
let (task, handle) = async_task::spawn(future, schedule, task);
task.schedule();
Ok(JoinHandle::new(handle))
kv_log_macro::trace!("spawn_local", {
task_id: wrapped.tag.id().0,
parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
});
let task = wrapped.tag.task().clone();
let smol_task = smol::Task::local(wrapped).into();
Ok(JoinHandle::new(smol_task, task))
}
/// Spawns a task locally with the configured settings.
#[cfg(all(target_arch = "wasm32", feature = "unstable"))]
pub fn local<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
where
F: Future<Output = T> + 'static,
T: 'static,
{
use futures_channel::oneshot::channel;
let (sender, receiver) = channel();
let wrapped = self.build(async move {
let res = future.await;
let _ = sender.send(res);
});
kv_log_macro::trace!("spawn_local", {
task_id: wrapped.tag.id().0,
parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
});
let task = wrapped.tag.task().clone();
wasm_bindgen_futures::spawn_local(wrapped);
Ok(JoinHandle::new(receiver, task))
}
/// Spawns a task locally with the configured settings.
#[cfg(all(target_arch = "wasm32", not(feature = "unstable")))]
pub(crate) fn local<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
where
F: Future<Output = T> + 'static,
T: 'static,
{
use futures_channel::oneshot::channel;
let (sender, receiver) = channel();
let wrapped = self.build(async move {
let res = future.await;
let _ = sender.send(res);
});
kv_log_macro::trace!("spawn_local", {
task_id: wrapped.tag.id().0,
parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
});
let task = wrapped.tag.task().clone();
wasm_bindgen_futures::spawn_local(wrapped);
Ok(JoinHandle::new(receiver, task))
}
/// Spawns a task with the configured settings, blocking on its execution.
#[cfg(not(target_os = "unknown"))]
pub fn blocking<F, T>(self, future: F) -> T
where
F: Future<Output = T>,
{
let wrapped = self.build(future);
// Log this `block_on` operation.
kv_log_macro::trace!("block_on", {
task_id: wrapped.tag.id().0,
parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
});
// Run the future as a task.
unsafe { TaskLocalsWrapper::set_current(&wrapped.tag, || smol::run(wrapped)) }
}
}
/// A runnable task.
pub(crate) struct Runnable(async_task::Task<Task>);
pin_project! {
/// Wrapper to add support for task locals.
struct SupportTaskLocals<F> {
tag: TaskLocalsWrapper,
#[pin]
future: F,
}
}
impl Runnable {
/// Runs the task by polling its future once.
pub fn run(self) {
impl<F: Future> Future for SupportTaskLocals<F> {
type Output = F::Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
unsafe {
Task::set_current(self.0.tag(), || abort_on_panic(|| self.0.run()));
TaskLocalsWrapper::set_current(&self.tag, || {
let this = self.project();
this.future.poll(cx)
})
}
}
}

4
src/task/current.rs
View file

@ -1,4 +1,4 @@
use crate::task::Task;
use crate::task::{Task, TaskLocalsWrapper};
/// Returns a handle to the current task.
///
@ -23,6 +23,6 @@ use crate::task::Task;
/// # })
/// ```
pub fn current() -> Task {
Task::get_current(|t| t.clone())
TaskLocalsWrapper::get_current(|t| t.task().clone())
.expect("`task::current()` called outside the context of a task")
}

13
src/task/executor/mod.rs
View file

@ -1,13 +0,0 @@
//! Task executor.
//!
//! API bindings between `crate::task` and this module are very simple:
//!
//! * The only export is the `schedule` function.
//! * The only import is the `crate::task::Runnable` type.
pub(crate) use pool::schedule;
use sleepers::Sleepers;
mod pool;
mod sleepers;

179
src/task/executor/pool.rs
View file

@ -1,179 +0,0 @@
use std::cell::Cell;
use std::iter;
use std::thread;
use std::time::Duration;
use crossbeam_deque::{Injector, Stealer, Worker};
use once_cell::sync::Lazy;
use once_cell::unsync::OnceCell;
use crate::task::executor::Sleepers;
use crate::task::Runnable;
use crate::utils::{abort_on_panic, random};
/// The state of an executor.
struct Pool {
/// The global queue of tasks.
injector: Injector<Runnable>,
/// Handles to local queues for stealing work from worker threads.
stealers: Vec<Stealer<Runnable>>,
/// Used for putting idle workers to sleep and notifying them when new tasks come in.
sleepers: Sleepers,
}
/// Global executor that runs spawned tasks.
static POOL: Lazy<Pool> = Lazy::new(|| {
let num_threads = num_cpus::get().max(1);
let mut stealers = Vec::new();
// Spawn worker threads.
for _ in 0..num_threads {
let worker = Worker::new_fifo();
stealers.push(worker.stealer());
let proc = Processor {
worker,
slot: Cell::new(None),
slot_runs: Cell::new(0),
};
thread::Builder::new()
.name("async-std/executor".to_string())
.spawn(|| {
let _ = PROCESSOR.with(|p| p.set(proc));
abort_on_panic(main_loop);
})
.expect("cannot start a thread driving tasks");
}
Pool {
injector: Injector::new(),
stealers,
sleepers: Sleepers::new(),
}
});
/// The state of a worker thread.
struct Processor {
/// The local task queue.
worker: Worker<Runnable>,
/// Contains the next task to run as an optimization that skips queues.
slot: Cell<Option<Runnable>>,
/// How many times in a row tasks have been taked from the slot rather than the queue.
slot_runs: Cell<u32>,
}
thread_local! {
/// Worker thread state.
static PROCESSOR: OnceCell<Processor> = OnceCell::new();
}
/// Schedules a new runnable task for execution.
pub(crate) fn schedule(task: Runnable) {
PROCESSOR.with(|proc| {
// If the current thread is a worker thread, store it into its task slot or push it into
// its local task queue. Otherwise, push it into the global task queue.
match proc.get() {
Some(proc) => {
// Replace the task in the slot.
if let Some(task) = proc.slot.replace(Some(task)) {
// If the slot already contained a task, push it into the local task queue.
proc.worker.push(task);
POOL.sleepers.notify_one();
}
}
None => {
POOL.injector.push(task);
POOL.sleepers.notify_one();
}
}
})
}
/// Main loop running a worker thread.
fn main_loop() {
/// Number of yields when no runnable task is found.
const YIELDS: u32 = 3;
/// Number of short sleeps when no runnable task in found.
const SLEEPS: u32 = 1;
// The number of times the thread didn't find work in a row.
let mut fails = 0;
loop {
// Try to find a runnable task.
match find_runnable() {
Some(task) => {
fails = 0;
// Run the found task.
task.run();
}
None => {
fails += 1;
// Yield the current thread or put it to sleep.
if fails <= YIELDS {
thread::yield_now();
} else if fails <= YIELDS + SLEEPS {
thread::sleep(Duration::from_micros(10));
} else {
POOL.sleepers.wait();
fails = 0;
}
}
}
}
}
/// Find the next runnable task.
fn find_runnable() -> Option<Runnable> {
/// Maximum number of times the slot can be used in a row.
const SLOT_LIMIT: u32 = 16;
PROCESSOR.with(|proc| {
let proc = proc.get().unwrap();
// Try taking a task from the slot.
let runs = proc.slot_runs.get();
if runs < SLOT_LIMIT {
if let Some(task) = proc.slot.take() {
proc.slot_runs.set(runs + 1);
return Some(task);
}
}
proc.slot_runs.set(0);
// Pop a task from the local queue, if not empty.
proc.worker.pop().or_else(|| {
// Otherwise, we need to look for a task elsewhere.
iter::repeat_with(|| {
// Try stealing a batch of tasks from the global queue.
POOL.injector
.steal_batch_and_pop(&proc.worker)
// Or try stealing a batch of tasks from one of the other threads.
.or_else(|| {
// First, pick a random starting point in the list of local queues.
let len = POOL.stealers.len();
let start = random(len as u32) as usize;
// Try stealing a batch of tasks from each local queue starting from the
// chosen point.
let (l, r) = POOL.stealers.split_at(start);
let stealers = r.iter().chain(l.iter());
stealers
.map(|s| s.steal_batch_and_pop(&proc.worker))
.collect()
})
})
// Loop while no task was stolen and any steal operation needs to be retried.
.find(|s| !s.is_retry())
// Extract the stolen task, if there is one.
.and_then(|s| s.success())
})
})
}

52
src/task/executor/sleepers.rs
View file

@ -1,52 +0,0 @@
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Condvar, Mutex};
/// The place where worker threads go to sleep.
///
/// Similar to how thread parking works, if a notification comes up while no threads are sleeping,
/// the next thread that attempts to go to sleep will pick up the notification immediately.
pub struct Sleepers {
/// How many threads are currently a sleep.
sleep: Mutex<usize>,
/// A condvar for notifying sleeping threads.
wake: Condvar,
/// Set to `true` if a notification came up while nobody was sleeping.
notified: AtomicBool,
}
impl Sleepers {
/// Creates a new `Sleepers`.
pub fn new() -> Sleepers {
Sleepers {
sleep: Mutex::new(0),
wake: Condvar::new(),
notified: AtomicBool::new(false),
}
}
/// Puts the current thread to sleep.
pub fn wait(&self) {
let mut sleep = self.sleep.lock().unwrap();
if !self.notified.swap(false, Ordering::SeqCst) {
*sleep += 1;
let _ = self.wake.wait(sleep).unwrap();
}
}
/// Notifies one thread.
pub fn notify_one(&self) {
if !self.notified.load(Ordering::SeqCst) {
let mut sleep = self.sleep.lock().unwrap();
if *sleep > 0 {
*sleep -= 1;
self.wake.notify_one();
} else {
self.notified.store(true, Ordering::SeqCst);
}
}
}
}

43
src/task/join_handle.rs
View file

@ -12,15 +12,23 @@ use crate::task::{Context, Poll, Task};
///
/// [spawned]: fn.spawn.html
#[derive(Debug)]
pub struct JoinHandle<T>(async_task::JoinHandle<T, Task>);
pub struct JoinHandle<T> {
handle: Option<InnerHandle<T>>,
task: Task,
}
unsafe impl<T> Send for JoinHandle<T> {}
unsafe impl<T> Sync for JoinHandle<T> {}
#[cfg(not(target_os = "unknown"))]
type InnerHandle<T> = async_task::JoinHandle<T, ()>;
#[cfg(target_arch = "wasm32")]
type InnerHandle<T> = futures_channel::oneshot::Receiver<T>;
impl<T> JoinHandle<T> {
/// Creates a new `JoinHandle`.
pub(crate) fn new(inner: async_task::JoinHandle<T, Task>) -> JoinHandle<T> {
JoinHandle(inner)
pub(crate) fn new(inner: InnerHandle<T>, task: Task) -> JoinHandle<T> {
JoinHandle {
handle: Some(inner),
task,
}
}
/// Returns a handle to the underlying task.
@ -39,7 +47,23 @@ impl<T> JoinHandle<T> {
/// #
/// # })
pub fn task(&self) -> &Task {
self.0.tag()
&self.task
}
/// Cancel this task.
#[cfg(not(target_os = "unknown"))]
pub async fn cancel(mut self) -> Option<T> {
let handle = self.handle.take().unwrap();
handle.cancel();
handle.await
}
/// Cancel this task.
#[cfg(target_arch = "wasm32")]
pub async fn cancel(mut self) -> Option<T> {
let mut handle = self.handle.take().unwrap();
handle.close();
handle.await.ok()
}
}
@ -47,10 +71,11 @@ impl<T> Future for JoinHandle<T> {
type Output = T;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match Pin::new(&mut self.0).poll(cx) {
match Pin::new(&mut self.handle.as_mut().unwrap()).poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(None) => panic!("cannot await the result of a panicked task"),
Poll::Ready(Some(val)) => Poll::Ready(val),
Poll::Ready(output) => {
Poll::Ready(output.expect("cannot await the result of a panicked task"))
}
}
}
}

17
src/task/mod.rs
View file

@ -138,26 +138,37 @@ cfg_default! {
pub use task_id::TaskId;
pub use join_handle::JoinHandle;
pub use sleep::sleep;
#[cfg(not(target_os = "unknown"))]
pub use spawn::spawn;
pub use task_local::{AccessError, LocalKey};
use builder::Runnable;
use task_local::LocalsMap;
pub(crate) use task_local::LocalsMap;
pub(crate) use task_locals_wrapper::TaskLocalsWrapper;
mod block_on;
mod builder;
mod current;
mod executor;
mod join_handle;
mod sleep;
#[cfg(not(target_os = "unknown"))]
mod spawn;
#[cfg(not(target_os = "unknown"))]
mod spawn_blocking;
mod task;
mod task_id;
mod task_local;
mod task_locals_wrapper;
#[cfg(not(target_os = "unknown"))]
#[cfg(any(feature = "unstable", test))]
pub use spawn_blocking::spawn_blocking;
#[cfg(not(target_os = "unknown"))]
#[cfg(not(any(feature = "unstable", test)))]
pub(crate) use spawn_blocking::spawn_blocking;
}
cfg_unstable! {
pub use spawn_local::spawn_local;
mod spawn_local;
}

91
src/task/spawn_blocking.rs
View file

@ -1,12 +1,4 @@
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use std::time::Duration;
use crossbeam_channel::{unbounded, Receiver, Sender};
use once_cell::sync::Lazy;
use crate::task::{JoinHandle, Task};
use crate::utils::abort_on_panic;
/// Spawns a blocking task.
///
@ -31,7 +23,8 @@ use crate::utils::abort_on_panic;
///
/// task::spawn_blocking(|| {
/// println!("long-running task here");
/// }).await;
/// })
/// .await;
/// #
/// # })
/// ```
@ -42,80 +35,8 @@ where
F: FnOnce() -> T + Send + 'static,
T: Send + 'static,
{
let schedule = |task| POOL.sender.send(task).unwrap();
let (task, handle) = async_task::spawn(async { f() }, schedule, Task::new(None));
task.schedule();
JoinHandle::new(handle)
}
type Runnable = async_task::Task<Task>;
/// The number of sleeping worker threads.
static SLEEPING: AtomicUsize = AtomicUsize::new(0);
struct Pool {
sender: Sender<Runnable>,
receiver: Receiver<Runnable>,
}
static POOL: Lazy<Pool> = Lazy::new(|| {
// Start a single worker thread waiting for the first task.
start_thread();
let (sender, receiver) = unbounded();
Pool { sender, receiver }
});
fn start_thread() {
SLEEPING.fetch_add(1, Ordering::SeqCst);
let timeout = Duration::from_secs(1);
thread::Builder::new()
.name("async-std/blocking".to_string())
.spawn(move || {
loop {
let mut task = match POOL.receiver.recv_timeout(timeout) {
Ok(task) => task,
Err(_) => {
// Check whether this is the last sleeping thread.
if SLEEPING.fetch_sub(1, Ordering::SeqCst) == 1 {
// If so, then restart the thread to make sure there is always at least
// one sleeping thread.
if SLEEPING.compare_and_swap(0, 1, Ordering::SeqCst) == 0 {
continue;
}
}
// Stop the thread.
return;
}
};
// If there are no sleeping threads, then start one to make sure there is always at
// least one sleeping thread.
if SLEEPING.fetch_sub(1, Ordering::SeqCst) == 1 {
start_thread();
}
loop {
// Run the task.
abort_on_panic(|| task.run());
// Try taking another task if there are any available.
task = match POOL.receiver.try_recv() {
Ok(task) => task,
Err(_) => break,
};
}
// If there is at least one sleeping thread, stop this thread instead of putting it
// to sleep.
if SLEEPING.load(Ordering::SeqCst) > 0 {
return;
}
SLEEPING.fetch_add(1, Ordering::SeqCst);
}
})
.expect("cannot start a blocking thread");
once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
let handle = smol::Task::blocking(async move { f() }).into();
JoinHandle::new(handle, Task::new(None))
}

28
src/task/spawn_local.rs Normal file
View file

@ -0,0 +1,28 @@
use std::future::Future;
use crate::task::{Builder, JoinHandle};
/// Spawns a task onto the thread-local executor.
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// #
/// use async_std::task;
///
/// let handle = task::spawn_local(async {
/// 1 + 2
/// });
///
/// assert_eq!(handle.await, 3);
/// #
/// # })
/// ```
pub fn spawn_local<F, T>(future: F) -> JoinHandle<T>
where
F: Future<Output = T> + 'static,
T: 'static,
{
Builder::new().local(future).expect("cannot spawn task")
}

152
src/task/task.rs
View file

@ -1,74 +1,32 @@
use std::cell::Cell;
use std::fmt;
use std::mem::ManuallyDrop;
use std::ptr;
use std::sync::atomic::{AtomicPtr, Ordering};
use std::sync::Arc;
use crate::task::{LocalsMap, TaskId};
use crate::utils::abort_on_panic;
use crate::task::TaskId;
thread_local! {
/// A pointer to the currently running task.
static CURRENT: Cell<*const Task> = Cell::new(ptr::null_mut());
}
/// The inner representation of a task handle.
struct Inner {
/// A handle to a task.
#[derive(Clone)]
pub struct Task {
/// The task ID.
id: TaskId,
/// The optional task name.
name: Option<Box<str>>,
/// The map holding task-local values.
locals: LocalsMap,
name: Option<Arc<String>>,
}
impl Inner {
#[inline]
fn new(name: Option<String>) -> Inner {
Inner {
id: TaskId::generate(),
name: name.map(String::into_boxed_str),
locals: LocalsMap::new(),
}
}
}
/// A handle to a task.
pub struct Task {
/// The inner representation.
///
/// This pointer is lazily initialized on first use. In most cases, the inner representation is
/// never touched and therefore we don't allocate it unless it's really needed.
inner: AtomicPtr<Inner>,
}
unsafe impl Send for Task {}
unsafe impl Sync for Task {}
impl Task {
/// Creates a new task handle.
///
/// If the task is unnamed, the inner representation of the task will be lazily allocated on
/// demand.
#[inline]
pub(crate) fn new(name: Option<String>) -> Task {
let inner = match name {
None => AtomicPtr::default(),
Some(name) => {
let raw = Arc::into_raw(Arc::new(Inner::new(Some(name))));
AtomicPtr::new(raw as *mut Inner)
}
};
Task { inner }
pub(crate) fn new(name: Option<Arc<String>>) -> Task {
Task {
id: TaskId::generate(),
name,
}
}
/// Gets the task's unique identifier.
#[inline]
pub fn id(&self) -> TaskId {
self.inner().id
self.id
}
/// Returns the name of this task.
@ -77,93 +35,7 @@ impl Task {
///
/// [`Builder::name`]: struct.Builder.html#method.name
pub fn name(&self) -> Option<&str> {
self.inner().name.as_ref().map(|s| &**s)
}
/// Returns the map holding task-local values.
pub(crate) fn locals(&self) -> &LocalsMap {
&self.inner().locals
}
/// Drops all task-local values.
///
/// This method is only safe to call at the end of the task.
#[inline]
pub(crate) unsafe fn drop_locals(&self) {
let raw = self.inner.load(Ordering::Acquire);
if let Some(inner) = raw.as_mut() {
// Abort the process if dropping task-locals panics.
abort_on_panic(|| {
inner.locals.clear();
});
}
}
/// Returns the inner representation, initializing it on first use.
fn inner(&self) -> &Inner {
loop {
let raw = self.inner.load(Ordering::Acquire);
if !raw.is_null() {
return unsafe { &*raw };
}
let new = Arc::into_raw(Arc::new(Inner::new(None))) as *mut Inner;
if self.inner.compare_and_swap(raw, new, Ordering::AcqRel) != raw {
unsafe {
drop(Arc::from_raw(new));
}
}
}
}
/// Set a reference to the current task.
pub(crate) unsafe fn set_current<F, R>(task: *const Task, f: F) -> R
where
F: FnOnce() -> R,
{
CURRENT.with(|current| {
let old_task = current.replace(task);
defer! {
current.set(old_task);
}
f()
})
}
/// Gets a reference to the current task.
pub(crate) fn get_current<F, R>(f: F) -> Option<R>
where
F: FnOnce(&Task) -> R,
{
let res = CURRENT.try_with(|current| unsafe { current.get().as_ref().map(f) });
match res {
Ok(Some(val)) => Some(val),
Ok(None) | Err(_) => None,
}
}
}
impl Drop for Task {
fn drop(&mut self) {
// Deallocate the inner representation if it was initialized.
let raw = *self.inner.get_mut();
if !raw.is_null() {
unsafe {
drop(Arc::from_raw(raw));
}
}
}
}
impl Clone for Task {
fn clone(&self) -> Task {
// We need to make sure the inner representation is initialized now so that this instance
// and the clone have raw pointers that point to the same `Arc<Inner>`.
let arc = unsafe { ManuallyDrop::new(Arc::from_raw(self.inner())) };
let raw = Arc::into_raw(Arc::clone(&arc));
Task {
inner: AtomicPtr::new(raw as *mut Inner),
}
self.name.as_ref().map(|s| s.as_str())
}
}

9
src/task/task_id.rs
View file

@ -1,5 +1,5 @@
use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::atomic::{AtomicUsize, Ordering};
/// A unique identifier for a task.
///
@ -13,15 +13,16 @@ use std::sync::atomic::{AtomicU64, Ordering};
/// })
/// ```
#[derive(Eq, PartialEq, Clone, Copy, Hash, Debug)]
pub struct TaskId(pub(crate) u64);
pub struct TaskId(pub(crate) usize);
impl TaskId {
/// Generates a new `TaskId`.
pub(crate) fn generate() -> TaskId {
static COUNTER: AtomicU64 = AtomicU64::new(1);
// TODO: find a good version to emulate u64 atomics on 32 bit systems.
static COUNTER: AtomicUsize = AtomicUsize::new(1);
let id = COUNTER.fetch_add(1, Ordering::Relaxed);
if id > u64::max_value() / 2 {
if id > usize::max_value() / 2 {
std::process::abort();
}
TaskId(id)

4
src/task/task_local.rs
View file

@ -3,7 +3,7 @@ use std::error::Error;
use std::fmt;
use std::sync::atomic::{AtomicU32, Ordering};
use crate::task::Task;
use crate::task::TaskLocalsWrapper;
/// The key for accessing a task-local value.
///
@ -98,7 +98,7 @@ impl<T: Send + 'static> LocalKey<T> {
where
F: FnOnce(&T) -> R,
{
Task::get_current(|task| unsafe {
TaskLocalsWrapper::get_current(|task| unsafe {
// Prepare the numeric key, initialization function, and the map of task-locals.
let key = self.key();
let init = || Box::new((self.__init)()) as Box<dyn Send>;

84
src/task/task_locals_wrapper.rs Normal file
View file

@ -0,0 +1,84 @@
use std::cell::Cell;
use std::ptr;
use crate::task::{LocalsMap, Task, TaskId};
use crate::utils::abort_on_panic;
thread_local! {
/// A pointer to the currently running task.
static CURRENT: Cell<*const TaskLocalsWrapper> = Cell::new(ptr::null_mut());
}
/// A wrapper to store task local data.
pub(crate) struct TaskLocalsWrapper {
/// The actual task details.
task: Task,
/// The map holding task-local values.
locals: LocalsMap,
}
impl TaskLocalsWrapper {
/// Creates a new task handle.
///
/// If the task is unnamed, the inner representation of the task will be lazily allocated on
/// demand.
#[inline]
pub(crate) fn new(task: Task) -> Self {
Self {
task,
locals: LocalsMap::new(),
}
}
/// Gets the task's unique identifier.
#[inline]
pub fn id(&self) -> TaskId {
self.task.id()
}
/// Returns a reference to the inner `Task`.
pub(crate) fn task(&self) -> &Task {
&self.task
}
/// Returns the map holding task-local values.
pub(crate) fn locals(&self) -> &LocalsMap {
&self.locals
}
/// Set a reference to the current task.
pub(crate) unsafe fn set_current<F, R>(task: *const TaskLocalsWrapper, f: F) -> R
where
F: FnOnce() -> R,
{
CURRENT.with(|current| {
let old_task = current.replace(task);
defer! {
current.set(old_task);
}
f()
})
}
/// Gets a reference to the current task.
pub(crate) fn get_current<F, R>(f: F) -> Option<R>
where
F: FnOnce(&TaskLocalsWrapper) -> R,
{
let res = CURRENT.try_with(|current| unsafe { current.get().as_ref().map(f) });
match res {
Ok(Some(val)) => Some(val),
Ok(None) | Err(_) => None,
}
}
}
impl Drop for TaskLocalsWrapper {
fn drop(&mut self) {
// Abort the process if dropping task-locals panics.
abort_on_panic(|| {
unsafe { self.locals.clear() };
});
}
}

2
src/task/yield_now.rs
View file

@ -1,5 +1,5 @@
use std::pin::Pin;
use std::future::Future;
use std::pin::Pin;
use crate::task::{Context, Poll};

33
src/utils.rs
View file

@ -21,7 +21,7 @@ pub fn abort_on_panic<T>(f: impl FnOnce() -> T) -> T {
}
/// Generates a random number in `0..n`.
#[cfg(any(feature = "unstable", feature = "default"))]
#[cfg(feature = "unstable")]
pub fn random(n: u32) -> u32 {
use std::cell::Cell;
use std::num::Wrapping;
@ -59,6 +59,37 @@ pub(crate) trait Context {
fn context(self, message: impl Fn() -> String) -> Self;
}
#[cfg(all(not(target_os = "unknown"), feature = "default"))]
pub(crate) type Timer = smol::Timer;
#[cfg(all(target_arch = "wasm32", feature = "default"))]
#[derive(Debug)]
pub(crate) struct Timer(futures_timer::Delay);
#[cfg(all(target_arch = "wasm32", feature = "default"))]
impl Timer {
pub(crate) fn after(dur: std::time::Duration) -> Self {
Timer(futures_timer::Delay::new(dur))
}
}
#[cfg(target_arch = "wasm32")]
use std::pin::Pin;
#[cfg(target_arch = "wasm32")]
use std::task::Poll;
#[cfg(target_arch = "wasm32")]
impl std::future::Future for Timer {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
match Pin::new(&mut self.0).poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(_) => Poll::Ready(()),
}
}
}
/// Defers evaluation of a block of code until the end of the scope.
#[cfg(feature = "default")]
#[doc(hidden)]

2
tests/addr.rs
View file

@ -1,3 +1,5 @@
#![cfg(not(target_os = "unknown"))]
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6};
use async_std::net::ToSocketAddrs;

2
tests/block_on.rs
View file

@ -1,3 +1,5 @@
#![cfg(not(target_os = "unknown"))]
use async_std::task;
#[test]

24
tests/buf_writer.rs
View file

@ -2,15 +2,19 @@ use async_std::io::{self, BufWriter, SeekFrom};
use async_std::prelude::*;
use async_std::task;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn test_buffered_writer() {
#![allow(clippy::cognitive_complexity)]
task::block_on(async {
let inner = Vec::new();
let mut writer = BufWriter::with_capacity(2, inner);
let inner: Vec<u8> = Vec::new();
let mut writer = BufWriter::<Vec<u8>>::with_capacity(2, inner);
writer.write(&[0, 1]).await.unwrap();
assert_eq!(writer.buffer(), []);
assert!(writer.buffer().is_empty());
assert_eq!(*writer.get_ref(), [0, 1]);
writer.write(&[2]).await.unwrap();
@ -22,7 +26,7 @@ fn test_buffered_writer() {
assert_eq!(*writer.get_ref(), [0, 1]);
writer.flush().await.unwrap();
assert_eq!(writer.buffer(), []);
assert!(writer.buffer().is_empty());
assert_eq!(*writer.get_ref(), [0, 1, 2, 3]);
writer.write(&[4]).await.unwrap();
@ -35,31 +39,33 @@ fn test_buffered_writer() {
assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5]);
writer.write(&[7, 8]).await.unwrap();
assert_eq!(writer.buffer(), []);
assert!(writer.buffer().is_empty());
assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8]);
writer.write(&[9, 10, 11]).await.unwrap();
assert_eq!(writer.buffer(), []);
assert!(writer.buffer().is_empty());
assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
writer.flush().await.unwrap();
assert_eq!(writer.buffer(), []);
assert!(writer.buffer().is_empty());
assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
})
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn test_buffered_writer_inner_into_inner_flushes() {
task::block_on(async {
let mut w = BufWriter::with_capacity(3, Vec::new());
let mut w = BufWriter::with_capacity(3, Vec::<u8>::new());
w.write(&[0, 1]).await.unwrap();
assert_eq!(*w.get_ref(), []);
assert!(w.get_ref().is_empty());
let w = w.into_inner().await.unwrap();
assert_eq!(w, [0, 1]);
})
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn test_buffered_writer_seek() {
task::block_on(async {
let mut w = BufWriter::with_capacity(3, io::Cursor::new(Vec::new()));

47
tests/channel.rs
View file

@ -6,13 +6,22 @@ use std::time::Duration;
use async_std::sync::channel;
use async_std::task;
use rand::{thread_rng, Rng};
use rand::{Rng, SeedableRng};
#[cfg(not(target_os = "unknown"))]
use async_std::task::spawn;
#[cfg(target_os = "unknown")]
use async_std::task::spawn_local as spawn;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
fn ms(ms: u64) -> Duration {
Duration::from_millis(ms)
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn smoke() {
task::block_on(async {
let (s, r) = channel(1);
@ -35,6 +44,7 @@ fn smoke() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn capacity() {
for i in 1..10 {
let (s, r) = channel::<()>(i);
@ -44,6 +54,7 @@ fn capacity() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn len_empty_full() {
#![allow(clippy::cognitive_complexity)]
task::block_on(async {
@ -86,11 +97,12 @@ fn len_empty_full() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn recv() {
task::block_on(async {
let (s, r) = channel(100);
task::spawn(async move {
spawn(async move {
assert_eq!(r.recv().await.unwrap(), 7);
task::sleep(ms(1000)).await;
assert_eq!(r.recv().await.unwrap(), 8);
@ -107,11 +119,12 @@ fn recv() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn send() {
task::block_on(async {
let (s, r) = channel(1);
task::spawn(async move {
spawn(async move {
s.send(7).await;
task::sleep(ms(1000)).await;
s.send(8).await;
@ -129,6 +142,7 @@ fn send() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn recv_after_disconnect() {
task::block_on(async {
let (s, r) = channel(100);
@ -147,6 +161,7 @@ fn recv_after_disconnect() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn len() {
const COUNT: usize = 25_000;
const CAP: usize = 1000;
@ -184,7 +199,7 @@ fn len() {
assert_eq!(s.len(), 0);
assert_eq!(r.len(), 0);
let child = task::spawn({
let child = spawn({
let r = r.clone();
async move {
for i in 0..COUNT {
@ -209,11 +224,12 @@ fn len() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn disconnect_wakes_receiver() {
task::block_on(async {
let (s, r) = channel::<()>(1);
let child = task::spawn(async move {
let child = spawn(async move {
assert!(r.recv().await.is_err());
});
@ -225,13 +241,14 @@ fn disconnect_wakes_receiver() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn spsc() {
const COUNT: usize = 100_000;
task::block_on(async {
let (s, r) = channel(3);
let child = task::spawn(async move {
let child = spawn(async move {
for i in 0..COUNT {
assert_eq!(r.recv().await.unwrap(), i);
}
@ -248,6 +265,7 @@ fn spsc() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn mpmc() {
const COUNT: usize = 25_000;
const TASKS: usize = 4;
@ -262,7 +280,7 @@ fn mpmc() {
for _ in 0..TASKS {
let r = r.clone();
let v = v.clone();
tasks.push(task::spawn(async move {
tasks.push(spawn(async move {
for _ in 0..COUNT {
let n = r.recv().await.unwrap();
v[n].fetch_add(1, Ordering::SeqCst);
@ -272,7 +290,7 @@ fn mpmc() {
for _ in 0..TASKS {
let s = s.clone();
tasks.push(task::spawn(async move {
tasks.push(spawn(async move {
for i in 0..COUNT {
s.send(i).await;
}
@ -290,6 +308,7 @@ fn mpmc() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn oneshot() {
const COUNT: usize = 10_000;
@ -297,8 +316,8 @@ fn oneshot() {
for _ in 0..COUNT {
let (s, r) = channel(1);
let c1 = task::spawn(async move { r.recv().await.unwrap() });
let c2 = task::spawn(async move { s.send(0).await });
let c1 = spawn(async move { r.recv().await.unwrap() });
let c2 = spawn(async move { s.send(0).await });
c1.await;
c2.await;
@ -307,6 +326,7 @@ fn oneshot() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn drops() {
const RUNS: usize = 100;
@ -321,17 +341,16 @@ fn drops() {
}
}
let mut rng = thread_rng();
for _ in 0..RUNS {
task::block_on(async {
let mut rng = rand_xorshift::XorShiftRng::seed_from_u64(0);
task::block_on(async move {
let steps = rng.gen_range(0, 10_000);
let additional = rng.gen_range(0, 50);
DROPS.store(0, Ordering::SeqCst);
let (s, r) = channel::<DropCounter>(50);
let child = task::spawn({
let child = spawn({
let r = r.clone();
async move {
for _ in 0..steps {

103
tests/condvar.rs Normal file
View file

@ -0,0 +1,103 @@
#![cfg(feature = "unstable")]
use std::sync::Arc;
use std::time::Duration;
use async_std::sync::{Condvar, Mutex};
use async_std::task::{self, JoinHandle};
#[cfg(not(target_os = "unknown"))]
use async_std::task::spawn;
#[cfg(target_os = "unknown")]
use async_std::task::spawn_local as spawn;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn wait_timeout_with_lock() {
task::block_on(async {
let pair = Arc::new((Mutex::new(false), Condvar::new()));
let pair2 = pair.clone();
spawn(async move {
let (m, c) = &*pair2;
let _g = m.lock().await;
task::sleep(Duration::from_millis(200)).await;
c.notify_one();
});
let (m, c) = &*pair;
let (_, wait_result) = c
.wait_timeout(m.lock().await, Duration::from_millis(50))
.await;
assert!(wait_result.timed_out());
})
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn wait_timeout_without_lock() {
task::block_on(async {
let m = Mutex::new(false);
let c = Condvar::new();
let (_, wait_result) = c
.wait_timeout(m.lock().await, Duration::from_millis(10))
.await;
assert!(wait_result.timed_out());
})
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn wait_timeout_until_timed_out() {
task::block_on(async {
let m = Mutex::new(false);
let c = Condvar::new();
let (_, wait_result) = c
.wait_timeout_until(m.lock().await, Duration::from_millis(100), |&mut started| {
started
})
.await;
assert!(wait_result.timed_out());
})
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn notify_all() {
task::block_on(async {
let mut tasks: Vec<JoinHandle<()>> = Vec::new();
let pair = Arc::new((Mutex::new(0u32), Condvar::new()));
for _ in 0..10 {
let pair = pair.clone();
tasks.push(spawn(async move {
let (m, c) = &*pair;
let mut count = m.lock().await;
while *count == 0 {
count = c.wait(count).await;
}
*count += 1;
}));
}
// Give some time for tasks to start up
task::sleep(Duration::from_millis(50)).await;
let (m, c) = &*pair;
{
let mut count = m.lock().await;
*count += 1;
c.notify_all();
}
for t in tasks {
t.await;
}
let count = m.lock().await;
assert_eq!(11, *count);
})
}

8
tests/io_timeout.rs
View file

@ -5,6 +5,14 @@ use async_std::task;
#[test]
#[should_panic(expected = "timed out")]
#[cfg(not(any(
target_os = "unknown",
target_arch = "arm",
target_arch = "mips",
target_arch = "powerpc",
target_arch = "powerpc64",
target_arch = "x86",
)))] // stdin tests fail when running through cross
fn io_timeout_timedout() {
task::block_on(async {
io::timeout(Duration::from_secs(1), async {

24
tests/mutex.rs
View file

@ -5,7 +5,16 @@ use async_std::sync::Mutex;
use async_std::task;
use futures::channel::mpsc;
#[cfg(not(target_os = "unknown"))]
use async_std::task::spawn;
#[cfg(target_os = "unknown")]
use async_std::task::spawn_local as spawn;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn smoke() {
task::block_on(async {
let m = Mutex::new(());
@ -15,18 +24,21 @@ fn smoke() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn try_lock() {
let m = Mutex::new(());
*m.try_lock().unwrap() = ();
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn into_inner() {
let m = Mutex::new(10);
assert_eq!(m.into_inner(), 10);
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn get_mut() {
let mut m = Mutex::new(10);
*m.get_mut() = 20;
@ -34,6 +46,7 @@ fn get_mut() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn contention() {
task::block_on(async {
let (tx, mut rx) = mpsc::unbounded();
@ -42,22 +55,29 @@ fn contention() {
let mutex = Arc::new(Mutex::new(0));
let num_tasks = 10000;
let mut handles = Vec::new();
for _ in 0..num_tasks {
let tx = tx.clone();
let mutex = mutex.clone();
task::spawn(async move {
handles.push(spawn(async move {
let mut lock = mutex.lock().await;
*lock += 1;
tx.unbounded_send(()).unwrap();
drop(lock);
});
}));
}
for _ in 0..num_tasks {
rx.next().await.unwrap();
}
for handle in handles.into_iter() {
handle.await;
}
dbg!("wait");
let lock = mutex.lock().await;
assert_eq!(num_tasks, *lock);
});

25
tests/rwlock.rs
View file

@ -10,6 +10,14 @@ use async_std::sync::RwLock;
use async_std::task;
use futures::channel::mpsc;
#[cfg(not(target_os = "unknown"))]
use async_std::task::spawn;
#[cfg(target_os = "unknown")]
use async_std::task::spawn_local as spawn;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
/// Generates a random number in `0..n`.
pub fn random(n: u32) -> u32 {
thread_local! {
@ -35,6 +43,7 @@ pub fn random(n: u32) -> u32 {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn smoke() {
task::block_on(async {
let lock = RwLock::new(());
@ -46,6 +55,7 @@ fn smoke() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn try_write() {
task::block_on(async {
let lock = RwLock::new(0isize);
@ -56,12 +66,14 @@ fn try_write() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn into_inner() {
let lock = RwLock::new(10);
assert_eq!(lock.into_inner(), 10);
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn into_inner_and_drop() {
struct Counter(Arc<AtomicUsize>);
@ -84,6 +96,7 @@ fn into_inner_and_drop() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn get_mut() {
let mut lock = RwLock::new(10);
*lock.get_mut() = 20;
@ -91,6 +104,7 @@ fn get_mut() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn contention() {
const N: u32 = 10;
const M: usize = 1000;
@ -104,7 +118,7 @@ fn contention() {
let tx = tx.clone();
let rw = rw.clone();
task::spawn(async move {
spawn(async move {
for _ in 0..M {
if random(N) == 0 {
drop(rw.write().await);
@ -116,7 +130,7 @@ fn contention() {
});
}
task::block_on(async {
task::block_on(async move {
for _ in 0..N {
rx.next().await.unwrap();
}
@ -124,6 +138,7 @@ fn contention() {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn writer_and_readers() {
#[derive(Default)]
struct Yield(Cell<bool>);
@ -146,7 +161,7 @@ fn writer_and_readers() {
let (tx, mut rx) = mpsc::unbounded();
// Spawn a writer task.
task::spawn({
spawn({
let lock = lock.clone();
async move {
let mut lock = lock.write().await;
@ -164,13 +179,13 @@ fn writer_and_readers() {
let mut readers = Vec::new();
for _ in 0..5 {
let lock = lock.clone();
readers.push(task::spawn(async move {
readers.push(spawn(async move {
let lock = lock.read().await;
assert!(*lock >= 0);
}));
}
task::block_on(async {
task::block_on(async move {
// Wait for readers to pass their asserts.
for r in readers {
r.await;

20
tests/stream.rs
View file

@ -8,14 +8,23 @@ use async_std::stream;
use async_std::sync::channel;
use async_std::task;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
#[cfg(not(target_os = "unknown"))]
use async_std::task::spawn;
#[cfg(target_os = "unknown")]
use async_std::task::spawn_local as spawn;
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
/// Checks that streams are merged fully even if one of the components
/// experiences delay.
fn merging_delayed_streams_work() {
let (sender, receiver) = channel::<i32>(10);
let mut s = receiver.merge(stream::empty());
let t = task::spawn(async move {
let t = spawn(async move {
let mut xs = Vec::new();
while let Some(x) = s.next().await {
xs.push(x);
@ -34,7 +43,7 @@ fn merging_delayed_streams_work() {
let (sender, receiver) = channel::<i32>(10);
let mut s = stream::empty().merge(receiver);
let t = task::spawn(async move {
let t = spawn(async move {
let mut xs = Vec::new();
while let Some(x) = s.next().await {
xs.push(x);
@ -85,16 +94,17 @@ fn explode<S: Stream>(s: S) -> Explode<S> {
}
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn merge_works_with_unfused_streams() {
let s1 = explode(stream::once(92));
let s2 = explode(stream::once(92));
let mut s = s1.merge(s2);
let xs = task::block_on(async move {
task::block_on(async move {
let mut xs = Vec::new();
while let Some(x) = s.next().await {
xs.push(x)
}
xs
assert_eq!(xs, vec![92, 92]);
});
assert_eq!(xs, vec![92, 92]);
}

11
tests/task_local.rs
View file

@ -3,7 +3,16 @@ use std::sync::atomic::{AtomicBool, Ordering};
use async_std::task;
use async_std::task_local;
#[cfg(not(target_os = "unknown"))]
use async_std::task::spawn;
#[cfg(target_os = "unknown")]
use async_std::task::spawn_local as spawn;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn drop_local() {
static DROP_LOCAL: AtomicBool = AtomicBool::new(false);
@ -20,7 +29,7 @@ fn drop_local() {
}
// Spawn a task that just touches its task-local.
let handle = task::spawn(async {
let handle = spawn(async {
LOCAL.with(|_| ());
});
let task = handle.task().clone();

2
tests/tcp.rs
View file

@ -1,3 +1,5 @@
#![cfg(not(target_os = "unknown"))]
use async_std::io;
use async_std::net::{TcpListener, TcpStream};
use async_std::prelude::*;

26
tests/timeout.rs Normal file
View file

@ -0,0 +1,26 @@
use std::time::Duration;
use async_std::future::timeout;
use async_std::task;
#[cfg(target_arch = "wasm32")]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);
#[test]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
fn timeout_future_many() {
task::block_on(async {
let futures = (0..100)
.map(|i| {
timeout(Duration::from_millis(i * 20), async move {
task::sleep(Duration::from_millis(i)).await;
Ok::<(), async_std::future::TimeoutError>(())
})
})
.collect::<Vec<_>>();
for future in futures {
future.await.unwrap().unwrap();
}
});
}

2
tests/udp.rs
View file

@ -1,3 +1,5 @@
#![cfg(not(target_os = "unknown"))]
use async_std::io;
use async_std::net::UdpSocket;
use async_std::task;

26
tests/uds.rs
View file

@ -1,4 +1,4 @@
#![cfg(unix)]
#![cfg(all(unix, not(target_os = "unknown")))]
use async_std::io;
use async_std::os::unix::net::{UnixDatagram, UnixListener, UnixStream};
@ -94,3 +94,27 @@ async fn ping_pong_client(socket: &std::path::PathBuf, iterations: u32) -> std::
}
Ok(())
}
#[test]
fn uds_clone() -> io::Result<()> {
task::block_on(async {
let tmp_dir = TempDir::new("socket_ping_pong").expect("Temp dir not created");
let sock_path = tmp_dir.as_ref().join("sock");
let input = UnixListener::bind(&sock_path).await?;
let mut writer = UnixStream::connect(&sock_path).await?;
let mut reader = input.incoming().next().await.unwrap()?;
writer.write(b"original").await.unwrap();
let mut original_buf = [0; 8];
reader.read(&mut original_buf).await?;
assert_eq!(&original_buf, b"original");
writer.clone().write(b"clone").await.unwrap();
let mut clone_buf = [0; 5];
reader.clone().read(&mut clone_buf).await?;
assert_eq!(&clone_buf, b"clone");
Ok(())
})
}

2
tests/verbose_errors.rs
View file

@ -1,3 +1,5 @@
#![cfg(not(target_os = "unknown"))]
use async_std::{fs, io, net::ToSocketAddrs, task};
#[test]

10
wasm-test.sh Executable file
View file

@ -0,0 +1,10 @@
#!/bin/sh
wasm-pack test --chrome --headless -- --features unstable --test buf_writer
wasm-pack test --chrome --headless -- --features unstable --test channel
wasm-pack test --chrome --headless -- --features unstable --test condvar
wasm-pack test --chrome --headless -- --features unstable --test mutex
wasm-pack test --chrome --headless -- --features unstable --test rwlock
wasm-pack test --chrome --headless -- --features unstable --test stream
wasm-pack test --chrome --headless -- --features unstable --test task_local
wasm-pack test --chrome --headless -- --features unstable --test timeout