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, 19 insertions, 26 deletions

diff --git a/core/examples/http_bench_json_ops.rs b/core/examples/http_bench_json_ops.rs
index bc96ce478..b1116d757 100644
--- a/core/examples/http_bench_json_ops.rs
+++ b/core/examples/http_bench_json_ops.rs
@@ -9,10 +9,8 @@ use deno_core::JsRuntime;
 use deno_core::OpState;
 use deno_core::RcRef;
 use deno_core::Resource;
+use deno_core::ResourceId;
 use deno_core::ZeroCopyBuf;
-use serde::Deserialize;
-use serde::Serialize;
-use serde_json::Value;
 use std::cell::RefCell;
 use std::convert::TryFrom;
 use std::env;
@@ -121,11 +119,6 @@ fn create_js_runtime() -> JsRuntime {
   runtime
 }
 
-#[derive(Deserialize, Serialize)]
-struct ResourceId {
-  rid: u32,
-}
-
 fn op_listen(
   state: &mut OpState,
   _args: (),
@@ -137,71 +130,71 @@ fn op_listen(
   std_listener.set_nonblocking(true)?;
   let listener = TcpListener::try_from(std_listener)?;
   let rid = state.resource_table.add(listener);
-  Ok(ResourceId { rid })
+  Ok(rid)
 }
 
 fn op_close(
   state: &mut OpState,
-  args: ResourceId,
+  rid: ResourceId,
   _buf: &mut [ZeroCopyBuf],
 ) -> Result<(), AnyError> {
-  log::debug!("close rid={}", args.rid);
+  log::debug!("close rid={}", rid);
   state
     .resource_table
-    .close(args.rid)
+    .close(rid)
     .map(|_| ())
     .ok_or_else(bad_resource_id)
 }
 
 async fn op_accept(
   state: Rc<RefCell<OpState>>,
-  args: ResourceId,
+  rid: ResourceId,
   _bufs: BufVec,
 ) -> Result<ResourceId, AnyError> {
-  log::debug!("accept rid={}", args.rid);
+  log::debug!("accept rid={}", rid);
 
   let listener = state
     .borrow()
     .resource_table
-    .get::<TcpListener>(args.rid)
+    .get::<TcpListener>(rid)
     .ok_or_else(bad_resource_id)?;
   let stream = listener.accept().await?;
   let rid = state.borrow_mut().resource_table.add(stream);
-  Ok(ResourceId { rid })
+  Ok(rid)
 }
 
 async fn op_read(
   state: Rc<RefCell<OpState>>,
-  args: ResourceId,
+  rid: ResourceId,
   mut bufs: BufVec,
-) -> Result<Value, AnyError> {
+) -> Result<usize, AnyError> {
   assert_eq!(bufs.len(), 1, "Invalid number of arguments");
-  log::debug!("read rid={}", args.rid);
+  log::debug!("read rid={}", rid);
 
   let stream = state
     .borrow()
     .resource_table
-    .get::<TcpStream>(args.rid)
+    .get::<TcpStream>(rid)
     .ok_or_else(bad_resource_id)?;
   let nread = stream.read(&mut bufs[0]).await?;
-  Ok(serde_json::json!({ "nread": nread }))
+  Ok(nread)
 }
 
 async fn op_write(
   state: Rc<RefCell<OpState>>,
-  args: ResourceId,
+  rid: ResourceId,
   bufs: BufVec,
-) -> Result<Value, AnyError> {
+) -> Result<usize, AnyError> {
   assert_eq!(bufs.len(), 1, "Invalid number of arguments");
-  log::debug!("write rid={}", args.rid);
+  log::debug!("write rid={}", rid);
 
   let stream = state
     .borrow()
     .resource_table
-    .get::<TcpStream>(args.rid)
+    .get::<TcpStream>(rid)
     .ok_or_else(bad_resource_id)?;
   let nwritten = stream.write(&bufs[0]).await?;
-  Ok(serde_json::json!({ "nwritten": nwritten }))
+  Ok(nwritten)
 }
 
 fn main() {