first pass at Tokio 0.2

Author: bartlomieju

Diff for: Cargo.lock

@@ -52,10 +52,9 @@ source-map-mappings = "0.5.0"
 sys-info = "0.5.8"
 tempfile = "3.1.0"
 termcolor = "1.0.5"
-tokio = "0.1.22"
+tokio = { version = "0.2.1", features = ["full"] }
 tokio-executor = "0.1.8"
-tokio-process = "0.2.4"
-tokio-rustls = "0.10.2"
+tokio-rustls = "0.12.0"
 url = "1.7.2"
 utime = "0.2.1"
 webpki = "0.21.0"

Diff for: cli/Cargo.toml

@@ -13,7 +13,7 @@ use futures::channel::oneshot;
 use futures::future::FutureExt;
 use std::future::Future;
 use std::time::Instant;
-use tokio::timer::Delay;
+use tokio;
 
 #[derive(Default)]
 pub struct GlobalTimer {
@@ -43,8 +43,8 @@ impl GlobalTimer {
     let (tx, rx) = oneshot::channel();
     self.tx = Some(tx);
 
-    let delay = futures::compat::Compat01As03::new(Delay::new(deadline))
-      .map_err(|err| panic!("Unexpected error in timeout {:?}", err));
+    let tokio_deadline = tokio::time::Instant::from_std(deadline);
+    let delay = tokio::time::delay_until(tokio_deadline);
     let rx = rx
       .map_err(|err| panic!("Unexpected error in receiving channel {:?}", err));
 

Diff for: cli/global_timer.rs

@@ -12,11 +12,7 @@ use futures::future::FutureExt;
 use futures::future::TryFutureExt;
 use std;
 use std::convert::From;
-use std::future::Future;
 use std::io::SeekFrom;
-use std::pin::Pin;
-use std::task::Context;
-use std::task::Poll;
 use tokio;
 
 pub fn init(i: &mut Isolate, s: &ThreadSafeState) {
@@ -92,21 +88,19 @@ fn op_open(
   }
 
   let is_sync = args.promise_id.is_none();
-  let op = futures::compat::Compat01As03::new(tokio::prelude::Future::map_err(
-    open_options.open(filename),
-    ErrBox::from,
-  ))
-  .and_then(move |fs_file| {
+
+  let fut = async move {
+    let fs_file = open_options.open(filename).map_err(ErrBox::from).await?;
     let mut table = state_.lock_resource_table();
     let rid = table.add("fsFile", Box::new(StreamResource::FsFile(fs_file)));
-    futures::future::ok(json!(rid))
-  });
+    Ok(json!(rid))
+  };
 
   if is_sync {
-    let buf = futures::executor::block_on(op)?;
+    let buf = futures::executor::block_on(fut)?;
     Ok(JsonOp::Sync(buf))
   } else {
-    Ok(JsonOp::Async(op.boxed()))
+    Ok(JsonOp::Async(fut.boxed()))
   }
 }
 
@@ -127,37 +121,6 @@ fn op_close(
   Ok(JsonOp::Sync(json!({})))
 }
 
-pub struct SeekFuture {
-  seek_from: SeekFrom,
-  rid: ResourceId,
-  state: ThreadSafeState,
-}
-
-impl Future for SeekFuture {
-  type Output = Result<u64, ErrBox>;
-
-  fn poll(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Self::Output> {
-    let inner = self.get_mut();
-    let mut table = inner.state.lock_resource_table();
-    let resource = table
-      .get_mut::<StreamResource>(inner.rid)
-      .ok_or_else(bad_resource)?;
-
-    let tokio_file = match resource {
-      StreamResource::FsFile(ref mut file) => file,
-      _ => return Poll::Ready(Err(bad_resource())),
-    };
-
-    use tokio::prelude::Async::*;
-
-    match tokio_file.poll_seek(inner.seek_from).map_err(ErrBox::from) {
-      Ok(Ready(v)) => Poll::Ready(Ok(v)),
-      Err(err) => Poll::Ready(Err(err)),
-      Ok(NotReady) => Poll::Pending,
-    }
-  }
-}
-
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 struct SeekArgs {
@@ -189,17 +152,26 @@ fn op_seek(
     }
   };
 
-  let fut = SeekFuture {
-    state: state.clone(),
-    seek_from,
-    rid,
+  let mut table = state.lock_resource_table();
+  let resource = table
+    .get_mut::<StreamResource>(rid)
+    .ok_or_else(bad_resource)?;
+
+  let tokio_file = match resource {
+    StreamResource::FsFile(ref mut file) => file,
+    _ => return Err(bad_resource()),
+  };
+  let mut file = futures::executor::block_on(tokio_file.try_clone())?;
+
+  let fut = async move {
+    file.seek(seek_from).await?;
+    Ok(json!({}))
   };
 
-  let op = fut.and_then(move |_| futures::future::ok(json!({})));
   if args.promise_id.is_none() {
-    let buf = futures::executor::block_on(op)?;
+    let buf = futures::executor::block_on(fut)?;
     Ok(JsonOp::Sync(buf))
   } else {
-    Ok(JsonOp::Async(op.boxed()))
+    Ok(JsonOp::Async(fut.boxed()))
   }
 }

Diff for: cli/ops/files.rs

@@ -8,18 +8,15 @@ use deno::ErrBox;
 use deno::Resource;
 use deno::*;
 use futures;
-use futures::compat::AsyncRead01CompatExt;
-use futures::compat::AsyncWrite01CompatExt;
 use futures::future::FutureExt;
-use futures::io::{AsyncRead, AsyncWrite};
 use std;
 use std::future::Future;
 use std::pin::Pin;
 use std::task::Context;
 use std::task::Poll;
 use tokio;
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::TcpStream;
-use tokio_process;
 use tokio_rustls::client::TlsStream as ClientTlsStream;
 use tokio_rustls::server::TlsStream as ServerTlsStream;
 
@@ -86,9 +83,9 @@ pub enum StreamResource {
   ServerTlsStream(Box<ServerTlsStream<TcpStream>>),
   ClientTlsStream(Box<ClientTlsStream<TcpStream>>),
   HttpBody(Box<HttpBody>),
-  ChildStdin(tokio_process::ChildStdin),
-  ChildStdout(tokio_process::ChildStdout),
-  ChildStderr(tokio_process::ChildStderr),
+  ChildStdin(tokio::process::ChildStdin),
+  ChildStdout(tokio::process::ChildStdout),
+  ChildStderr(tokio::process::ChildStderr),
 }
 
 impl Resource for StreamResource {}
@@ -111,22 +108,15 @@ impl DenoAsyncRead for StreamResource {
   ) -> Poll<Result<usize, ErrBox>> {
     let inner = self.get_mut();
     let mut f: Box<dyn AsyncRead + Unpin> = match inner {
-      StreamResource::FsFile(f) => Box::new(AsyncRead01CompatExt::compat(f)),
-      StreamResource::Stdin(f) => Box::new(AsyncRead01CompatExt::compat(f)),
-      StreamResource::TcpStream(f) => Box::new(AsyncRead01CompatExt::compat(f)),
-      StreamResource::ClientTlsStream(f) => {
-        Box::new(AsyncRead01CompatExt::compat(f))
-      }
-      StreamResource::ServerTlsStream(f) => {
-        Box::new(AsyncRead01CompatExt::compat(f))
-      }
-      StreamResource::HttpBody(f) => Box::new(f),
-      StreamResource::ChildStdout(f) => {
-        Box::new(AsyncRead01CompatExt::compat(f))
-      }
-      StreamResource::ChildStderr(f) => {
-        Box::new(AsyncRead01CompatExt::compat(f))
-      }
+      StreamResource::FsFile(f) => Box::new(f),
+      StreamResource::Stdin(f) => Box::new(f),
+      StreamResource::TcpStream(f) => Box::new(f),
+      StreamResource::ClientTlsStream(f) => Box::new(f),
+      StreamResource::ServerTlsStream(f) => Box::new(f),
+      StreamResource::ChildStdout(f) => Box::new(f),
+      StreamResource::ChildStderr(f) => Box::new(f),
+      // TODO: temporarily disabled, pending release of Reqwest
+      // StreamResource::HttpBody(f) => Box::new(f),
       _ => {
         return Poll::Ready(Err(bad_resource()));
       }
@@ -247,21 +237,13 @@ impl DenoAsyncWrite for StreamResource {
   ) -> Poll<Result<usize, ErrBox>> {
     let inner = self.get_mut();
     let mut f: Box<dyn AsyncWrite + Unpin> = match inner {
-      StreamResource::FsFile(f) => Box::new(AsyncWrite01CompatExt::compat(f)),
-      StreamResource::Stdout(f) => Box::new(AsyncWrite01CompatExt::compat(f)),
-      StreamResource::Stderr(f) => Box::new(AsyncWrite01CompatExt::compat(f)),
-      StreamResource::TcpStream(f) => {
-        Box::new(AsyncWrite01CompatExt::compat(f))
-      }
-      StreamResource::ClientTlsStream(f) => {
-        Box::new(AsyncWrite01CompatExt::compat(f))
-      }
-      StreamResource::ServerTlsStream(f) => {
-        Box::new(AsyncWrite01CompatExt::compat(f))
-      }
-      StreamResource::ChildStdin(f) => {
-        Box::new(AsyncWrite01CompatExt::compat(f))
-      }
+      StreamResource::FsFile(f) => Box::new(f),
+      StreamResource::Stdout(f) => Box::new(f),
+      StreamResource::Stderr(f) => Box::new(f),
+      StreamResource::TcpStream(f) => Box::new(f),
+      StreamResource::ClientTlsStream(f) => Box::new(f),
+      StreamResource::ServerTlsStream(f) => Box::new(f),
+      StreamResource::ChildStdin(f) => Box::new(f),
       _ => {
         return Poll::Ready(Err(bad_resource()));
       }

Diff for: cli/ops/io.rs

@@ -9,8 +9,6 @@ use deno::Resource;
 use deno::*;
 use futures::future::FutureExt;
 use futures::future::TryFutureExt;
-use futures::stream::StreamExt;
-use futures::stream::TryStreamExt;
 use std;
 use std::convert::From;
 use std::future::Future;
@@ -20,7 +18,6 @@ use std::pin::Pin;
 use std::task::Context;
 use std::task::Poll;
 use tokio;
-use tokio::net::tcp::Incoming;
 use tokio::net::TcpListener;
 use tokio::net::TcpStream;
 
@@ -74,50 +71,44 @@ impl Future for Accept {
         ErrBox::from(e)
       })?;
 
-    let mut listener =
-      futures::compat::Compat01As03::new(&mut listener_resource.listener)
-        .map_err(ErrBox::from);
+    let listener = &mut listener_resource.listener;
 
     if inner.accept_state == AcceptState::Eager {
       // Similar to try_ready!, but also track/untrack accept task
       // in TcpListener resource.
       // In this way, when the listener is closed, the task can be
       // notified to error out (instead of stuck forever).
-      match listener.poll_next_unpin(cx) {
-        Poll::Ready(Some(Ok(stream))) => {
+      match listener.poll_accept(cx).map_err(ErrBox::from) {
+        Poll::Ready(Ok((stream, addr))) => {
           inner.accept_state = AcceptState::Done;
-          let addr = stream.peer_addr().unwrap();
           return Poll::Ready(Ok((stream, addr)));
         }
         Poll::Pending => {
           inner.accept_state = AcceptState::Pending;
           return Poll::Pending;
         }
-        Poll::Ready(Some(Err(e))) => {
+        Poll::Ready(Err(e)) => {
           inner.accept_state = AcceptState::Done;
           return Poll::Ready(Err(e));
         }
-        _ => unreachable!(),
       }
     }
 
-    match listener.poll_next_unpin(cx) {
-      Poll::Ready(Some(Ok(stream))) => {
+    match listener.poll_accept(cx).map_err(ErrBox::from) {
+      Poll::Ready(Ok((stream, addr))) => {
         listener_resource.untrack_task();
         inner.accept_state = AcceptState::Done;
-        let addr = stream.peer_addr().unwrap();
         Poll::Ready(Ok((stream, addr)))
       }
       Poll::Pending => {
         listener_resource.track_task(cx)?;
         Poll::Pending
       }
-      Poll::Ready(Some(Err(e))) => {
+      Poll::Ready(Err(e)) => {
         listener_resource.untrack_task();
         inner.accept_state = AcceptState::Done;
         Poll::Ready(Err(e))
       }
-      _ => unreachable!(),
     }
   }
 }
@@ -184,32 +175,20 @@ fn op_dial(
   let state_ = state.clone();
   state.check_net(&args.hostname, args.port)?;
 
-  let op = resolve_addr(&args.hostname, args.port).and_then(move |addr| {
-    futures::compat::Compat01As03::new(TcpStream::connect(&addr))
-      .map_err(ErrBox::from)
-      .and_then(move |tcp_stream| {
-        let local_addr = match tcp_stream.local_addr() {
-          Ok(v) => v,
-          Err(e) => return futures::future::err(ErrBox::from(e)),
-        };
-        let remote_addr = match tcp_stream.peer_addr() {
-          Ok(v) => v,
-          Err(e) => return futures::future::err(ErrBox::from(e)),
-        };
-        let mut table = state_.lock_resource_table();
-        let rid = table
-          .add("tcpStream", Box::new(StreamResource::TcpStream(tcp_stream)));
-        futures::future::ok((rid, local_addr, remote_addr))
-      })
-      .map_err(ErrBox::from)
-      .and_then(move |(rid, local_addr, remote_addr)| {
-        futures::future::ok(json!({
-          "rid": rid,
-          "localAddr": local_addr.to_string(),
-          "remoteAddr": remote_addr.to_string(),
-        }))
-      })
-  });
+  let op = async move {
+    let addr = resolve_addr(&args.hostname, args.port).await?;
+    let tcp_stream = TcpStream::connect(&addr).await?;
+    let local_addr = tcp_stream.local_addr()?;
+    let remote_addr = tcp_stream.peer_addr()?;
+    let mut table = state_.lock_resource_table();
+    let rid =
+      table.add("tcpStream", Box::new(StreamResource::TcpStream(tcp_stream)));
+    Ok(json!({
+      "rid": rid,
+      "localAddr": local_addr.to_string(),
+      "remoteAddr": remote_addr.to_string(),
+    }))
+  };
 
   Ok(JsonOp::Async(op.boxed()))
 }
@@ -259,7 +238,7 @@ struct ListenArgs {
 
 #[allow(dead_code)]
 struct TcpListenerResource {
-  listener: Incoming,
+  listener: TcpListener,
   waker: Option<futures::task::AtomicWaker>,
   local_addr: SocketAddr,
 }
@@ -324,11 +303,11 @@ fn op_listen(
 
   let addr =
     futures::executor::block_on(resolve_addr(&args.hostname, args.port))?;
-  let listener = TcpListener::bind(&addr)?;
+  let listener = futures::executor::block_on(TcpListener::bind(&addr))?;
   let local_addr = listener.local_addr()?;
   let local_addr_str = local_addr.to_string();
   let listener_resource = TcpListenerResource {
-    listener: listener.incoming(),
+    listener,
     waker: None,
     local_addr,
   };