Merge pull request #833 from OSSystems/topic/fix_channel_doc

channel doc: Fix misleading reference to None return on Receiver

.github/workflows/ci.yml

@@ -29,6 +29,24 @@ jobs:
         toolchain: ${{ matrix.rust }}
         override: true
 
+    - name: Cache cargo registry
+      uses: actions/cache@v2
+      with:
+        path: ~/.cargo/registry
+        key: ${{ matrix.os }}-${{ matrix.rust }}-cargo-registry-${{ hashFiles('**/Cargo.toml') }}
+
+    - name: Cache cargo index
+      uses: actions/cache@v2
+      with:
+        path: ~/.cargo/git
+        key: ${{ matrix.os }}-${{ matrix.rust }}-cargo-index-${{ hashFiles('**/Cargo.toml') }}
+
+    - name: Cache cargo build
+      uses: actions/cache@v2
+      with:
+        path: target
+        key: ${{ matrix.os }}-${{ matrix.rust }}-cargo-build-target-${{ hashFiles('**/Cargo.toml') }}
+
     - name: check
       uses: actions-rs/cargo@v1
       with:
@@ -41,6 +59,14 @@ jobs:
         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
       if: matrix.rust == 'nightly'
@@ -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,50 @@ jobs:
         command:  check
         args: --no-default-features --features alloc --target thumbv7m-none-eabi -Z avoid-dev-deps
 
+  check_tokio_02_feature:
+    name: Check tokio02 feature
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@master
+    - name: check tokio02
+      uses: actions-rs/cargo@v1
+      with:
+        command:  check
+        args: --all --features tokio02
+
+  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 +181,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

CHANGELOG.md

