1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-12-22 15:24:46 -05:00

Split core http benchmark into 'bin_ops' and 'json_ops' variants (#7147)

This commit is contained in:
Bert Belder 2020-08-21 17:14:47 +02:00
parent 999e5cf3d4
commit 29e3f4cd3a
No known key found for this signature in database
GPG key ID: 7A77887B2E2ED461
7 changed files with 459 additions and 16 deletions

View file

@ -1,5 +1,4 @@
# Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
[package]
name = "deno_core"
version = "0.54.0"
@ -20,14 +19,18 @@ lazy_static = "1.4.0"
libc = "0.2.74"
log = "0.4.11"
rusty_v8 = "0.8.1"
serde_json = { version = "1.0.57", features = [ "preserve_order" ] }
serde_json = { version = "1.0.57", features = ["preserve_order"] }
smallvec = "1.4.2"
url = "2.1.1"
[[example]]
name = "deno_core_http_bench"
path = "examples/http_bench.rs"
name = "http_bench_bin_ops"
path = "examples/http_bench_bin_ops.rs"
# These dependendencies are only used for deno_core_http_bench.
[[example]]
name = "http_bench_json_ops"
path = "examples/http_bench_json_ops.rs"
# These dependendencies are only used for the 'http_bench_*_ops' examples.
[dev-dependencies]
tokio = { version = "0.2.22", features = ["full"] }

View file

@ -18,7 +18,9 @@ function to trigger the "dispatch" callback in Rust. The user is responsible for
encoding both the request and response into a Uint8Array.
Documentation for this crate is thin at the moment. Please see
[http_bench.rs](https://github.com/denoland/deno/blob/master/core/examples/http_bench.rs)
[http_bench_bin_ops.rs](https://github.com/denoland/deno/blob/master/core/examples/http_bench_bin_ops.rs)
and
[http_bench_json_ops.rs](https://github.com/denoland/deno/blob/master/core/examples/http_bench_json_ops.rs)
as a simple example of usage.
TypeScript support and a lot of other functionality is not available at this

View file

@ -0,0 +1,145 @@
// 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.
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)),
);
const promiseMap = new Map();
let nextPromiseId = 1;
function assert(cond) {
if (!cond) {
throw Error("assert");
}
}
function createResolvable() {
let resolve;
let reject;
const promise = new Promise((res, rej) => {
resolve = res;
reject = rej;
});
promise.resolve = resolve;
promise.reject = reject;
return promise;
}
const scratch32 = new Int32Array(3);
const scratchBytes = new Uint8Array(
scratch32.buffer,
scratch32.byteOffset,
scratch32.byteLength,
);
assert(scratchBytes.byteLength === 3 * 4);
function send(promiseId, opId, rid, ...zeroCopy) {
scratch32[0] = promiseId;
scratch32[1] = rid;
scratch32[2] = -1;
return Deno.core.dispatch(opId, scratchBytes, ...zeroCopy);
}
/** Returns Promise<number> */
function sendAsync(opId, rid, ...zeroCopy) {
const promiseId = nextPromiseId++;
const p = createResolvable();
const buf = send(promiseId, opId, rid, ...zeroCopy);
if (buf) {
const record = recordFromBuf(buf);
// Sync result.
p.resolve(record.result);
} else {
// Async result.
promiseMap.set(promiseId, p);
}
return p;
}
/** Returns i32 number */
function sendSync(opId, rid) {
const buf = send(0, opId, rid);
const record = recordFromBuf(buf);
return record[2];
}
function recordFromBuf(buf) {
assert(buf.byteLength === 3 * 4);
return new Int32Array(buf.buffer, buf.byteOffset, buf.byteLength / 4);
}
function handleAsyncMsgFromRust(buf) {
const record = recordFromBuf(buf);
const p = promiseMap.get(record[0]);
promiseMap.delete(record[0]);
p.resolve(record[2]);
}
/** Listens on 0.0.0.0:4500, returns rid. */
function listen() {
return sendSync(ops["listen"], -1);
}
/** Accepts a connection, returns rid. */
function accept(rid) {
return sendAsync(ops["accept"], rid);
}
/**
* Reads a packet from the rid, presumably an http request. data is ignored.
* Returns bytes read.
*/
function read(rid, data) {
return sendAsync(ops["read"], rid, data);
}
/** Writes a fixed HTTP response to the socket rid. Returns bytes written. */
function write(rid, data) {
return sendAsync(ops["write"], rid, data);
}
function close(rid) {
return sendSync(ops["close"], rid);
}
async function serve(rid) {
while (true) {
const nread = await read(rid, requestBuf);
if (nread <= 0) {
break;
}
const nwritten = await write(rid, responseBuf);
if (nwritten < 0) {
break;
}
}
close(rid);
}
let ops;
async function main() {
ops = Deno.core.ops();
for (const opName in ops) {
Deno.core.setAsyncHandler(ops[opName], handleAsyncMsgFromRust);
}
const listenerRid = listen();
Deno.core.print(`http_bench_bin_ops listening on http://127.0.0.1:4544/\n`);
for (;;) {
const rid = await accept(listenerRid);
// Deno.core.print(`accepted ${rid}`);
if (rid < 0) {
Deno.core.print(`accept error ${rid}`);
return;
}
serve(rid);
}
}
main();

View file

@ -0,0 +1,284 @@
#[macro_use]
extern crate log;
use deno_core::CoreIsolate;
use deno_core::CoreIsolateState;
use deno_core::Op;
use deno_core::ResourceTable;
use deno_core::Script;
use deno_core::StartupData;
use deno_core::ZeroCopyBuf;
use futures::future::poll_fn;
use futures::future::FutureExt;
use futures::future::TryFuture;
use futures::future::TryFutureExt;
use std::cell::RefCell;
use std::convert::TryInto;
use std::env;
use std::fmt::Debug;
use std::io::Error;
use std::io::ErrorKind;
use std::mem::size_of;
use std::net::SocketAddr;
use std::pin::Pin;
use std::ptr;
use std::rc::Rc;
use tokio::io::AsyncRead;
use tokio::io::AsyncWrite;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
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) {}
}
#[derive(Copy, Clone, Debug, PartialEq)]
struct Record {
pub promise_id: u32,
pub rid: u32,
pub result: i32,
}
type RecordBuf = [u8; size_of::<Record>()];
impl From<&[u8]> for Record {
fn from(buf: &[u8]) -> Self {
assert_eq!(buf.len(), size_of::<RecordBuf>());
unsafe { *(buf as *const _ as *const RecordBuf) }.into()
}
}
impl From<RecordBuf> for Record {
fn from(buf: RecordBuf) -> Self {
unsafe {
#[allow(clippy::cast_ptr_alignment)]
ptr::read_unaligned(&buf as *const _ as *const Self)
}
}
}
impl From<Record> for RecordBuf {
fn from(record: Record) -> Self {
unsafe { ptr::read(&record as *const _ as *const Self) }
}
}
pub fn isolate_new() -> CoreIsolate {
let startup_data = StartupData::Script(Script {
source: include_str!("http_bench_bin_ops.js"),
filename: "http_bench_bin_ops.js",
});
let mut isolate = CoreIsolate::new(startup_data, false);
fn register_sync_op<F>(
isolate: &mut CoreIsolate,
name: &'static str,
handler: F,
) where
F: 'static
+ Fn(
Rc<RefCell<ResourceTable>>,
u32,
&mut [ZeroCopyBuf],
) -> Result<u32, Error>,
{
let core_handler = move |state: &mut CoreIsolateState,
zero_copy_bufs: &mut [ZeroCopyBuf]|
-> Op {
assert!(!zero_copy_bufs.is_empty());
let record = Record::from(zero_copy_bufs[0].as_ref());
let is_sync = record.promise_id == 0;
assert!(is_sync);
let resource_table = state.resource_table.clone();
let result: i32 =
match handler(resource_table, record.rid, &mut zero_copy_bufs[1..]) {
Ok(r) => r as i32,
Err(_) => -1,
};
let buf = RecordBuf::from(Record { result, ..record })[..].into();
Op::Sync(buf)
};
isolate.register_op(name, core_handler);
}
fn register_async_op<F>(
isolate: &mut CoreIsolate,
name: &'static str,
handler: impl Fn(Rc<RefCell<ResourceTable>>, u32, &mut [ZeroCopyBuf]) -> F
+ Copy
+ 'static,
) where
F: TryFuture,
F::Ok: TryInto<i32>,
<F::Ok as TryInto<i32>>::Error: Debug,
{
let core_handler = move |state: &mut CoreIsolateState,
zero_copy_bufs: &mut [ZeroCopyBuf]|
-> Op {
assert!(!zero_copy_bufs.is_empty());
let record = Record::from(zero_copy_bufs[0].as_ref());
let is_sync = record.promise_id == 0;
assert!(!is_sync);
let mut zero_copy = zero_copy_bufs[1..].to_vec();
let resource_table = state.resource_table.clone();
let fut = async move {
let op = handler(resource_table, record.rid, &mut zero_copy);
let result = op
.map_ok(|r| r.try_into().expect("op result does not fit in i32"))
.unwrap_or_else(|_| -1)
.await;
RecordBuf::from(Record { result, ..record })[..].into()
};
Op::Async(fut.boxed_local())
};
isolate.register_op(name, core_handler);
}
register_sync_op(&mut isolate, "listen", op_listen);
register_async_op(&mut isolate, "accept", op_accept);
register_async_op(&mut isolate, "read", op_read);
register_async_op(&mut isolate, "write", op_write);
register_sync_op(&mut isolate, "close", op_close);
isolate
}
fn op_close(
resource_table: Rc<RefCell<ResourceTable>>,
rid: u32,
_buf: &mut [ZeroCopyBuf],
) -> Result<u32, Error> {
debug!("close rid={}", rid);
let resource_table = &mut resource_table.borrow_mut();
resource_table
.close(rid)
.map(|_| 0)
.ok_or_else(bad_resource)
}
fn op_listen(
resource_table: Rc<RefCell<ResourceTable>>,
_rid: u32,
_buf: &mut [ZeroCopyBuf],
) -> Result<u32, Error> {
debug!("listen");
let addr = "127.0.0.1:4544".parse::<SocketAddr>().unwrap();
let std_listener = std::net::TcpListener::bind(&addr)?;
let listener = TcpListener::from_std(std_listener)?;
let resource_table = &mut resource_table.borrow_mut();
let rid = resource_table.add("tcpListener", Box::new(listener));
Ok(rid)
}
fn op_accept(
resource_table: Rc<RefCell<ResourceTable>>,
rid: u32,
_buf: &mut [ZeroCopyBuf],
) -> impl TryFuture<Ok = u32, Error = Error> {
debug!("accept rid={}", rid);
poll_fn(move |cx| {
let resource_table = &mut resource_table.borrow_mut();
let listener = resource_table
.get_mut::<TcpListener>(rid)
.ok_or_else(bad_resource)?;
listener.poll_accept(cx).map_ok(|(stream, _addr)| {
resource_table.add("tcpStream", Box::new(stream))
})
})
}
fn op_read(
resource_table: Rc<RefCell<ResourceTable>>,
rid: u32,
bufs: &mut [ZeroCopyBuf],
) -> impl TryFuture<Ok = usize, Error = Error> {
assert_eq!(bufs.len(), 1, "Invalid number of arguments");
let mut buf = bufs[0].clone();
debug!("read rid={}", rid);
poll_fn(move |cx| {
let resource_table = &mut resource_table.borrow_mut();
let stream = resource_table
.get_mut::<TcpStream>(rid)
.ok_or_else(bad_resource)?;
Pin::new(stream).poll_read(cx, &mut buf)
})
}
fn op_write(
resource_table: Rc<RefCell<ResourceTable>>,
rid: u32,
bufs: &mut [ZeroCopyBuf],
) -> impl TryFuture<Ok = usize, Error = Error> {
assert_eq!(bufs.len(), 1, "Invalid number of arguments");
let buf = bufs[0].clone();
debug!("write rid={}", rid);
poll_fn(move |cx| {
let resource_table = &mut resource_table.borrow_mut();
let stream = resource_table
.get_mut::<TcpStream>(rid)
.ok_or_else(bad_resource)?;
Pin::new(stream).poll_write(cx, &buf)
})
}
fn bad_resource() -> Error {
Error::new(ErrorKind::NotFound, "bad resource id")
}
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 isolate = isolate_new();
let mut runtime = tokio::runtime::Builder::new()
.basic_scheduler()
.enable_all()
.build()
.unwrap();
runtime.block_on(isolate).expect("unexpected isolate error");
}
#[test]
fn test_record_from() {
let expected = Record {
promise_id: 1,
rid: 3,
result: 4,
};
let buf = RecordBuf::from(expected);
if cfg!(target_endian = "little") {
assert_eq!(buf, [1u8, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0]);
}
let actual = Record::from(buf);
assert_eq!(actual, expected);
}

