7 files changed, 459 insertions, 16 deletions

diff --git a/core/Cargo.toml b/core/Cargo.toml
index 0e9f53d75..74c80872e 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -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"
+
+[[example]]
+name = "http_bench_json_ops"
+path = "examples/http_bench_json_ops.rs"
 
-# These dependendencies are only used for deno_core_http_bench.
+# These dependendencies are only used for the 'http_bench_*_ops' examples.
 [dev-dependencies]
 tokio = { version = "0.2.22", features = ["full"] }
diff --git a/core/README.md b/core/README.md
index e1a469159..94fe29de9 100644
--- a/core/README.md
+++ b/core/README.md
@@ -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
diff --git a/core/examples/http_bench_bin_ops.js b/core/examples/http_bench_bin_ops.js
new file mode 100644
index 000000000..b4d7add22
--- /dev/null
+++ b/core/examples/http_bench_bin_ops.js
@@ -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();
diff --git a/core/examples/http_bench_bin_ops.rs b/core/examples/http_bench_bin_ops.rs
new file mode 100644
index 000000000..366779e8c
--- /dev/null
+++ b/core/examples/http_bench_bin_ops.rs
@@ -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);
+}
diff --git a/core/examples/http_bench.js b/core/examples/http_bench_json_ops.js
index de6b5a9a4..865b04d7e 100644
--- a/core/examples/http_bench.js
+++ b/core/examples/http_bench_json_ops.js
@@ -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);
diff --git a/core/examples/http_bench.rs b/core/examples/http_bench_json_ops.rs
index bf9d8d137..f0fc5f94c 100644
--- a/core/examples/http_bench.rs
+++ b/core/examples/http_bench_json_ops.rs
@@ -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);
diff --git a/tools/http_benchmark.py b/tools/http_benchmark.py
index fbd5dc746..17420fe84 100755
--- a/tools/http_benchmark.py
+++ b/tools/http_benchmark.py
@@ -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),