diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e9fcdcc..8f519e5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,6 +58,14 @@ jobs: with: command: check args: --features unstable --all --bins --examples --tests + + - name: check wasm + uses: actions-rs/cargo@v1 + with: + command: check + target: wasm32-unknown-unknown + override: true + args: --features unstable --all --bins --tests - name: check bench uses: actions-rs/cargo@v1 diff --git a/Cargo.toml b/Cargo.toml index d49eb95..630b373 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ version = "1.5.0" authors = [ "Stjepan Glavina ", "Yoshua Wuyts ", + "Friedel Ziegelmayer ", "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" diff --git a/src/future/future/delay.rs b/src/future/future/delay.rs index 641084f..b6c30bc 100644 --- a/src/future/future/delay.rs +++ b/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 DelayFuture { pub fn new(future: F, dur: Duration) -> DelayFuture { - let delay = Delay::new(dur); + let delay = Timer::after(dur); DelayFuture { future, delay } } diff --git a/src/future/timeout.rs b/src/future/timeout.rs index 05aaa45..4a9d93c 100644 --- a/src/future/timeout.rs +++ b/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(dur: Duration, f: F) -> Result where F: Future, { - let f = TimeoutFuture { - future: f, - delay: Delay::new(dur), - }; - f.await + TimeoutFuture::new(f, dur).await } pin_project! { @@ -46,14 +42,17 @@ pin_project! { #[pin] future: F, #[pin] - delay: Delay, + delay: Timer, } } impl TimeoutFuture { #[allow(dead_code)] pub(super) fn new(future: F, dur: Duration) -> TimeoutFuture { - TimeoutFuture { future: future, delay: Delay::new(dur) } + TimeoutFuture { + future, + delay: Timer::after(dur), + } } } diff --git a/src/io/mod.rs b/src/io/mod.rs index dd97567..f5dd9e2 100644 --- a/src/io/mod.rs +++ b/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; } diff --git a/src/io/read/mod.rs b/src/io/read/mod.rs index 8aade18..388237c 100644 --- a/src/io/read/mod.rs +++ b/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(_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(()) - }) + }); } } diff --git a/src/io/read/take.rs b/src/io/read/take.rs index 09b02c2..ba9a9e3 100644 --- a/src/io/read/take.rs +++ b/src/io/read/take.rs @@ -218,7 +218,7 @@ impl BufRead for Take { } } -#[cfg(test)] +#[cfg(all(test, not(target_os = "unknown")))] mod tests { use crate::io; use crate::prelude::*; diff --git a/src/io/timeout.rs b/src/io/timeout.rs index 6e22dbf..ce33fea 100644 --- a/src/io/timeout.rs +++ b/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>, { 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, } } diff --git a/src/lib.rs b/src/lib.rs index 7e0e98d..408a7ab 100644 --- a/src/lib.rs +++ b/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; diff --git a/src/net/mod.rs b/src/net/mod.rs index fe83d3b..1814073 100644 --- a/src/net/mod.rs +++ b/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; diff --git a/src/net/tcp/listener.rs b/src/net/tcp/listener.rs index 9e15d40..72c5d3a 100644 --- a/src/net/tcp/listener.rs +++ b/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, + watcher: Async, } impl TcpListener { @@ -79,11 +79,9 @@ impl TcpListener { let addrs = addrs.to_socket_addrs().await?; for addr in addrs { - match mio::net::TcpListener::bind(&addr) { - Ok(mio_listener) => { - return Ok(TcpListener { - watcher: Watcher::new(mio_listener), - }); + match Async::::bind(&addr) { + Ok(listener) => { + return Ok(TcpListener { watcher: listener }); } Err(err) => last_err = Some(err), } @@ -114,13 +112,9 @@ impl TcpListener { /// # Ok(()) }) } /// ``` pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> { - let (io, addr) = - future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.accept_std())) - .await?; - - let mio_stream = mio::net::TcpStream::from_stream(io)?; + let (stream, addr) = self.watcher.accept().await?; let stream = TcpStream { - watcher: Arc::new(Watcher::new(mio_stream)), + watcher: Arc::new(stream), }; Ok((stream, addr)) } @@ -206,9 +200,8 @@ impl<'a> Stream for Incoming<'a> { impl From for TcpListener { /// Converts a `std::net::TcpListener` into its asynchronous equivalent. fn from(listener: std::net::TcpListener) -> TcpListener { - let mio_listener = mio::net::TcpListener::from_std(listener).unwrap(); TcpListener { - watcher: Watcher::new(mio_listener), + watcher: Async::new(listener).expect("TcpListener is known to be good"), } } } @@ -230,29 +223,31 @@ cfg_unix! { impl IntoRawFd for TcpListener { fn into_raw_fd(self) -> RawFd { - self.watcher.into_inner().into_raw_fd() + self.watcher.into_raw_fd() } } } cfg_windows! { - // use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle}; - // - // impl AsRawSocket for TcpListener { - // fn as_raw_socket(&self) -> RawSocket { - // self.raw_socket - // } - // } - // - // impl FromRawSocket for TcpListener { - // unsafe fn from_raw_socket(handle: RawSocket) -> TcpListener { - // net::TcpListener::from_raw_socket(handle).try_into().unwrap() - // } - // } - // - // impl IntoRawSocket for TcpListener { - // fn into_raw_socket(self) -> RawSocket { - // self.raw_socket - // } - // } + use crate::os::windows::io::{ + AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket, + }; + + impl AsRawSocket for TcpListener { + fn as_raw_socket(&self) -> RawSocket { + self.watcher.as_raw_socket() + } + } + + impl FromRawSocket for TcpListener { + unsafe fn from_raw_socket(handle: RawSocket) -> TcpListener { + std::net::TcpListener::from_raw_socket(handle).into() + } + } + + impl IntoRawSocket for TcpListener { + fn into_raw_socket(self) -> RawSocket { + self.watcher.into_raw_socket() + } + } } diff --git a/src/net/tcp/stream.rs b/src/net/tcp/stream.rs index 1f50e8f..b854143 100644 --- a/src/net/tcp/stream.rs +++ b/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>, + pub(super) watcher: Arc>, } impl TcpStream { @@ -75,28 +75,16 @@ impl TcpStream { let addrs = addrs.to_socket_addrs().await?; for addr in addrs { - // mio's TcpStream::connect is non-blocking and may just be in progress - // when it returns with `Ok`. We therefore wait for write readiness to - // be sure the connection has either been established or there was an - // error which we check for afterwards. - let watcher = match mio::net::TcpStream::connect(&addr) { - Ok(s) => Watcher::new(s), + match Async::::connect(&addr).await { + Ok(stream) => { + return Ok(TcpStream { + watcher: Arc::new(stream), + }); + } Err(e) => { last_err = Some(e); continue; } - }; - - future::poll_fn(|cx| watcher.poll_write_ready(cx)).await; - - match watcher.get_ref().take_error() { - Ok(None) => { - return Ok(TcpStream { - watcher: Arc::new(watcher), - }); - } - Ok(Some(e)) => last_err = Some(e), - Err(e) => last_err = Some(e), } } @@ -214,7 +202,7 @@ impl TcpStream { /// # Ok(()) }) } /// ``` pub async fn peek(&self, buf: &mut [u8]) -> io::Result { - 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> { - 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> { - self.watcher - .poll_write_with(cx, |mut inner| inner.write(buf)) + Pin::new(&mut &*self.watcher).poll_write(cx, buf) } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - 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> { - self.shutdown(std::net::Shutdown::Write)?; - Poll::Ready(Ok(())) + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut &*self.watcher).poll_close(cx) } } impl From for TcpStream { /// Converts a `std::net::TcpStream` into its asynchronous equivalent. fn from(stream: std::net::TcpStream) -> TcpStream { - let mio_stream = mio::net::TcpStream::from_stream(stream).unwrap(); TcpStream { - watcher: Arc::new(Watcher::new(mio_stream)), + watcher: Arc::new(Async::new(stream).expect("TcpStream is known to be good")), } } } @@ -403,23 +388,28 @@ cfg_unix! { } cfg_windows! { - // use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle}; - // - // impl AsRawSocket for TcpStream { - // fn as_raw_socket(&self) -> RawSocket { - // self.raw_socket - // } - // } - // - // impl FromRawSocket for TcpStream { - // unsafe fn from_raw_socket(handle: RawSocket) -> TcpStream { - // net::TcpStream::from_raw_socket(handle).try_into().unwrap() - // } - // } - // - // impl IntoRawSocket for TcpListener { - // fn into_raw_socket(self) -> RawSocket { - // self.raw_socket - // } - // } + use crate::os::windows::io::{ + RawSocket, AsRawSocket, FromRawSocket, IntoRawSocket + }; + + impl AsRawSocket for TcpStream { + fn as_raw_socket(&self) -> RawSocket { + self.watcher.get_ref().as_raw_socket() + } + } + + impl FromRawSocket for TcpStream { + unsafe fn from_raw_socket(handle: RawSocket) -> TcpStream { + std::net::TcpStream::from_raw_socket(handle).into() + } + } + + impl IntoRawSocket for TcpStream { + fn into_raw_socket(self) -> RawSocket { + // TODO(stjepang): This does not mean `RawFd` is now the sole owner of the file + // descriptor because it's possible that there are other clones of this `TcpStream` + // using it at the same time. We should probably document that behavior. + self.as_raw_socket() + } + } } diff --git a/src/net/udp/mod.rs b/src/net/udp/mod.rs index 774478d..30cceb7 100644 --- a/src/net/udp/mod.rs +++ b/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, + watcher: Async, } impl UdpSocket { @@ -69,16 +69,12 @@ impl UdpSocket { /// ``` pub async fn bind(addrs: A) -> io::Result { let mut last_err = None; - let addrs = addrs - .to_socket_addrs() - .await?; + let addrs = addrs.to_socket_addrs().await?; for addr in addrs { - match mio::net::UdpSocket::bind(&addr) { - Ok(mio_socket) => { - return Ok(UdpSocket { - watcher: Watcher::new(mio_socket), - }); + match Async::::bind(&addr) { + Ok(socket) => { + return Ok(UdpSocket { watcher: socket }); } Err(err) => last_err = Some(err), } @@ -153,12 +149,10 @@ impl UdpSocket { } }; - future::poll_fn(|cx| { - self.watcher - .poll_write_with(cx, |inner| inner.send_to(buf, &addr)) - }) - .await - .context(|| format!("could not send packet to {}", addr)) + self.watcher + .send_to(buf, addr) + .await + .context(|| format!("could not send packet to {}", addr)) } /// Receives data from the socket. @@ -181,22 +175,7 @@ impl UdpSocket { /// # Ok(()) }) } /// ``` pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { - future::poll_fn(|cx| { - self.watcher - .poll_read_with(cx, |inner| inner.recv_from(buf)) - }) - .await - .context(|| { - use std::fmt::Write; - - let mut error = String::from("could not receive data on "); - if let Ok(addr) = self.local_addr() { - let _ = write!(&mut error, "{}", addr); - } else { - error.push_str("socket"); - } - error - }) + self.watcher.recv_from(buf).await } /// Connects the UDP socket to a remote address. @@ -267,19 +246,7 @@ impl UdpSocket { /// # Ok(()) }) } /// ``` pub async fn send(&self, buf: &[u8]) -> io::Result { - future::poll_fn(|cx| self.watcher.poll_write_with(cx, |inner| inner.send(buf))) - .await - .context(|| { - use std::fmt::Write; - - let mut error = String::from("could not send data on "); - if let Ok(addr) = self.local_addr() { - let _ = write!(&mut error, "{}", addr); - } else { - error.push_str("socket"); - } - error - }) + self.watcher.send(buf).await } /// Receives data from the socket. @@ -303,19 +270,7 @@ impl UdpSocket { /// # Ok(()) }) } /// ``` pub async fn recv(&self, buf: &mut [u8]) -> io::Result { - future::poll_fn(|cx| self.watcher.poll_read_with(cx, |inner| inner.recv(buf))) - .await - .context(|| { - use std::fmt::Write; - - let mut error = String::from("could not receive data on "); - if let Ok(addr) = self.local_addr() { - let _ = write!(&mut error, "{}", addr); - } else { - error.push_str("socket"); - } - error - }) + self.watcher.recv(buf).await } /// Gets the value of the `SO_BROADCAST` option for this socket. @@ -498,9 +453,8 @@ impl UdpSocket { impl From for UdpSocket { /// Converts a `std::net::UdpSocket` into its asynchronous equivalent. fn from(socket: std::net::UdpSocket) -> UdpSocket { - let mio_socket = mio::net::UdpSocket::from_socket(socket).unwrap(); UdpSocket { - watcher: Watcher::new(mio_socket), + watcher: Async::new(socket).expect("UdpSocket is known to be good"), } } } @@ -522,29 +476,31 @@ cfg_unix! { impl IntoRawFd for UdpSocket { fn into_raw_fd(self) -> RawFd { - self.watcher.into_inner().into_raw_fd() + self.watcher.into_raw_fd() } } } cfg_windows! { - // use crate::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle}; - // - // impl AsRawSocket for UdpSocket { - // fn as_raw_socket(&self) -> RawSocket { - // self.raw_socket - // } - // } - // - // impl FromRawSocket for UdpSocket { - // unsafe fn from_raw_socket(handle: RawSocket) -> UdpSocket { - // net::UdpSocket::from_raw_socket(handle).into() - // } - // } - // - // impl IntoRawSocket for UdpSocket { - // fn into_raw_socket(self) -> RawSocket { - // self.raw_socket - // } - // } + use crate::os::windows::io::{ + RawSocket, AsRawSocket, IntoRawSocket, FromRawSocket + }; + + impl AsRawSocket for UdpSocket { + fn as_raw_socket(&self) -> RawSocket { + self.watcher.get_ref().as_raw_socket() + } + } + + impl FromRawSocket for UdpSocket { + unsafe fn from_raw_socket(handle: RawSocket) -> UdpSocket { + std::net::UdpSocket::from_raw_socket(handle).into() + } + } + + impl IntoRawSocket for UdpSocket { + fn into_raw_socket(self) -> RawSocket { + self.watcher.into_raw_socket() + } + } } diff --git a/src/os/unix/net/datagram.rs b/src/os/unix/net/datagram.rs index 5a2d6ec..6a30b02 100644 --- a/src/os/unix/net/datagram.rs +++ b/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, + watcher: Async, } 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>(path: P) -> io::Result { let path = path.as_ref().to_owned(); - let socket = spawn_blocking(move || mio_uds::UnixDatagram::bind(path)).await?; - Ok(UnixDatagram::new(socket)) + let socket = Async::::bind(path)?; + 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 { - 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 - .poll_read_with(cx, |inner| inner.recv_from(buf)) - }) - .await + self.watcher.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 { - 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>(&self, buf: &[u8], path: P) -> io::Result { - future::poll_fn(|cx| { - self.watcher - .poll_write_with(cx, |inner| inner.send_to(buf, path.as_ref())) - }) - .await + self.watcher.send_to(buf, path.as_ref()).await } /// Sends data on the socket to the socket's peer. @@ -267,7 +257,7 @@ impl UnixDatagram { /// # Ok(()) }) } /// ``` pub async fn send(&self, buf: &[u8]) -> io::Result { - 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 for UnixDatagram { +impl From for UnixDatagram { /// Converts a `std::os::unix::net::UnixDatagram` into its asynchronous equivalent. - fn from(datagram: std::os::unix::net::UnixDatagram) -> UnixDatagram { - let mio_datagram = mio_uds::UnixDatagram::from_datagram(datagram).unwrap(); + fn from(datagram: StdUnixDatagram) -> UnixDatagram { UnixDatagram { - watcher: Watcher::new(mio_datagram), + watcher: Async::new(datagram).expect("UnixDatagram is known to be good"), } } } impl AsRawFd for UnixDatagram { fn as_raw_fd(&self) -> RawFd { - self.watcher.get_ref().as_raw_fd() + self.watcher.as_raw_fd() } } impl FromRawFd for UnixDatagram { unsafe fn from_raw_fd(fd: RawFd) -> UnixDatagram { - let datagram = std::os::unix::net::UnixDatagram::from_raw_fd(fd); - datagram.into() + let raw = StdUnixDatagram::from_raw_fd(fd); + let datagram = Async::::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() } } diff --git a/src/os/unix/net/listener.rs b/src/os/unix/net/listener.rs index 9f6bdcb..4099bd6 100644 --- a/src/os/unix/net/listener.rs +++ b/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, + watcher: Async, } impl UnixListener { @@ -69,11 +68,9 @@ impl UnixListener { /// ``` pub async fn bind>(path: P) -> io::Result { let path = path.as_ref().to_owned(); - let listener = spawn_blocking(move || mio_uds::UnixListener::bind(path)).await?; + let listener = Async::::bind(path)?; - Ok(UnixListener { - watcher: Watcher::new(listener), - }) + Ok(UnixListener { watcher: 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 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? { - 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 + let (stream, addr) = self.watcher.accept().await?; + + Ok((UnixStream { watcher: stream }, addr)) } /// Returns a stream of incoming connections. @@ -206,19 +183,18 @@ impl Stream for Incoming<'_> { } } -impl From for UnixListener { +impl From for UnixListener { /// Converts a `std::os::unix::net::UnixListener` into its asynchronous equivalent. - fn from(listener: std::os::unix::net::UnixListener) -> UnixListener { - let mio_listener = mio_uds::UnixListener::from_listener(listener).unwrap(); + fn from(listener: StdUnixListener) -> UnixListener { UnixListener { - watcher: Watcher::new(mio_listener), + watcher: Async::new(listener).expect("UnixListener is known to be good"), } } } impl AsRawFd for UnixListener { fn as_raw_fd(&self) -> RawFd { - self.watcher.get_ref().as_raw_fd() + self.watcher.as_raw_fd() } } @@ -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() } } diff --git a/src/os/unix/net/stream.rs b/src/os/unix/net/stream.rs index a1c83f1..7320c85 100644 --- a/src/os/unix/net/stream.rs +++ b/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, + pub(super) watcher: Async, } impl UnixStream { @@ -57,15 +56,9 @@ impl UnixStream { /// ``` pub async fn connect>(path: P) -> io::Result { let path = path.as_ref().to_owned(); + let stream = Async::::connect(path).await?; - spawn_blocking(move || { - let std_stream = std::os::unix::net::UnixStream::connect(path)?; - let mio_stream = mio_uds::UnixStream::from_stream(std_stream)?; - Ok(UnixStream { - watcher: Watcher::new(mio_stream), - }) - }) - .await + Ok(UnixStream { watcher: stream }) } /// Creates an unnamed pair of connected sockets. @@ -84,13 +77,9 @@ impl UnixStream { /// # Ok(()) }) } /// ``` pub fn pair() -> io::Result<(UnixStream, UnixStream)> { - let (a, b) = mio_uds::UnixStream::pair()?; - let a = UnixStream { - watcher: Watcher::new(a), - }; - let b = UnixStream { - watcher: Watcher::new(b), - }; + let (a, b) = Async::::pair()?; + let a = UnixStream { watcher: a }; + let b = UnixStream { watcher: b }; Ok((a, b)) } @@ -169,7 +158,7 @@ impl Read for &UnixStream { cx: &mut Context<'_>, buf: &mut [u8], ) -> Poll> { - 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> { - self.watcher - .poll_write_with(cx, |mut inner| inner.write(buf)) + Pin::new(&mut &self.watcher).poll_write(cx, buf) } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - 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> { - Poll::Ready(Ok(())) + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut &self.watcher).poll_close(cx) } } @@ -227,19 +215,17 @@ impl fmt::Debug for UnixStream { } } -impl From for UnixStream { +impl From for UnixStream { /// Converts a `std::os::unix::net::UnixStream` into its asynchronous equivalent. - fn from(stream: std::os::unix::net::UnixStream) -> UnixStream { - let mio_stream = mio_uds::UnixStream::from_stream(stream).unwrap(); - UnixStream { - watcher: Watcher::new(mio_stream), - } + fn from(stream: StdUnixStream) -> UnixStream { + let stream = Async::new(stream).expect("UnixStream is known to be good"); + UnixStream { watcher: 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() } } diff --git a/src/os/windows/io.rs b/src/os/windows/io.rs index e83d557..30d37a0 100644 --- a/src/os/windows/io.rs +++ b/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; + } } diff --git a/src/path/path.rs b/src/path/path.rs index dfe9426..185bfaf 100644 --- a/src/path/path.rs +++ b/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(self).await } @@ -607,6 +608,7 @@ impl Path { /// # /// # Ok(()) }) } /// ``` + #[cfg(not(target_os = "unknown"))] pub async fn symlink_metadata(&self) -> io::Result { fs::symlink_metadata(self).await } @@ -632,6 +634,7 @@ impl Path { /// # /// # Ok(()) }) } /// ``` + #[cfg(not(target_os = "unknown"))] pub async fn canonicalize(&self) -> io::Result { fs::canonicalize(self).await } @@ -654,6 +657,7 @@ impl Path { /// # /// # Ok(()) }) } /// ``` + #[cfg(not(target_os = "unknown"))] pub async fn read_link(&self) -> io::Result { 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::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 diff --git a/src/rt/mod.rs b/src/rt/mod.rs index 2149d24..d5d0d61 100644 --- a/src/rt/mod.rs +++ b/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}; -pub use runtime::Runtime; - -mod reactor; -mod runtime; +/// Dummy runtime struct. +pub struct Runtime {} /// The global runtime. pub static RUNTIME: Lazy = Lazy::new(|| { - thread::Builder::new() - .name("async-std/runtime".to_string()) - .spawn(|| abort_on_panic(|| RUNTIME.run())) - .expect("cannot start a runtime thread"); - - Runtime::new() + // Create an executor thread pool. + let num_threads = num_cpus::get().max(1); + for _ in 0..num_threads { + thread::Builder::new() + .name("async-std/runtime".to_string()) + .spawn(|| smol::run(future::pending::<()>())) + .expect("cannot start a runtime thread"); + } + Runtime {} }); diff --git a/src/rt/reactor.rs b/src/rt/reactor.rs deleted file mode 100644 index 2a35b72..0000000 --- a/src/rt/reactor.rs +++ /dev/null @@ -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, - - /// Tasks that are blocked on writing to this I/O handle. - writers: Mutex, -} - -/// 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, - - /// A collection of registered I/O handles. - entries: Mutex>>, - - /// 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, -} - -/// 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, -} - -impl Reactor { - /// Creates a new reactor for polling I/O events. - pub fn new() -> io::Result { - 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> { - 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) -> io::Result { - 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 { - /// Data associated with the I/O handle. - entry: Arc, - - /// The I/O event source. - source: Option, -} - -impl Watcher { - /// 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 { - 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> - where - F: FnMut(&'a T) -> io::Result, - { - // 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> - where - F: FnMut(&'a T) -> io::Result, - { - // 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 Drop for Watcher { - 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 fmt::Debug for Watcher { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Watcher") - .field("entry", &self.entry) - .field("source", &self.source) - .finish() - } -} diff --git a/src/rt/runtime.rs b/src/rt/runtime.rs deleted file mode 100644 index a0d88b9..0000000 --- a/src/rt/runtime.rs +++ /dev/null @@ -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> = OnceCell::new(); - - /// This flag is set to true whenever `task::yield_now()` is invoked. - static YIELD_NOW: Cell = Cell::new(false); -} - -struct Scheduler { - /// Set to `true` while a machine is polling the reactor. - polling: bool, - - /// Idle processors. - processors: Vec, - - /// Running machines. - machines: Vec>, -} - -/// An async runtime. -pub struct Runtime { - /// The reactor. - reactor: Reactor, - - /// The global queue of tasks. - injector: Injector, - - /// Handles to local queues for stealing work. - stealers: Vec>, - - /// The scheduler state. - sched: Mutex, -} - -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> { - 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 { - 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>, -} - -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 { - 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, - - /// Contains the next task to run as an optimization that skips the queue. - slot: Option, -} - -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 { - self.slot.take().or_else(|| self.worker.pop()) - } - - /// Steals a task from the global queue. - fn steal_from_global(&self, rt: &Runtime) -> Steal { - rt.injector.steal_batch_and_pop(&self.worker) - } - - /// Steals a task from other processors. - fn steal_from_others(&self, rt: &Runtime) -> Steal { - // 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() - } -} diff --git a/src/stream/interval.rs b/src/stream/interval.rs index be94b06..4e5c92b 100644 --- a/src/stream/interval.rs +++ b/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(())) } } diff --git a/src/stream/stream/delay.rs b/src/stream/stream/delay.rs index ff4c937..0ba42b0 100644 --- a/src/stream/stream/delay.rs +++ b/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 Delay { pub(super) fn new(stream: S, dur: Duration) -> Self { Delay { stream, - delay: futures_timer::Delay::new(dur), + delay: Timer::after(dur), delay_done: false, } } diff --git a/src/stream/stream/throttle.rs b/src/stream/stream/throttle.rs index 554ca30..2f9333a 100644 --- a/src/stream/stream/throttle.rs +++ b/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 Throttle { stream, duration, blocked: false, - delay: Delay::new(Duration::default()), + delay: Timer::after(Duration::default()), } } } @@ -59,7 +59,7 @@ impl Stream for Throttle { 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)) } } diff --git a/src/stream/stream/timeout.rs b/src/stream/stream/timeout.rs index ce658c8..28e52ae 100644 --- a/src/stream/stream/timeout.rs +++ b/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 Timeout { pub(crate) fn new(stream: S, dur: Duration) -> Self { - let delay = Delay::new(dur); + let delay = Timer::after(dur); Self { stream, delay } } diff --git a/src/sync/barrier.rs b/src/sync/barrier.rs index 2822d54..86e9a2d 100644 --- a/src/sync/barrier.rs +++ b/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; diff --git a/src/sync/mod.rs b/src/sync/mod.rs index 1531f8c..bccc6ec 100644 --- a/src/sync/mod.rs +++ b/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; -} diff --git a/src/sync/spin_lock.rs b/src/sync/spin_lock.rs deleted file mode 100644 index 854b7e0..0000000 --- a/src/sync/spin_lock.rs +++ /dev/null @@ -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 { - locked: AtomicBool, - value: UnsafeCell, -} - -unsafe impl Send for Spinlock {} -unsafe impl Sync for Spinlock {} - -impl Spinlock { - /// Returns a new spinlock initialized with `value`. - pub const fn new(value: T) -> Spinlock { - 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, -} - -unsafe impl Send for SpinlockGuard<'_, T> {} -unsafe impl 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); - }) -} diff --git a/src/task/block_on.rs b/src/task/block_on.rs index 4bade5b..fa66f91 100644 --- a/src/task/block_on.rs +++ b/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 kv_log_macro::trace; - -use crate::task::{Context, Poll, Task, Waker}; +use crate::task::Builder; /// 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(future: F) -> T where F: Future, { - // Create a new task handle. - 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)) } + Builder::new().blocking(future) } -/// Blocks the current thread on a future's result. -fn run(future: F) -> T +/// Spawns a task and waits for it to finish. +#[cfg(target_os = "unknown")] +pub fn block_on(future: F) where - F: Future, + F: Future + 'static, + T: 'static, { - thread_local! { - // May hold a pre-allocated parker that can be reused for efficiency. - // - // Note that each invocation of `block` needs its own parker. In particular, if `block` - // recursively calls itself, we must make sure that each recursive call uses a distinct - // parker instance. - static CACHE: Cell>> = Cell::new(None); - } - - // Virtual table for wakers based on `Arc`. - 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 = cache.take().unwrap_or_else(|| Arc::new(Parker::new())); - let ptr = (&*arc_parker as *const Parker) as *const (); - - // Create a waker and task context. - let waker = unsafe { ManuallyDrop::new(Waker::from_raw(RawWaker::new(ptr, &VTABLE))) }; - let cx = &mut Context::from_waker(&waker); - - loop { - if let Poll::Ready(t) = future.as_mut().poll(cx) { - // Save the parker for the next invocation of `block`. - cache.set(Some(arc_parker)); - return t; - } - - arc_parker.park(); - } - }) + Builder::new().local(future).unwrap(); } diff --git a/src/task/builder.rs b/src/task/builder.rs index f1fef59..cbb3187 100644 --- a/src/task/builder.rs +++ b/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}; -use crate::utils::abort_on_panic; +use crate::task::{JoinHandle, Task, TaskLocalsWrapper}; /// Task builder that configures the settings of a new task. #[derive(Debug, Default)] @@ -27,52 +28,151 @@ impl Builder { self } + fn build(self, future: F) -> SupportTaskLocals + where + F: Future, + { + 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(self, future: F) -> io::Result> where F: Future + Send + 'static, T: Send + 'static, { - // Create a new task handle. - let task = Task::new(self.name); + let wrapped = self.build(future); + + kv_log_macro::trace!("spawn", { + task_id: wrapped.tag.id().0, + parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0), + }); + + let task = wrapped.tag.task().clone(); + let smol_task = smol::Task::spawn(wrapped).into(); + + Ok(JoinHandle::new(smol_task, task)) + } + + /// Spawns a task locally with the configured settings. + #[cfg(all(not(target_os = "unknown"), feature = "unstable"))] + pub fn local(self, future: F) -> io::Result> + where + F: Future + 'static, + T: 'static, + { + let wrapped = self.build(future); + + kv_log_macro::trace!("spawn_local", { + task_id: wrapped.tag.id().0, + parent_task_id: TaskLocalsWrapper::get_current(|t| t.id().0).unwrap_or(0), + }); + + let task = wrapped.tag.task().clone(); + let smol_task = smol::Task::local(wrapped).into(); + + Ok(JoinHandle::new(smol_task, task)) + } + + /// Spawns a task locally with the configured settings. + #[cfg(all(target_arch = "wasm32", feature = "unstable"))] + pub fn local(self, future: F) -> io::Result> + where + F: Future + '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(self, future: F) -> io::Result> + where + F: Future + '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)) + } - // Log this `spawn` operation. - trace!("spawn", { - task_id: task.id().0, - parent_task_id: Task::get_current(|t| t.id().0).unwrap_or(0), + /// Spawns a task with the configured settings, blocking on its execution. + #[cfg(not(target_os = "unknown"))] + pub fn blocking(self, future: F) -> T + where + F: Future, + { + 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), }); - 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 - }; - - let schedule = move |t| RUNTIME.schedule(Runnable(t)); - let (task, handle) = async_task::spawn(future, schedule, task); - task.schedule(); - Ok(JoinHandle::new(handle)) + // Run the future as a task. + unsafe { TaskLocalsWrapper::set_current(&wrapped.tag, || smol::run(wrapped)) } + } +} + +pin_project! { + /// Wrapper to add support for task locals. + struct SupportTaskLocals { + tag: TaskLocalsWrapper, + #[pin] + future: F, } } -/// A runnable task. -pub struct Runnable(async_task::Task); +impl Future for SupportTaskLocals { + type Output = F::Output; -impl Runnable { - /// Runs the task by polling its future once. - pub fn run(self) { + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { 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) + }) } } } diff --git a/src/task/current.rs b/src/task/current.rs index 0dc3699..e4624e1 100644 --- a/src/task/current.rs +++ b/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") } diff --git a/src/task/join_handle.rs b/src/task/join_handle.rs index d929d11..110b827 100644 --- a/src/task/join_handle.rs +++ b/src/task/join_handle.rs @@ -12,12 +12,23 @@ use crate::task::{Context, Poll, Task}; /// /// [spawned]: fn.spawn.html #[derive(Debug)] -pub struct JoinHandle(async_task::JoinHandle); +pub struct JoinHandle { + handle: Option>, + task: Task, +} + +#[cfg(not(target_os = "unknown"))] +type InnerHandle = async_task::JoinHandle; +#[cfg(target_arch = "wasm32")] +type InnerHandle = futures_channel::oneshot::Receiver; impl JoinHandle { /// Creates a new `JoinHandle`. - pub(crate) fn new(inner: async_task::JoinHandle) -> JoinHandle { - JoinHandle(inner) + pub(crate) fn new(inner: InnerHandle, task: Task) -> JoinHandle { + JoinHandle { + handle: Some(inner), + task, + } } /// Returns a handle to the underlying task. @@ -36,7 +47,23 @@ impl JoinHandle { /// # /// # }) 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 { + let handle = self.handle.take().unwrap(); + handle.cancel(); + handle.await + } + + /// Cancel this task. + #[cfg(target_arch = "wasm32")] + pub async fn cancel(mut self) -> Option { + let mut handle = self.handle.take().unwrap(); + handle.close(); + handle.await.ok() } } @@ -44,10 +71,11 @@ impl Future for JoinHandle { type Output = T; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - match Pin::new(&mut self.0).poll(cx) { + match Pin::new(&mut self.handle.as_mut().unwrap()).poll(cx) { Poll::Pending => Poll::Pending, - Poll::Ready(None) => panic!("cannot await the result of a panicked task"), - Poll::Ready(Some(val)) => Poll::Ready(val), + Poll::Ready(output) => { + Poll::Ready(output.expect("cannot await the result of a panicked task")) + } } } } diff --git a/src/task/mod.rs b/src/task/mod.rs index 56224a3..eefc7c2 100644 --- a/src/task/mod.rs +++ b/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; +} diff --git a/src/task/spawn_blocking.rs b/src/task/spawn_blocking.rs index 27143f7..e9ed0c5 100644 --- a/src/task/spawn_blocking.rs +++ b/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(); - let (task, handle) = async_task::spawn(async { f() }, schedule, Task::new(None)); - task.schedule(); - JoinHandle::new(handle) -} - -type Runnable = async_task::Task; - -struct Pool { - sender: Sender, - receiver: Receiver, -} - -/// The number of sleeping worker threads. -static SLEEPING: AtomicUsize = AtomicUsize::new(0); - -static POOL: Lazy = 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; - } + once_cell::sync::Lazy::force(&crate::rt::RUNTIME); - SLEEPING.fetch_add(1, Ordering::SeqCst); - } - }) - .expect("cannot start a blocking thread"); + let handle = smol::Task::blocking(async move { f() }).into(); + JoinHandle::new(handle, Task::new(None)) } diff --git a/src/task/spawn_local.rs b/src/task/spawn_local.rs new file mode 100644 index 0000000..04da020 --- /dev/null +++ b/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(future: F) -> JoinHandle +where + F: Future + 'static, + T: 'static, +{ + Builder::new().local(future).expect("cannot spawn task") +} diff --git a/src/task/task.rs b/src/task/task.rs index bcec2e0..eba99c7 100644 --- a/src/task/task.rs +++ b/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::utils::abort_on_panic; +use crate::task::TaskId; -thread_local! { - /// A pointer to the currently running task. - static CURRENT: Cell<*const Task> = Cell::new(ptr::null_mut()); -} - -/// The inner representation of a task handle. -struct Inner { +/// A handle to a task. +#[derive(Clone)] +pub struct Task { /// The task ID. id: TaskId, /// The optional task name. - name: Option>, - - /// The map holding task-local values. - locals: LocalsMap, -} - -impl Inner { - #[inline] - fn new(name: Option) -> Inner { - Inner { - id: TaskId::generate(), - name: name.map(String::into_boxed_str), - locals: LocalsMap::new(), - } - } + name: Option>, } -/// 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, -} - -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) -> Task { - let inner = match name { - None => AtomicPtr::default(), - Some(name) => { - let raw = Arc::into_raw(Arc::new(Inner::new(Some(name)))); - AtomicPtr::new(raw as *mut Inner) - } - }; - Task { inner } + pub(crate) fn new(name: Option>) -> Task { + Task { + id: TaskId::generate(), + name, + } } /// Gets the task's unique identifier. #[inline] pub fn id(&self) -> TaskId { - self.inner().id + self.id } /// Returns the name of this task. @@ -77,93 +35,7 @@ impl Task { /// /// [`Builder::name`]: struct.Builder.html#method.name pub fn name(&self) -> Option<&str> { - self.inner().name.as_ref().map(|s| &**s) - } - - /// Returns the map holding task-local values. - pub(crate) fn locals(&self) -> &LocalsMap { - &self.inner().locals - } - - /// Drops all task-local values. - /// - /// This method is only safe to call at the end of the task. - #[inline] - pub(crate) unsafe fn drop_locals(&self) { - let raw = self.inner.load(Ordering::Acquire); - if let Some(inner) = raw.as_mut() { - // Abort the process if dropping task-locals panics. - abort_on_panic(|| { - inner.locals.clear(); - }); - } - } - - /// Returns the inner representation, initializing it on first use. - fn inner(&self) -> &Inner { - loop { - let raw = self.inner.load(Ordering::Acquire); - if !raw.is_null() { - return unsafe { &*raw }; - } - - let new = Arc::into_raw(Arc::new(Inner::new(None))) as *mut Inner; - if self.inner.compare_and_swap(raw, new, Ordering::AcqRel) != raw { - unsafe { - drop(Arc::from_raw(new)); - } - } - } - } - - /// Set a reference to the current task. - pub(crate) unsafe fn set_current(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: F) -> Option - 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`. - let arc = unsafe { ManuallyDrop::new(Arc::from_raw(self.inner())) }; - let raw = Arc::into_raw(Arc::clone(&arc)); - Task { - inner: AtomicPtr::new(raw as *mut Inner), - } + self.name.as_ref().map(|s| s.as_str()) } } diff --git a/src/task/task_local.rs b/src/task/task_local.rs index 72e53d7..4e2ba83 100644 --- a/src/task/task_local.rs +++ b/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 LocalKey { 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; diff --git a/src/task/task_locals_wrapper.rs b/src/task/task_locals_wrapper.rs new file mode 100644 index 0000000..2a7ddb7 --- /dev/null +++ b/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(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: F) -> Option + 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() }; + }); + } +} diff --git a/src/task/yield_now.rs b/src/task/yield_now.rs index bdb08eb..2b1fd0b 100644 --- a/src/task/yield_now.rs +++ b/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(()) diff --git a/src/utils.rs b/src/utils.rs index 4bdbd92..ef068cb 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -21,7 +21,7 @@ pub fn abort_on_panic(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 { + 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)] diff --git a/tests/addr.rs b/tests/addr.rs index aada557..fcd5aa1 100644 --- a/tests/addr.rs +++ b/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; diff --git a/tests/block_on.rs b/tests/block_on.rs index c422d06..28902b0 100644 --- a/tests/block_on.rs +++ b/tests/block_on.rs @@ -1,3 +1,5 @@ +#![cfg(not(target_os = "unknown"))] + use async_std::task; #[test] diff --git a/tests/buf_writer.rs b/tests/buf_writer.rs index 5df90e0..442cf8a 100644 --- a/tests/buf_writer.rs +++ b/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 mut writer = BufWriter::with_capacity(2, inner); + let inner: Vec = Vec::new(); + let mut writer = BufWriter::>::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::::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())); diff --git a/tests/channel.rs b/tests/channel.rs index f302906..a218ea2 100644 --- a/tests/channel.rs +++ b/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 c2 = task::spawn(async move { s.send(0).await }); + let c1 = spawn(async move { r.recv().await.unwrap() }); + let c2 = spawn(async move { s.send(0).await }); c1.await; c2.await; @@ -307,6 +326,7 @@ fn oneshot() { } #[test] +#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)] fn drops() { const RUNS: usize = 100; @@ -321,17 +341,16 @@ fn drops() { } } - let mut rng = thread_rng(); - for _ in 0..RUNS { - task::block_on(async { + let mut rng = rand_xorshift::XorShiftRng::seed_from_u64(0); + task::block_on(async move { let steps = rng.gen_range(0, 10_000); let additional = rng.gen_range(0, 50); DROPS.store(0, Ordering::SeqCst); let (s, r) = channel::(50); - let child = task::spawn({ + let child = spawn({ let r = r.clone(); async move { for _ in 0..steps { diff --git a/tests/condvar.rs b/tests/condvar.rs index c4d680f..76574a1 100644 --- a/tests/condvar.rs +++ b/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> = 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; { diff --git a/tests/io_timeout.rs b/tests/io_timeout.rs index 85a17ab..fa30a68 100644 --- a/tests/io_timeout.rs +++ b/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 { diff --git a/tests/mutex.rs b/tests/mutex.rs index fd1c07b..76f42e2 100644 --- a/tests/mutex.rs +++ b/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); }); diff --git a/tests/rwlock.rs b/tests/rwlock.rs index 370dcb9..1d33a45 100644 --- a/tests/rwlock.rs +++ b/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); @@ -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); @@ -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; diff --git a/tests/stream.rs b/tests/stream.rs index 42a6191..3576cb9 100644 --- a/tests/stream.rs +++ b/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::(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::(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: S) -> Explode { } #[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]); } diff --git a/tests/task_local.rs b/tests/task_local.rs index 813185c..b5345fe 100644 --- a/tests/task_local.rs +++ b/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(); diff --git a/tests/tcp.rs b/tests/tcp.rs index d92cff0..f21737e 100644 --- a/tests/tcp.rs +++ b/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::*; diff --git a/tests/timeout.rs b/tests/timeout.rs new file mode 100644 index 0000000..8ad358a --- /dev/null +++ b/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::>(); + + for future in futures { + future.await.unwrap().unwrap(); + } + }); +} diff --git a/tests/udp.rs b/tests/udp.rs index 319dc74..15404f8 100644 --- a/tests/udp.rs +++ b/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; diff --git a/tests/uds.rs b/tests/uds.rs index 3ab4d6b..038ac0e 100644 --- a/tests/uds.rs +++ b/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}; diff --git a/tests/verbose_errors.rs b/tests/verbose_errors.rs index 17d4261..2876183 100644 --- a/tests/verbose_errors.rs +++ b/tests/verbose_errors.rs @@ -1,3 +1,5 @@ +#![cfg(not(target_os = "unknown"))] + use async_std::{fs, io, net::ToSocketAddrs, task}; #[test] diff --git a/wasm-test.sh b/wasm-test.sh new file mode 100755 index 0000000..3d8be9f --- /dev/null +++ b/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