@@ -7,6 +7,75 @@ and this project adheres to [Semantic Versioning](https://book.async.rs/overview
 
 ## [Unreleased]
 
+# [1.6.2] - 2020-06-19
+
+## Added
+
+- Add `UdpSocket::peer_addr` ([#816](https://github.com/async-rs/async-std/pull/816))
+
+## Changed
+
+## Fixed
+
+- Ensure the reactor is running for sockets and timers ([#819](https://github.com/async-rs/async-std/pull/819)).
+- Avoid excessive polling in `flatten` and `flat_map` ([#701](https://github.com/async-rs/async-std/pull/701))
+
+
+# [1.6.1] - 2020-06-11
+
+## Added
+
+- Added `tokio02` feature flag, to allow compatability usage with tokio@0.2 ([#804](https://github.com/async-rs/async-std/pull/804)).
+
+## Changed
+
+- Removed unstable `stdio` lock methods, due to their unsoundness ([#807](https://github.com/async-rs/async-std/pull/807)).
+
+## Fixed
+
+- Fixed wrong slice index for file reading ([#802](https://github.com/async-rs/async-std/pull/802)).
+- Fixed recursive calls to `block_on` ([#799](https://github.com/async-rs/async-std/pull/799)) and ([#809](https://github.com/async-rs/async-std/pull/809)).
+- Remove `default` feature requirement for the `unstable` feature ([#806](https://github.com/async-rs/async-std/pull/806)).
+
+# [1.6.0] - 2020-05-22
+
+See `1.6.0-beta.1` and `1.6.0-beta.2`.
+
+# [1.6.0-beta.2] - 2020-05-19
+
+## Added 
+
+- Added an environment variable to configure the thread pool size of the runtime. ([#774](https://github.com/async-rs/async-std/pull/774))
+- Implement `Clone` for `UnixStream` ([#772](https://github.com/async-rs/async-std/pull/772))
+
+## 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))
+
+## Fixed
+
+- Use `smol::block_on` to handle drop of `File`, avoiding nested executor panic. ([#768](https://github.com/async-rs/async-std/pull/768))
+
+# [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)
@@ -677,7 +746,12 @@ task::blocking(async {
 
 - Initial beta release
 
-[Unreleased]: https://github.com/async-rs/async-std/compare/v1.5.0...HEAD
+[Unreleased]: https://github.com/async-rs/async-std/compare/v1.6.2...HEAD
+[1.6.2]: https://github.com/async-rs/async-std/compare/v1.6.1...v1.6.2
+[1.6.1]: https://github.com/async-rs/async-std/compare/v1.6.0...v1.6.1
+[1.6.0]: https://github.com/async-rs/async-std/compare/v1.5.0...v1.6.0
+[1.6.0-beta.2]: https://github.com/async-rs/async-std/compare/v1.6.0-beta.1...v1.6.0-beta.2
+[1.6.0-beta.1]: https://github.com/async-rs/async-std/compare/v1.5.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

Cargo.toml

@@ -1,9 +1,10 @@
 [package]
 name = "async-std"
-version = "1.5.0"
+version = "1.6.2"
 authors = [
   "Stjepan Glavina <stjepang@gmail.com>",
   "Yoshua Wuyts <yoshuawuyts@gmail.com>",
+  "Friedel Ziegelmayer <me@dignifiedquire.com>",
   "Contributors to async-std",
 ]
 edition = "2018"
@@ -24,19 +25,17 @@ rustdoc-args = ["--cfg", "feature=\"docs\""]
 default = [
   "std",
   "async-task",
-  "crossbeam-channel",
-  "crossbeam-deque",
-  "crossbeam-queue",
-  "futures-timer",
   "kv-log-macro",
   "log",
-  "mio",
-  "mio-uds",
   "num_cpus",
   "pin-project-lite",
+  "smol",
 ]
 docs = ["attributes", "unstable", "default"]
-unstable = ["std", "broadcaster", "futures-timer"]
+unstable = [
+  "std",
+  "futures-timer",
+]
 attributes = ["async-attributes"]
 std = [
   "alloc",
@@ -47,40 +46,53 @@ std = [
   "once_cell",
   "pin-utils",
   "slab",
+  "wasm-bindgen-futures",
+  "futures-channel",
+  "async-mutex",
 ]
 alloc = [
   "futures-core/alloc",
   "pin-project-lite",
 ]
+tokio02 = ["smol/tokio02"]
 
 [dependencies]
 async-attributes = { version = "1.1.1", optional = true }
-async-task = { version = "1.3.1", optional = true }
+async-task = { version = "3.0.0", optional = true }
-broadcaster = { version = "1.0.0", optional = true }
+async-mutex = { version = "1.1.3", optional = true }
-crossbeam-channel = { version = "0.4.2", optional = true }
-crossbeam-deque = { version = "0.7.3", optional = true }
-crossbeam-queue = { version = "0.2.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 }
-futures-timer = { version = "3.0.2", optional = true }
+kv-log-macro = { version = "1.0.6", 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.3", optional = true }
-mio = { version = "0.6.19", optional = true }
-mio-uds = { version = "0.6.7", 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 }
+futures-timer = { version = "3.0.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.17", 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.4"
+rand_xorshift = "0.2.0"
 
 [[test]]
 name = "stream"
@@ -89,3 +101,7 @@ required-features = ["unstable"]
 [[example]]
 name = "tcp-ipv4-and-6-echo"
 required-features = ["unstable"]
+
+[[example]]
+name = "surf-web"
+required-features = ["surf"]

README.md

@@ -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>

src/collections/binary_heap/extend.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T: Ord> stream::Extend<T> for BinaryHeap<T> {
+impl<T: Ord + Send> stream::Extend<T> for BinaryHeap<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         self.reserve(stream.size_hint().0);

src/collections/binary_heap/from_stream.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T: Ord> FromStream<T> for BinaryHeap<T> {
+impl<T: Ord + Send> FromStream<T> for BinaryHeap<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/collections/btree_map/extend.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<K: Ord, V> stream::Extend<(K, V)> for BTreeMap<K, V> {
+impl<K: Ord + Send, V: Send> stream::Extend<(K, V)> for BTreeMap<K, V> {
     fn extend<'a, S: IntoStream<Item = (K, V)> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         Box::pin(stream.into_stream().for_each(move |(k, v)| {
             self.insert(k, v);
         }))

src/collections/btree_map/from_stream.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<K: Ord, V> FromStream<(K, V)> for BTreeMap<K, V> {
+impl<K: Ord + Send, V: Send> FromStream<(K, V)> for BTreeMap<K, V> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = (K, V)> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/collections/btree_set/extend.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T: Ord> stream::Extend<T> for BTreeSet<T> {
+impl<T: Ord + Send> stream::Extend<T> for BTreeSet<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         Box::pin(stream.into_stream().for_each(move |item| {
             self.insert(item);
         }))

src/collections/btree_set/from_stream.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T: Ord> FromStream<T> for BTreeSet<T> {
+impl<T: Ord + Send> FromStream<T> for BTreeSet<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/collections/hash_map/extend.rs

@@ -7,13 +7,17 @@ use crate::stream::{self, IntoStream};
 
 impl<K, V, H> stream::Extend<(K, V)> for HashMap<K, V, H>
 where
-    K: Eq + Hash,
+    K: Eq + Hash + Send,
-    H: BuildHasher + Default,
+    V: Send,
+    H: BuildHasher + Default + Send,
 {
     fn extend<'a, S: IntoStream<Item = (K, V)> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         // The following is adapted from the hashbrown source code:

src/collections/hash_map/from_stream.rs

@@ -7,13 +7,17 @@ use crate::stream::{self, FromStream, IntoStream};
 
 impl<K, V, H> FromStream<(K, V)> for HashMap<K, V, H>
 where
-    K: Eq + Hash,
+    K: Eq + Hash + Send,
-    H: BuildHasher + Default,
+    H: BuildHasher + Default + Send,
+    V: Send,
 {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = (K, V)> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/collections/hash_set/extend.rs

@@ -7,13 +7,16 @@ use crate::stream::{self, IntoStream};
 
 impl<T, H> stream::Extend<T> for HashSet<T, H>
 where
-    T: Eq + Hash,
+    T: Eq + Hash + Send,
-    H: BuildHasher + Default,
+    H: BuildHasher + Default + Send,
 {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         // The Extend impl for HashSet in the standard library delegates to the internal HashMap.
         // Thus, this impl is just a copy of the async Extend impl for HashMap in this crate.
 

src/collections/hash_set/from_stream.rs

@@ -7,13 +7,16 @@ use crate::stream::{self, FromStream, IntoStream};
 
 impl<T, H> FromStream<T> for HashSet<T, H>
 where
-    T: Eq + Hash,
+    T: Eq + Hash + Send,
-    H: BuildHasher + Default,
+    H: BuildHasher + Default + Send,
 {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/collections/linked_list/extend.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T> stream::Extend<T> for LinkedList<T> {
+impl<T: Send> stream::Extend<T> for LinkedList<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
         Box::pin(stream.for_each(move |item| self.push_back(item)))
     }

src/collections/linked_list/from_stream.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T> FromStream<T> for LinkedList<T> {
+impl<T: Send> FromStream<T> for LinkedList<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/collections/vec_deque/extend.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T> stream::Extend<T> for VecDeque<T> {
+impl<T: Send> stream::Extend<T> for VecDeque<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         self.reserve(stream.size_hint().0);

src/collections/vec_deque/from_stream.rs

@@ -4,11 +4,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T> FromStream<T> for VecDeque<T> {
+impl<T: Send> FromStream<T> for VecDeque<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/fs/file.rs

@@ -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());
     }
 }
 
@@ -673,7 +673,7 @@ impl LockGuard<State> {
                 if available > 0 || self.cache.is_empty() {
                     // Copy data from the cache into the buffer.
                     let n = cmp::min(available, buf.len());
-                    buf[..n].copy_from_slice(&self.cache[start..n]);
+                    buf[..n].copy_from_slice(&self.cache[start..(start + n)]);
 
                     // Move the read cursor forward.
                     self.mode = Mode::Reading(start + n);
@@ -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();
+        });
+    }
+}

src/future/future/delay.rs

@@ -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_after, 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 }
     }

src/future/mod.rs

@@ -61,10 +61,10 @@ cfg_std! {
     mod ready;
 }
 
-cfg_default! {
+#[cfg(any(feature = "unstable", feature = "default"))]
-    pub use timeout::{timeout, TimeoutError};
+pub use timeout::{timeout, TimeoutError};
-    mod timeout;
+#[cfg(any(feature = "unstable", feature = "default"))]
-}
+mod timeout;
 
 cfg_unstable! {
     pub use into_future::IntoFuture;

src/future/timeout.rs

@@ -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_after, 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 {
+    TimeoutFuture::new(f, dur).await
-        future: f,
-        delay: Delay::new(dur),
-    };
-    f.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),
+        }
     }
 }
 

src/io/buf_read/mod.rs

@@ -29,7 +29,7 @@ extension_trait! {
 
         ```
         # #[allow(unused_imports)]
-        use async_std::prelude::*;
+        use async_std::io::prelude::*;
         ```
 
         [`std::io::BufRead`]: https://doc.rust-lang.org/std/io/trait.BufRead.html

src/io/mod.rs

@@ -307,22 +307,24 @@ 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! {
-    pub use stderr::StderrLock;
-    pub use stdin::StdinLock;
-    pub use stdout::StdoutLock;
-}

src/io/read/mod.rs

@@ -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(())
+        });
-        })
     }
 }

src/io/read/take.rs

@@ -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::*;

src/io/stderr.rs

@@ -5,11 +5,6 @@ use std::future::Future;
 use crate::io::{self, Write};
 use crate::task::{spawn_blocking, Context, JoinHandle, Poll};
 
-cfg_unstable! {
-    use once_cell::sync::Lazy;
-    use std::io::Write as _;
-}
-
 /// Constructs a new handle to the standard error of the current process.
 ///
 /// This function is an async version of [`std::io::stderr`].
@@ -58,22 +53,6 @@ pub fn stderr() -> Stderr {
 #[derive(Debug)]
 pub struct Stderr(Mutex<State>);
 
-/// A locked reference to the Stderr handle.
-///
-/// This handle implements the [`Write`] traits, and is constructed via the [`Stderr::lock`]
-/// method.
-///
-/// [`Write`]: trait.Read.html
-/// [`Stderr::lock`]: struct.Stderr.html#method.lock
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-#[derive(Debug)]
-pub struct StderrLock<'a>(std::io::StderrLock<'a>);
-
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-unsafe impl Send for StderrLock<'_> {}
-
 /// The state of the asynchronous stderr.
 ///
 /// The stderr can be either idle or busy performing an asynchronous operation.
@@ -108,42 +87,14 @@ enum Operation {
     Flush(io::Result<()>),
 }
 
-impl Stderr {
-    /// Locks this handle to the standard error stream, returning a writable guard.
-    ///
-    /// The lock is released when the returned lock goes out of scope. The returned guard also implements the Write trait for writing data.
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
-    /// #
-    /// use async_std::io;
-    /// use async_std::prelude::*;
-    ///
-    /// let stderr = io::stderr();
-    /// let mut handle = stderr.lock().await;
-    ///
-    /// handle.write_all(b"hello world").await?;
-    /// #
-    /// # Ok(()) }) }
-    /// ```
-    #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-    #[cfg(any(feature = "unstable", feature = "docs"))]
-    pub async fn lock(&self) -> StderrLock<'static> {
-        static STDERR: Lazy<std::io::Stderr> = Lazy::new(std::io::stderr);
-
-        spawn_blocking(move || StderrLock(STDERR.lock())).await
-    }
-}
-
 impl Write for Stderr {
     fn poll_write(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        let state = &mut *self.0.lock().unwrap();
+        let mut state_guard = self.0.lock().unwrap();
+        let state = &mut *state_guard;
 
         loop {
             match state {
@@ -187,8 +138,9 @@ impl Write for Stderr {
         }
     }
 
-    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        let state = &mut *self.0.lock().unwrap();
+        let mut state_guard = self.0.lock().unwrap();
+        let state = &mut *state_guard;
 
         loop {
             match state {
@@ -239,23 +191,3 @@ cfg_windows! {
         }
     }
 }
-
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-impl io::Write for StderrLock<'_> {
-    fn poll_write(
-        mut self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<io::Result<usize>> {
-        Poll::Ready(self.0.write(buf))
-    }
-
-    fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        Poll::Ready(self.0.flush())
-    }
-
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        self.poll_flush(cx)
-    }
-}

src/io/stdin.rs

@@ -7,11 +7,6 @@ use crate::io::{self, Read};
 use crate::task::{spawn_blocking, Context, JoinHandle, Poll};
 use crate::utils::Context as _;
 
-cfg_unstable! {
-    use once_cell::sync::Lazy;
-    use std::io::Read as _;
-}
-
 /// Constructs a new handle to the standard input of the current process.
 ///
 /// This function is an async version of [`std::io::stdin`].
@@ -61,21 +56,6 @@ pub fn stdin() -> Stdin {
 #[derive(Debug)]
 pub struct Stdin(Mutex<State>);
 
-/// A locked reference to the Stdin handle.
-///
-/// This handle implements the [`Read`] traits, and is constructed via the [`Stdin::lock`] method.
-///
-/// [`Read`]: trait.Read.html
-/// [`Stdin::lock`]: struct.Stdin.html#method.lock
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-#[cfg(feature = "unstable")]
-#[derive(Debug)]
-pub struct StdinLock<'a>(std::io::StdinLock<'a>);
-
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-unsafe impl Send for StdinLock<'_> {}
-
 /// The state of the asynchronous stdin.
 ///
 /// The stdin can be either idle or busy performing an asynchronous operation.
@@ -165,44 +145,16 @@ impl Stdin {
         .await
         .context(|| String::from("could not read line on stdin"))
     }
-
-    /// Locks this handle to the standard input stream, returning a readable guard.
-    ///
-    /// The lock is released when the returned lock goes out of scope. The returned guard also implements the Read trait for accessing the underlying data.
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
-    /// #
-    /// use async_std::io;
-    /// use async_std::prelude::*;
-    ///
-    /// let mut buffer = String::new();
-    ///
-    /// let stdin = io::stdin();
-    /// let mut handle = stdin.lock().await;
-    ///
-    /// handle.read_to_string(&mut buffer).await?;
-    /// #
-    /// # Ok(()) }) }
-    /// ```
-    #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-    #[cfg(any(feature = "unstable", feature = "docs"))]
-    pub async fn lock(&self) -> StdinLock<'static> {
-        static STDIN: Lazy<std::io::Stdin> = Lazy::new(std::io::stdin);
-
-        spawn_blocking(move || StdinLock(STDIN.lock())).await
-    }
 }
 
 impl Read for Stdin {
     fn poll_read(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &mut [u8],
     ) -> Poll<io::Result<usize>> {
-        let state = &mut *self.0.lock().unwrap();
+        let mut state_guard = self.0.lock().unwrap();
+        let state = &mut *state_guard;
 
         loop {
             match state {
@@ -265,15 +217,3 @@ cfg_windows! {
         }
     }
 }
-
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-impl Read for StdinLock<'_> {
-    fn poll_read(
-        mut self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
-        Poll::Ready(self.0.read(buf))
-    }
-}

src/io/stdout.rs

@@ -5,11 +5,6 @@ use std::future::Future;
 use crate::io::{self, Write};
 use crate::task::{spawn_blocking, Context, JoinHandle, Poll};
 
-cfg_unstable! {
-    use once_cell::sync::Lazy;
-    use std::io::Write as _;
-}
-
 /// Constructs a new handle to the standard output of the current process.
 ///
 /// This function is an async version of [`std::io::stdout`].
@@ -58,22 +53,6 @@ pub fn stdout() -> Stdout {
 #[derive(Debug)]
 pub struct Stdout(Mutex<State>);
 
-/// A locked reference to the Stderr handle.
-///
-/// This handle implements the [`Write`] traits, and is constructed via the [`Stdout::lock`]
-/// method.
-///
-/// [`Write`]: trait.Read.html
-/// [`Stdout::lock`]: struct.Stdout.html#method.lock
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-#[derive(Debug)]
-pub struct StdoutLock<'a>(std::io::StdoutLock<'a>);
-
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-unsafe impl Send for StdoutLock<'_> {}
-
 /// The state of the asynchronous stdout.
 ///
 /// The stdout can be either idle or busy performing an asynchronous operation.
@@ -108,42 +87,14 @@ enum Operation {
     Flush(io::Result<()>),
 }
 
-impl Stdout {
-    /// Locks this handle to the standard error stream, returning a writable guard.
-    ///
-    /// The lock is released when the returned lock goes out of scope. The returned guard also implements the Write trait for writing data.
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
-    /// #
-    /// use async_std::io;
-    /// use async_std::prelude::*;
-    ///
-    /// let stdout = io::stdout();
-    /// let mut handle = stdout.lock().await;
-    ///
-    /// handle.write_all(b"hello world").await?;
-    /// #
-    /// # Ok(()) }) }
-    /// ```
-    #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-    #[cfg(any(feature = "unstable", feature = "docs"))]
-    pub async fn lock(&self) -> StdoutLock<'static> {
-        static STDOUT: Lazy<std::io::Stdout> = Lazy::new(std::io::stdout);
-
-        spawn_blocking(move || StdoutLock(STDOUT.lock())).await
-    }
-}
-
 impl Write for Stdout {
     fn poll_write(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        let state = &mut *self.0.lock().unwrap();
+        let mut state_guard = self.0.lock().unwrap();
+        let state = &mut *state_guard;
 
         loop {
             match state {
@@ -187,8 +138,9 @@ impl Write for Stdout {
         }
     }
 
-    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        let state = &mut *self.0.lock().unwrap();
+        let mut state_guard = self.0.lock().unwrap();
+        let state = &mut *state_guard;
 
         loop {
             match state {
@@ -239,23 +191,3 @@ cfg_windows! {
         }
     }
 }
-
-#[cfg(feature = "unstable")]
-#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-impl Write for StdoutLock<'_> {
-    fn poll_write(
-        mut self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<io::Result<usize>> {
-        Poll::Ready(self.0.write(buf))
-    }
-
-    fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        Poll::Ready(self.0.flush())
-    }
-
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        self.poll_flush(cx)
-    }
-}

src/io/timeout.rs

@@ -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_after, 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,
     }
 }
 

src/lib.rs

@@ -138,7 +138,8 @@
 //!
 //! Call an async function from the main function:
 //!
-//! ```
+#![cfg_attr(feature = "attributes", doc = "```")]
+#![cfg_attr(not(feature = "attributes"), doc = "```ignore")]
 //! async fn say_hello() {
 //!     println!("Hello, world!");
 //! }
@@ -151,7 +152,8 @@
 //!
 //! Await two futures concurrently, and return a tuple of their output:
 //!
-//! ```
+#![cfg_attr(feature = "attributes", doc = "```")]
+#![cfg_attr(not(feature = "attributes"), doc = "```ignore")]
 //! use async_std::prelude::*;
 //!
 //! #[async_std::main]
@@ -164,7 +166,8 @@
 //!
 //! Create a UDP server that echoes back each received message to the sender:
 //!
-//! ```no_run
+#![cfg_attr(feature = "attributes", doc = "```no_run")]
+#![cfg_attr(not(feature = "attributes"), doc = "```ignore")]
 //! use async_std::net::UdpSocket;
 //!
 //! #[async_std::main]
@@ -194,7 +197,7 @@
 //!
 //! ```toml
 //! [dependencies.async-std]
-//! version = "1.0.0"
+//! version = "1.6.2"
 //! features = ["unstable"]
 //! ```
 //!
@@ -207,16 +210,25 @@
 //!
 //! ```toml
 //! [dependencies.async-std]
-//! version = "1.0.0"
+//! version = "1.6.2"
 //! features = ["attributes"]
 //! ```
 //!
+//! Compatibility with the `tokio` runtime is possible using the `tokio02`
+//! Cargo feature:
+//!
+//! ```toml
+//! [dependencies.async-std]
+//! version = "1.6.2"
+//! features = ["tokio02"]
+//! ```
+//!
 //! Additionally it's possible to only use the core traits and combinators by
 //! only enabling the `std` Cargo feature:
 //!
 //! ```toml
 //! [dependencies.async-std]
-//! version = "1.0.0"
+//! version = "1.6.2"
 //! default-features = false
 //! features = ["std"]
 //! ```
@@ -226,10 +238,25 @@
 //!
 //! ```toml
 //! [dependencies.async-std]
-//! version = "1.5.0"
+//! version = "1.6.2"
 //! 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))]
@@ -267,14 +294,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;

src/net/mod.rs

@@ -61,10 +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;
+#[cfg(not(target_os = "unknown"))]
 mod tcp;
+#[cfg(not(target_os = "unknown"))]
 mod udp;

src/net/tcp/listener.rs

@@ -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::rt::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 {
@@ -75,15 +75,15 @@ impl TcpListener {
     ///
     /// [`local_addr`]: #method.local_addr
     pub async fn bind<A: ToSocketAddrs>(addrs: A) -> io::Result<TcpListener> {
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         let mut last_err = None;
         let addrs = addrs.to_socket_addrs().await?;
 
         for addr in addrs {
-            match mio::net::TcpListener::bind(&addr) {
+            match Async::<std::net::TcpListener>::bind(&addr) {
-                Ok(mio_listener) => {
+                Ok(listener) => {
-                    return Ok(TcpListener {
+                    return Ok(TcpListener { watcher: listener });
-                        watcher: Watcher::new(mio_listener),
-                    });
                 }
                 Err(err) => last_err = Some(err),
             }
@@ -114,13 +114,9 @@ impl TcpListener {
     /// # Ok(()) }) }
     /// ```
     pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
-        let (io, addr) =
+        let (stream, addr) = self.watcher.accept().await?;
-            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 = TcpStream {
-            watcher: Arc::new(Watcher::new(mio_stream)),
+            watcher: Arc::new(stream),
         };
         Ok((stream, addr))
     }
@@ -206,9 +202,10 @@ 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();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         TcpListener {
-            watcher: Watcher::new(mio_listener),
+            watcher: Async::new(listener).expect("TcpListener is known to be good"),
         }
     }
 }
@@ -230,29 +227,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};
+    use crate::os::windows::io::{
-    //
+        AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket,
-    // impl AsRawSocket for TcpListener {
+    };
-    //     fn as_raw_socket(&self) -> RawSocket {
+
-    //         self.raw_socket
+    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 {
+
-    //         net::TcpListener::from_raw_socket(handle).try_into().unwrap()
+    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.raw_socket
+    impl IntoRawSocket for TcpListener {
-    //     }
+        fn into_raw_socket(self) -> RawSocket {
-    // }
+            self.watcher.into_raw_socket()
+        }
+    }
 }

src/net/tcp/stream.rs

@@ -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::rt::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 {
@@ -71,32 +71,22 @@ impl TcpStream {
     /// # Ok(()) }) }
     /// ```
     pub async fn connect<A: ToSocketAddrs>(addrs: A) -> io::Result<TcpStream> {
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         let mut last_err = None;
         let addrs = addrs.to_socket_addrs().await?;
 
         for addr in addrs {
-            // mio's TcpStream::connect is non-blocking and may just be in progress
+            match Async::<std::net::TcpStream>::connect(&addr).await {
-            // when it returns with `Ok`. We therefore wait for write readiness to
+                Ok(stream) => {
-            // be sure the connection has either been established or there was an
+                    return Ok(TcpStream {
-            // error which we check for afterwards.
+                        watcher: Arc::new(stream),
-            let watcher = match mio::net::TcpStream::connect(&addr) {
+                    });
-                Ok(s) => Watcher::new(s),
+                }
                 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 +204,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 +307,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 +343,25 @@ impl Write for &TcpStream {
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        self.watcher
+        Pin::new(&mut &*self.watcher).poll_write(cx, buf)
-            .poll_write_with(cx, |mut inner| inner.write(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<()>> {
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        self.shutdown(std::net::Shutdown::Write)?;
+        Pin::new(&mut &*self.watcher).poll_close(cx)
-        Poll::Ready(Ok(()))
     }
 }
 
 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();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         TcpStream {
-            watcher: Arc::new(Watcher::new(mio_stream)),
+            watcher: Arc::new(Async::new(stream).expect("TcpStream is known to be good")),
         }
     }
 }
@@ -403,23 +392,28 @@ cfg_unix! {
 }
 
 cfg_windows! {
-    // use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
+    use crate::os::windows::io::{
-    //
+        RawSocket, AsRawSocket, FromRawSocket, IntoRawSocket
-    // impl AsRawSocket for TcpStream {
+    };
-    //     fn as_raw_socket(&self) -> RawSocket {
+
-    //         self.raw_socket
+    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 {
+
-    //         net::TcpStream::from_raw_socket(handle).try_into().unwrap()
+    impl FromRawSocket for TcpStream {
-    //     }
+        unsafe fn from_raw_socket(handle: RawSocket) -> TcpStream {
-    // }
+            std::net::TcpStream::from_raw_socket(handle).into()
-    //
+        }
-    // impl IntoRawSocket for TcpListener {
+    }
-    //     fn into_raw_socket(self) -> RawSocket {
+
-    //         self.raw_socket
+    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()
+        }
+    }
 }

src/net/udp/mod.rs

@@ -2,9 +2,9 @@ use std::io;
 use std::net::SocketAddr;
 use std::net::{Ipv4Addr, Ipv6Addr};
 
-use crate::future;
+use smol::Async;
+
 use crate::net::ToSocketAddrs;
-use crate::rt::Watcher;
 use crate::utils::Context as _;
 
 /// A UDP socket.
@@ -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 {
@@ -68,17 +68,15 @@ impl UdpSocket {
     /// # Ok(()) }) }
     /// ```
     pub async fn bind<A: ToSocketAddrs>(addrs: A) -> io::Result<UdpSocket> {
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         let mut last_err = None;
-        let addrs = addrs
+        let addrs = addrs.to_socket_addrs().await?;
-            .to_socket_addrs()
-            .await?;
 
         for addr in addrs {
-            match mio::net::UdpSocket::bind(&addr) {
+            match Async::<std::net::UdpSocket>::bind(&addr) {
-                Ok(mio_socket) => {
+                Ok(socket) => {
-                    return Ok(UdpSocket {
+                    return Ok(UdpSocket { watcher: socket });
-                        watcher: Watcher::new(mio_socket),
-                    });
                 }
                 Err(err) => last_err = Some(err),
             }
@@ -92,6 +90,32 @@ impl UdpSocket {
         }))
     }
 
+    /// Returns the peer address that this listener is connected to.
+    ///
+    /// This can be useful, for example, when connect to port 0 to figure out which port was
+    /// actually connected.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
+    /// #
+    /// use async_std::net::UdpSocket;
+    ///
+    /// let socket1 = UdpSocket::bind("127.0.0.1:0").await?;
+    /// let socket2 = UdpSocket::bind("127.0.0.1:0").await?;
+    /// socket1.connect(socket2.local_addr()?).await?;
+    /// let addr = socket1.peer_addr()?;
+    /// #
+    /// # Ok(()) }) }
+    /// ```
+    pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+        self.watcher
+            .get_ref()
+            .peer_addr()
+            .context(|| String::from("could not get peer address"))
+    }
+
     /// Returns the local address that this listener is bound to.
     ///
     /// This can be useful, for example, when binding to port 0 to figure out which port was
@@ -153,12 +177,10 @@ impl UdpSocket {
             }
         };
 
-        future::poll_fn(|cx| {
+        self.watcher
-            self.watcher
+            .send_to(buf, addr)
-                .poll_write_with(cx, |inner| inner.send_to(buf, &addr))
+            .await
-        })
+            .context(|| format!("could not send packet to {}", addr))
-        .await
-        .context(|| format!("could not send packet to {}", addr))
     }
 
     /// Receives data from the socket.
@@ -181,22 +203,7 @@ impl UdpSocket {
     /// # Ok(()) }) }
     /// ```
     pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
-        future::poll_fn(|cx| {
+        self.watcher.recv_from(buf).await
-            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
-        })
     }
 
     /// Connects the UDP socket to a remote address.
@@ -267,19 +274,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)))
+        self.watcher.send(buf).await
-            .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
-            })
     }
 
     /// Receives data from the socket.
@@ -303,19 +298,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)))
+        self.watcher.recv(buf).await
-            .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
-            })
     }
 
     /// Gets the value of the `SO_BROADCAST` option for this socket.
@@ -498,9 +481,10 @@ 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();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         UdpSocket {
-            watcher: Watcher::new(mio_socket),
+            watcher: Async::new(socket).expect("UdpSocket is known to be good"),
         }
     }
 }
@@ -522,29 +506,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};
+    use crate::os::windows::io::{
-    //
+        RawSocket, AsRawSocket, IntoRawSocket, FromRawSocket
-    // impl AsRawSocket for UdpSocket {
+    };
-    //     fn as_raw_socket(&self) -> RawSocket {
+
-    //         self.raw_socket
+    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 {
+
-    //         net::UdpSocket::from_raw_socket(handle).into()
+    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.raw_socket
+    impl IntoRawSocket for UdpSocket {
-    //     }
+        fn into_raw_socket(self) -> RawSocket {
-    // }
+            self.watcher.into_raw_socket()
+        }
+    }
 }

src/option/from_stream.rs

@@ -4,7 +4,7 @@ use crate::prelude::*;
 use crate::stream::{FromStream, IntoStream};
 use std::convert::identity;
 
-impl<T, V> FromStream<Option<T>> for Option<V>
+impl<T: Send, V> FromStream<Option<T>> for Option<V>
 where
     V: FromStream<T>,
 {
@@ -14,7 +14,10 @@ where
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = Option<T>> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/os/unix/net/datagram.rs

@@ -2,16 +2,14 @@
 
 use std::fmt;
 use std::net::Shutdown;
+use std::os::unix::net::UnixDatagram as StdUnixDatagram;
 
-use mio_uds;
+use smol::Async;
 
 use super::SocketAddr;
-use crate::future;
 use crate::io;
-use crate::rt::Watcher;
 use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use crate::path::Path;
-use crate::task::spawn_blocking;
 
 /// A Unix datagram socket.
 ///
@@ -42,13 +40,15 @@ 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 {
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         UnixDatagram {
-            watcher: Watcher::new(socket),
+            watcher: Async::new(socket).expect("UnixDatagram is known to be good"),
         }
     }
 
@@ -66,9 +66,11 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixDatagram> {
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         let path = path.as_ref().to_owned();
-        let socket = spawn_blocking(move || mio_uds::UnixDatagram::bind(path)).await?;
+        let socket = Async::<StdUnixDatagram>::bind(path)?;
-        Ok(UnixDatagram::new(socket))
+        Ok(UnixDatagram { watcher: socket })
     }
 
     /// Creates a Unix datagram which is not bound to any address.
@@ -85,7 +87,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub fn unbound() -> io::Result<UnixDatagram> {
-        let socket = mio_uds::UnixDatagram::unbound()?;
+        let socket = StdUnixDatagram::unbound()?;
         Ok(UnixDatagram::new(socket))
     }
 
@@ -105,7 +107,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))
@@ -197,11 +199,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
-        future::poll_fn(|cx| {
+        self.watcher.recv_from(buf).await
-            self.watcher
-                .poll_read_with(cx, |inner| inner.recv_from(buf))
-        })
-        .await
     }
 
     /// Receives data from the socket.
@@ -222,7 +220,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.
@@ -242,11 +240,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.send_to(buf, path.as_ref()).await
-            self.watcher
-                .poll_write_with(cx, |inner| inner.send_to(buf, path.as_ref()))
-        })
-        .await
     }
 
     /// Sends data on the socket to the socket's peer.
@@ -267,7 +261,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.
@@ -312,31 +306,35 @@ 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 {
+    fn from(datagram: StdUnixDatagram) -> UnixDatagram {
-        let mio_datagram = mio_uds::UnixDatagram::from_datagram(datagram).unwrap();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         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);
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
-        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()
     }
 }

src/os/unix/net/listener.rs

@@ -1,20 +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 mio_uds;
+use smol::Async;
 
 use super::SocketAddr;
 use super::UnixStream;
-use crate::future;
 use crate::io;
-use crate::rt::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.
 ///
@@ -50,7 +50,7 @@ use crate::task::{spawn_blocking, Context, Poll};
 /// # Ok(()) }) }
 /// ```
 pub struct UnixListener {
-    watcher: Watcher<mio_uds::UnixListener>,
+    watcher: Async<StdUnixListener>,
 }
 
 impl UnixListener {
@@ -68,12 +68,12 @@ impl UnixListener {
     /// # Ok(()) }) }
     /// ```
     pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixListener> {
-        let path = path.as_ref().to_owned();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
-        let listener = spawn_blocking(move || mio_uds::UnixListener::bind(path)).await?;
 
-        Ok(UnixListener {
+        let path = path.as_ref().to_owned();
-            watcher: Watcher::new(listener),
+        let listener = Async::<StdUnixListener>::bind(path)?;
-        })
+
+        Ok(UnixListener { watcher: listener })
     }
 
     /// Accepts a new incoming connection to this listener.
@@ -93,29 +93,14 @@ impl UnixListener {
     /// # Ok(()) }) }
     /// ```
     pub async fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
-        future::poll_fn(|cx| {
+        let (stream, addr) = self.watcher.accept().await?;
-            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,
-                }
-            }));
 
-            match res? {
+        Ok((
-                Some((io, addr)) => {
+            UnixStream {
-                    let mio_stream = mio_uds::UnixStream::from_stream(io)?;
+                watcher: Arc::new(stream),
-                    let stream = UnixStream {
+            },
-                        watcher: Watcher::new(mio_stream),
+            addr,
-                    };
+        ))
-                    Poll::Ready(Ok((stream, addr)))
-                }
-                // This should never happen since `None` is converted to `WouldBlock`
-                None => unreachable!(),
-            }
-        })
-        .await
     }
 
     /// Returns a stream of incoming connections.
@@ -206,19 +191,20 @@ 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 {
+    fn from(listener: StdUnixListener) -> UnixListener {
-        let mio_listener = mio_uds::UnixListener::from_listener(listener).unwrap();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
         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()
     }
 }
 
@@ -231,6 +217,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()
     }
 }

src/os/unix/net/stream.rs

@@ -1,18 +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 mio_uds;
+use smol::Async;
 
 use super::SocketAddr;
 use crate::io::{self, Read, Write};
-use crate::rt::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.
 ///
@@ -37,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 {
@@ -56,16 +57,12 @@ impl UnixStream {
     /// # Ok(()) }) }
     /// ```
     pub async fn connect<P: AsRef<Path>>(path: P) -> io::Result<UnixStream> {
-        let path = path.as_ref().to_owned();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
 
-        spawn_blocking(move || {
+        let path = path.as_ref().to_owned();
-            let std_stream = std::os::unix::net::UnixStream::connect(path)?;
+        let stream = Arc::new(Async::<StdUnixStream>::connect(path).await?);
-            let mio_stream = mio_uds::UnixStream::from_stream(std_stream)?;
+
-            Ok(UnixStream {
+        Ok(UnixStream { watcher: stream })
-                watcher: Watcher::new(mio_stream),
-            })
-        })
-        .await
     }
 
     /// Creates an unnamed pair of connected sockets.
@@ -84,12 +81,14 @@ impl UnixStream {
     /// # Ok(()) }) }
     /// ```
     pub fn pair() -> io::Result<(UnixStream, UnixStream)> {
-        let (a, b) = mio_uds::UnixStream::pair()?;
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
+        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))
     }
@@ -169,7 +168,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)
     }
 }
 
@@ -197,16 +196,15 @@ impl Write for &UnixStream {
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        self.watcher
+        Pin::new(&mut &*self.watcher).poll_write(cx, buf)
-            .poll_write_with(cx, |mut inner| inner.write(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<()>> {
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        Poll::Ready(Ok(()))
+        Pin::new(&mut &*self.watcher).poll_close(cx)
     }
 }
 
@@ -227,19 +225,21 @@ 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 {
+    fn from(stream: StdUnixStream) -> UnixStream {
-        let mio_stream = mio_uds::UnixStream::from_stream(stream).unwrap();
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
+        let stream = Async::new(stream).expect("UnixStream is known to be good");
         UnixStream {
-            watcher: Watcher::new(mio_stream),
+            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()
     }
 }
 
@@ -252,6 +252,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()
     }
 }

src/os/windows/io.rs

@@ -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;
+    }
 }

src/path/path.rs

@@ -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

src/path/pathbuf.rs

@@ -323,7 +323,10 @@ impl<P: AsRef<Path>> stream::Extend<P> for PathBuf {
     fn extend<'a, S: IntoStream<Item = P> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -337,11 +340,14 @@ impl<P: AsRef<Path>> stream::Extend<P> for PathBuf {
 }
 
 #[cfg(feature = "unstable")]
-impl<'b, P: AsRef<Path> + 'b> FromStream<P> for PathBuf {
+impl<'b, P: AsRef<Path> + 'b + Send> FromStream<P> for PathBuf {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = P> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/result/from_stream.rs

@@ -5,6 +5,8 @@ use crate::stream::{FromStream, IntoStream};
 
 impl<T, E, V> FromStream<Result<T, E>> for Result<V, E>
 where
+    T: Send,
+    E: Send,
     V: FromStream<T>,
 {
     /// Takes each element in the stream: if it is an `Err`, no further
@@ -30,7 +32,10 @@ where
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = Result<T, E>> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/rt/mod.rs

@@ -1,23 +1,34 @@
 //! The runtime.
 
+use std::env;
 use std::thread;
 
 use once_cell::sync::Lazy;
 
-use crate::utils::abort_on_panic;
+use crate::future;
 
-pub use reactor::{Reactor, Watcher};
+/// Dummy runtime struct.
-pub use runtime::Runtime;
+pub struct Runtime {}
-
-mod reactor;
-mod runtime;
 
 /// The global runtime.
 pub static RUNTIME: Lazy<Runtime> = Lazy::new(|| {
-    thread::Builder::new()
+    // Create an executor thread pool.
-        .name("async-std/runtime".to_string())
-        .spawn(|| abort_on_panic(|| RUNTIME.run()))
-        .expect("cannot start a runtime thread");
 
-    Runtime::new()
+    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(|| crate::task::block_on(future::pending::<()>()))
+            .expect("cannot start a runtime thread");
+    }
+    Runtime {}
 });

src/rt/reactor.rs

@@ -1,354 +0,0 @@
-use std::fmt;
-use std::sync::{Arc, Mutex};
-use std::time::Duration;
-
-use mio::{self, Evented};
-use slab::Slab;
-
-use crate::io;
-use crate::rt::RUNTIME;
-use crate::task::{Context, Poll, Waker};
-
-/// 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>,
-
-    /// Tasks that are blocked on writing to this I/O handle.
-    writers: Mutex<Writers>,
-}
-
-/// The state of a networking driver.
-pub struct Reactor {
-    /// A mio instance that polls for new events.
-    poller: mio::Poll,
-
-    /// A list into which mio stores events.
-    events: Mutex<mio::Events>,
-
-    /// 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,
-}
-
-/// 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>,
-}
-
-impl Reactor {
-    /// Creates a new reactor for polling I/O events.
-    pub fn new() -> io::Result<Reactor> {
-        let poller = mio::Poll::new()?;
-        let notify_reg = mio::Registration::new2();
-
-        let mut reactor = Reactor {
-            poller,
-            events: Mutex::new(mio::Events::with_capacity(1000)),
-            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(())
-    }
-
-    /// Notifies the reactor so that polling stops blocking.
-    pub fn notify(&self) -> io::Result<()> {
-        self.notify_reg.1.set_readiness(mio::Ready::readable())
-    }
-
-    /// Waits on the poller for new events and wakes up tasks blocked on I/O handles.
-    ///
-    /// Returns `Ok(true)` if at least one new task was woken.
-    pub fn poll(&self, timeout: Option<Duration>) -> io::Result<bool> {
-        let mut events = self.events.lock().unwrap();
-
-        // Block on the poller until at least one new event comes in.
-        self.poller.poll(&mut events, timeout)?;
-
-        // Lock the entire entry table while we're processing new events.
-        let entries = self.entries.lock().unwrap();
-
-        // The number of woken tasks.
-        let mut progress = false;
-
-        for event in events.iter() {
-            let token = event.token();
-
-            if token == self.notify_token {
-                // If this is the notification token, we just need the notification state.
-                self.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.
-                    let reader_interests = mio::Ready::all() - mio::Ready::writable();
-                    if !(readiness & reader_interests).is_empty() {
-                        let mut readers = entry.readers.lock().unwrap();
-                        readers.ready = true;
-                        for w in readers.wakers.drain(..) {
-                            w.wake();
-                            progress = true;
-                        }
-                    }
-
-                    // Wake up writer tasks blocked on this I/O handle.
-                    let writer_interests = mio::Ready::all() - mio::Ready::readable();
-                    if !(readiness & writer_interests).is_empty() {
-                        let mut writers = entry.writers.lock().unwrap();
-                        writers.ready = true;
-                        for w in writers.wakers.drain(..) {
-                            w.wake();
-                            progress = true;
-                        }
-                    }
-                }
-            }
-        }
-
-        Ok(progress)
-    }
-}
-
-/// 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: RUNTIME
-                .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());
-        }
-
-        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());
-        }
-
-        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();
-        RUNTIME
-            .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 {
-            RUNTIME
-                .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()
-    }
-}

