feat: Minimal op registration in isolate

core/examples/http_bench.js

@@ -1,11 +1,70 @@
 // 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 OP_LISTEN = 1;
-const OP_ACCEPT = 2;
-const OP_READ = 3;
-const OP_WRITE = 4;
-const OP_CLOSE = 5;
+
+/** This structure is used to collect all ops
+ * and assign ids to them after we get them
+ * from Rust.
+ *
+ * @type {Map<string, HttpOp>}
+ */
+const opRegistry = new Map();
+
+class HttpOp {
+  constructor(name) {
+    if (typeof opRegistry.get(name) !== "undefined") {
+      throw new Error(`Duplicate op: ${name}`);
+    }
+
+    this.name = name;
+    this.opId = 0;
+    opRegistry.set(name, this);
+  }
+
+  setOpId(opId) {
+    this.opId = opId;
+  }
+
+  static handleAsyncMsgFromRust(opId, buf) {
+    const record = recordFromBuf(buf);
+    const { promiseId } = record;
+    const p = promiseMap.get(promiseId);
+    promiseMap.delete(promiseId);
+    p.resolve(record);
+  }
+
+  static sendSync(opId, arg, zeroCopy) {
+    const buf = send(0, opId, arg, zeroCopy);
+    return recordFromBuf(buf);
+  }
+
+  static sendAsync(opId, arg, zeroCopy = null) {
+    const promiseId = nextPromiseId++;
+    const p = createResolvable();
+    promiseMap.set(promiseId, p);
+    send(promiseId, opId, arg, zeroCopy);
+    return p;
+  }
+
+  /** Returns i32 number */
+  sendSync(arg, zeroCopy = null) {
+    const res = HttpOp.sendSync(this.opId, arg, zeroCopy);
+    return res.result;
+  }
+
+  /** Returns Promise<number> */
+  async sendAsync(arg, zeroCopy = null) {
+    const res = await HttpOp.sendAsync(this.opId, arg, zeroCopy);
+    return res.result;
+  }
+}
+
+const OP_LISTEN = new HttpOp("listen");
+const OP_ACCEPT = new HttpOp("accept");
+const OP_READ = new HttpOp("read");
+const OP_WRITE = new HttpOp("write");
+const OP_CLOSE = new HttpOp("close");
+
 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"
@@ -44,15 +103,6 @@ function send(promiseId, opId, arg, zeroCopy = null) {
   return Deno.core.dispatch(opId, scratchBytes, zeroCopy);
 }
 
-/** Returns Promise<number> */
-function sendAsync(opId, arg, zeroCopy = null) {
-  const promiseId = nextPromiseId++;
-  const p = createResolvable();
-  promiseMap.set(promiseId, p);
-  send(promiseId, opId, arg, zeroCopy);
-  return p;
-}
-
 function recordFromBuf(buf) {
   assert(buf.byteLength === 3 * 4);
   const buf32 = new Int32Array(buf.buffer, buf.byteOffset, buf.byteLength / 4);
@@ -63,46 +113,31 @@ function recordFromBuf(buf) {
   };
 }
 
