Move JSON ops to deno_core (#7336)

1 files changed, 105 insertions, 118 deletions

diff --git a/core/examples/http_bench_bin_ops.rs b/core/examples/http_bench_bin_ops.rs
index 366779e8c..f93b8b079 100644
--- a/core/examples/http_bench_bin_ops.rs
+++ b/core/examples/http_bench_bin_ops.rs
@@ -1,10 +1,12 @@
 #[macro_use]
 extern crate log;
 
+use deno_core::js_check;
+use deno_core::BasicState;
+use deno_core::BufVec;
 use deno_core::CoreIsolate;
-use deno_core::CoreIsolateState;
 use deno_core::Op;
-use deno_core::ResourceTable;
+use deno_core::OpRegistry;
 use deno_core::Script;
 use deno_core::StartupData;
 use deno_core::ZeroCopyBuf;
@@ -12,7 +14,6 @@ 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;
@@ -27,6 +28,7 @@ use tokio::io::AsyncRead;
 use tokio::io::AsyncWrite;
 use tokio::net::TcpListener;
 use tokio::net::TcpStream;
+use tokio::runtime;
 
 struct Logger;
 
@@ -46,9 +48,9 @@ impl log::Log for Logger {
 
 #[derive(Copy, Clone, Debug, PartialEq)]
 struct Record {
-  pub promise_id: u32,
-  pub rid: u32,
-  pub result: i32,
+  promise_id: u32,
+  rid: u32,
+  result: i32,
 }
 
 type RecordBuf = [u8; size_of::<Record>()];
@@ -75,131 +77,64 @@ impl From<Record> for RecordBuf {
   }
 }
 
-pub fn isolate_new() -> CoreIsolate {
+fn create_isolate() -> CoreIsolate {
+  let state = BasicState::new();
+  register_op_bin_sync(&state, "listen", op_listen);
+  register_op_bin_sync(&state, "close", op_close);
+  register_op_bin_async(&state, "accept", op_accept);
+  register_op_bin_async(&state, "read", op_read);
+  register_op_bin_async(&state, "write", op_write);
+
   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)
+  CoreIsolate::new(state, startup_data, false)
 }
 
 fn op_listen(
-  resource_table: Rc<RefCell<ResourceTable>>,
+  state: &BasicState,
   _rid: u32,
-  _buf: &mut [ZeroCopyBuf],
+  _bufs: &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));
+  let rid = state
+    .resource_table
+    .borrow_mut()
+    .add("tcpListener", Box::new(listener));
   Ok(rid)
 }
 
+fn op_close(
+  state: &BasicState,
+  rid: u32,
+  _bufs: &mut [ZeroCopyBuf],
+) -> Result<u32, Error> {
+  debug!("close rid={}", rid);
+  state
+    .resource_table
+    .borrow_mut()
+    .close(rid)
+    .map(|_| 0)
+    .ok_or_else(bad_resource_id)
+}
+
 fn op_accept(
-  resource_table: Rc<RefCell<ResourceTable>>,
+  state: Rc<BasicState>,
   rid: u32,
-  _buf: &mut [ZeroCopyBuf],
+  _bufs: BufVec,
 ) -> impl TryFuture<Ok = u32, Error = Error> {
   debug!("accept rid={}", rid);
 
   poll_fn(move |cx| {
-    let resource_table = &mut resource_table.borrow_mut();
+    let resource_table = &mut state.resource_table.borrow_mut();
     let listener = resource_table
       .get_mut::<TcpListener>(rid)
-      .ok_or_else(bad_resource)?;
+      .ok_or_else(bad_resource_id)?;
     listener.poll_accept(cx).map_ok(|(stream, _addr)| {
       resource_table.add("tcpStream", Box::new(stream))
     })
@@ -207,9 +142,9 @@ fn op_accept(
 }
 
 fn op_read(
-  resource_table: Rc<RefCell<ResourceTable>>,
+  state: Rc<BasicState>,
   rid: u32,
-  bufs: &mut [ZeroCopyBuf],
+  bufs: BufVec,
 ) -> impl TryFuture<Ok = usize, Error = Error> {
   assert_eq!(bufs.len(), 1, "Invalid number of arguments");
   let mut buf = bufs[0].clone();
@@ -217,33 +152,85 @@ fn op_read(
   debug!("read rid={}", rid);
 
   poll_fn(move |cx| {
-    let resource_table = &mut resource_table.borrow_mut();
+    let resource_table = &mut state.resource_table.borrow_mut();
     let stream = resource_table
       .get_mut::<TcpStream>(rid)
-      .ok_or_else(bad_resource)?;
+      .ok_or_else(bad_resource_id)?;
     Pin::new(stream).poll_read(cx, &mut buf)
   })
 }
 
 fn op_write(
-  resource_table: Rc<RefCell<ResourceTable>>,
+  state: Rc<BasicState>,
   rid: u32,
-  bufs: &mut [ZeroCopyBuf],
+  bufs: BufVec,
 ) -> 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 resource_table = &mut state.resource_table.borrow_mut();
     let stream = resource_table
       .get_mut::<TcpStream>(rid)
-      .ok_or_else(bad_resource)?;
+      .ok_or_else(bad_resource_id)?;
     Pin::new(stream).poll_write(cx, &buf)
   })
 }
 
-fn bad_resource() -> Error {
+fn register_op_bin_sync<F>(state: &BasicState, name: &'static str, op_fn: F)
+where
+  F: Fn(&BasicState, u32, &mut [ZeroCopyBuf]) -> Result<u32, Error> + 'static,
+{
+  let base_op_fn = move |state: Rc<BasicState>, mut bufs: BufVec| -> Op {
+    let record = Record::from(bufs[0].as_ref());
+    let is_sync = record.promise_id == 0;
+    assert!(is_sync);
+
+    let zero_copy_bufs = &mut bufs[1..];
+    let result: i32 = match op_fn(&state, record.rid, zero_copy_bufs) {
+      Ok(r) => r as i32,
+      Err(_) => -1,
+    };
+    let buf = RecordBuf::from(Record { result, ..record })[..].into();
+    Op::Sync(buf)
+  };
+
+  state.register_op(name, base_op_fn);
+}
+
+fn register_op_bin_async<F, R>(state: &BasicState, name: &'static str, op_fn: F)
+where
+  F: Fn(Rc<BasicState>, u32, BufVec) -> R + Copy + 'static,
+  R: TryFuture,
+  R::Ok: TryInto<i32>,
+  <R::Ok as TryInto<i32>>::Error: Debug,
+{
+  let base_op_fn = move |state: Rc<BasicState>, bufs: BufVec| -> Op {
+    let mut bufs_iter = bufs.into_iter();
+    let record_buf = bufs_iter.next().unwrap();
+    let zero_copy_bufs = bufs_iter.collect::<BufVec>();
+
+    let record = Record::from(record_buf.as_ref());
+    let is_sync = record.promise_id == 0;
+    assert!(!is_sync);
+
+    let fut = async move {
+      let op = op_fn(state, record.rid, zero_copy_bufs);
+      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())
+  };
+
+  state.register_op(name, base_op_fn);
+}
+
+fn bad_resource_id() -> Error {
   Error::new(ErrorKind::NotFound, "bad resource id")
 }
 
@@ -259,13 +246,13 @@ fn main() {
   // 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()
+  let isolate = create_isolate();
+  let mut runtime = runtime::Builder::new()
     .basic_scheduler()
     .enable_all()
     .build()
     .unwrap();
-  runtime.block_on(isolate).expect("unexpected isolate error");
+  js_check(runtime.block_on(isolate));
 }
 
 #[test]