src/rt/runtime.rs

@@ -1,415 +0,0 @@
-use std::cell::Cell;
-use std::io;
-use std::iter;
-use std::ptr;
-use std::sync::atomic::{self, Ordering};
-use std::sync::{Arc, Mutex};
-use std::thread;
-use std::time::Duration;
-
-use crossbeam_deque::{Injector, Steal, Stealer, Worker};
-use crossbeam_utils::thread::scope;
-use once_cell::unsync::OnceCell;
-
-use crate::rt::Reactor;
-use crate::sync::Spinlock;
-use crate::task::Runnable;
-use crate::utils::{abort_on_panic, random};
-
-thread_local! {
-    /// A reference to the current machine, if the current thread runs tasks.
-    static MACHINE: OnceCell<Arc<Machine>> = OnceCell::new();
-
-    /// This flag is set to true whenever `task::yield_now()` is invoked.
-    static YIELD_NOW: Cell<bool> = Cell::new(false);
-}
-
-struct Scheduler {
-    /// Set to `true` while a machine is polling the reactor.
-    polling: bool,
-
-    /// Idle processors.
-    processors: Vec<Processor>,
-
-    /// Running machines.
-    machines: Vec<Arc<Machine>>,
-}
-
-/// An async runtime.
-pub struct Runtime {
-    /// The reactor.
-    reactor: Reactor,
-
-    /// The global queue of tasks.
-    injector: Injector<Runnable>,
-
-    /// Handles to local queues for stealing work.
-    stealers: Vec<Stealer<Runnable>>,
-
-    /// The scheduler state.
-    sched: Mutex<Scheduler>,
-}
-
-impl Runtime {
-    /// Creates a new runtime.
-    pub fn new() -> Runtime {
-        let cpus = num_cpus::get().max(1);
-        let processors: Vec<_> = (0..cpus).map(|_| Processor::new()).collect();
-        let stealers = processors.iter().map(|p| p.worker.stealer()).collect();
-
-        Runtime {
-            reactor: Reactor::new().unwrap(),
-            injector: Injector::new(),
-            stealers,
-            sched: Mutex::new(Scheduler {
-                processors,
-                machines: Vec::new(),
-                polling: false,
-            }),
-        }
-    }
-
-    /// Returns a reference to the reactor.
-    pub fn reactor(&self) -> &Reactor {
-        &self.reactor
-    }
-
-    /// Flushes the task slot so that tasks get run more fairly.
-    pub fn yield_now(&self) {
-        YIELD_NOW.with(|flag| flag.set(true));
-    }
-
-    /// Schedules a task.
-    pub fn schedule(&self, task: Runnable) {
-        MACHINE.with(|machine| {
-            // If the current thread is a worker thread, schedule it onto the current machine.
-            // Otherwise, push it into the global task queue.
-            match machine.get() {
-                None => {
-                    self.injector.push(task);
-                    self.notify();
-                }
-                Some(m) => m.schedule(&self, task),
-            }
-        });
-    }
-
-    /// Runs the runtime on the current thread.
-    pub fn run(&self) {
-        scope(|s| {
-            let mut idle = 0;
-            let mut delay = 0;
-
-            loop {
-                // Get a list of new machines to start, if any need to be started.
-                for m in self.make_machines() {
-                    idle = 0;
-
-                    s.builder()
-                        .name("async-std/machine".to_string())
-                        .spawn(move |_| {
-                            abort_on_panic(|| {
-                                let _ = MACHINE.with(|machine| machine.set(m.clone()));
-                                m.run(self);
-                            })
-                        })
-                        .expect("cannot start a machine thread");
-                }
-
-                // Sleep for a bit longer if the scheduler state hasn't changed in a while.
-                if idle > 10 {
-                    delay = (delay * 2).min(10_000);
-                } else {
-                    idle += 1;
-                    delay = 1000;
-                }
-
-                thread::sleep(Duration::from_micros(delay));
-            }
-        })
-        .unwrap();
-    }
-
-    /// Returns a list of machines that need to be started.
-    fn make_machines(&self) -> Vec<Arc<Machine>> {
-        let mut sched = self.sched.lock().unwrap();
-        let mut to_start = Vec::new();
-
-        // If no machine has been polling the reactor in a while, that means the runtime is
-        // overloaded with work and we need to start another machine.
-        if !sched.polling {
-            if let Some(p) = sched.processors.pop() {
-                let m = Arc::new(Machine::new(p));
-                to_start.push(m.clone());
-                sched.machines.push(m);
-            }
-        }
-
-        to_start
-    }
-
-    /// Unparks a thread polling the reactor.
-    fn notify(&self) {
-        atomic::fence(Ordering::SeqCst);
-        self.reactor.notify().unwrap();
-    }
-
-    /// Attempts to poll the reactor without blocking on it.
-    ///
-    /// Returns `Ok(true)` if at least one new task was woken.
-    ///
-    /// This function might not poll the reactor at all so do not rely on it doing anything. Only
-    /// use for optimization.
-    fn quick_poll(&self) -> io::Result<bool> {
-        if let Ok(sched) = self.sched.try_lock() {
-            if !sched.polling {
-                return self.reactor.poll(Some(Duration::from_secs(0)));
-            }
-        }
-        Ok(false)
-    }
-}
-
-/// A thread running a processor.
-struct Machine {
-    /// Holds the processor until it gets stolen.
-    processor: Spinlock<Option<Processor>>,
-}
-
-impl Machine {
-    /// Creates a new machine running a processor.
-    fn new(p: Processor) -> Machine {
-        Machine {
-            processor: Spinlock::new(Some(p)),
-        }
-    }
-
-    /// Schedules a task onto the machine.
-    fn schedule(&self, rt: &Runtime, task: Runnable) {
-        match self.processor.lock().as_mut() {
-            None => {
-                rt.injector.push(task);
-                rt.notify();
-            }
-            Some(p) => p.schedule(rt, task),
-        }
-    }
-
-    /// Finds the next runnable task.
-    fn find_task(&self, rt: &Runtime) -> Steal<Runnable> {
-        let mut retry = false;
-
-        // First try finding a task in the local queue or in the global queue.
-        if let Some(p) = self.processor.lock().as_mut() {
-            if let Some(task) = p.pop_task() {
-                return Steal::Success(task);
-            }
-
-            match p.steal_from_global(rt) {
-                Steal::Empty => {}
-                Steal::Retry => retry = true,
-                Steal::Success(task) => return Steal::Success(task),
-            }
-        }
-
-        // Try polling the reactor, but don't block on it.
-        let progress = rt.quick_poll().unwrap();
-
-        // Try finding a task in the local queue, which might hold tasks woken by the reactor. If
-        // the local queue is still empty, try stealing from other processors.
-        if let Some(p) = self.processor.lock().as_mut() {
-            if progress {
-                if let Some(task) = p.pop_task() {
-                    return Steal::Success(task);
-                }
-            }
-
-            match p.steal_from_others(rt) {
-                Steal::Empty => {}
-                Steal::Retry => retry = true,
-                Steal::Success(task) => return Steal::Success(task),
-            }
-        }
-
-        if retry { Steal::Retry } else { Steal::Empty }
-    }
-
-    /// Runs the machine on the current thread.
-    fn run(&self, rt: &Runtime) {
-        /// 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 = 10;
-        /// Number of runs in a row before the global queue is inspected.
-        const RUNS: u32 = 64;
-
-        // The number of times the thread found work in a row.
-        let mut runs = 0;
-        // The number of times the thread didn't find work in a row.
-        let mut fails = 0;
-
-        loop {
-            // Check if `task::yield_now()` was invoked and flush the slot if so.
-            YIELD_NOW.with(|flag| {
-                if flag.replace(false) {
-                    if let Some(p) = self.processor.lock().as_mut() {
-                        p.flush_slot(rt);
-                    }
-                }
-            });
-
-            // After a number of runs in a row, do some work to ensure no task is left behind
-            // indefinitely. Poll the reactor, steal tasks from the global queue, and flush the
-            // task slot.
-            if runs >= RUNS {
-                runs = 0;
-                rt.quick_poll().unwrap();
-
-                if let Some(p) = self.processor.lock().as_mut() {
-                    if let Steal::Success(task) = p.steal_from_global(rt) {
-                        p.schedule(rt, task);
-                    }
-
-                    p.flush_slot(rt);
-                }
-            }
-
-            // Try to find a runnable task.
-            if let Steal::Success(task) = self.find_task(rt) {
-                task.run();
-                runs += 1;
-                fails = 0;
-                continue;
-            }
-
-            fails += 1;
-
-            // Yield the current thread a few times.
-            if fails <= YIELDS {
-                thread::yield_now();
-                continue;
-            }
-
-            // Put the current thread to sleep a few times.
-            if fails <= YIELDS + SLEEPS {
-                let opt_p = self.processor.lock().take();
-                thread::sleep(Duration::from_micros(10));
-                *self.processor.lock() = opt_p;
-                continue;
-            }
-
-            let mut sched = rt.sched.lock().unwrap();
-
-            // One final check for available tasks while the scheduler is locked.
-            if let Some(task) = iter::repeat_with(|| self.find_task(rt))
-                .find(|s| !s.is_retry())
-                .and_then(|s| s.success())
-            {
-                self.schedule(rt, task);
-                continue;
-            }
-
-            // If another thread is already blocked on the reactor, there is no point in keeping
-            // the current thread around since there is too little work to do.
-            if sched.polling {
-                break;
-            }
-
-            // Take out the machine associated with the current thread.
-            let m = match sched
-                .machines
-                .iter()
-                .position(|elem| ptr::eq(&**elem, self))
-            {
-                None => break, // The processor was stolen.
-                Some(pos) => sched.machines.swap_remove(pos),
-            };
-
-            // Unlock the schedule poll the reactor until new I/O events arrive.
-            sched.polling = true;
-            drop(sched);
-            rt.reactor.poll(None).unwrap();
-
-            // Lock the scheduler again and re-register the machine.
-            sched = rt.sched.lock().unwrap();
-            sched.polling = false;
-            sched.machines.push(m);
-
-            runs = 0;
-            fails = 0;
-        }
-
-        // When shutting down the thread, take the processor out if still available.
-        let opt_p = self.processor.lock().take();
-
-        // Return the processor to the scheduler and remove the machine.
-        if let Some(p) = opt_p {
-            let mut sched = rt.sched.lock().unwrap();
-            sched.processors.push(p);
-            sched.machines.retain(|elem| !ptr::eq(&**elem, self));
-        }
-    }
-}
-
-struct Processor {
-    /// The local task queue.
-    worker: Worker<Runnable>,
-
-    /// Contains the next task to run as an optimization that skips the queue.
-    slot: Option<Runnable>,
-}
-
-impl Processor {
-    /// Creates a new processor.
-    fn new() -> Processor {
-        Processor {
-            worker: Worker::new_fifo(),
-            slot: None,
-        }
-    }
-
-    /// Schedules a task to run on this processor.
-    fn schedule(&mut self, rt: &Runtime, task: Runnable) {
-        match self.slot.replace(task) {
-            None => {}
-            Some(task) => {
-                self.worker.push(task);
-                rt.notify();
-            }
-        }
-    }
-
-    /// Flushes a task from the slot into the local queue.
-    fn flush_slot(&mut self, rt: &Runtime) {
-        if let Some(task) = self.slot.take() {
-            self.worker.push(task);
-            rt.notify();
-        }
-    }
-
-    /// Pops a task from this processor.
-    fn pop_task(&mut self) -> Option<Runnable> {
-        self.slot.take().or_else(|| self.worker.pop())
-    }
-
-    /// Steals a task from the global queue.
-    fn steal_from_global(&self, rt: &Runtime) -> Steal<Runnable> {
-        rt.injector.steal_batch_and_pop(&self.worker)
-    }
-
-    /// Steals a task from other processors.
-    fn steal_from_others(&self, rt: &Runtime) -> Steal<Runnable> {
-        // Pick a random starting point in the list of queues.
-        let len = rt.stealers.len();
-        let start = random(len as u32) as usize;
-
-        // Create an iterator over stealers that starts from the chosen point.
-        let (l, r) = rt.stealers.split_at(start);
-        let stealers = r.iter().chain(l.iter());
-
-        // Try stealing a batch of tasks from each queue.
-        stealers
-            .map(|s| s.steal_batch_and_pop(&self.worker))
-            .collect()
-    }
-}

