Add waker to StreamResource to fix hang on close bugs (denoland#4293)

cli/js/tests/net_test.ts

@@ -336,3 +336,46 @@ unitTest(
  conn.close();
  }
 );
+
+unitTest(
+ {
+ perms: { net: true }
+ },
+ async function netHangsOnClose() {
+ let acceptedConn: Deno.Conn;
+ const resolvable = createResolvable();
+
+ async function iteratorReq(listener: Deno.Listener): Promise<void> {
+ const p = new Uint8Array(10);
+ const conn = await listener.accept();
+ acceptedConn = conn;
+
+ try {
+ while (true) {
+ const nread = await conn.read(p);
+ if (nread === Deno.EOF) {
+ break;
+ }
+ await conn.write(new Uint8Array([1, 2, 3]));
+ }
+ } catch (err) {
+ assert(!!err);
+ assert(err instanceof Deno.errors.BadResource);
+ }
+
+ resolvable.resolve();
+ }
+
+ const addr = { hostname: "127.0.0.1", port: 4500 };
+ const listener = Deno.listen(addr);
+ iteratorReq(listener);
+ const conn = await Deno.connect(addr);
+ await conn.write(new Uint8Array([1, 2, 3, 4]));
+ const buf = new Uint8Array(10);
+ await conn.read(buf);
+ conn!.close();
+ acceptedConn!.close();
+ listener.close();
+ await resolvable;
+ }
+);

cli/ops/fetch.rs

@@ -1,6 +1,6 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 use super::dispatch_json::{Deserialize, JsonOp, Value};
-use super::io::StreamResource;
+use super::io::{StreamResource, StreamResourceHolder};
 use crate::http_util::{create_http_client, HttpBody};
 use crate::op_error::OpError;
 use crate::state::State;
