Optimize read and write ops (#2259)

cli/BUILD.gn

@@ -64,6 +64,7 @@ ts_sources = [
   "../js/deno.ts",
   "../js/dir.ts",
   "../js/dispatch.ts",
+  "../js/dispatch_minimal.ts",
   "../js/dom_types.ts",
   "../js/errors.ts",
   "../js/event.ts",

cli/dispatch_minimal.rs

@@ -0,0 +1,159 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+// Do not add flatbuffer dependencies to this module.
+//! Connects to js/dispatch_minimal.ts sendAsyncMinimal This acts as a faster
+//! alternative to flatbuffers using a very simple list of int32s to lay out
+//! messages. The first i32 is used to determine if a message a flatbuffer
+//! message or a "minimal" message.
+use crate::state::ThreadSafeState;
+use deno::Buf;
+use deno::Op;
+use deno::PinnedBuf;
+use futures::Future;
+
+const DISPATCH_MINIMAL_TOKEN: i32 = 0xCAFE;
+const OP_READ: i32 = 1;
+const OP_WRITE: i32 = 2;
+
+#[derive(Copy, Clone, Debug, PartialEq)]
+// This corresponds to RecordMinimal on the TS side.
+pub struct Record {
+  pub promise_id: i32,
+  pub op_id: i32,
+  pub arg: i32,
+  pub result: i32,
+}
+
+impl Into<Buf> for Record {
+  fn into(self) -> Buf {
+    let vec = vec![
+      DISPATCH_MINIMAL_TOKEN,
+      self.promise_id,
+      self.op_id,
+      self.arg,
+      self.result,
+    ];
+    let buf32 = vec.into_boxed_slice();
+    let ptr = Box::into_raw(buf32) as *mut [u8; 5 * 4];
+    unsafe { Box::from_raw(ptr) }
+  }
+}
+
+pub fn parse_min_record(bytes: &[u8]) -> Option<Record> {
+  if bytes.len() % std::mem::size_of::<i32>() != 0 {
+    return None;
+  }
+  let p = bytes.as_ptr();
+  #[allow(clippy::cast_ptr_alignment)]
+  let p32 = p as *const i32;
+  let s = unsafe { std::slice::from_raw_parts(p32, bytes.len() / 4) };
+
+  if s.len() < 5 {
+    return None;
+  }
+  let ptr = s.as_ptr();
+  let ints = unsafe { std::slice::from_raw_parts(ptr, 5) };
+  if ints[0] != DISPATCH_MINIMAL_TOKEN {
+    return None;
+  }
+  Some(Record {
+    promise_id: ints[1],
+    op_id: ints[2],
+    arg: ints[3],
+    result: ints[4],
+  })
+}
+
+#[test]
+fn test_parse_min_record() {
+  let buf = vec![
+    0xFE, 0xCA, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0,
+  ];
+  assert_eq!(
+    parse_min_record(&buf),
+    Some(Record {
+      promise_id: 1,
+      op_id: 2,
+      arg: 3,
+      result: 4,
+    })
+  );
+
+  let buf = vec![];
+  assert_eq!(parse_min_record(&buf), None);
+
+  let buf = vec![5];
+  assert_eq!(parse_min_record(&buf), None);
+}
+
+pub fn dispatch_minimal(
+  state: &ThreadSafeState,
+  mut record: Record,
+  zero_copy: Option<PinnedBuf>,
+) -> Op {
+  let is_sync = record.promise_id == 0;
+  let min_op = match record.op_id {
+    OP_READ => ops::read(record.arg, zero_copy),
+    OP_WRITE => ops::write(record.arg, zero_copy),
+    _ => unimplemented!(),
+  };
+
+  let state = state.clone();
+
+  let fut = Box::new(min_op.then(move |result| -> Result<Buf, ()> {
+    match result {
+      Ok(r) => {
+        record.result = r;
+      }
+      Err(err) => {
+        // TODO(ry) The dispatch_minimal doesn't properly pipe errors back to
+        // the caller.
+        debug!("swallowed err {}", err);
+        record.result = -1;
+      }
+    }
+    let buf: Buf = record.into();
+    state.metrics_op_completed(buf.len());
+    Ok(buf)
+  }));
+  if is_sync {
+    Op::Sync(fut.wait().unwrap())
+  } else {
+    Op::Async(fut)
+  }
+}
+
+mod ops {
+  use crate::errors;
+  use crate::resources;
+  use crate::tokio_write;
+  use deno::PinnedBuf;
+  use futures::Future;
+
+  type MinimalOp = dyn Future<Item = i32, Error = errors::DenoError> + Send;
+
+  pub fn read(rid: i32, zero_copy: Option<PinnedBuf>) -> Box<MinimalOp> {
+    debug!("read rid={}", rid);
+    let zero_copy = zero_copy.unwrap();
+    match resources::lookup(rid as u32) {
+      None => Box::new(futures::future::err(errors::bad_resource())),
+      Some(resource) => Box::new(
+        tokio::io::read(resource, zero_copy)
+          .map_err(errors::DenoError::from)
+          .and_then(move |(_resource, _buf, nread)| Ok(nread as i32)),
+      ),
+    }
+  }
+
+  pub fn write(rid: i32, zero_copy: Option<PinnedBuf>) -> Box<MinimalOp> {
+    debug!("write rid={}", rid);
+    let zero_copy = zero_copy.unwrap();
+    match resources::lookup(rid as u32) {
+      None => Box::new(futures::future::err(errors::bad_resource())),
+      Some(resource) => Box::new(
+        tokio_write::write(resource, zero_copy)
+          .map_err(errors::DenoError::from)
+          .and_then(move |(_resource, _buf, nwritten)| Ok(nwritten as i32)),
+      ),
+    }
+  }
+}

cli/main.rs

@@ -15,6 +15,7 @@ extern crate nix;
 mod ansi;
 pub mod compiler;
 pub mod deno_dir;
+mod dispatch_minimal;
 pub mod errors;
 pub mod flags;
 mod fs;

cli/ops.rs

@@ -2,6 +2,8 @@
 use atty;
 use crate::ansi;
 use crate::compiler::get_compiler_config;
+use crate::dispatch_minimal::dispatch_minimal;
+use crate::dispatch_minimal::parse_min_record;
 use crate::errors;
 use crate::errors::{DenoError, DenoResult, ErrorKind};
 use crate::fs as deno_fs;
@@ -74,10 +76,6 @@ fn empty_buf() -> Buf {
   Box::new([])
 }
 
-/// Processes raw messages from JavaScript.
-/// This functions invoked every time Deno.core.dispatch() is called.
-/// control corresponds to the first argument of Deno.core.dispatch().
-/// data corresponds to the second argument of Deno.core.dispatch().
 pub fn dispatch_all(
   state: &ThreadSafeState,
   control: &[u8],
@@ -86,6 +84,25 @@ pub fn dispatch_all(
 ) -> Op {
   let bytes_sent_control = control.len();
   let bytes_sent_zero_copy = zero_copy.as_ref().map(|b| b.len()).unwrap_or(0);
+  let op = if let Some(min_record) = parse_min_record(control) {
+    dispatch_minimal(state, min_record, zero_copy)
+  } else {
+    dispatch_all_legacy(state, control, zero_copy, op_selector)
+  };
+  state.metrics_op_dispatched(bytes_sent_control, bytes_sent_zero_copy);
+  op
+}
+
+/// Processes raw messages from JavaScript.
+/// This functions invoked every time Deno.core.dispatch() is called.
+/// control corresponds to the first argument of Deno.core.dispatch().
+/// data corresponds to the second argument of Deno.core.dispatch().
+pub fn dispatch_all_legacy(
+  state: &ThreadSafeState,
+  control: &[u8],
+  zero_copy: Option<PinnedBuf>,
+  op_selector: OpSelector,
+) -> Op {
   let base = msg::get_root_as_base(&control);
   let is_sync = base.sync();
   let inner_type = base.inner_type();
@@ -99,7 +116,6 @@ pub fn dispatch_all(
   let op: Box<OpWithError> = op_func(state, &base, zero_copy);
 
   let state = state.clone();
-  state.metrics_op_dispatched(bytes_sent_control, bytes_sent_zero_copy);
 
   let fut = Box::new(
     op.or_else(move |err: DenoError| -> Result<Buf, ()> {

js/dispatch.ts

@@ -4,25 +4,53 @@ import * as flatbuffers from "./flatbuffers";
 import * as msg from "gen/cli/msg_generated";
 import * as errors from "./errors";
 import * as util from "./util";
+import {
+  nextPromiseId,
+  recordFromBufMinimal,
+  handleAsyncMsgFromRustMinimal
+} from "./dispatch_minimal";
 
-let nextCmdId = 0;
 const promiseTable = new Map<number, util.Resolvable<msg.Base>>();
 
-export function handleAsyncMsgFromRust(ui8: Uint8Array): void {
-  const bb = new flatbuffers.ByteBuffer(ui8);
+interface FlatbufferRecord {
+  promiseId: number;
+  base: msg.Base;
+}
+
+function flatbufferRecordFromBuf(buf: Uint8Array): FlatbufferRecord {
+  const bb = new flatbuffers.ByteBuffer(buf);
   const base = msg.Base.getRootAsBase(bb);
-  const cmdId = base.cmdId();
-  const promise = promiseTable.get(cmdId);
-  util.assert(promise != null, `Expecting promise in table. ${cmdId}`);
-  promiseTable.delete(cmdId);
-  const err = errors.maybeError(base);
-  if (err != null) {
-    promise!.reject(err);
+  return {
+    promiseId: base.cmdId(),
+    base
+  };
+}
+
+export function handleAsyncMsgFromRust(ui8: Uint8Array): void {
+  const buf32 = new Int32Array(ui8.buffer, ui8.byteOffset, ui8.byteLength / 4);
+  const recordMin = recordFromBufMinimal(buf32);
+  if (recordMin) {
+    // Fast and new
+    handleAsyncMsgFromRustMinimal(ui8, recordMin);
   } else {
-    promise!.resolve(base);
+    // Legacy
+    let { promiseId, base } = flatbufferRecordFromBuf(ui8);
+    const promise = promiseTable.get(promiseId);
+    util.assert(promise != null, `Expecting promise in table. ${promiseId}`);
+    promiseTable.delete(promiseId);
+    const err = errors.maybeError(base);
+    if (err != null) {
+      promise!.reject(err);
+    } else {
+      promise!.resolve(base);
+    }
   }
 }
 
+function ui8FromArrayBufferView(abv: ArrayBufferView): Uint8Array {
+  return new Uint8Array(abv.buffer, abv.byteOffset, abv.byteLength);
+}
+
 function sendInternal(
   builder: flatbuffers.Builder,
   innerType: msg.Any,
@@ -30,20 +58,20 @@ function sendInternal(
   zeroCopy: undefined | ArrayBufferView,
   sync = true
 ): [number, null | Uint8Array] {
-  const cmdId = nextCmdId++;
-  const message = msg.Base.createBase(
-    builder,
-    cmdId,
-    sync,
-    0,
-    0,
-    innerType,
-    inner
-  );
-  builder.finish(message);
+  const cmdId = nextPromiseId();
+  msg.Base.startBase(builder);
+  msg.Base.addInner(builder, inner);
+  msg.Base.addInnerType(builder, innerType);
+  msg.Base.addSync(builder, sync);
+  msg.Base.addCmdId(builder, cmdId);
+  builder.finish(msg.Base.endBase(builder));
 
   const control = builder.asUint8Array();
-  const response = core.dispatch(control, zeroCopy);
+
+  const response = core.dispatch(
+    control,
+    zeroCopy ? ui8FromArrayBufferView(zeroCopy) : undefined
+  );
 
   builder.inUse = false;
   return [cmdId, response];

js/dispatch_minimal.ts

@@ -0,0 +1,77 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+// Do not add flatbuffer dependencies to this module.
+import * as util from "./util";
+import { core } from "./core";
+
+const DISPATCH_MINIMAL_TOKEN = 0xcafe;
+const promiseTableMin = new Map<number, util.Resolvable<number>>();
+let _nextPromiseId = 0;
+
+export function nextPromiseId(): number {
+  return _nextPromiseId++;
+}
+
+export interface RecordMinimal {
+  promiseId: number;
+  opId: number;
+  arg: number;
+  result: number;
+}
+
+/** Determines if a message has the "minimal" serialization format. If false, it
+ * is flatbuffer encoded.
+ */
+export function hasMinimalToken(i32: Int32Array): boolean {
+  return i32[0] == DISPATCH_MINIMAL_TOKEN;
+}
+
+export function recordFromBufMinimal(buf32: Int32Array): null | RecordMinimal {
+  if (hasMinimalToken(buf32)) {
+    return {
+      promiseId: buf32[1],
+      opId: buf32[2],
+      arg: buf32[3],
+      result: buf32[4]
+    };
+  }
+  return null;
+}
+
+const scratch32 = new Int32Array(5);
+const scratchBytes = new Uint8Array(
+  scratch32.buffer,
+  scratch32.byteOffset,
+  scratch32.byteLength
+);
+util.assert(scratchBytes.byteLength === scratch32.length * 4);
+
+export function handleAsyncMsgFromRustMinimal(
+  ui8: Uint8Array,
+  record: RecordMinimal
+): void {
+  // Fast and new
+  util.log("minimal handleAsyncMsgFromRust ", ui8.length);
+  const { promiseId, result } = record;
+  const promise = promiseTableMin.get(promiseId);
+  promiseTableMin.delete(promiseId);
+  promise!.resolve(result);
+}
+
+export function sendAsyncMinimal(
+  opId: number,
+  arg: number,
+  zeroCopy: Uint8Array
+): Promise<number> {
+  const promiseId = nextPromiseId(); // AKA cmdId
+
+  scratch32[0] = DISPATCH_MINIMAL_TOKEN;
+  scratch32[1] = promiseId;
+  scratch32[2] = opId;
+  scratch32[3] = arg;
+
+  const promise = util.createResolvable<number>();
+  promiseTableMin.set(promiseId, promise);
+
+  core.dispatch(scratchBytes, zeroCopy);
+  return promise;
+}

js/files.ts

@@ -11,10 +11,14 @@ import {
   SyncSeeker
 } from "./io";
 import * as dispatch from "./dispatch";
+import { sendAsyncMinimal } from "./dispatch_minimal";
 import * as msg from "gen/cli/msg_generated";
 import { assert } from "./util";
 import * as flatbuffers from "./flatbuffers";
 
+const OP_READ = 1;
+const OP_WRITE = 2;
+
 function reqOpen(
   filename: string,
   mode: OpenMode
@@ -101,7 +105,14 @@ export function readSync(rid: number, p: Uint8Array): ReadResult {
  *       })();
  */
 export async function read(rid: number, p: Uint8Array): Promise<ReadResult> {
-  return resRead(await dispatch.sendAsync(...reqRead(rid, p)));
+  const nread = await sendAsyncMinimal(OP_READ, rid, p);
+  if (nread < 0) {
+    throw new Error("read error");
+  } else if (nread == 0) {
+    return { nread, eof: true };
+  } else {
+    return { nread, eof: false };
+  }
 }
 
 function reqWrite(
@@ -147,7 +158,12 @@ export function writeSync(rid: number, p: Uint8Array): number {
  *
  */
 export async function write(rid: number, p: Uint8Array): Promise<number> {
-  return resWrite(await dispatch.sendAsync(...reqWrite(rid, p)));
+  let result = await sendAsyncMinimal(OP_WRITE, rid, p);
+  if (result < 0) {
+    throw new Error("write error");
+  } else {
+    return result;
+  }
 }
 
 function reqSeek(