src/stream/extend.rs

@@ -34,7 +34,9 @@ pub trait Extend<A> {
     fn extend<'a, T: IntoStream<Item = A> + 'a>(
         &'a mut self,
         stream: T,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>>;
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <T as IntoStream>::IntoStream: Send;
 }
 
 /// Extends a collection with the contents of a stream.
@@ -69,6 +71,7 @@ pub async fn extend<'a, C, T, S>(collection: &mut C, stream: S)
 where
     C: Extend<T>,
     S: IntoStream<Item = T> + 'a,
+    <S as IntoStream>::IntoStream: Send,
 {
     Extend::extend(collection, stream).await
 }

src/stream/from_stream.rs

@@ -72,7 +72,10 @@ use crate::stream::IntoStream;
 /// impl FromStream<i32> for MyCollection {
 ///     fn from_stream<'a, S: IntoStream<Item = i32> + 'a>(
 ///         stream: S,
-///     ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+///     ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+///    where
+///        <S as IntoStream>::IntoStream: Send,
+///    {
 ///         let stream = stream.into_stream();
 ///
 ///         Box::pin(async move {
@@ -107,12 +110,12 @@ use crate::stream::IntoStream;
 /// assert_eq!(c.0, vec![5, 5, 5, 5, 5]);
 /// #
 /// # Ok(()) }) }
-///```
+/// ```
 ///
 /// [`IntoStream`]: trait.IntoStream.html
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-pub trait FromStream<T> {
+pub trait FromStream<T: Send> {
     /// Creates a value from a stream.
     ///
     /// # Examples
@@ -135,5 +138,7 @@ pub trait FromStream<T> {
     /// ```
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>>;
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send;
 }

src/stream/interval.rs

@@ -1,10 +1,10 @@
+use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
-use crate::future::Future;
 use crate::stream::Stream;
-use futures_timer::Delay;
+use crate::utils::{timer_after, 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,
 }
 
@@ -72,7 +72,7 @@ impl Stream for Interval {
             return Poll::Pending;
         }
         let interval = self.interval;
-        self.delay.reset(interval);
+        let _ = std::mem::replace(&mut self.delay, timer_after(interval));
         Poll::Ready(Some(()))
     }
 }

src/stream/stream/delay.rs

@@ -6,6 +6,7 @@ use pin_project_lite::pin_project;
 
 use crate::stream::Stream;
 use crate::task::{Context, Poll};
+use crate::utils::{timer_after, 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,
         }
     }

