foster/denoland-deno
1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-11-24 15:19:26 -05:00
Code Issues Packages 1 Wiki Activity

refactor(ext/http): rewrite hyper integration and fix bug (#12332)

Browse source 
Fixes: #12193
This commit is contained in:
Bert Belder 2021-10-04 18:50:40 -07:00
parent f1b1a3f389
commit 5b1e537446
No known key found for this signature in database
GPG key ID: 7A77887B2E2ED461
6 changed files with 650 additions and 648 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
cli/tests/unit/http_test.ts
View file

@ -7,6 +7,7 @@ import {
assert,
assertEquals,
assertRejects,
assertStrictEquals,
assertThrows,
deferred,
delay,
@ -386,7 +387,7 @@ unitTest(
Deno.errors.Http,
"connection closed",
);
// The error from `op_http_request_next` reroutes to `respondWith()`.
// The error from `op_http_accept` reroutes to `respondWith()`.
assertEquals(await nextRequestPromise, null);
listener.close();
})();
@ -865,6 +866,7 @@ unitTest(
const writer = writable.getWriter();
async function writeResponse() {
await delay(50);
await writer.write(
new TextEncoder().encode(
"written to the writable side of a TransformStream",
@ -999,3 +1001,39 @@ unitTest(
await Promise.all([server(), client()]);
},
);
// https://github.com/denoland/deno/pull/12332
unitTest(
{ permissions: { net: true } },
async function httpConnConcurrentNextRequestCalls() {
const hostname = "localhost";
const port = 4501;
async function server() {
const listener = Deno.listen({ hostname, port });
const tcpConn = await listener.accept();
const httpConn = Deno.serveHttp(tcpConn);
const promises = new Array(10).fill(null).map(async (_, i) => {
const event = await httpConn.nextRequest();
assert(event);
const { pathname } = new URL(event.request.url);
assertStrictEquals(pathname, `/${i}`);
const response = new Response(`Response #${i}`);
await event.respondWith(response);
});
await Promise.all(promises);
httpConn.close();
listener.close();
}
async function client() {
for (let i = 0; i < 10; i++) {
const response = await fetch(`http://${hostname}:${port}/${i}`);
const body = await response.text();
assertStrictEquals(body, `Response #${i}`);
}
}
await Promise.all([server(), delay(100).then(client)]);
},
);

277
ext/http/01_http.js
View file

@ -27,6 +27,7 @@
Set,
SetPrototypeAdd,
SetPrototypeDelete,
SetPrototypeHas,
SetPrototypeValues,
StringPrototypeIncludes,
StringPrototypeToLowerCase,
@ -42,6 +43,8 @@
class HttpConn {
#rid = 0;
#closed = false;
// This set holds resource ids of resources
// that were created during lifecycle of this request.
// When the connection is closed these resources should be closed
@ -62,10 +65,11 @@
let nextRequest;
try {
nextRequest = await core.opAsync(
"op_http_request_next",
"op_http_accept",
this.#rid,
);
} catch (error) {
this.close();
// A connection error seen here would cause disrupted responses to throw
// a generic `BadResource` error. Instead store this error and replace
// those with it.
@ -79,26 +83,26 @@
}
throw error;
}
if (nextRequest === null) return null;
if (nextRequest == null) {
this.close();
return null;
}
const [
requestRid,
responseSenderRid,
streamRid,
method,
headersList,
url,
] = nextRequest;
SetPrototypeAdd(this.managedResources, streamRid);
/** @type {ReadableStream<Uint8Array> | undefined} */
let body = null;
if (typeof requestRid === "number") {
SetPrototypeAdd(this.managedResources, requestRid);
// There might be a body, but we don't expose it for GET/HEAD requests.
// It will be closed automatically once the request has been handled and
// the response has been sent.
if (method !== "GET" && method !== "HEAD") {
body = createRequestBodyStream(this, requestRid);
}
// There might be a body, but we don't expose it for GET/HEAD requests.
// It will be closed automatically once the request has been handled and
// the response has been sent.
if (method !== "GET" && method !== "HEAD") {
body = createRequestBodyStream(streamRid);
}
const innerRequest = newInnerRequest(
@ -111,22 +115,21 @@
const signal = abortSignal.newSignal();
const request = fromInnerRequest(innerRequest, signal, "immutable");
SetPrototypeAdd(this.managedResources, responseSenderRid);
const respondWith = createRespondWith(
this,
responseSenderRid,
requestRid,
);
const respondWith = createRespondWith(this, streamRid);
return { request, respondWith };
}
/** @returns {void} */
close() {
for (const rid of SetPrototypeValues(this.managedResources)) {
core.tryClose(rid);
if (!this.#closed) {
this.#closed = true;
core.close(this.#rid);
for (const rid of SetPrototypeValues(this.managedResources)) {
SetPrototypeDelete(this.managedResources, rid);
core.close(rid);
}
}
core.close(this.#rid);
}
[SymbolAsyncIterator]() {
@ -136,97 +139,86 @@
async next() {
const reqEvt = await httpConn.nextRequest();
// Change with caution, current form avoids a v8 deopt
return { value: reqEvt, done: reqEvt === null };
return { value: reqEvt ?? undefined, done: reqEvt === null };
},
};
}
}
function readRequest(requestRid, zeroCopyBuf) {
return core.opAsync(
"op_http_request_read",
requestRid,
zeroCopyBuf,
);
function readRequest(streamRid, buf) {
return core.opAsync("op_http_read", streamRid, buf);
}
function createRespondWith(httpConn, responseSenderRid, requestRid) {
function createRespondWith(httpConn, streamRid) {
return async function respondWith(resp) {
if (resp instanceof Promise) {
resp = await resp;
}
try {
if (resp instanceof Promise) {
resp = await resp;
}
if (!(resp instanceof Response)) {
throw new TypeError(
"First argument to respondWith must be a Response or a promise resolving to a Response.",
);
}
if (!(resp instanceof Response)) {
throw new TypeError(
"First argument to respondWith must be a Response or a promise resolving to a Response.",
);
}
const innerResp = toInnerResponse(resp);
const innerResp = toInnerResponse(resp);
// If response body length is known, it will be sent synchronously in a
// single op, in other case a "response body" resource will be created and
// we'll be streaming it.
/** @type {ReadableStream<Uint8Array> | Uint8Array | null} */
let respBody = null;
if (innerResp.body !== null) {
if (innerResp.body.unusable()) throw new TypeError("Body is unusable.");
if (innerResp.body.streamOrStatic instanceof ReadableStream) {
if (
innerResp.body.length === null ||
innerResp.body.source instanceof Blob
) {
respBody = innerResp.body.stream;
} else {
const reader = innerResp.body.stream.getReader();
const r1 = await reader.read();
if (r1.done) {
respBody = new Uint8Array(0);
// If response body length is known, it will be sent synchronously in a
// single op, in other case a "response body" resource will be created and
// we'll be streaming it.
/** @type {ReadableStream<Uint8Array> | Uint8Array | null} */
let respBody = null;
if (innerResp.body !== null) {
if (innerResp.body.unusable()) {
throw new TypeError("Body is unusable.");
}
if (innerResp.body.streamOrStatic instanceof ReadableStream) {
if (
innerResp.body.length === null ||
innerResp.body.source instanceof Blob
) {
respBody = innerResp.body.stream;
} else {
respBody = r1.value;
const r2 = await reader.read();
if (!r2.done) throw new TypeError("Unreachable");
const reader = innerResp.body.stream.getReader();
const r1 = await reader.read();
if (r1.done) {
respBody = new Uint8Array(0);
} else {
respBody = r1.value;
const r2 = await reader.read();
if (!r2.done) throw new TypeError("Unreachable");
}
}
} else {
innerResp.body.streamOrStatic.consumed = true;
respBody = innerResp.body.streamOrStatic.body;
}
} else {
innerResp.body.streamOrStatic.consumed = true;
respBody = innerResp.body.streamOrStatic.body;
respBody = new Uint8Array(0);
}
} else {
respBody = new Uint8Array(0);
}
const isStreamingResponseBody =
!(typeof respBody === "string" || respBody instanceof Uint8Array);
SetPrototypeDelete(httpConn.managedResources, responseSenderRid);
let responseBodyRid;
try {
responseBodyRid = await core.opAsync(
"op_http_response",
[
responseSenderRid,
try {
await core.opAsync("op_http_write_headers", [
streamRid,
innerResp.status ?? 200,
innerResp.headerList,
],
(respBody instanceof Uint8Array || typeof respBody === "string")
? respBody
: null,
);
} catch (error) {
const connError = httpConn[connErrorSymbol];
if (error instanceof BadResource && connError != null) {
// deno-lint-ignore no-ex-assign
error = new connError.constructor(connError.message);
], isStreamingResponseBody ? null : respBody);
} catch (error) {
const connError = httpConn[connErrorSymbol];
if (error instanceof BadResource && connError != null) {
// deno-lint-ignore no-ex-assign
error = new connError.constructor(connError.message);
}
if (respBody !== null && respBody instanceof ReadableStream) {
await respBody.cancel(error);
}
throw error;
}
if (respBody !== null && respBody instanceof ReadableStream) {
await respBody.cancel(error);
}
throw error;
}
// If `respond` returns a responseBodyRid, we should stream the body
// to that resource.
if (responseBodyRid !== null) {
SetPrototypeAdd(httpConn.managedResources, responseBodyRid);
try {
if (isStreamingResponseBody) {
if (respBody === null || !(respBody instanceof ReadableStream)) {
throw new TypeError("Unreachable");
}
@ -239,11 +231,7 @@
break;
}
try {
await core.opAsync(
"op_http_response_write",
responseBodyRid,
value,
);
await core.opAsync("op_http_write", streamRid, value);
} catch (error) {
const connError = httpConn[connErrorSymbol];
if (error instanceof BadResource && connError != null) {
@ -254,61 +242,55 @@
throw error;
}
}
} finally {
// Once all chunks are sent, and the request body is closed, we can
// close the response body.
SetPrototypeDelete(httpConn.managedResources, responseBodyRid);
try {
await core.opAsync("op_http_response_close", responseBodyRid);
} catch { /* pass */ }
await core.opAsync("op_http_shutdown", streamRid);
} catch (error) {
await reader.cancel(error);
throw error;
}
}
}
const ws = resp[_ws];
if (ws) {
if (typeof requestRid !== "number") {
throw new TypeError(
"This request can not be upgraded to a websocket connection.",
const ws = resp[_ws];
if (ws) {
const wsRid = await core.opAsync(
"op_http_upgrade_websocket",
streamRid,
);
ws[_rid] = wsRid;
ws[_protocol] = resp.headers.get("sec-websocket-protocol");
httpConn.close();
if (ws[_readyState] === WebSocket.CLOSING) {
await core.opAsync("op_ws_close", { rid: wsRid });
ws[_readyState] = WebSocket.CLOSED;
const errEvent = new ErrorEvent("error");
ws.dispatchEvent(errEvent);
const event = new CloseEvent("close");
ws.dispatchEvent(event);
core.tryClose(wsRid);
} else {
ws[_readyState] = WebSocket.OPEN;
const event = new Event("open");
ws.dispatchEvent(event);
ws[_eventLoop]();
}
}
const wsRid = await core.opAsync(
"op_http_upgrade_websocket",
requestRid,
);
ws[_rid] = wsRid;
ws[_protocol] = resp.headers.get("sec-websocket-protocol");
if (ws[_readyState] === WebSocket.CLOSING) {
await core.opAsync("op_ws_close", { rid: wsRid });
ws[_readyState] = WebSocket.CLOSED;
const errEvent = new ErrorEvent("error");
ws.dispatchEvent(errEvent);
const event = new CloseEvent("close");
ws.dispatchEvent(event);
core.tryClose(wsRid);
} else {
ws[_readyState] = WebSocket.OPEN;
const event = new Event("open");
ws.dispatchEvent(event);
ws[_eventLoop]();
} finally {
if (SetPrototypeHas(httpConn.managedResources, streamRid)) {
SetPrototypeDelete(httpConn.managedResources, streamRid);
core.close(streamRid);
}
} else if (typeof requestRid === "number") {
// Try to close "request" resource. It might have been already consumed,
// but if it hasn't been we need to close it here to avoid resource
// leak.
SetPrototypeDelete(httpConn.managedResources, requestRid);
core.tryClose(requestRid);
}
};
}
function createRequestBodyStream(httpConn, requestRid) {
function createRequestBodyStream(streamRid) {
return new ReadableStream({
type: "bytes",
async pull(controller) {
@ -316,32 +298,21 @@
// This is the largest possible size for a single packet on a TLS
// stream.
const chunk = new Uint8Array(16 * 1024 + 256);
const read = await readRequest(
requestRid,
chunk,
);
const read = await readRequest(streamRid, chunk);
if (read > 0) {
// We read some data. Enqueue it onto the stream.
controller.enqueue(TypedArrayPrototypeSubarray(chunk, 0, read));
} else {
// We have reached the end of the body, so we close the stream.
controller.close();
SetPrototypeDelete(httpConn.managedResources, requestRid);
core.close(requestRid);
}
} catch (err) {
// There was an error while reading a chunk of the body, so we
// error.
controller.error(err);
controller.close();
SetPrototypeDelete(httpConn.managedResources, requestRid);
core.close(requestRid);
}
},
cancel() {
SetPrototypeDelete(httpConn.managedResources, requestRid);
core.close(requestRid);
},
});
}

945
ext/http/lib.rs
View file

File diff suppressed because it is too large Load diff

26
ext/websocket/lib.rs
View file

@ -34,12 +34,13 @@ use std::sync::Arc;
use tokio::net::TcpStream;
use tokio_rustls::rustls::RootCertStore;
use tokio_rustls::TlsConnector;
use tokio_tungstenite::client_async;
use tokio_tungstenite::tungstenite::{
handshake::client::Response, protocol::frame::coding::CloseCode,
protocol::CloseFrame, Message,
protocol::CloseFrame, protocol::Role, Message,
};
use tokio_tungstenite::MaybeTlsStream;
use tokio_tungstenite::{client_async, WebSocketStream};
use tokio_tungstenite::WebSocketStream;
pub use tokio_tungstenite; // Re-export tokio_tungstenite
@ -72,6 +73,27 @@ pub enum WebSocketStreamType {
},
}
pub async fn ws_create_server_stream(
state: &Rc<RefCell<OpState>>,
transport: hyper::upgrade::Upgraded,
) -> Result<ResourceId, AnyError> {
let ws_stream =
WebSocketStream::from_raw_socket(transport, Role::Server, None).await;
let (ws_tx, ws_rx) = ws_stream.split();
let ws_resource = WsStreamResource {
stream: WebSocketStreamType::Server {
tx: AsyncRefCell::new(ws_tx),
rx: AsyncRefCell::new(ws_rx),
},
cancel: Default::default(),
};
let resource_table = &mut state.borrow_mut().resource_table;
let rid = resource_table.add(ws_resource);
Ok(rid)
}
pub struct WsStreamResource {
pub stream: WebSocketStreamType,
// When a `WsStreamResource` resource is closed, all pending 'read' ops are

5
runtime/errors.rs
View file

@ -17,6 +17,7 @@ use deno_fetch::reqwest;
use std::env;
use std::error::Error;
use std::io;
use std::sync::Arc;
fn get_dlopen_error_class(error: &dlopen::Error) -> &'static str {
use dlopen::Error::*;
@ -163,6 +164,10 @@ pub fn get_error_class_name(e: &AnyError) -> Option<&'static str> {
.map(get_dlopen_error_class)
})
.or_else(|| e.downcast_ref::<hyper::Error>().map(get_hyper_error_class))
.or_else(|| {
e.downcast_ref::<Arc<hyper::Error>>()
.map(|e| get_hyper_error_class(&**e))
})
.or_else(|| {
e.downcast_ref::<deno_core::Canceled>().map(|e| {
let io_err: io::Error = e.to_owned().into();

5
runtime/ops/http.rs
View file

@ -6,6 +6,7 @@ use deno_core::op_sync;
use deno_core::Extension;
use deno_core::OpState;
use deno_core::ResourceId;
use deno_http::http_create_conn_resource;
use deno_net::io::TcpStreamResource;
use deno_net::ops_tls::TlsStreamResource;
@ -29,7 +30,7 @@ fn op_http_start(
let (read_half, write_half) = resource.into_inner();
let tcp_stream = read_half.reunite(write_half)?;
let addr = tcp_stream.local_addr()?;
return deno_http::start_http(state, tcp_stream, addr, "http");
return http_create_conn_resource(state, tcp_stream, addr, "http");
}
if let Ok(resource_rc) = state
@ -41,7 +42,7 @@ fn op_http_start(
let (read_half, write_half) = resource.into_inner();
let tls_stream = read_half.reunite(write_half);
let addr = tls_stream.get_ref().0.local_addr()?;
return deno_http::start_http(state, tls_stream, addr, "https");
return http_create_conn_resource(state, tls_stream, addr, "https");
}
Err(bad_resource_id())