Optimize read and write ops (#2259)

3 files changed, 146 insertions, 25 deletions

diff --git a/js/dispatch.ts b/js/dispatch.ts
index d5e10ad09..64bdfda17 100644
--- a/js/dispatch.ts
+++ b/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];
diff --git a/js/dispatch_minimal.ts b/js/dispatch_minimal.ts
new file mode 100644
index 000000000..17d328110
--- /dev/null
+++ b/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;
+}
diff --git a/js/files.ts b/js/files.ts
index 18ee612bc..1479e4146 100644
--- a/js/files.ts
+++ b/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(