src/stream/stream/flat_map.rs

@@ -50,14 +50,15 @@ where
         let mut this = self.project();
         loop {
             if let Some(inner) = this.inner_stream.as_mut().as_pin_mut() {
-                if let item @ Some(_) = futures_core::ready!(inner.poll_next(cx)) {
+                match futures_core::ready!(inner.poll_next(cx)) {
-                    return Poll::Ready(item);
+                    item @ Some(_) => return Poll::Ready(item),
+                    None => this.inner_stream.set(None),
                 }
             }
 
             match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
+                inner @ Some(_) => this.inner_stream.set(inner.map(IntoStream::into_stream)),
                 None => return Poll::Ready(None),
-                Some(inner) => this.inner_stream.set(Some(inner.into_stream())),
             }
         }
     }

src/stream/stream/flatten.rs

@@ -52,14 +52,15 @@ where
         let mut this = self.project();
         loop {
             if let Some(inner) = this.inner_stream.as_mut().as_pin_mut() {
-                if let item @ Some(_) = futures_core::ready!(inner.poll_next(cx)) {
+                match futures_core::ready!(inner.poll_next(cx)) {
-                    return Poll::Ready(item);
+                    item @ Some(_) => return Poll::Ready(item),
+                    None => this.inner_stream.set(None),
                 }
             }
 
             match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
+                inner @ Some(_) => this.inner_stream.set(inner.map(IntoStream::into_stream)),
                 None => return Poll::Ready(None),
-                Some(inner) => this.inner_stream.set(Some(inner.into_stream())),
             }
         }
     }

src/stream/stream/max.rs

@@ -1,7 +1,6 @@
 use core::cmp::{Ord, Ordering};
-use core::marker::PhantomData;
-use core::pin::Pin;
 use core::future::Future;
+use core::pin::Pin;
 
 use pin_project_lite::pin_project;
 
