Merge pull request #757 from dignifiedquire/feat/smol

.github/workflows/ci.yml

@@ -59,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'

Cargo.toml

@@ -4,6 +4,7 @@ version = "1.5.0"
 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,13 @@ 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",
 ]
 docs = ["attributes", "unstable", "default"]
-unstable = ["std", "broadcaster", "futures-timer"]
+unstable = ["std", "broadcaster"]
 attributes = ["async-attributes"]
 std = [
   "alloc",
@@ -47,6 +42,10 @@ std = [
   "once_cell",
   "pin-utils",
   "slab",
+  "smol",
+  "wasm-timer",
+  "wasm-bindgen-futures",
+  "futures-channel",
 ]
 alloc = [
   "futures-core/alloc",
@@ -55,32 +54,38 @@ alloc = [
 
 [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 }
-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.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 }
 
+[target.'cfg(not(target_os = "unknown"))'.dependencies]
+smol = { version = "0.1.1", optional = true }
+
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+wasm-timer = { version = "0.2.4", optional = true }
+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"

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;
 
 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/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;
 
 /// 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/mod.rs

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

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

@@ -267,14 +267,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 {
@@ -79,11 +79,9 @@ impl TcpListener {
         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 +112,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 +200,8 @@ impl<'a> Stream for Incoming<'a> {
 impl From<std::net::TcpListener> for TcpListener {
     /// Converts a `std::net::TcpListener` into its asynchronous equivalent.
     fn from(listener: std::net::TcpListener) -> TcpListener {
-        let mio_listener = mio::net::TcpListener::from_std(listener).unwrap();
         TcpListener {
-            watcher: Watcher::new(mio_listener),
+            watcher: Async::new(listener).expect("TcpListener is known to be good"),
         }
     }
 }
@@ -230,29 +223,31 @@ cfg_unix! {
 
     impl IntoRawFd for TcpListener {
         fn into_raw_fd(self) -> RawFd {
-            self.watcher.into_inner().into_raw_fd()
+            self.watcher.into_raw_fd()
         }
     }
 }
 
 cfg_windows! {
-    // use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
+    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 {
@@ -75,28 +75,16 @@ impl TcpStream {
         let addrs = addrs.to_socket_addrs().await?;
 
         for addr in addrs {
-            // mio's TcpStream::connect is non-blocking and may just be in progress
+            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 +202,7 @@ impl TcpStream {
     /// # Ok(()) }) }
     /// ```
     pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
-        future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.peek(buf))).await
+        self.watcher.peek(buf).await
     }
 
     /// Gets the value of the `TCP_NODELAY` option on this socket.
@@ -317,7 +305,7 @@ impl Read for &TcpStream {
         cx: &mut Context<'_>,
         buf: &mut [u8],
     ) -> Poll<io::Result<usize>> {
-        self.watcher.poll_read_with(cx, |mut inner| inner.read(buf))
+        Pin::new(&mut &*self.watcher).poll_read(cx, buf)
     }
 }
 
@@ -353,26 +341,23 @@ impl Write for &TcpStream {
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        self.watcher
+        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();
         TcpStream {
-            watcher: Arc::new(Watcher::new(mio_stream)),
+            watcher: Arc::new(Async::new(stream).expect("TcpStream is known to be good")),
         }
     }
 }
@@ -403,23 +388,28 @@ cfg_unix! {
 }
 
 cfg_windows! {
-    // use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
+    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 {
@@ -69,16 +69,12 @@ impl UdpSocket {
     /// ```
     pub async fn bind<A: ToSocketAddrs>(addrs: A) -> io::Result<UdpSocket> {
         let mut last_err = None;
-        let addrs = addrs
+        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),
             }
@@ -153,12 +149,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 +175,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 +246,7 @@ impl UdpSocket {
     /// # Ok(()) }) }
     /// ```
     pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
-        future::poll_fn(|cx| self.watcher.poll_write_with(cx, |inner| inner.send(buf)))
+        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 +270,7 @@ impl UdpSocket {
     /// # Ok(()) }) }
     /// ```
     pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
-        future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.recv(buf)))
+        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 +453,8 @@ impl UdpSocket {
 impl From<std::net::UdpSocket> for UdpSocket {
     /// Converts a `std::net::UdpSocket` into its asynchronous equivalent.
     fn from(socket: std::net::UdpSocket) -> UdpSocket {
-        let mio_socket = mio::net::UdpSocket::from_socket(socket).unwrap();
         UdpSocket {
-            watcher: Watcher::new(mio_socket),
+            watcher: Async::new(socket).expect("UdpSocket is known to be good"),
         }
     }
 }
@@ -522,29 +476,31 @@ cfg_unix! {
 
     impl IntoRawFd for UdpSocket {
         fn into_raw_fd(self) -> RawFd {
-            self.watcher.into_inner().into_raw_fd()
+            self.watcher.into_raw_fd()
         }
     }
 }
 
 cfg_windows! {
-    // use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
+    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/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,13 @@ use crate::task::spawn_blocking;
 /// # Ok(()) }) }
 /// ```
 pub struct UnixDatagram {
-    watcher: Watcher<mio_uds::UnixDatagram>,
+    watcher: Async<StdUnixDatagram>,
 }
 
 impl UnixDatagram {
-    fn new(socket: mio_uds::UnixDatagram) -> UnixDatagram {
+    fn new(socket: StdUnixDatagram) -> UnixDatagram {
         UnixDatagram {
-            watcher: Watcher::new(socket),
+            watcher: Async::new(socket).expect("UnixDatagram is known to be good"),
         }
     }
 
@@ -67,8 +65,8 @@ impl UnixDatagram {
     /// ```
     pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixDatagram> {
         let path = path.as_ref().to_owned();
-        let socket = spawn_blocking(move || mio_uds::UnixDatagram::bind(path)).await?;
+        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 +83,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub fn unbound() -> io::Result<UnixDatagram> {
-        let socket = mio_uds::UnixDatagram::unbound()?;
+        let socket = StdUnixDatagram::unbound()?;
         Ok(UnixDatagram::new(socket))
     }
 
@@ -105,7 +103,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub fn pair() -> io::Result<(UnixDatagram, UnixDatagram)> {
-        let (a, b) = mio_uds::UnixDatagram::pair()?;
+        let (a, b) = StdUnixDatagram::pair()?;
         let a = UnixDatagram::new(a);
         let b = UnixDatagram::new(b);
         Ok((a, b))
@@ -197,11 +195,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
-        future::poll_fn(|cx| {
+        self.watcher.recv_from(buf).await
-            self.watcher
-                .poll_read_with(cx, |inner| inner.recv_from(buf))
-        })
-        .await
     }
 
     /// Receives data from the socket.
@@ -222,7 +216,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
-        future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.recv(buf))).await
+        self.watcher.recv(buf).await
     }
 
     /// Sends data on the socket to the specified address.
@@ -242,11 +236,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub async fn send_to<P: AsRef<Path>>(&self, buf: &[u8], path: P) -> io::Result<usize> {
-        future::poll_fn(|cx| {
+        self.watcher.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 +257,7 @@ impl UnixDatagram {
     /// # Ok(()) }) }
     /// ```
     pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
-        future::poll_fn(|cx| self.watcher.poll_write_with(cx, |inner| inner.send(buf))).await
+        self.watcher.send(buf).await
     }
 
     /// Shut down the read, write, or both halves of this connection.
@@ -312,31 +302,31 @@ impl fmt::Debug for UnixDatagram {
     }
 }
 
-impl From<std::os::unix::net::UnixDatagram> for UnixDatagram {
+impl From<StdUnixDatagram> for UnixDatagram {
     /// Converts a `std::os::unix::net::UnixDatagram` into its asynchronous equivalent.
-    fn from(datagram: std::os::unix::net::UnixDatagram) -> UnixDatagram {
+    fn from(datagram: StdUnixDatagram) -> UnixDatagram {
-        let mio_datagram = mio_uds::UnixDatagram::from_datagram(datagram).unwrap();
         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);
+        let raw = StdUnixDatagram::from_raw_fd(fd);
-        datagram.into()
+        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,19 @@
 //! 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::task::{Context, Poll};
 
 /// A Unix domain socket server, listening for connections.
 ///
@@ -50,7 +49,7 @@ use crate::task::{spawn_blocking, Context, Poll};
 /// # Ok(()) }) }
 /// ```
 pub struct UnixListener {
-    watcher: Watcher<mio_uds::UnixListener>,
+    watcher: Async<StdUnixListener>,
 }
 
 impl UnixListener {
@@ -69,11 +68,9 @@ impl UnixListener {
     /// ```
     pub async fn bind<P: AsRef<Path>>(path: P) -> io::Result<UnixListener> {
         let path = path.as_ref().to_owned();
-        let listener = spawn_blocking(move || mio_uds::UnixListener::bind(path)).await?;
+        let listener = Async::<StdUnixListener>::bind(path)?;
 
-        Ok(UnixListener {
+        Ok(UnixListener { watcher: listener })
-            watcher: Watcher::new(listener),
-        })
     }
 
     /// Accepts a new incoming connection to this listener.
@@ -93,29 +90,9 @@ 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((UnixStream { watcher: stream }, addr))
-                Some((io, addr)) => {
-                    let mio_stream = mio_uds::UnixStream::from_stream(io)?;
-                    let stream = UnixStream {
-                        watcher: Watcher::new(mio_stream),
-                    };
-                    Poll::Ready(Ok((stream, addr)))
-                }
-                // This should never happen since `None` is converted to `WouldBlock`
-                None => unreachable!(),
-            }
-        })
-        .await
     }
 
     /// Returns a stream of incoming connections.
@@ -206,19 +183,18 @@ impl Stream for Incoming<'_> {
     }
 }
 
-impl From<std::os::unix::net::UnixListener> for UnixListener {
+impl From<StdUnixListener> for UnixListener {
     /// Converts a `std::os::unix::net::UnixListener` into its asynchronous equivalent.
-    fn from(listener: std::os::unix::net::UnixListener) -> UnixListener {
+    fn from(listener: StdUnixListener) -> UnixListener {
-        let mio_listener = mio_uds::UnixListener::from_listener(listener).unwrap();
         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 +207,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,17 @@
 //! 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::task::{Context, Poll};
 
 /// A Unix stream socket.
 ///
@@ -38,7 +37,7 @@ use crate::task::{spawn_blocking, Context, Poll};
 /// # Ok(()) }) }
 /// ```
 pub struct UnixStream {
-    pub(super) watcher: Watcher<mio_uds::UnixStream>,
+    pub(super) watcher: Async<StdUnixStream>,
 }
 
 impl UnixStream {
@@ -57,15 +56,9 @@ impl UnixStream {
     /// ```
     pub async fn connect<P: AsRef<Path>>(path: P) -> io::Result<UnixStream> {
         let path = path.as_ref().to_owned();
+        let stream = Async::<StdUnixStream>::connect(path).await?;
 
-        spawn_blocking(move || {
+        Ok(UnixStream { watcher: stream })
-            let std_stream = std::os::unix::net::UnixStream::connect(path)?;
-            let mio_stream = mio_uds::UnixStream::from_stream(std_stream)?;
-            Ok(UnixStream {
-                watcher: Watcher::new(mio_stream),
-            })
-        })
-        .await
     }
 
     /// Creates an unnamed pair of connected sockets.
@@ -84,13 +77,9 @@ impl UnixStream {
     /// # Ok(()) }) }
     /// ```
     pub fn pair() -> io::Result<(UnixStream, UnixStream)> {
-        let (a, b) = mio_uds::UnixStream::pair()?;
+        let (a, b) = Async::<StdUnixStream>::pair()?;
-        let a = UnixStream {
+        let a = UnixStream { watcher: a };
-            watcher: Watcher::new(a),
+        let b = UnixStream { watcher: b };
-        };
-        let b = UnixStream {
-            watcher: Watcher::new(b),
-        };
         Ok((a, b))
     }
 
@@ -169,7 +158,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 +186,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 +215,17 @@ impl fmt::Debug for UnixStream {
     }
 }
 
-impl From<std::os::unix::net::UnixStream> for UnixStream {
+impl From<StdUnixStream> for UnixStream {
     /// Converts a `std::os::unix::net::UnixStream` into its asynchronous equivalent.
-    fn from(stream: std::os::unix::net::UnixStream) -> UnixStream {
+    fn from(stream: StdUnixStream) -> UnixStream {
-        let mio_stream = mio_uds::UnixStream::from_stream(stream).unwrap();
+        let stream = Async::new(stream).expect("UnixStream is known to be good");
-        UnixStream {
+        UnixStream { watcher: stream }
-            watcher: Watcher::new(mio_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 +238,6 @@ impl FromRawFd for UnixStream {
 
 impl IntoRawFd for UnixStream {
     fn into_raw_fd(self) -> RawFd {
-        self.watcher.into_inner().into_raw_fd()
+        self.watcher.into_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/rt/mod.rs

@@ -4,20 +4,20 @@ 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())
+    let num_threads = num_cpus::get().max(1);
-        .spawn(|| abort_on_panic(|| RUNTIME.run()))
+    for _ in 0..num_threads {
-        .expect("cannot start a runtime thread");
+        thread::Builder::new()
-
+            .name("async-std/runtime".to_string())
-    Runtime::new()
+            .spawn(|| smol::run(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/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;
 
 /// 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;
 
 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/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;
 
 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;
 
 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/sync/barrier.rs

@@ -202,7 +202,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/mod.rs

@@ -194,8 +194,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/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,12 @@
 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 +28,151 @@ impl Builder {
         self
     }
 
+    fn build<F, T>(self, future: F) -> SupportTaskLocals<F>
+    where
+        F: Future<Output = T>,
+    {
+        let name = self.name.map(Arc::new);
+
+        // Create a new task handle.
+        let task = Task::new(name);
+
+        #[cfg(not(target_os = "unknown"))]
+        once_cell::sync::Lazy::force(&crate::rt::RUNTIME);
+
+        let tag = TaskLocalsWrapper::new(task.clone());
+
+        SupportTaskLocals { tag, future }
+    }
+
     /// Spawns a task with the configured settings.
+    #[cfg(not(target_os = "unknown"))]
     pub fn spawn<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
     where
         F: Future<Output = T> + Send + 'static,
         T: Send + 'static,
     {
-        // Create a new task handle.
+        let 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() });
-            }
 
-            // Log completion on exit.
+        Ok(JoinHandle::new(smol_task, task))
-            defer! {
+    }
-                trace!("completed", {
-                    task_id: Task::get_current(|t| t.id().0),
-                });
-            }
 
-            future.await
+    /// Spawns a task locally with the configured settings.
-        };
+    #[cfg(all(not(target_os = "unknown"), feature = "unstable"))]
+    pub fn local<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
+    where
+        F: Future<Output = T> + 'static,
+        T: 'static,
+    {
+        let wrapped = self.build(future);
 
-        let schedule = move |t| RUNTIME.schedule(Runnable(t));
+        kv_log_macro::trace!("spawn_local", {
-        let (task, handle) = async_task::spawn(future, schedule, task);
+            task_id: wrapped.tag.id().0,
-        task.schedule();
+            parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
-        Ok(JoinHandle::new(handle))
+        });
+
+        let task = wrapped.tag.task().clone();
+        let smol_task = smol::Task::local(wrapped).into();
+
+        Ok(JoinHandle::new(smol_task, task))
+    }
+
+    /// Spawns a task locally with the configured settings.
+    #[cfg(all(target_arch = "wasm32", feature = "unstable"))]
+    pub fn local<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
+    where
+        F: Future<Output = T> + 'static,
+        T: 'static,
+    {
+        use futures_channel::oneshot::channel;
+        let (sender, receiver) = channel();
+
+        let wrapped = self.build(async move {
+            let res = future.await;
+            let _ = sender.send(res);
+        });
+        kv_log_macro::trace!("spawn_local", {
+            task_id: wrapped.tag.id().0,
+            parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
+        });
+
+        let task = wrapped.tag.task().clone();
+        wasm_bindgen_futures::spawn_local(wrapped);
+
+        Ok(JoinHandle::new(receiver, task))
+    }
+
+    /// Spawns a task locally with the configured settings.
+    #[cfg(all(target_arch = "wasm32", not(feature = "unstable")))]
+    pub(crate) fn local<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
+    where
+        F: Future<Output = T> + 'static,
+        T: 'static,
+    {
+        use futures_channel::oneshot::channel;
+        let (sender, receiver) = channel();
+
+        let wrapped = self.build(async move {
+            let res = future.await;
+            let _ = sender.send(res);
+        });
+
+        kv_log_macro::trace!("spawn_local", {
+            task_id: wrapped.tag.id().0,
+            parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
+        });
+
+        let task = wrapped.tag.task().clone();
+        wasm_bindgen_futures::spawn_local(wrapped);
+
+        Ok(JoinHandle::new(receiver, task))
+    }
+
+    /// Spawns a task with the configured settings, blocking on its execution.
+    #[cfg(not(target_os = "unknown"))]
+    pub fn blocking<F, T>(self, future: F) -> T
+    where
+        F: Future<Output = T>,
+    {
+        let wrapped = self.build(future);
+
+        // Log this `block_on` operation.
+        kv_log_macro::trace!("block_on", {
+            task_id: wrapped.tag.id().0,
+            parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0),
+        });
+
+        // Run the future as a task.
+        unsafe { TaskLocalsWrapper::set_current(&wrapped.tag, || smol::run(wrapped)) }
     }
 }
 
-/// A runnable task.
+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,37 @@ cfg_default! {
     pub use task_id::TaskId;
     pub use join_handle::JoinHandle;
     pub use sleep::sleep;
+    #[cfg(not(target_os = "unknown"))]
     pub use spawn::spawn;
     pub use task_local::{AccessError, LocalKey};
 
-    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! {
+    pub use spawn_local::spawn_local;
+
+    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,28 @@
+use std::future::Future;
+
+use crate::task::{Builder, JoinHandle};
+
+/// Spawns a task onto the thread-local executor.
+///
+/// # Examples
+///
+/// ```
+/// # async_std::task::block_on(async {
+/// #
+/// use async_std::task;
+///
+/// let handle = task::spawn_local(async {
+///     1 + 2
+/// });
+///
+/// assert_eq!(handle.await, 3);
+/// #
+/// # })
+/// ```
+pub fn spawn_local<F, T>(future: F) -> JoinHandle<T>
+where
+    F: Future<Output = T> + 'static,
+    T: 'static,
+{
+    Builder::new().local(future).expect("cannot spawn task")
+}

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_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/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,37 @@ pub(crate) trait Context {
     fn context(self, message: impl Fn() -> String) -> Self;
 }
 
+#[cfg(all(not(target_os = "unknown"), feature = "default"))]
+pub(crate) type Timer = smol::Timer;
+
+#[cfg(all(target_arch = "wasm32", feature = "default"))]
+#[derive(Debug)]
+pub(crate) struct Timer(wasm_timer::Delay);
+
+#[cfg(all(target_arch = "wasm32", feature = "default"))]
+impl Timer {
+    pub(crate) fn after(dur: std::time::Duration) -> Self {
+        Timer(wasm_timer::Delay::new(dur))
+    }
+}
+
+#[cfg(target_arch = "wasm32")]
+use std::pin::Pin;
+#[cfg(target_arch = "wasm32")]
+use std::task::Poll;
+
+#[cfg(target_arch = "wasm32")]
+impl std::future::Future for Timer {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
+        match Pin::new(&mut self.0).poll(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(_) => Poll::Ready(()),
+        }
+    }
+}
+
 /// Defers evaluation of a block of code until the end of the scope.
 #[cfg(feature = "default")]
 #[doc(hidden)]

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,3 +1,5 @@
+#![cfg(not(target_os = "unknown"))]
+
 use async_std::task;
 
 #[test]

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/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(100))
             .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,7 @@ use async_std::task;
 
 #[test]
 #[should_panic(expected = "timed out")]
+#[cfg(not(target_os = "unknown"))]
 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

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

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..100)
+            .map(|i| {
+                timeout(Duration::from_millis(i * 10), 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;

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

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