-/** Returns i32 number */
-function sendSync(opId, arg) {
-  const buf = send(0, opId, arg);
-  const record = recordFromBuf(buf);
-  return record.result;
-}
-
-function handleAsyncMsgFromRust(opId, buf) {
-  const record = recordFromBuf(buf);
-  const { promiseId, result } = record;
-  const p = promiseMap.get(promiseId);
-  promiseMap.delete(promiseId);
-  p.resolve(result);
-}
-
 /** Listens on 0.0.0.0:4500, returns rid. */
 function listen() {
-  return sendSync(OP_LISTEN, -1);
+  return OP_LISTEN.sendSync(-1);
 }
 
 /** Accepts a connection, returns rid. */
 async function accept(rid) {
-  return await sendAsync(OP_ACCEPT, rid);
+  return await OP_ACCEPT.sendAsync(rid);
 }
 
 /**
  * Reads a packet from the rid, presumably an http request. data is ignored.
  * Returns bytes read.
  */
 async function read(rid, data) {
-  return await sendAsync(OP_READ, rid, data);
+  return await OP_READ.sendAsync(rid, data);
 }
 
 /** Writes a fixed HTTP response to the socket rid. Returns bytes written. */
 async function write(rid, data) {
-  return await sendAsync(OP_WRITE, rid, data);
+  return await OP_WRITE.sendAsync(rid, data);
 }
 
 function close(rid) {
-  return sendSync(OP_CLOSE, rid);
+  return OP_CLOSE.sendSync(rid);
 }
 
 async function serve(rid) {
@@ -121,7 +156,19 @@ async function serve(rid) {
 }
 
 async function main() {
-  Deno.core.setAsyncHandler(handleAsyncMsgFromRust);
+  Deno.core.setAsyncHandler(HttpOp.handleAsyncMsgFromRust);
+  // Initialize ops by getting their ids from Rust
+  // and assign id for each of our ops.
+  const opsMap = Deno.core.getOps();
+  for (const [name, opId] of Object.entries(opsMap)) {
+    const op = opRegistry.get(name);
+
+    if (!op) {
+      throw new Error(`Unknown op: ${name}`);
+    }
+
+    op.setOpId(opId);
+  }
 
   Deno.core.print("http_bench.js start\n");
 

core/examples/http_bench.rs

@@ -36,12 +36,6 @@ impl log::Log for Logger {
   fn flush(&self) {}
 }
 
-const OP_LISTEN: OpId = 1;
-const OP_ACCEPT: OpId = 2;
-const OP_READ: OpId = 3;
-const OP_WRITE: OpId = 4;
-const OP_CLOSE: OpId = 5;
-
 #[derive(Clone, Debug, PartialEq)]
 pub struct Record {
   pub promise_id: i32,
@@ -104,67 +98,45 @@ fn test_record_from() {
   // TODO test From<&[u8]> for Record
 }
 
-pub type HttpBenchOp = dyn Future<Item = i32, Error = std::io::Error> + Send;
-
-fn dispatch(
-  op_id: OpId,
-  control: &[u8],
-  zero_copy_buf: Option<PinnedBuf>,
-) -> CoreOp {
-  let record = Record::from(control);
-  let is_sync = record.promise_id == 0;
-  let http_bench_op = match op_id {
-    OP_LISTEN => {
-      assert!(is_sync);
-      op_listen()
-    }
-    OP_CLOSE => {
-      assert!(is_sync);
-      let rid = record.arg;
-      op_close(rid)
-    }
-    OP_ACCEPT => {
-      assert!(!is_sync);
-      let listener_rid = record.arg;
-      op_accept(listener_rid)
-    }
-    OP_READ => {
-      assert!(!is_sync);
-      let rid = record.arg;
-      op_read(rid, zero_copy_buf)
-    }
-    OP_WRITE => {
-      assert!(!is_sync);
-      let rid = record.arg;
-      op_write(rid, zero_copy_buf)
-    }
-    _ => panic!("bad op {}", op_id),
-  };
-  let mut record_a = record.clone();
-  let mut record_b = record.clone();
-
-  let fut = Box::new(
-    http_bench_op
-      .and_then(move |result| {
-        record_a.result = result;
-        Ok(record_a)
-      })
-      .or_else(|err| -> Result<Record, ()> {
-        eprintln!("unexpected err {}", err);
-        record_b.result = -1;
-        Ok(record_b)
-      })
-      .then(|result| -> Result<Buf, ()> {
-        let record = result.unwrap();
-        Ok(record.into())
-      }),
-  );
+pub type HttpOp = dyn Future<Item = i32, Error = std::io::Error> + Send;
+
+pub type HttpOpHandler =
+  fn(record: Record, zero_copy_buf: Option<PinnedBuf>) -> Box<HttpOp>;
+
+fn http_op(handler: HttpOpHandler) -> Box<CoreOpHandler> {
+  let serialized_op =
+    move |control: &[u8], zero_copy_buf: Option<PinnedBuf>| -> CoreOp {
+      let record = Record::from(control);
+      let is_sync = record.promise_id == 0;
+      let op = handler(record.clone(), zero_copy_buf);
+
+      let mut record_a = record.clone();
+      let mut record_b = record.clone();
+
+      let fut = Box::new(
+        op.and_then(move |result| {
+          record_a.result = result;
+          Ok(record_a)
+        })
+        .or_else(|err| -> Result<Record, ()> {
+          eprintln!("unexpected err {}", err);
+          record_b.result = -1;
+          Ok(record_b)
+        })
+        .then(|result| -> Result<Buf, ()> {
+          let record = result.unwrap();
+          Ok(record.into())
+        }),
+      );
+
+      if is_sync {
+        Op::Sync(fut.wait().unwrap())
+      } else {
+        Op::Async(fut)
+      }
+    };
 
-  if is_sync {
-    Op::Sync(fut.wait().unwrap())
-  } else {
-    Op::Async(fut)
-  }
+  Box::new(serialized_op)
 }
 
 fn main() {
@@ -181,7 +153,11 @@ fn main() {
     });
 
     let mut isolate = deno::Isolate::new(startup_data, false);
-    isolate.set_dispatch(dispatch);
+    isolate.register_op("listen", http_op(op_listen));
+    isolate.register_op("accept", http_op(op_accept));
+    isolate.register_op("read", http_op(op_read));
+    isolate.register_op("write", http_op(op_write));
+    isolate.register_op("close", http_op(op_close));
 
     isolate.then(|r| {
       js_check(r);
@@ -225,7 +201,8 @@ fn new_rid() -> i32 {
   rid as i32
 }
 
-fn op_accept(listener_rid: i32) -> Box<HttpBenchOp> {
+fn op_accept(record: Record, _zero_copy_buf: Option<PinnedBuf>) -> Box<HttpOp> {
+  let listener_rid = record.arg;
   debug!("accept {}", listener_rid);
   Box::new(
     futures::future::poll_fn(move || {
@@ -248,9 +225,11 @@ fn op_accept(listener_rid: i32) -> Box<HttpBenchOp> {
   )
 }
 
-fn op_listen() -> Box<HttpBenchOp> {
+fn op_listen(
+  _record: Record,
+  _zero_copy_buf: Option<PinnedBuf>,
+) -> Box<HttpOp> {
   debug!("listen");
-
   Box::new(lazy(move || {
     let addr = "127.0.0.1:4544".parse::<SocketAddr>().unwrap();
     let listener = tokio::net::TcpListener::bind(&addr).unwrap();
@@ -262,8 +241,9 @@ fn op_listen() -> Box<HttpBenchOp> {
   }))
 }
 
-fn op_close(rid: i32) -> Box<HttpBenchOp> {
+fn op_close(record: Record, _zero_copy_buf: Option<PinnedBuf>) -> Box<HttpOp> {
   debug!("close");
+  let rid = record.arg;
   Box::new(lazy(move || {
     let mut table = RESOURCE_TABLE.lock().unwrap();
     let r = table.remove(&rid);
@@ -272,7 +252,8 @@ fn op_close(rid: i32) -> Box<HttpBenchOp> {
   }))
 }
 
-fn op_read(rid: i32, zero_copy_buf: Option<PinnedBuf>) -> Box<HttpBenchOp> {
+fn op_read(record: Record, zero_copy_buf: Option<PinnedBuf>) -> Box<HttpOp> {
+  let rid = record.arg;
   debug!("read rid={}", rid);
   let mut zero_copy_buf = zero_copy_buf.unwrap();
   Box::new(
@@ -293,7 +274,8 @@ fn op_read(rid: i32, zero_copy_buf: Option<PinnedBuf>) -> Box<HttpBenchOp> {
   )
 }
 
-fn op_write(rid: i32, zero_copy_buf: Option<PinnedBuf>) -> Box<HttpBenchOp> {
+fn op_write(record: Record, zero_copy_buf: Option<PinnedBuf>) -> Box<HttpOp> {
+  let rid = record.arg;
   debug!("write rid={}", rid);
   let zero_copy_buf = zero_copy_buf.unwrap();
   Box::new(