@@ -11,29 +10,23 @@ use crate::task::{Context, Poll};
 pin_project! {
     #[doc(hidden)]
     #[allow(missing_debug_implementations)]
-    pub struct MaxFuture<S, F, T> {
+    pub struct MaxFuture<S, T> {
         #[pin]
         stream: S,
-        _compare: PhantomData<F>,
         max: Option<T>,
     }
 }
 
-impl<S, F, T> MaxFuture<S, F, T> {
+impl<S, T> MaxFuture<S, T> {
     pub(super) fn new(stream: S) -> Self {
-        Self {
+        Self { stream, max: None }
-            stream,
-            _compare: PhantomData,
-            max: None,
-        }
     }
 }
 
-impl<S, F> Future for MaxFuture<S, F, S::Item>
+impl<S> Future for MaxFuture<S, S::Item>
 where
     S: Stream,
     S::Item: Ord,
-    F: FnMut(&S::Item, &S::Item) -> Ordering,
 {
     type Output = Option<S::Item>;
 

src/stream/stream/min.rs

@@ -1,7 +1,6 @@
 use core::cmp::{Ord, Ordering};
-use core::marker::PhantomData;
-use core::pin::Pin;
 use core::future::Future;
+use core::pin::Pin;
 
 use pin_project_lite::pin_project;
 
@@ -11,29 +10,23 @@ use crate::task::{Context, Poll};
 pin_project! {
     #[doc(hidden)]
     #[allow(missing_debug_implementations)]
-    pub struct MinFuture<S, F, T> {
+    pub struct MinFuture<S, T> {
         #[pin]
         stream: S,
-        _compare: PhantomData<F>,
         min: Option<T>,
     }
 }
 
-impl<S, F, T> MinFuture<S, F, T> {
+impl<S, T> MinFuture<S, T> {
     pub(super) fn new(stream: S) -> Self {
-        Self {
+        Self { stream, min: None }
-            stream,
-            _compare: PhantomData,
-            min: None,
-        }
     }
 }
 
-impl<S, F> Future for MinFuture<S, F, S::Item>
+impl<S> Future for MinFuture<S, S::Item>
 where
     S: Stream,
     S::Item: Ord,
-    F: FnMut(&S::Item, &S::Item) -> Ordering,
 {
     type Output = Option<S::Item>;
 

src/stream/stream/mod.rs

@@ -1011,7 +1011,7 @@ extension_trait! {
 
             # Examples
 
-            ```ignore
+            ```
             # fn main() { async_std::task::block_on(async {
             #
             use async_std::prelude::*;
@@ -1028,12 +1028,12 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn max<F>(
+        fn max(
             self,
-        ) -> impl Future<Output = Option<Self::Item>> [MaxFuture<Self, F, Self::Item>]
+        ) -> impl Future<Output = Option<Self::Item>> [MaxFuture<Self, Self::Item>]
         where
             Self: Sized,
-            F: FnMut(&Self::Item, &Self::Item) -> Ordering,
+            Self::Item: Ord,
         {
             MaxFuture::new(self)
         }
@@ -1044,7 +1044,7 @@ extension_trait! {
 
             # Examples
 
-            ```ignore
+            ```
             # fn main() { async_std::task::block_on(async {
             #
             use async_std::prelude::*;
@@ -1061,12 +1061,12 @@ extension_trait! {
             # }) }
             ```
         "#]
-        fn min<F>(
+        fn min(
             self,
-        ) -> impl Future<Output = Option<Self::Item>> [MinFuture<Self, F, Self::Item>]
+        ) -> impl Future<Output = Option<Self::Item>> [MinFuture<Self, Self::Item>]
         where
             Self: Sized,
-            F: FnMut(&Self::Item, &Self::Item) -> Ordering,
+            Self::Item: Ord,
         {
             MinFuture::new(self)
         }
@@ -1888,10 +1888,11 @@ extension_trait! {
         #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
         fn collect<'a, B>(
             self,
-        ) -> impl Future<Output = B> + 'a [Pin<Box<dyn Future<Output = B> + 'a>>]
+        ) -> impl Future<Output = B> + 'a [Pin<Box<dyn Future<Output = B> + 'a + Send>>]
         where
-            Self: Sized + 'a,
+            Self: Sized + 'a + Send,
             B: FromStream<Self::Item>,
+            Self::Item: Send,
         {
             FromStream::from_stream(self)
         }

src/stream/stream/throttle.rs

@@ -2,11 +2,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_after, 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(*this.duration);
+                let _ = std::mem::replace(&mut *this.delay, timer_after(*this.duration));
                 Poll::Ready(Some(v))
             }
         }

src/stream/stream/timeout.rs

@@ -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_after, 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 }
     }

src/string/extend.rs

@@ -8,7 +8,10 @@ impl stream::Extend<char> for String {
     fn extend<'a, S: IntoStream<Item = char> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
         self.reserve(stream.size_hint().0);
 
@@ -26,7 +29,10 @@ impl<'b> stream::Extend<&'b char> for String {
     fn extend<'a, S: IntoStream<Item = &'b char> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -43,7 +49,10 @@ impl<'b> stream::Extend<&'b str> for String {
     fn extend<'a, S: IntoStream<Item = &'b str> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -60,7 +69,10 @@ impl stream::Extend<String> for String {
     fn extend<'a, S: IntoStream<Item = String> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -77,7 +89,10 @@ impl<'b> stream::Extend<Cow<'b, str>> for String {
     fn extend<'a, S: IntoStream<Item = Cow<'b, str>> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/string/from_stream.rs

@@ -8,7 +8,10 @@ impl FromStream<char> for String {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = char> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -23,7 +26,10 @@ impl<'b> FromStream<&'b char> for String {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = &'b char> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -38,7 +44,10 @@ impl<'b> FromStream<&'b str> for String {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = &'b str> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -53,7 +62,10 @@ impl FromStream<String> for String {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = String> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -68,7 +80,10 @@ impl<'b> FromStream<Cow<'b, str>> for String {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = Cow<'b, str>> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/sync/barrier.rs

@@ -1,6 +1,4 @@
-use broadcaster::BroadcastChannel;
+use crate::sync::{Condvar,Mutex};
-
-use crate::sync::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)>,
+    cvar: Condvar,
-    n: usize,
+    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 {
+    pub fn new(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;
-        }
-
         Barrier {
             state: Mutex::new(BarrierState {
-                waker,
                 count: 0,
                 generation_id: 1,
             }),
-            n,
+            cvar: Condvar::new(),
-            wait,
+            num_tasks: n,
         }
     }
 
@@ -143,35 +129,20 @@ impl Barrier {
     /// # });
     /// ```
     pub async fn wait(&self) -> BarrierWaitResult {
-        let mut lock = self.state.lock().await;
+        let mut state = self.state.lock().await;
-        let local_gen = lock.generation_id;
+        let local_gen = state.generation_id;
+        state.count += 1;
 
-        lock.count += 1;
+        if state.count < self.num_tasks {
-
+            while local_gen == state.generation_id && state.count < self.num_tasks {
-        if lock.count < self.n {
+                state = self.cvar.wait(state).await;
-            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;
             }
 
             BarrierWaitResult(false)
         } else {
-            lock.count = 0;
+            state.count = 0;
-            lock.generation_id = lock.generation_id.wrapping_add(1);
+            state.generation_id = state.generation_id.wrapping_add(1);
-
+            self.cvar.notify_all();
-            lock.waker
-                .send(&(lock.generation_id, lock.count))
-                .await
-                .expect("there should be at least one receiver");
-
             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;

src/sync/channel.rs

@@ -22,8 +22,8 @@ use crate::sync::WakerSet;
 /// This channel has a buffer that can hold at most `cap` messages at a time.
 ///
 /// Senders and receivers can be cloned. When all senders associated with a channel get dropped, it
-/// becomes closed. Receive operations on a closed and empty channel return `None` instead of
+/// becomes closed. Receive operations on a closed and empty channel return [RecvError] instead of
-/// trying to await a message.
+/// trying to await a message when using [Receiver::recv] or `None` when used as a [Stream].
 ///
 /// # Panics
 ///
@@ -376,7 +376,7 @@ impl<T> Receiver<T> {
     ///
     /// If the channel is empty and still has senders, this method
     /// will wait until a message is sent into it. Once all senders
-    /// have been dropped it will return `None`.
+    /// have been dropped it will return [RecvError].
     ///
     /// # Examples
     ///
@@ -992,6 +992,7 @@ impl<T> Drop for Channel<T> {
 /// An error returned from the `try_send` method.
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[derive(PartialEq, Eq)]
 pub enum TrySendError<T> {
     /// The channel is full but not disconnected.
     Full(T),
@@ -1023,7 +1024,7 @@ impl<T> Display for TrySendError<T> {
 /// An error returned from the `try_recv` method.
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq)]
 pub enum TryRecvError {
     /// The channel is empty but not disconnected.
     Empty,
@@ -1046,7 +1047,7 @@ impl Display for TryRecvError {
 /// An error returned from the `recv` method.
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct RecvError;
 
 impl Error for RecvError {}

src/sync/condvar.rs

@@ -2,7 +2,7 @@ use std::fmt;
 use std::pin::Pin;
 use std::time::Duration;
 
-use super::mutex::{guard_lock, MutexGuard};
+use super::MutexGuard;
 use crate::future::{timeout, Future};
 use crate::sync::WakerSet;
 use crate::task::{Context, Poll};
@@ -120,7 +120,7 @@ impl Condvar {
     /// ```
     #[allow(clippy::needless_lifetimes)]
     pub async fn wait<'a, T>(&self, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> {
-        let mutex = guard_lock(&guard);
+        let mutex = MutexGuard::source(&guard);
 
         self.await_notify(guard).await;
 
@@ -228,7 +228,7 @@ impl Condvar {
         guard: MutexGuard<'a, T>,
         dur: Duration,
     ) -> (MutexGuard<'a, T>, WaitTimeoutResult) {
-        let mutex = guard_lock(&guard);
+        let mutex = MutexGuard::source(&guard);
         match timeout(dur, self.wait(guard)).await {
             Ok(guard) => (guard, WaitTimeoutResult(false)),
             Err(_) => (mutex.lock().await, WaitTimeoutResult(true)),
@@ -281,7 +281,7 @@ impl Condvar {
     where
         F: FnMut(&mut T) -> bool,
     {
-        let mutex = guard_lock(&guard);
+        let mutex = MutexGuard::source(&guard);
         match timeout(dur, self.wait_until(guard, condition)).await {
             Ok(guard) => (guard, WaitTimeoutResult(false)),
             Err(_) => (mutex.lock().await, WaitTimeoutResult(true)),

src/sync/mod.rs

@@ -176,10 +176,11 @@
 #[doc(inline)]
 pub use std::sync::{Arc, Weak};
 
-pub use mutex::{Mutex, MutexGuard};
+#[doc(inline)]
+pub use async_mutex::{Mutex, MutexGuard};
+
 pub use rwlock::{RwLock, RwLockReadGuard, RwLockWriteGuard};
 
-mod mutex;
 mod rwlock;
 
 cfg_unstable! {
@@ -194,8 +195,3 @@ cfg_unstable! {
 
 pub(crate) mod waker_set;
 pub(crate) use waker_set::WakerSet;
-
-cfg_default! {
-    pub(crate) mod spin_lock;
-    pub(crate) use spin_lock::Spinlock;
-}

src/sync/mutex.rs

@@ -1,294 +0,0 @@
-use std::cell::UnsafeCell;
-use std::fmt;
-use std::ops::{Deref, DerefMut};
-use std::pin::Pin;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::future::Future;
-
-use crate::sync::WakerSet;
-use crate::task::{Context, Poll};
-
-/// A mutual exclusion primitive for protecting shared data.
-///
-/// This type is an async version of [`std::sync::Mutex`].
-///
-/// [`std::sync::Mutex`]: https://doc.rust-lang.org/std/sync/struct.Mutex.html
-///
-/// # Examples
-///
-/// ```
-/// # async_std::task::block_on(async {
-/// #
-/// use async_std::sync::{Arc, Mutex};
-/// use async_std::task;
-///
-/// let m = Arc::new(Mutex::new(0));
-/// let mut tasks = vec![];
-///
-/// for _ in 0..10 {
-///     let m = m.clone();
-///     tasks.push(task::spawn(async move {
-///         *m.lock().await += 1;
-///     }));
-/// }
-///
-/// for t in tasks {
-///     t.await;
-/// }
-/// assert_eq!(*m.lock().await, 10);
-/// #
-/// # })
-/// ```
-pub struct Mutex<T: ?Sized> {
-    locked: AtomicBool,
-    wakers: WakerSet,
-    value: UnsafeCell<T>,
-}
-
-unsafe impl<T: ?Sized + Send> Send for Mutex<T> {}
-unsafe impl<T: ?Sized + Send> Sync for Mutex<T> {}
-
-impl<T> Mutex<T> {
-    /// Creates a new mutex.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use async_std::sync::Mutex;
-    ///
-    /// let mutex = Mutex::new(0);
-    /// ```
-    pub fn new(t: T) -> Mutex<T> {
-        Mutex {
-            locked: AtomicBool::new(false),
-            wakers: WakerSet::new(),
-            value: UnsafeCell::new(t),
-        }
-    }
-}
-
-impl<T: ?Sized> Mutex<T> {
-    /// Acquires the lock.
-    ///
-    /// Returns a guard that releases the lock when dropped.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// # async_std::task::block_on(async {
-    /// #
-    /// use async_std::sync::{Arc, Mutex};
-    /// use async_std::task;
-    ///
-    /// let m1 = Arc::new(Mutex::new(10));
-    /// let m2 = m1.clone();
-    ///
-    /// task::spawn(async move {
-    ///     *m1.lock().await = 20;
-    /// })
-    /// .await;
-    ///
-    /// assert_eq!(*m2.lock().await, 20);
-    /// #
-    /// # })
-    /// ```
-    pub async fn lock(&self) -> MutexGuard<'_, T> {
-        pub struct LockFuture<'a, T: ?Sized> {
-            mutex: &'a Mutex<T>,
-            opt_key: Option<usize>,
-        }
-
-        impl<'a, T: ?Sized> Future for LockFuture<'a, T> {
-            type Output = MutexGuard<'a, T>;
-
-            fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-                loop {
-                    // If the current task is in the set, remove it.
-                    if let Some(key) = self.opt_key.take() {
-                        self.mutex.wakers.remove(key);
-                    }
-
-                    // Try acquiring the lock.
-                    match self.mutex.try_lock() {
-                        Some(guard) => return Poll::Ready(guard),
-                        None => {
-                            // Insert this lock operation.
-                            self.opt_key = Some(self.mutex.wakers.insert(cx));
-
-                            // If the mutex is still locked, return.
-                            if self.mutex.locked.load(Ordering::SeqCst) {
-                                return Poll::Pending;
-                            }
-                        }
-                    }
-                }
-            }
-        }
-
-        impl<T: ?Sized> Drop for LockFuture<'_, T> {
-            fn drop(&mut self) {
-                // If the current task is still in the set, that means it is being cancelled now.
-                if let Some(key) = self.opt_key {
-                    self.mutex.wakers.cancel(key);
-                }
-            }
-        }
-
-        LockFuture {
-            mutex: self,
-            opt_key: None,
-        }
-        .await
-    }
-
-    /// Attempts to acquire the lock.
-    ///
-    /// If the lock could not be acquired at this time, then [`None`] is returned. Otherwise, a
-    /// guard is returned that releases the lock when dropped.
-    ///
-    /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// # async_std::task::block_on(async {
-    /// #
-    /// use async_std::sync::{Arc, Mutex};
-    /// use async_std::task;
-    ///
-    /// let m1 = Arc::new(Mutex::new(10));
-    /// let m2 = m1.clone();
-    ///
-    /// task::spawn(async move {
-    ///     if let Some(mut guard) = m1.try_lock() {
-    ///         *guard = 20;
-    ///     } else {
-    ///         println!("try_lock failed");
-    ///     }
-    /// })
-    /// .await;
-    ///
-    /// assert_eq!(*m2.lock().await, 20);
-    /// #
-    /// # })
-    /// ```
-    #[inline]
-    pub fn try_lock(&self) -> Option<MutexGuard<'_, T>> {
-        if !self.locked.swap(true, Ordering::SeqCst) {
-            Some(MutexGuard(self))
-        } else {
-            None
-        }
-    }
-
-    /// Consumes the mutex, returning the underlying data.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use async_std::sync::Mutex;
-    ///
-    /// let mutex = Mutex::new(10);
-    /// assert_eq!(mutex.into_inner(), 10);
-    /// ```
-    pub fn into_inner(self) -> T where T: Sized {
-        self.value.into_inner()
-    }
-
-    /// Returns a mutable reference to the underlying data.
-    ///
-    /// Since this call borrows the mutex mutably, no actual locking takes place -- the mutable
-    /// borrow statically guarantees no locks exist.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// # async_std::task::block_on(async {
-    /// #
-    /// use async_std::sync::Mutex;
-    ///
-    /// let mut mutex = Mutex::new(0);
-    /// *mutex.get_mut() = 10;
-    /// assert_eq!(*mutex.lock().await, 10);
-    /// #
-    /// # })
-    /// ```
-    pub fn get_mut(&mut self) -> &mut T {
-        unsafe { &mut *self.value.get() }
-    }
-}
-
-impl<T: ?Sized + fmt::Debug> fmt::Debug for Mutex<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        struct Locked;
-        impl fmt::Debug for Locked {
-            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-                f.write_str("<locked>")
-            }
-        }
-
-        match self.try_lock() {
-            None => f.debug_struct("Mutex").field("data", &Locked).finish(),
-            Some(guard) => f.debug_struct("Mutex").field("data", &&*guard).finish(),
-        }
-    }
-}
-
-impl<T> From<T> for Mutex<T> {
-    fn from(val: T) -> Mutex<T> {
-        Mutex::new(val)
-    }
-}
-
-impl<T: ?Sized + Default> Default for Mutex<T> {
-    fn default() -> Mutex<T> {
-        Mutex::new(Default::default())
-    }
-}
-
-/// A guard that releases the lock when dropped.
-pub struct MutexGuard<'a, T: ?Sized>(&'a Mutex<T>);
-
-unsafe impl<T: ?Sized + Send> Send for MutexGuard<'_, T> {}
-unsafe impl<T: ?Sized + Sync> Sync for MutexGuard<'_, T> {}
-
-impl<T: ?Sized> Drop for MutexGuard<'_, T> {
-    fn drop(&mut self) {
-        // Use `SeqCst` ordering to synchronize with `WakerSet::insert()` and `WakerSet::update()`.
-        self.0.locked.store(false, Ordering::SeqCst);
-
-        // Notify a blocked `lock()` operation if none were notified already.
-        self.0.wakers.notify_any();
-    }
-}
-
-impl<T: ?Sized +fmt::Debug> fmt::Debug for MutexGuard<'_, T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        fmt::Debug::fmt(&**self, f)
-    }
-}
-
-impl<T: ?Sized + fmt::Display> fmt::Display for MutexGuard<'_, T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        (**self).fmt(f)
-    }
-}
-
-impl<T: ?Sized> Deref for MutexGuard<'_, T> {
-    type Target = T;
-
-    fn deref(&self) -> &T {
-        unsafe { &*self.0.value.get() }
-    }
-}
-
-impl<T: ?Sized> DerefMut for MutexGuard<'_, T> {
-    fn deref_mut(&mut self) -> &mut 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
-}

src/sync/spin_lock.rs

@@ -1,89 +0,0 @@
-use std::cell::UnsafeCell;
-use std::ops::{Deref, DerefMut};
-use std::sync::atomic::{AtomicBool, Ordering};
-
-use crossbeam_utils::Backoff;
-
-/// A simple spinlock.
-#[derive(Debug)]
-pub struct Spinlock<T> {
-    locked: AtomicBool,
-    value: UnsafeCell<T>,
-}
-
-unsafe impl<T: Send> Send for Spinlock<T> {}
-unsafe impl<T: Send> Sync for Spinlock<T> {}
-
-impl<T> Spinlock<T> {
-    /// Returns a new spinlock initialized with `value`.
-    pub const fn new(value: T) -> Spinlock<T> {
-        Spinlock {
-            locked: AtomicBool::new(false),
-            value: UnsafeCell::new(value),
-        }
-    }
-
-    /// Locks the spinlock.
-    pub fn lock(&self) -> SpinlockGuard<'_, T> {
-        let backoff = Backoff::new();
-        while self.locked.compare_and_swap(false, true, Ordering::Acquire) {
-            backoff.snooze();
-        }
-        SpinlockGuard { parent: self }
-    }
-}
-
-/// A guard holding a spinlock locked.
-#[derive(Debug)]
-pub struct SpinlockGuard<'a, T> {
-    parent: &'a Spinlock<T>,
-}
-
-unsafe impl<T: Send> Send for SpinlockGuard<'_, T> {}
-unsafe impl<T: Sync> Sync for SpinlockGuard<'_, T> {}
-
-impl<'a, T> Drop for SpinlockGuard<'a, T> {
-    fn drop(&mut self) {
-        self.parent.locked.store(false, Ordering::Release);
-    }
-}
-
-impl<'a, T> Deref for SpinlockGuard<'a, T> {
-    type Target = T;
-
-    fn deref(&self) -> &T {
-        unsafe { &*self.parent.value.get() }
-    }
-}
-
-impl<'a, T> DerefMut for SpinlockGuard<'a, T> {
-    fn deref_mut(&mut self) -> &mut T {
-        unsafe { &mut *self.parent.value.get() }
-    }
-}
-
-#[test]
-fn spinlock() {
-    use std::sync::Arc;
-
-    use crate::sync::{Spinlock};
-    use crate::task;
-
-    task::block_on(async {
-
-        let m = Arc::new(Spinlock::new(0));
-        let mut tasks = vec![];
-
-        for _ in 0..10 {
-            let m = m.clone();
-            tasks.push(task::spawn(async move {
-                *m.lock() += 1;
-            }));
-        }
-
-        for t in tasks {
-            t.await;
-        }
-        assert_eq!(*m.lock(), 10);
-    })
-}

src/task/block_on.rs

@@ -1,13 +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 crate::task::Builder;
-use kv_log_macro::trace;
-
-use crate::task::{Context, Poll, Task, Waker};
 
 /// Spawns a task and blocks the current thread on its result.
 ///
@@ -32,99 +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.
+    Builder::new().blocking(future)
-    let task = Task::new(None);
-
-    // Log this `block_on` operation.
-    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! {
-            trace!("completed", {
-                task_id: Task::get_current(|t| t.id().0),
-            });
-        }
-
-        future.await
-    };
-
-    // Run the future as a task.
-    unsafe { Task::set_current(&task, || run(future)) }
 }
 
-/// Blocks the current thread on a future's result.
+/// Spawns a task and waits for it to finish.
-fn run<F, T>(future: F) -> T
+#[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! {
+    Builder::new().local(future).unwrap();
-        // 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();
-        }
-    })
 }