View file

@ -58,7 +58,7 @@ async function main() {
Deno.core.ops();
const listenerRid = listen();
Deno.core.print(`listening http://127.0.0.1:4544/ rid=${listenerRid}\n`);
Deno.core.print(`http_bench_json_ops listening on http://127.0.0.1:4544/\n`);
for (;;) {
const rid = await accept(listenerRid);

View file

@ -39,8 +39,8 @@ impl log::Log for Logger {
pub fn isolate_new() -> CoreIsolate {
let startup_data = StartupData::Script(Script {
source: include_str!("http_bench.js"),
filename: "http_bench.js",
source: include_str!("http_bench_json_ops.js"),
filename: "http_bench_json_ops.js",
});
let mut isolate = CoreIsolate::new(startup_data, false);

View file

@ -85,8 +85,13 @@ def deno_http_proxy(deno_exe, hyper_hello_exe):
origin_cmd=http_proxy_origin(hyper_hello_exe, origin_port))
def deno_core_http_bench(exe):
print "http_benchmark testing deno_core_http_bench"
def core_http_bin_ops(exe):
print "http_benchmark testing CORE http_bench_bin_ops"
return run([exe], 4544)
def core_http_json_ops(exe):
print "http_benchmark testing CORE http_bench_json_ops"
return run([exe], 4544)
@ -134,10 +139,12 @@ def hyper_http(hyper_hello_exe):
def http_benchmark(build_dir):
hyper_hello_exe = os.path.join(build_dir, "test_server")
deno_core_http_bench_exe = os.path.join(build_dir,
"examples/deno_core_http_bench")
deno_exe = os.path.join(build_dir, "deno")
hyper_hello_exe = os.path.join(build_dir, "test_server")
core_http_bin_ops_exe = os.path.join(build_dir,
"examples/http_bench_bin_ops")
core_http_json_ops_exe = os.path.join(build_dir,
"examples/http_bench_json_ops")
return {
# "deno_tcp" was once called "deno"
"deno_tcp": deno_tcp(deno_exe),
@ -146,8 +153,10 @@ def http_benchmark(build_dir):
# TODO(ry) deno_proxy disabled to make fetch() standards compliant.
# "deno_proxy": deno_http_proxy(deno_exe, hyper_hello_exe),
"deno_proxy_tcp": deno_tcp_proxy(deno_exe, hyper_hello_exe),
# "deno_core_http_bench" was once called "deno_core_single"
"deno_core_http_bench": deno_core_http_bench(deno_core_http_bench_exe),
# "core_http_bin_ops" was once called "deno_core_single"
# "core_http_bin_ops" was once called "deno_core_http_bench"
"core_http_bin_ops": core_http_bin_ops(core_http_bin_ops_exe),
"core_http_json_ops": core_http_json_ops(core_http_json_ops_exe),
# "node_http" was once called "node"
"node_http": node_http(),
"node_proxy": node_http_proxy(hyper_hello_exe),