refactor: new optimized op-layer using serde_v8 (#9843)

- Improves op performance.
- Handle op-metadata (errors, promise IDs) explicitly in the op-layer vs
  per op-encoding (aka: out-of-payload).
- Remove shared queue & custom "asyncHandlers", all async values are
  returned in batches via js_recv_cb.
- The op-layer should be thought of as simple function calls with little
  indirection or translation besides the conceptually straightforward
  serde_v8 bijections.
- Preserve concepts of json/bin/min as semantic groups of their
  inputs/outputs instead of their op-encoding strategy, preserving these
  groups will also facilitate partial transitions over to v8 Fast API for the
  "min" and "bin" groups

1 files changed, 117 insertions, 37 deletions

diff --git a/core/ops.rs b/core/ops.rs
index 212a713ad..3af60d072 100644
--- a/core/ops.rs
+++ b/core/ops.rs
@@ -6,10 +6,12 @@ use crate::error::AnyError;
 use crate::gotham_state::GothamState;
 use crate::resources::ResourceTable;
 use crate::runtime::GetErrorClassFn;
-use crate::BufVec;
 use crate::ZeroCopyBuf;
 use futures::Future;
 use indexmap::IndexMap;
+use rusty_v8 as v8;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
 use serde_json::json;
 use serde_json::Value;
 use std::cell::RefCell;
@@ -20,12 +22,50 @@ use std::ops::DerefMut;
 use std::pin::Pin;
 use std::rc::Rc;
 
-pub type OpAsyncFuture = Pin<Box<dyn Future<Output = Box<[u8]>>>>;
-pub type OpFn = dyn Fn(Rc<RefCell<OpState>>, BufVec) -> Op + 'static;
+pub use erased_serde::Serialize as Serializable;
+pub type PromiseId = u64;
+pub type OpAsyncFuture = Pin<Box<dyn Future<Output = OpResponse>>>;
+pub type OpFn =
+  dyn Fn(Rc<RefCell<OpState>>, OpPayload, Option<ZeroCopyBuf>) -> Op + 'static;
 pub type OpId = usize;
 
+pub struct OpPayload<'a, 'b, 'c> {
+  pub(crate) scope: Option<&'a mut v8::HandleScope<'b>>,
+  pub(crate) value: Option<v8::Local<'c, v8::Value>>,
+}
+
+impl<'a, 'b, 'c> OpPayload<'a, 'b, 'c> {
+  pub fn new(
+    scope: &'a mut v8::HandleScope<'b>,
+    value: v8::Local<'c, v8::Value>,
+  ) -> Self {
+    Self {
+      scope: Some(scope),
+      value: Some(value),
+    }
+  }
+
+  pub fn empty() -> Self {
+    Self {
+      scope: None,
+      value: None,
+    }
+  }
+
+  pub fn deserialize<T: DeserializeOwned>(self) -> Result<T, AnyError> {
+    serde_v8::from_v8(self.scope.unwrap(), self.value.unwrap())
+      .map_err(AnyError::from)
+      .map_err(|e| type_error(format!("Error parsing args: {}", e)))
+  }
+}
+
+pub enum OpResponse {
+  Value(Box<dyn Serializable>),
+  Buffer(Box<[u8]>),
+}
+
 pub enum Op {
-  Sync(Box<[u8]>),
+  Sync(OpResponse),
   Async(OpAsyncFuture),
   /// AsyncUnref is the variation of Async, which doesn't block the program
   /// exiting.
@@ -33,6 +73,32 @@ pub enum Op {
   NotFound,
 }
 
+#[derive(Serialize)]
+pub struct OpResult<R>(Option<R>, Option<OpError>);
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct OpError {
+  class_name: &'static str,
+  message: String,
+}
+
+pub fn serialize_op_result<R: Serialize + 'static>(
+  result: Result<R, AnyError>,
+  state: Rc<RefCell<OpState>>,
+) -> OpResponse {
+  OpResponse::Value(Box::new(match result {
+    Ok(v) => OpResult::<R>(Some(v), None),
+    Err(err) => OpResult::<R>(
+      None,
+      Some(OpError {
+        class_name: (state.borrow().get_error_class_fn)(&err),
+        message: err.to_string(),
+      }),
+    ),
+  }))
+}
+
 /// Maintains the resources and ops inside a JS runtime.
 pub struct OpState {
   pub resource_table: ResourceTable,
@@ -73,41 +139,43 @@ pub struct OpTable(IndexMap<String, Rc<OpFn>>);
 impl OpTable {
   pub fn register_op<F>(&mut self, name: &str, op_fn: F) -> OpId
   where
-    F: Fn(Rc<RefCell<OpState>>, BufVec) -> Op + 'static,
+    F: Fn(Rc<RefCell<OpState>>, OpPayload, Option<ZeroCopyBuf>) -> Op + 'static,
   {
     let (op_id, prev) = self.0.insert_full(name.to_owned(), Rc::new(op_fn));
     assert!(prev.is_none());
     op_id
   }
 
+  pub fn op_entries(state: Rc<RefCell<OpState>>) -> Vec<(String, OpId)> {
+    state.borrow().op_table.0.keys().cloned().zip(0..).collect()
+  }
+
   pub fn route_op(
     op_id: OpId,
     state: Rc<RefCell<OpState>>,
-    bufs: BufVec,
+    payload: OpPayload,
+    buf: Option<ZeroCopyBuf>,
   ) -> Op {
-    if op_id == 0 {
-      let ops: HashMap<String, OpId> =
-        state.borrow().op_table.0.keys().cloned().zip(0..).collect();
-      let buf = serde_json::to_vec(&ops).map(Into::into).unwrap();
-      Op::Sync(buf)
-    } else {
-      let op_fn = state
-        .borrow()
-        .op_table
-        .0
-        .get_index(op_id)
-        .map(|(_, op_fn)| op_fn.clone());
-      match op_fn {
-        Some(f) => (f)(state, bufs),
-        None => Op::NotFound,
-      }
+    let op_fn = state
+      .borrow()
+      .op_table
+      .0
+      .get_index(op_id)
+      .map(|(_, op_fn)| op_fn.clone());
+    match op_fn {
+      Some(f) => (f)(state, payload, buf),
+      None => Op::NotFound,
     }
   }
 }
 
 impl Default for OpTable {
   fn default() -> Self {
-    fn dummy(_state: Rc<RefCell<OpState>>, _bufs: BufVec) -> Op {
+    fn dummy(
+      _state: Rc<RefCell<OpState>>,
+      _p: OpPayload,
+      _b: Option<ZeroCopyBuf>,
+    ) -> Op {
       unreachable!()
     }
     Self(once(("ops".to_owned(), Rc::new(dummy) as _)).collect())
@@ -164,24 +232,36 @@ mod tests {
     let bar_id;
     {
       let op_table = &mut state.borrow_mut().op_table;
-      foo_id = op_table.register_op("foo", |_, _| Op::Sync(b"oof!"[..].into()));
+      foo_id = op_table.register_op("foo", |_, _, _| {
+        Op::Sync(OpResponse::Buffer(b"oof!"[..].into()))
+      });
       assert_eq!(foo_id, 1);
-      bar_id = op_table.register_op("bar", |_, _| Op::Sync(b"rab!"[..].into()));
+      bar_id = op_table.register_op("bar", |_, _, _| {
+        Op::Sync(OpResponse::Buffer(b"rab!"[..].into()))
+      });
       assert_eq!(bar_id, 2);
     }
 
-    let foo_res = OpTable::route_op(foo_id, state.clone(), Default::default());
-    assert!(matches!(foo_res, Op::Sync(buf) if &*buf == b"oof!"));
-    let bar_res = OpTable::route_op(bar_id, state.clone(), Default::default());
-    assert!(matches!(bar_res, Op::Sync(buf) if &*buf == b"rab!"));
-
-    let catalog_res = OpTable::route_op(0, state, Default::default());
-    let mut catalog_entries = match catalog_res {
-      Op::Sync(buf) => serde_json::from_slice::<HashMap<String, OpId>>(&buf)
-        .map(|map| map.into_iter().collect::<Vec<_>>())
-        .unwrap(),
-      _ => panic!("unexpected `Op` variant"),
-    };
+    let foo_res = OpTable::route_op(
+      foo_id,
+      state.clone(),
+      OpPayload::empty(),
+      Default::default(),
+    );
+    assert!(
+      matches!(foo_res, Op::Sync(OpResponse::Buffer(buf)) if &*buf == b"oof!")
+    );
+    let bar_res = OpTable::route_op(
+      bar_id,
+      state.clone(),
+      OpPayload::empty(),
+      Default::default(),
+    );
+    assert!(
+      matches!(bar_res, Op::Sync(OpResponse::Buffer(buf)) if &*buf == b"rab!")
+    );
+
+    let mut catalog_entries = OpTable::op_entries(state);
     catalog_entries.sort_by(|(_, id1), (_, id2)| id1.partial_cmp(id2).unwrap());
     assert_eq!(
       catalog_entries,