src/task/builder.rs

@@ -1,11 +1,13 @@
+use std::cell::Cell;
 use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
 
-use kv_log_macro::trace;
+use pin_project_lite::pin_project;
 
 use crate::io;
-use crate::rt::RUNTIME;
+use crate::task::{JoinHandle, Task, TaskLocalsWrapper};
-use crate::task::{JoinHandle, Task};
-use crate::utils::abort_on_panic;
 
 /// Task builder that configures the settings of a new task.
 #[derive(Debug, Default)]
@@ -27,52 +29,174 @@ 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 wrapped = self.build(future);
-        let task = Task::new(self.name);
 
-        // Log this `spawn` operation.
+        kv_log_macro::trace!("spawn", {
-        trace!("spawn", {
+            task_id: wrapped.tag.id().0,
-            task_id: task.id().0,
+            parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
-            parent_task_id: Task::get_current(|t| t.id().0).unwrap_or(0),
         });
 
-        let future = async move {
+        let task = wrapped.tag.task().clone();
-            // Drop task-locals on exit.
+        let smol_task = smol::Task::spawn(wrapped).into();
-            defer! {
+
-                Task::get_current(|t| unsafe { t.drop_locals() });
+        Ok(JoinHandle::new(smol_task, task))
+    }
+
+    /// 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);
+
+        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),
+        });
+
+        thread_local! {
+            /// Tracks the number of nested block_on calls.
+            static NUM_NESTED_BLOCKING: Cell<usize> = Cell::new(0);
+        }
+
+        // Run the future as a task.
+        NUM_NESTED_BLOCKING.with(|num_nested_blocking| {
+            let count = num_nested_blocking.get();
+            let should_run = count == 0;
+            // increase the count
+            num_nested_blocking.replace(count + 1);
+
+            unsafe {
+                TaskLocalsWrapper::set_current(&wrapped.tag, || {
+                    let res = if should_run {
+                        // The first call should use run.
+                        smol::run(wrapped)
+                    } else {
+                        smol::block_on(wrapped)
+                    };
+                    num_nested_blocking.replace(num_nested_blocking.get() - 1);
+                    res
+                })
             }
-
+        })
-            // Log completion on exit.
-            defer! {
-                trace!("completed", {
-                    task_id: Task::get_current(|t| t.id().0),
-                });
-            }
-
-            future.await
-        };
-
-        let schedule = move |t| RUNTIME.schedule(Runnable(t));
-        let (task, handle) = async_task::spawn(future, schedule, task);
-        task.schedule();
-        Ok(JoinHandle::new(handle))
     }
 }
 
-/// A runnable task.
+pin_project! {
-pub struct Runnable(async_task::Task<Task>);
+    /// Wrapper to add support for task locals.
+    struct SupportTaskLocals<F> {
+        tag: TaskLocalsWrapper,
+        #[pin]
+        future: F,
+    }
+}
 
-impl Runnable {
+impl<F: Future> Future for SupportTaskLocals<F> {
-    /// Runs the task by polling its future once.
+    type Output = F::Output;
-    pub fn run(self) {
+
+    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)
+            })
         }
     }
 }

src/task/current.rs

@@ -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")
 }

src/task/join_handle.rs

@@ -12,12 +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,
+}
+
+#[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> {
+    pub(crate) fn new(inner: InnerHandle<T>, task: Task) -> JoinHandle<T> {
-        JoinHandle(inner)
+        JoinHandle {
+            handle: Some(inner),
+            task,
+        }
     }
 
     /// Returns a handle to the underlying task.
@@ -36,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()
     }
 }
 
@@ -44,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(output) => {
-            Poll::Ready(Some(val)) => Poll::Ready(val),
+                Poll::Ready(output.expect("cannot await the result of a panicked task"))
+            }
         }
     }
 }

src/task/mod.rs

@@ -138,25 +138,39 @@ 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};
 
-    pub(crate) use builder::Runnable;
     pub(crate) use task_local::LocalsMap;
+    pub(crate) use task_locals_wrapper::TaskLocalsWrapper;
 
     mod block_on;
     mod builder;
     mod current;
     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! {
+    #[cfg(feature = "default")]
+    pub use spawn_local::spawn_local;
+
+    #[cfg(feature = "default")]
+    mod spawn_local;
+}

src/task/spawn_blocking.rs

@@ -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.
 ///
@@ -43,80 +35,8 @@ where
     F: FnOnce() -> T + Send + 'static,
     T: Send + 'static,
 {
-    let schedule = |task| POOL.sender.send(task).unwrap();
+    once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
-    let (task, handle) = async_task::spawn(async { f() }, schedule, Task::new(None));
+
-    task.schedule();
+    let handle = smol::Task::blocking(async move { f() }).into();
-    JoinHandle::new(handle)
+    JoinHandle::new(handle, Task::new(None))
-}
-
-type Runnable = async_task::Task<Task>;
-
-struct Pool {
-    sender: Sender<Runnable>,
-    receiver: Receiver<Runnable>,
-}
-
-/// The number of sleeping worker threads.
-static SLEEPING: AtomicUsize = AtomicUsize::new(0);
-
-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");
 }

src/task/spawn_local.rs

@@ -0,0 +1,31 @@
+use std::future::Future;
+
+use crate::task::{Builder, JoinHandle};
+
+/// Spawns a task onto the thread-local executor.
+///
+/// # Examples
+///
+/// ```
+/// # #[cfg(feature = "unstable")]
+/// # async_std::task::block_on(async {
+/// #
+/// use async_std::task;
+///
+/// let handle = task::spawn_local(async {
+///     1 + 2
+/// });
+///
+/// assert_eq!(handle.await, 3);
+/// #
+/// # })
+/// ```
+#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[inline]
+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")
+}

src/task/task.rs

@@ -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::task::TaskId;
-use crate::utils::abort_on_panic;
 
