New scheduler resilient to blocking

Author: Stjepan Glavina

Cargo.toml

@@ -26,6 +26,7 @@ default = [
   "async-task",
   "crossbeam-channel",
   "crossbeam-deque",
+  "crossbeam-queue",
   "futures-timer",
   "kv-log-macro",
   "log",
@@ -56,6 +57,7 @@ async-task = { version = "1.0.0", optional = true }
 broadcaster = { version = "0.2.6", optional = true, default-features = false, features = ["default-channels"] }
 crossbeam-channel = { version = "0.4.0", optional = true }
 crossbeam-deque = { version = "0.7.2", optional = true }
+crossbeam-queue = { version = "0.2.0", optional = true }
 crossbeam-utils = { version = "0.7.0", optional = true }
 futures-core = { version = "0.3.1", optional = true }
 futures-io = { version = "0.3.1", optional = true }

src/lib.rs

@@ -246,6 +246,8 @@ cfg_std! {
     pub mod stream;
     pub mod sync;
     pub mod task;
+
+    pub(crate) mod rt;
 }
 
 cfg_default! {

src/net/mod.rs

@@ -66,6 +66,5 @@ pub use tcp::{Incoming, TcpListener, TcpStream};
 pub use udp::UdpSocket;
 
 mod addr;
-pub(crate) mod driver;
 mod tcp;
 mod udp;

src/net/tcp/listener.rs

@@ -4,7 +4,7 @@ use std::pin::Pin;
 
 use crate::future;
 use crate::io;
-use crate::net::driver::Watcher;
+use crate::rt::Watcher;
 use crate::net::{TcpStream, ToSocketAddrs};
 use crate::stream::Stream;
 use crate::task::{Context, Poll};

src/net/tcp/stream.rs

@@ -4,7 +4,7 @@ use std::pin::Pin;
 
 use crate::future;
 use crate::io::{self, Read, Write};
-use crate::net::driver::Watcher;
+use crate::rt::Watcher;
 use crate::net::ToSocketAddrs;
 use crate::task::{spawn_blocking, Context, Poll};
 use crate::utils::Context as _;

src/net/udp/mod.rs

@@ -3,8 +3,8 @@ use std::net::SocketAddr;
 use std::net::{Ipv4Addr, Ipv6Addr};
 
 use crate::future;
-use crate::net::driver::Watcher;
 use crate::net::ToSocketAddrs;
+use crate::rt::Watcher;
 use crate::utils::Context as _;
 
 /// A UDP socket.
@@ -102,7 +102,7 @@ impl UdpSocket {
     /// ```no_run
     /// # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
     /// #
-    ///	use async_std::net::UdpSocket;
+    /// use async_std::net::UdpSocket;
     ///
     /// let socket = UdpSocket::bind("127.0.0.1:0").await?;
     /// let addr = socket.local_addr()?;

src/os/unix/net/datagram.rs

@@ -8,7 +8,7 @@ use mio_uds;
 use super::SocketAddr;
 use crate::future;
 use crate::io;
-use crate::net::driver::Watcher;
+use crate::rt::Watcher;
 use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use crate::path::Path;
 use crate::task::spawn_blocking;

src/os/unix/net/listener.rs

@@ -10,7 +10,7 @@ use super::SocketAddr;
 use super::UnixStream;
 use crate::future;
 use crate::io;
-use crate::net::driver::Watcher;
+use crate::rt::Watcher;
 use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use crate::path::Path;
 use crate::stream::Stream;

src/os/unix/net/stream.rs

@@ -9,7 +9,7 @@ use mio_uds;
 
 use super::SocketAddr;
 use crate::io::{self, Read, Write};
-use crate::net::driver::Watcher;
+use crate::rt::Watcher;
 use crate::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use crate::path::Path;
 use crate::task::{spawn_blocking, Context, Poll};

src/rt/mod.rs

@@ -0,0 +1,23 @@
+//! The runtime.
+
+use std::thread;
+
+use once_cell::sync::Lazy;
+
+use crate::utils::abort_on_panic;
+
+pub use reactor::{Reactor, Watcher};
+pub use runtime::Runtime;
+
+mod reactor;
+mod runtime;
+
+/// The global runtime.
+pub static RUNTIME: Lazy<Runtime> = 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()
+});