@@ -80,7 +80,9 @@ pub fn op_fetch(
  let mut state = state_.borrow_mut();
  let rid = state.resource_table.add(
  "httpBody",
- Box::new(StreamResource::HttpBody(Box::new(body))),
+ Box::new(StreamResourceHolder::new(StreamResource::HttpBody(
+ Box::new(body),
+ ))),
  );
 
  let json_res = json!({

cli/ops/fs.rs

@@ -1,7 +1,7 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 // Some deserializer fields are only used on Unix and Windows build fails without it
 use super::dispatch_json::{blocking_json, Deserialize, JsonOp, Value};
-use super::io::{FileMetadata, StreamResource};
+use super::io::{FileMetadata, StreamResource, StreamResourceHolder};
 use crate::fs as deno_fs;
 use crate::op_error::OpError;
 use crate::ops::dispatch_json::JsonResult;
@@ -153,7 +153,10 @@ fn op_open(
  let mut state = state_.borrow_mut();
  let rid = state.resource_table.add(
  "fsFile",
- Box::new(StreamResource::FsFile(fs_file, FileMetadata::default())),
+ Box::new(StreamResourceHolder::new(StreamResource::FsFile(
+ fs_file,
+ FileMetadata::default(),
+ ))),
  );
  Ok(json!(rid))
  };
@@ -198,12 +201,12 @@ fn op_seek(
  };
 
  let state = state.borrow();
- let resource = state
+ let resource_holder = state
  .resource_table
- .get::<StreamResource>(rid)
+ .get::<StreamResourceHolder>(rid)
  .ok_or_else(OpError::bad_resource_id)?;
 
- let tokio_file = match resource {
+ let tokio_file = match resource_holder.resource {
  StreamResource::FsFile(ref file, _) => file,
  _ => return Err(OpError::bad_resource_id()),
  };

cli/ops/io.rs

@@ -7,7 +7,9 @@ use deno_core::*;
 use futures::future::poll_fn;
 use futures::future::FutureExt;
 use futures::ready;
+use std::collections::HashMap;
 use std::pin::Pin;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::task::Context;
 use std::task::Poll;
 use tokio::io::{AsyncRead, AsyncWrite};
@@ -56,15 +58,23 @@ pub fn init(i: &mut Isolate, s: &State) {
  );
 }
 
-pub fn get_stdio() -> (StreamResource, StreamResource, StreamResource) {
- let stdin = StreamResource::Stdin(tokio::io::stdin(), TTYMetadata::default());
- let stdout = StreamResource::Stdout({
+pub fn get_stdio() -> (
+ StreamResourceHolder,
+ StreamResourceHolder,
+ StreamResourceHolder,
+) {
+ let stdin = StreamResourceHolder::new(StreamResource::Stdin(
+ tokio::io::stdin(),
+ TTYMetadata::default(),
+ ));
+ let stdout = StreamResourceHolder::new(StreamResource::Stdout({
  let stdout = STDOUT_HANDLE
  .try_clone()
  .expect("Unable to clone stdout handle");
  tokio::fs::File::from_std(stdout)
- });
- let stderr = StreamResource::Stderr(tokio::io::stderr());
+ }));
+ let stderr =
+ StreamResourceHolder::new(StreamResource::Stderr(tokio::io::stderr()));
 
  (stdin, stdout, stderr)
 }
@@ -87,6 +97,51 @@ pub struct FileMetadata {
  pub tty: TTYMetadata,
 }
 
+pub struct StreamResourceHolder {
+ pub resource: StreamResource,
+ waker: HashMap<usize, futures::task::AtomicWaker>,
+ waker_counter: AtomicUsize,
+}
+
+impl StreamResourceHolder {
+ pub fn new(resource: StreamResource) -> StreamResourceHolder {
+ StreamResourceHolder {
+ resource,
+ // Atleast one task is expecter for the resource
+ waker: HashMap::with_capacity(1),
+ // Tracks wakers Ids
+ waker_counter: AtomicUsize::new(0),
+ }
+ }
+}
+
+impl Drop for StreamResourceHolder {
+ fn drop(&mut self) {
+ self.wake_tasks();
+ }
+}
+
+impl StreamResourceHolder {
+ pub fn track_task(&mut self, cx: &Context) -> Result<usize, OpError> {
+ let waker = futures::task::AtomicWaker::new();
+ waker.register(cx.waker());
+ // Its OK if it overflows
+ let task_waker_id = self.waker_counter.fetch_add(1, Ordering::Relaxed);
+ self.waker.insert(task_waker_id, waker);
+ Ok(task_waker_id)
+ }
+
+ pub fn wake_tasks(&mut self) {
+ for waker in self.waker.values() {
+ waker.wake();
+ }
+ }
+
+ pub fn untrack_task(&mut self, task_waker_id: usize) {
+ self.waker.remove(&task_waker_id);
+ }
+}
+
 pub enum StreamResource {
  Stdin(tokio::io::Stdin, TTYMetadata),
  Stdout(tokio::fs::File),
@@ -150,10 +205,27 @@ pub fn op_read(
 
  poll_fn(move |cx| {
  let resource_table = &mut state.borrow_mut().resource_table;
- let resource = resource_table
- .get_mut::<StreamResource>(rid as u32)
+ let resource_holder = resource_table
+ .get_mut::<StreamResourceHolder>(rid as u32)
  .ok_or_else(OpError::bad_resource_id)?;
- let nread = ready!(resource.poll_read(cx, &mut buf.as_mut()[..]))?;
+
+ let mut task_tracker_id: Option<usize> = None;
+ let nread = match resource_holder
+ .resource
+ .poll_read(cx, &mut buf.as_mut()[..])
+ .map_err(OpError::from)
+ {
+ Poll::Ready(t) => {
+ if let Some(id) = task_tracker_id {
+ resource_holder.untrack_task(id);
+ }
+ t
+ }
+ Poll::Pending => {
+ task_tracker_id.replace(resource_holder.track_task(cx)?);
+ return Poll::Pending;
+ }
+ }?;
  Poll::Ready(Ok(nread as i32))
  })
  .boxed_local()
@@ -233,10 +305,10 @@ pub fn op_write(
  async move {
  let nwritten = poll_fn(|cx| {
  let resource_table = &mut state.borrow_mut().resource_table;
- let resource = resource_table
- .get_mut::<StreamResource>(rid as u32)
+ let resource_holder = resource_table
+ .get_mut::<StreamResourceHolder>(rid as u32)
  .ok_or_else(OpError::bad_resource_id)?;
- resource.poll_write(cx, &buf.as_ref()[..])
+ resource_holder.resource.poll_write(cx, &buf.as_ref()[..])
  })
  .await?;
 
@@ -246,10 +318,10 @@ pub fn op_write(
  // https://github.com/denoland/deno/issues/3565
  poll_fn(|cx| {
  let resource_table = &mut state.borrow_mut().resource_table;
- let resource = resource_table
- .get_mut::<StreamResource>(rid as u32)
+ let resource_holder = resource_table
+ .get_mut::<StreamResourceHolder>(rid as u32)
  .ok_or_else(OpError::bad_resource_id)?;
- resource.poll_flush(cx)
+ resource_holder.resource.poll_flush(cx)
  })
  .await?;
 

cli/ops/net.rs

@@ -1,6 +1,6 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 use super::dispatch_json::{Deserialize, JsonOp, Value};
-use super::io::StreamResource;
+use super::io::{StreamResource, StreamResourceHolder};
 use crate::op_error::OpError;
 use crate::resolve_addr::resolve_addr;
 use crate::state::State;
@@ -78,9 +78,12 @@ fn op_accept(
  let local_addr = tcp_stream.local_addr()?;
  let remote_addr = tcp_stream.peer_addr()?;
  let mut state = state_.borrow_mut();
- let rid = state
- .resource_table
- .add("tcpStream", Box::new(StreamResource::TcpStream(tcp_stream)));
+ let rid = state.resource_table.add(
+ "tcpStream",
+ Box::new(StreamResourceHolder::new(StreamResource::TcpStream(
+ tcp_stream,
+ ))),
+ );
  Ok(json!({
  "rid": rid,
  "localAddr": {
@@ -207,9 +210,12 @@ fn op_connect(
  let local_addr = tcp_stream.local_addr()?;
  let remote_addr = tcp_stream.peer_addr()?;
  let mut state = state_.borrow_mut();
- let rid = state
- .resource_table
- .add("tcpStream", Box::new(StreamResource::TcpStream(tcp_stream)));
+ let rid = state.resource_table.add(
+ "tcpStream",
+ Box::new(StreamResourceHolder::new(StreamResource::TcpStream(
+ tcp_stream,
+ ))),
+ );
  Ok(json!({
  "rid": rid,
  "localAddr": {
@@ -251,11 +257,11 @@ fn op_shutdown(
  };
 
  let mut state = state.borrow_mut();
- let resource = state
+ let resource_holder = state
  .resource_table
- .get_mut::<StreamResource>(rid)
+ .get_mut::<StreamResourceHolder>(rid)
  .ok_or_else(OpError::bad_resource_id)?;
- match resource {
+ match resource_holder.resource {
  StreamResource::TcpStream(ref mut stream) => {
  TcpStream::shutdown(stream, shutdown_mode).map_err(OpError::from)?;
  }

cli/ops/process.rs

@@ -1,6 +1,6 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 use super::dispatch_json::{Deserialize, JsonOp, Value};
-use super::io::StreamResource;
+use super::io::{StreamResource, StreamResourceHolder};
 use crate::op_error::OpError;
 use crate::signal::kill;
 use crate::state::State;
@@ -24,11 +24,11 @@ pub fn init(i: &mut Isolate, s: &State) {
 
 fn clone_file(rid: u32, state: &State) -> Result<std::fs::File, OpError> {
  let mut state = state.borrow_mut();
- let repr = state
+ let repr_holder = state
  .resource_table
- .get_mut::<StreamResource>(rid)
+ .get_mut::<StreamResourceHolder>(rid)
  .ok_or_else(OpError::bad_resource_id)?;
- let file = match repr {
+ let file = match repr_holder.resource {
  StreamResource::FsFile(ref mut file, _) => file,
  _ => return Err(OpError::bad_resource_id()),
  };
@@ -127,7 +127,9 @@ fn op_run(
  Some(child_stdin) => {
  let rid = table.add(
  "childStdin",
- Box::new(StreamResource::ChildStdin(child_stdin)),
+ Box::new(StreamResourceHolder::new(StreamResource::ChildStdin(
+ child_stdin,
+ ))),
  );
  Some(rid)
  }
@@ -138,7 +140,9 @@ fn op_run(
  Some(child_stdout) => {
  let rid = table.add(
  "childStdout",
- Box::new(StreamResource::ChildStdout(child_stdout)),
+ Box::new(StreamResourceHolder::new(StreamResource::ChildStdout(
+ child_stdout,
+ ))),
  );
  Some(rid)
  }
@@ -149,7 +153,9 @@ fn op_run(
  Some(child_stderr) => {
  let rid = table.add(
  "childStderr",
- Box::new(StreamResource::ChildStderr(child_stderr)),
+ Box::new(StreamResourceHolder::new(StreamResource::ChildStderr(
+ child_stderr,
+ ))),
  );
  Some(rid)
  }

cli/ops/tls.rs

@@ -1,6 +1,6 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 use super::dispatch_json::{Deserialize, JsonOp, Value};
-use super::io::StreamResource;
+use super::io::{StreamResource, StreamResourceHolder};
 use crate::op_error::OpError;
 use crate::resolve_addr::resolve_addr;
 use crate::state::State;
@@ -85,7 +85,9 @@ pub fn op_connect_tls(
  let mut state = state_.borrow_mut();
  let rid = state.resource_table.add(
  "clientTlsStream",
- Box::new(StreamResource::ClientTlsStream(Box::new(tls_stream))),
+ Box::new(StreamResourceHolder::new(StreamResource::ClientTlsStream(
+ Box::new(tls_stream),
+ ))),
  );
  Ok(json!({
  "rid": rid,
@@ -318,7 +320,9 @@ fn op_accept_tls(
  let mut state = state.borrow_mut();
  state.resource_table.add(
  "serverTlsStream",
- Box::new(StreamResource::ServerTlsStream(Box::new(tls_stream))),
+ Box::new(StreamResourceHolder::new(StreamResource::ServerTlsStream(
+ Box::new(tls_stream),
+ ))),
  )
  };
  Ok(json!({