-thread_local! {
+/// A handle to a task.
-    /// A pointer to the currently running task.
+#[derive(Clone)]
-    static CURRENT: Cell<*const Task> = Cell::new(ptr::null_mut());
+pub struct Task {
-}
-
-/// The inner representation of a task handle.
-struct Inner {
     /// The task ID.
     id: TaskId,
 
     /// The optional task name.
-    name: Option<Box<str>>,
+    name: Option<Arc<String>>,
-
-    /// The map holding task-local values.
-    locals: LocalsMap,
 }
 
-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 {
+    pub(crate) fn new(name: Option<Arc<String>>) -> Task {
-        let inner = match name {
+        Task {
-            None => AtomicPtr::default(),
+            id: TaskId::generate(),
-            Some(name) => {
+            name,
-                let raw = Arc::into_raw(Arc::new(Inner::new(Some(name))));
+        }
-                AtomicPtr::new(raw as *mut Inner)
-            }
-        };
-        Task { inner }
     }
 
     /// 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)
+        self.name.as_ref().map(|s| s.as_str())
-    }
-
-    /// 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),
-        }
     }
 }
 

src/task/task_id.rs

@@ -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)

src/task/task_local.rs

@@ -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>;

src/task/task_locals_wrapper.rs

@@ -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() };
+        });
+    }
+}

src/task/yield_now.rs

@@ -43,10 +43,6 @@ impl Future for YieldNow {
         if !self.0 {
             self.0 = true;
             cx.waker().wake_by_ref();
-
-            #[cfg(feature = "default")]
-            crate::rt::RUNTIME.yield_now();
-
             Poll::Pending
         } else {
             Poll::Ready(())

src/unit/extend.rs

@@ -4,10 +4,13 @@ use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
 impl stream::Extend<()> for () {
-    fn extend<'a, T: IntoStream<Item = ()> + 'a>(
+    fn extend<'a, S: IntoStream<Item = ()> + 'a>(
         &'a mut self,
-        stream: T,
+        stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

src/unit/from_stream.rs

@@ -7,7 +7,10 @@ impl FromStream<()> for () {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = ()> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         Box::pin(stream.into_stream().for_each(drop))
     }
 }

src/utils.rs

@@ -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,51 @@ pub(crate) trait Context {
     fn context(self, message: impl Fn() -> String) -> Self;
 }
 
+#[cfg(all(not(target_os = "unknown"), feature = "default"))]
+mod timer {
+    pub type Timer = smol::Timer;
+}
+
+#[cfg(any(feature = "unstable", feature = "default"))]
+pub(crate) fn timer_after(dur: std::time::Duration) -> timer::Timer {
+    #[cfg(not(target_os = "unknown"))]
+    once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
+    Timer::after(dur)
+}
+
+#[cfg(any(
+    all(target_arch = "wasm32", feature = "default"),
+    all(feature = "unstable", not(feature = "default"))
+))]
+mod timer {
+    use std::pin::Pin;
+    use std::task::Poll;
+
+    #[derive(Debug)]
+    pub(crate) struct Timer(futures_timer::Delay);
+
+    impl Timer {
+        pub(crate) fn after(dur: std::time::Duration) -> Self {
+            Timer(futures_timer::Delay::new(dur))
+        }
+    }
+
+    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(()),
+            }
+        }
+    }
+}
+
+#[cfg(any(feature = "unstable", feature = "default"))]
+pub(crate) use timer::*;
+
 /// Defers evaluation of a block of code until the end of the scope.
 #[cfg(feature = "default")]
 #[doc(hidden)]

src/vec/extend.rs

@@ -3,11 +3,14 @@ use std::pin::Pin;
 use crate::prelude::*;
 use crate::stream::{self, IntoStream};
 
-impl<T> stream::Extend<T> for Vec<T> {
+impl<T: Send> stream::Extend<T> for Vec<T> {
     fn extend<'a, S: IntoStream<Item = T> + 'a>(
         &'a mut self,
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = ()> + 'a + Send>>
+    where
+        <S as IntoStream>::IntoStream: Send,
+    {
         let stream = stream.into_stream();
 
         self.reserve(stream.size_hint().0);

src/vec/from_stream.rs

@@ -6,13 +6,13 @@ use std::sync::Arc;
 use crate::prelude::*;
 use crate::stream::{self, FromStream, IntoStream};
 
-impl<T> FromStream<T> for Vec<T> {
+impl<T: Send> FromStream<T> for Vec<T> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T>>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>>
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>>
     where
-        <S as IntoStream>::IntoStream: 'a,
+        <S as IntoStream>::IntoStream: 'a + Send,
     {
         let stream = stream.into_stream();
 
@@ -24,11 +24,14 @@ impl<T> FromStream<T> for Vec<T> {
     }
 }
 
-impl<'b, T: Clone> FromStream<T> for Cow<'b, [T]> {
+impl<'b, T: Clone + Send> FromStream<T> for Cow<'b, [T]> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -37,11 +40,14 @@ impl<'b, T: Clone> FromStream<T> for Cow<'b, [T]> {
     }
 }
 
-impl<T> FromStream<T> for Box<[T]> {
+impl<T: Send> FromStream<T> for Box<[T]> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -50,11 +56,14 @@ impl<T> FromStream<T> for Box<[T]> {
     }
 }
 
-impl<T> FromStream<T> for Rc<[T]> {
+impl<T: Send> FromStream<T> for Rc<[T]> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {
@@ -63,11 +72,14 @@ impl<T> FromStream<T> for Rc<[T]> {
     }
 }
 
-impl<T> FromStream<T> for Arc<[T]> {
+impl<T: Send> FromStream<T> for Arc<[T]> {
     #[inline]
     fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
         stream: S,
-    ) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
+    ) -> Pin<Box<dyn Future<Output = Self> + 'a + Send>> 
+    where
+        <S as IntoStream>::IntoStream: Send
+    {
         let stream = stream.into_stream();
 
         Box::pin(async move {

tests/addr.rs

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

tests/block_on.rs

@@ -1,16 +1,63 @@
-use async_std::task;
+#![cfg(not(target_os = "unknown"))]
+
+use async_std::{future::ready, task::block_on};
 
 #[test]
 fn smoke() {
-    let res = task::block_on(async { 1 + 2 });
+    let res = block_on(async { 1 + 2 });
     assert_eq!(res, 3);
 }
 
 #[test]
 #[should_panic = "boom"]
 fn panic() {
-    task::block_on(async {
+    block_on(async {
         // This panic should get propagated into the parent thread.
         panic!("boom");
     });
 }
+
+#[cfg(feature = "unstable")]
+#[test]
+fn nested_block_on_local() {
+    use async_std::task::spawn_local;
+
+    let x = block_on(async {
+        let a = block_on(async { block_on(async { ready(3).await }) });
+        let b = spawn_local(async { block_on(async { ready(2).await }) }).await;
+        let c = block_on(async { block_on(async { ready(1).await }) });
+        a + b + c
+    });
+
+    assert_eq!(x, 3 + 2 + 1);
+
+    let y = block_on(async {
+        let a = block_on(async { block_on(async { ready(3).await }) });
+        let b = spawn_local(async { block_on(async { ready(2).await }) }).await;
+        let c = block_on(async { block_on(async { ready(1).await }) });
+        a + b + c
+    });
+
+    assert_eq!(y, 3 + 2 + 1);
+}
+
+#[test]
+fn nested_block_on() {
+    let x = block_on(async {
+        let a = block_on(async { block_on(async { ready(3).await }) });
+        let b = block_on(async { block_on(async { ready(2).await }) });
+        let c = block_on(async { block_on(async { ready(1).await }) });
+        a + b + c
+    });
+
+    assert_eq!(x, 3 + 2 + 1);
+
+    let y = block_on(async {
+        let a = block_on(async { block_on(async { ready(3).await }) });
+        let b = block_on(async { block_on(async { ready(2).await }) });
+        let c = block_on(async { block_on(async { ready(1).await }) });
+        a + b + c
+    });
+
+    assert_eq!(y, 3 + 2 + 1);
+}

tests/buf_writer.rs

@@ -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 inner: Vec<u8> = Vec::new();
-        let mut writer = BufWriter::with_capacity(2, inner);
+        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()));

tests/channel.rs

@@ -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 c1 = spawn(async move { r.recv().await.unwrap() });
-            let c2 = task::spawn(async move { s.send(0).await });
+            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 {

tests/collect.rs

@@ -0,0 +1,20 @@
+#[cfg(feature = "unstable")]
+#[test]
+fn test_send() -> async_std::io::Result<()> {
+    use async_std::prelude::*;
+    use async_std::{stream, task};
+
+    task::block_on(async {
+        fn test_send_trait<T: Send>(_: &T) {}
+
+        let stream = stream::repeat(1u8).take(10);
+        test_send_trait(&stream);
+
+        let fut = stream.collect::<Vec<_>>();
+
+        // This line triggers a compilation error
+        test_send_trait(&fut);
+
+        Ok(())
+    })
+}

tests/condvar.rs

@@ -5,28 +5,38 @@ 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();
 
-        task::spawn(async move {
+        spawn(async move {
             let (m, c) = &*pair2;
             let _g = m.lock().await;
-            task::sleep(Duration::from_millis(20)).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(10))
+            .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);
@@ -40,13 +50,14 @@ fn wait_timeout_without_lock() {
 }
 
 #[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(10), |&mut started| {
+            .wait_timeout_until(m.lock().await, Duration::from_millis(100), |&mut started| {
                 started
             })
             .await;
@@ -55,6 +66,7 @@ fn wait_timeout_until_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();
@@ -62,7 +74,7 @@ fn notify_all() {
 
         for _ in 0..10 {
             let pair = pair.clone();
-            tasks.push(task::spawn(async move {
+            tasks.push(spawn(async move {
                 let (m, c) = &*pair;
                 let mut count = m.lock().await;
                 while *count == 0 {
@@ -73,7 +85,7 @@ fn notify_all() {
         }
 
         // Give some time for tasks to start up
-        task::sleep(Duration::from_millis(5)).await;
+        task::sleep(Duration::from_millis(50)).await;
 
         let (m, c) = &*pair;
         {

tests/io_timeout.rs

@@ -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 {

tests/mutex.rs

@@ -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);
     });

tests/rwlock.rs

@@ -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;

tests/stream.rs

@@ -1,3 +1,5 @@
+use std::convert::identity;
+use std::marker::Unpin;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -8,14 +10,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 +45,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 +96,88 @@ 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]);
+    });
+}
+
+struct S<T>(T);
+
+impl<T: Stream + Unpin> Stream for S<T> {
+    type Item = T::Item;
+
+    fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
+        unsafe { Pin::new_unchecked(&mut self.0) }.poll_next(ctx)
+    }
+}
+
+struct StrictOnce {
+    polled: bool,
+}
+
+impl Stream for StrictOnce {
+    type Item = ();
+
+    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context) -> Poll<Option<Self::Item>> {
+        assert!(!self.polled, "Polled after completion!");
+        self.polled = true;
+        Poll::Ready(None)
+    }
+}
+
+struct Interchanger {
+    polled: bool,
+}
+
+impl Stream for Interchanger {
+    type Item = S<Box<dyn Stream<Item = ()> + Unpin>>;
+
+    fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
+        if self.polled {
+            self.polled = false;
+            ctx.waker().wake_by_ref();
+            Poll::Pending
+        } else {
+            self.polled = true;
+            Poll::Ready(Some(S(Box::new(StrictOnce { polled: false }))))
+        }
+    }
+}
+
+#[test]
+fn flat_map_doesnt_poll_completed_inner_stream() {
+    task::block_on(async {
+        assert_eq!(
+            Interchanger { polled: false }
+                .take(2)
+                .flat_map(identity)
+                .count()
+                .await,
+            0
+        );
+    });
+}
+
+#[test]
+fn flatten_doesnt_poll_completed_inner_stream() {
+    task::block_on(async {
+        assert_eq!(
+            Interchanger { polled: false }
+                .take(2)
+                .flatten()
+                .count()
+                .await,
+            0
+        );
     });
-    assert_eq!(xs, vec![92, 92]);
 }

tests/task_local.rs

@@ -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();

tests/tcp.rs

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

tests/timeout.rs

@@ -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..10)
+            .map(|i| {
+                timeout(Duration::from_millis(i * 50), async move {
+                    task::sleep(Duration::from_millis(i)).await;
+                    Ok::<(), async_std::future::TimeoutError>(())
+                })
+            })
+            .collect::<Vec<_>>();
+
+        for future in futures {
+            future.await.unwrap().unwrap();
+        }
+    });
+}

tests/udp.rs

@@ -1,3 +1,5 @@
+#![cfg(not(target_os = "unknown"))]
+
 use async_std::io;
 use async_std::net::UdpSocket;
 use async_std::task;
@@ -17,7 +19,7 @@ fn send_recv() -> io::Result<()> {
 
         socket1.connect(socket2.local_addr()?).await?;
         socket2.connect(socket1.local_addr()?).await?;
-
+        assert_eq!(socket1.peer_addr()?, socket2.local_addr()?);
         socket1.send(THE_MERCHANT_OF_VENICE).await?;
 
         let mut buf = [0u8; 1024];

tests/uds.rs

@@ -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(())
+    })
+}

tests/verbose_errors.rs

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

wasm-test.sh

@@ -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