fix: hang in Deno.serveHttp() (denoland#10923)

Waiting on next request in Deno.serveHttp() API hanged when responses were using ReadableStream. This was caused by op_http_request_next op that was never woken after response was fully written. This commit adds waker field to DenoService which is called after response is finished.
Dragon-152 · Jun 14, 2021 · 1e1959f · 1e1959f
1 parent 5814315
commit 1e1959f
Show file tree

Hide file tree

Showing 4 changed files with 94 additions and 11 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/cli/tests/unit/http_test.ts b/cli/tests/unit/http_test.ts
@@ -1,11 +1,42 @@
 // Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+import { chunkedBodyReader } from "../../../test_util/std/http/_io.ts";
+import { BufReader, BufWriter } from "../../../test_util/std/io/bufio.ts";
+import { Buffer } from "../../../test_util/std/io/buffer.ts";
+import { TextProtoReader } from "../../../test_util/std/textproto/mod.ts";
 import {
  assert,
  assertEquals,
  assertThrowsAsync,
  unitTest,
 } from "./test_util.ts";
 
+async function writeRequestAndReadResponse(conn: Deno.Conn): Promise<string> {
+ const encoder = new TextEncoder();
+ const decoder = new TextDecoder();
+
+ const w = new BufWriter(conn);
+ const r = new BufReader(conn);
+ const body = `GET / HTTP/1.1\r\nHost: 127.0.0.1:4501\r\n\r\n`;
+ const writeResult = await w.write(encoder.encode(body));
+ assertEquals(body.length, writeResult);
+ await w.flush();
+ const tpr = new TextProtoReader(r);
+ const statusLine = await tpr.readLine();
+ assert(statusLine !== null);
+ const headers = await tpr.readMIMEHeader();
+ assert(headers !== null);
+
+ const chunkedReader = chunkedBodyReader(headers, r);
+ const buf = new Uint8Array(5);
+ const dest = new Buffer();
+ let result: number | null;
+ while ((result = await chunkedReader.read(buf)) !== null) {
+ const len = Math.min(buf.byteLength, result);
+ await dest.write(buf.subarray(0, len));
+ }
+ return decoder.decode(dest.bytes());
+}
+
 unitTest({ perms: { net: true } }, async function httpServerBasic() {
  const promise = (async () => {
  const listener = Deno.listen({ port: 4501 });
@@ -373,3 +404,49 @@ unitTest(
  await delay(300);
  },
 );
+
+unitTest(
+ { perms: { net: true } },
+ // Issue: https://github.com/denoland/deno/issues/10870
+ async function httpServerHang() {
+ // Quick and dirty way to make a readable stream from a string. Alternatively,
+ // `readableStreamFromReader(file)` could be used.
+ function stream(s: string): ReadableStream<Uint8Array> {
+ return new Response(s).body!;
+ }
+
+ const httpConns: Deno.HttpConn[] = [];
+ const promise = (async () => {
+ let count = 0;
+ const listener = Deno.listen({ port: 4501 });
+ for await (const conn of listener) {
+ (async () => {
+ const httpConn = Deno.serveHttp(conn);
+ httpConns.push(httpConn);
+ for await (const { respondWith } of httpConn) {
+ respondWith(new Response(stream("hello")));
+
+ count++;
+ if (count >= 2) {
+ listener.close();
+ }
+ }
+ })();
+ }
+ })();
+
+ const clientConn = await Deno.connect({ port: 4501 });
+
+ const r1 = await writeRequestAndReadResponse(clientConn);
+ assertEquals(r1, "hello");
+
+ const r2 = await writeRequestAndReadResponse(clientConn);
+ assertEquals(r2, "hello");
+
+ clientConn.close();
+ await promise;
+ for (const conn of httpConns) {
+ conn.close();
+ }
+ },
+);
diff --git a/runtime/Cargo.toml b/runtime/Cargo.toml
@@ -55,7 +55,7 @@ dlopen = "0.1.8"
 encoding_rs = "0.8.28"
 filetime = "0.2.14"
 http = "0.2.3"
-hyper = { version = "0.14.5", features = ["server", "stream", "http1", "http2", "runtime"] }
+hyper = { version = "0.14.9", features = ["server", "stream", "http1", "http2", "runtime"] }
 indexmap = "1.6.2"
 lazy_static = "1.4.0"
 libc = "0.2.93"

diff --git a/runtime/ops/http.rs b/runtime/ops/http.rs
@@ -66,6 +66,7 @@ struct ServiceInner {
 #[derive(Clone, Default)]
 struct Service {
  inner: Rc<RefCell<Option<ServiceInner>>>,
+ waker: Rc<deno_core::futures::task::AtomicWaker>,
 }
 
 impl HyperService<Request<Body>> for Service {
@@ -160,15 +161,16 @@ async fn op_http_request_next(
  let cancel = RcRef::map(conn_resource.clone(), |r| &r.cancel);
 
  poll_fn(|cx| {
+ conn_resource.deno_service.waker.register(cx.waker());
  let connection_closed = match conn_resource.poll(cx) {
  Poll::Pending => false,
  Poll::Ready(Ok(())) => {
- // close ConnResource
- state
+ // try to close ConnResource, but don't unwrap as it might
+ // already be closed
+ let _ = state
  .borrow_mut()
  .resource_table
- .take::<ConnResource>(conn_rid)
- .unwrap();
+ .take::<ConnResource>(conn_rid);
  true
  }
  Poll::Ready(Err(e)) => {
@@ -188,7 +190,6 @@ async fn op_http_request_next(
  }
  }
  };
-
  if let Some(request_resource) =
  conn_resource.deno_service.inner.borrow_mut().take()
  {
@@ -409,6 +410,9 @@ async fn op_http_response(
  })
  .await?;
 
+ if maybe_response_body_rid.is_none() {
+ conn_resource.deno_service.waker.wake();
+ }
  Ok(maybe_response_body_rid)
 }
 
@@ -430,11 +434,13 @@ async fn op_http_response_close(
  .ok_or_else(bad_resource_id)?;
  drop(resource);
 
- poll_fn(|cx| match conn_resource.poll(cx) {
+ let r = poll_fn(|cx| match conn_resource.poll(cx) {
  Poll::Ready(x) => Poll::Ready(x),
  Poll::Pending => Poll::Ready(Ok(())),
  })
- .await
+ .await;
+ conn_resource.deno_service.waker.wake();
+ r
 }
 
 async fn op_http_request_read(