From f418be4f57594e88e5bcc6f384f1f6ad39f3918f Mon Sep 17 00:00:00 2001 From: Divy Srivastava Date: Wed, 18 Jan 2023 07:09:54 -0800 Subject: Optimize http_bench_json_ops example (#16505) --- core/examples/http_bench_json_ops.js | 51 ------ core/examples/http_bench_json_ops.rs | 182 --------------------- .../http_bench_json_ops/http_bench_json_ops.js | 54 ++++++ core/examples/http_bench_json_ops/main.rs | 174 ++++++++++++++++++++ 4 files changed, 228 insertions(+), 233 deletions(-) delete mode 100644 core/examples/http_bench_json_ops.js delete mode 100644 core/examples/http_bench_json_ops.rs create mode 100644 core/examples/http_bench_json_ops/http_bench_json_ops.js create mode 100644 core/examples/http_bench_json_ops/main.rs (limited to 'core/examples') diff --git a/core/examples/http_bench_json_ops.js b/core/examples/http_bench_json_ops.js deleted file mode 100644 index 28c491064..000000000 --- a/core/examples/http_bench_json_ops.js +++ /dev/null @@ -1,51 +0,0 @@ -// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. -// This is not a real HTTP server. We read blindly one time into 'requestBuf', -// then write this fixed 'responseBuf'. The point of this benchmark is to -// exercise the event loop in a simple yet semi-realistic way. -Deno.core.initializeAsyncOps(); - -const requestBuf = new Uint8Array(64 * 1024); -const responseBuf = new Uint8Array( - "HTTP/1.1 200 OK\r\nContent-Length: 12\r\n\r\nHello World\n" - .split("") - .map((c) => c.charCodeAt(0)), -); - -/** Listens on 0.0.0.0:4570, returns rid. */ -function listen() { - return Deno.core.ops.op_listen(); -} - -/** Accepts a connection, returns rid. */ -function accept(serverRid) { - return Deno.core.ops.op_accept(serverRid); -} - -async function serve(rid) { - try { - while (true) { - await Deno.core.read(rid, requestBuf); - await Deno.core.writeAll(rid, responseBuf); - } - } catch (e) { - if ( - !e.message.includes("Broken pipe") && - !e.message.includes("Connection reset by peer") - ) { - throw e; - } - } - Deno.core.close(rid); -} - -async function main() { - const listenerRid = listen(); - Deno.core.print(`http_bench_ops listening on http://127.0.0.1:4570/\n`); - - while (true) { - const rid = await accept(listenerRid); - serve(rid); - } -} - -main(); diff --git a/core/examples/http_bench_json_ops.rs b/core/examples/http_bench_json_ops.rs deleted file mode 100644 index 9a55a0823..000000000 --- a/core/examples/http_bench_json_ops.rs +++ /dev/null @@ -1,182 +0,0 @@ -// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. -use deno_core::anyhow::Error; -use deno_core::op; -use deno_core::AsyncRefCell; -use deno_core::AsyncResult; -use deno_core::CancelHandle; -use deno_core::CancelTryFuture; -use deno_core::JsRuntime; -use deno_core::OpState; -use deno_core::RcRef; -use deno_core::Resource; -use deno_core::ResourceId; -use std::cell::RefCell; -use std::env; -use std::net::SocketAddr; -use std::rc::Rc; -use tokio::io::AsyncReadExt; -use tokio::io::AsyncWriteExt; - -// This is a hack to make the `#[op]` macro work with -// deno_core examples. -// You can remove this: -use deno_core::*; - -struct Logger; - -impl log::Log for Logger { - fn enabled(&self, metadata: &log::Metadata) -> bool { - metadata.level() <= log::max_level() - } - - fn log(&self, record: &log::Record) { - if self.enabled(record.metadata()) { - println!("{} - {}", record.level(), record.args()); - } - } - - fn flush(&self) {} -} - -// Note: a `tokio::net::TcpListener` doesn't need to be wrapped in a cell, -// because it only supports one op (`accept`) which does not require a mutable -// reference to the listener. -struct TcpListener { - inner: tokio::net::TcpListener, - cancel: CancelHandle, -} - -impl TcpListener { - async fn accept(self: Rc) -> Result { - let cancel = RcRef::map(&self, |r| &r.cancel); - let stream = self.inner.accept().try_or_cancel(cancel).await?.0.into(); - Ok(stream) - } -} - -impl Resource for TcpListener { - fn close(self: Rc) { - self.cancel.cancel(); - } -} - -impl TryFrom for TcpListener { - type Error = std::io::Error; - fn try_from( - std_listener: std::net::TcpListener, - ) -> Result { - tokio::net::TcpListener::try_from(std_listener).map(|tokio_listener| Self { - inner: tokio_listener, - cancel: Default::default(), - }) - } -} - -struct TcpStream { - rd: AsyncRefCell, - wr: AsyncRefCell, - // When a `TcpStream` resource is closed, all pending 'read' ops are - // canceled, while 'write' ops are allowed to complete. Therefore only - // 'read' futures are attached to this cancel handle. - cancel: CancelHandle, -} - -impl TcpStream { - async fn read(self: Rc, data: &mut [u8]) -> Result { - let mut rd = RcRef::map(&self, |r| &r.rd).borrow_mut().await; - let cancel = RcRef::map(self, |r| &r.cancel); - let nread = rd.read(data).try_or_cancel(cancel).await?; - Ok(nread) - } - - async fn write(self: Rc, data: &[u8]) -> Result { - let mut wr = RcRef::map(self, |r| &r.wr).borrow_mut().await; - let nwritten = wr.write(data).await?; - Ok(nwritten) - } -} - -impl Resource for TcpStream { - deno_core::impl_readable_byob!(); - deno_core::impl_writable!(); - - fn close(self: Rc) { - self.cancel.cancel() - } -} - -impl From for TcpStream { - fn from(s: tokio::net::TcpStream) -> Self { - let (rd, wr) = s.into_split(); - Self { - rd: rd.into(), - wr: wr.into(), - cancel: Default::default(), - } - } -} - -fn create_js_runtime() -> JsRuntime { - let ext = deno_core::Extension::builder("my_ext") - .ops(vec![op_listen::decl(), op_accept::decl()]) - .build(); - - JsRuntime::new(deno_core::RuntimeOptions { - extensions: vec![ext], - ..Default::default() - }) -} - -#[op] -fn op_listen(state: &mut OpState) -> Result { - log::debug!("listen"); - let addr = "127.0.0.1:4570".parse::().unwrap(); - let std_listener = std::net::TcpListener::bind(addr)?; - std_listener.set_nonblocking(true)?; - let listener = TcpListener::try_from(std_listener)?; - let rid = state.resource_table.add(listener); - Ok(rid) -} - -#[op] -async fn op_accept( - state: Rc>, - rid: ResourceId, -) -> Result { - log::debug!("accept rid={}", rid); - - let listener = state.borrow().resource_table.get::(rid)?; - let stream = listener.accept().await?; - let rid = state.borrow_mut().resource_table.add(stream); - Ok(rid) -} - -fn main() { - log::set_logger(&Logger).unwrap(); - log::set_max_level( - env::args() - .find(|a| a == "-D") - .map(|_| log::LevelFilter::Debug) - .unwrap_or(log::LevelFilter::Warn), - ); - - // NOTE: `--help` arg will display V8 help and exit - deno_core::v8_set_flags(env::args().collect()); - - let mut js_runtime = create_js_runtime(); - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(); - - let future = async move { - js_runtime - .execute_script( - "http_bench_json_ops.js", - include_str!("http_bench_json_ops.js"), - ) - .unwrap(); - js_runtime.run_event_loop(false).await - }; - runtime.block_on(future).unwrap(); -} diff --git a/core/examples/http_bench_json_ops/http_bench_json_ops.js b/core/examples/http_bench_json_ops/http_bench_json_ops.js new file mode 100644 index 000000000..9650804c7 --- /dev/null +++ b/core/examples/http_bench_json_ops/http_bench_json_ops.js @@ -0,0 +1,54 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This is not a real HTTP server. We read blindly one time into 'requestBuf', +// then write this fixed 'responseBuf'. The point of this benchmark is to +// exercise the event loop in a simple yet semi-realistic way. +Deno.core.initializeAsyncOps(); + +const { ops } = Deno.core; + +const requestBuf = new Uint8Array(64 * 1024); +const responseBuf = new Uint8Array( + "HTTP/1.1 200 OK\r\nContent-Length: 12\r\n\r\nHello World\n" + .split("") + .map((c) => c.charCodeAt(0)), +); + +/** Listens on 0.0.0.0:4570, returns rid. */ +function listen() { + return ops.op_listen(); +} + +/** Accepts a connection, returns rid. */ +function accept(serverRid) { + return ops.op_accept(serverRid); +} + +function read(serverRid, buf) { + return ops.op_read_socket(serverRid, buf); +} + +async function serve(rid) { + try { + while (true) { + await read(rid, requestBuf); + if (!ops.op_try_write(rid, responseBuf)) { + await Deno.core.writeAll(rid, responseBuf); + } + } + } catch { + // pass + } + Deno.core.close(rid); +} + +async function main() { + const listenerRid = listen(); + Deno.core.print(`http_bench_ops listening on http://127.0.0.1:4570/\n`); + + while (true) { + const rid = await accept(listenerRid); + serve(rid); + } +} + +main(); diff --git a/core/examples/http_bench_json_ops/main.rs b/core/examples/http_bench_json_ops/main.rs new file mode 100644 index 000000000..f0bbec0d9 --- /dev/null +++ b/core/examples/http_bench_json_ops/main.rs @@ -0,0 +1,174 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +use deno_core::anyhow::Error; +use deno_core::op; +use deno_core::AsyncRefCell; +use deno_core::AsyncResult; +use deno_core::JsRuntime; +use deno_core::OpState; +use deno_core::Resource; +use deno_core::ResourceId; +use deno_core::ZeroCopyBuf; +use std::cell::RefCell; +use std::env; +use std::net::SocketAddr; +use std::rc::Rc; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; + +// This is a hack to make the `#[op]` macro work with +// deno_core examples. +// You can remove this: +use deno_core::*; + +// Note: a `tokio::net::TcpListener` doesn't need to be wrapped in a cell, +// because it only supports one op (`accept`) which does not require a mutable +// reference to the listener. +struct TcpListener { + inner: tokio::net::TcpListener, +} + +impl TcpListener { + async fn accept(self: Rc) -> Result { + let stream = self.inner.accept().await?.0.into(); + Ok(stream) + } +} + +impl Resource for TcpListener { + fn close(self: Rc) {} +} + +impl TryFrom for TcpListener { + type Error = std::io::Error; + fn try_from( + std_listener: std::net::TcpListener, + ) -> Result { + tokio::net::TcpListener::try_from(std_listener).map(|tokio_listener| Self { + inner: tokio_listener, + }) + } +} + +struct TcpStream { + rd: AsyncRefCell, + wr: AsyncRefCell, +} + +impl TcpStream { + async fn read(self: Rc, data: &mut [u8]) -> Result { + let mut rd = RcRef::map(&self, |r| &r.rd).borrow_mut().await; + let nread = rd.read(data).await?; + Ok(nread) + } + + async fn write(self: Rc, data: &[u8]) -> Result { + let mut wr = RcRef::map(self, |r| &r.wr).borrow_mut().await; + let nwritten = wr.write(data).await?; + Ok(nwritten) + } + + fn try_write(self: Rc, data: &[u8]) -> Result { + let wr = RcRef::map(self, |r| &r.wr) + .try_borrow_mut() + .ok_or_else(|| Error::msg("Failed to acquire lock on TcpStream"))?; + let nwritten = wr.try_write(data)?; + Ok(nwritten) + } +} + +impl Resource for TcpStream { + deno_core::impl_readable_byob!(); + deno_core::impl_writable!(); + + fn close(self: Rc) {} +} + +impl From for TcpStream { + fn from(s: tokio::net::TcpStream) -> Self { + let (rd, wr) = s.into_split(); + Self { + rd: rd.into(), + wr: wr.into(), + } + } +} + +fn create_js_runtime() -> JsRuntime { + let ext = deno_core::Extension::builder("my_ext") + .ops(vec![ + op_listen::decl(), + op_accept::decl(), + op_try_write::decl(), + op_read_socket::decl(), + ]) + .build(); + + JsRuntime::new(deno_core::RuntimeOptions { + extensions: vec![ext], + will_snapshot: false, + ..Default::default() + }) +} + +#[op] +async fn op_read_socket( + state: Rc>, + rid: ResourceId, + mut data: ZeroCopyBuf, +) -> Result { + let resource = state.borrow_mut().resource_table.get::(rid)?; + let nread = resource.read(&mut data).await?; + Ok(nread as u32) +} + +#[op] +fn op_listen(state: &mut OpState) -> Result { + let addr = "127.0.0.1:4570".parse::().unwrap(); + let std_listener = std::net::TcpListener::bind(addr)?; + std_listener.set_nonblocking(true)?; + let listener = TcpListener::try_from(std_listener)?; + let rid = state.resource_table.add(listener); + Ok(rid) +} + +#[op] +async fn op_accept( + state: Rc>, + rid: ResourceId, +) -> Result { + let listener = state.borrow().resource_table.get::(rid)?; + let stream = listener.accept().await?; + let rid = state.borrow_mut().resource_table.add(stream); + Ok(rid) +} + +#[op(fast)] +fn op_try_write( + state: &mut OpState, + rid: u32, + value: &[u8], +) -> Result { + let stream = state.resource_table.get::(rid)?; + Ok(stream.try_write(value).is_ok()) +} + +fn main() { + // NOTE: `--help` arg will display V8 help and exit + deno_core::v8_set_flags(env::args().collect()); + + let mut js_runtime = create_js_runtime(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .unwrap(); + let future = async move { + js_runtime + .execute_script( + "http_bench_json_ops.js", + include_str!("http_bench_json_ops.js"), + ) + .unwrap(); + js_runtime.run_event_loop(false).await + }; + runtime.block_on(future).unwrap(); +} -- cgit v1.2.3