Clean up core/shared_queue.js (#5237)

1 files changed, 203 insertions, 0 deletions

diff --git a/core/core.js b/core/core.js
new file mode 100644
index 000000000..4c6f708bb
--- /dev/null
+++ b/core/core.js
@@ -0,0 +1,203 @@
+// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
+/*
+SharedQueue Binary Layout
++-------------------------------+-------------------------------+
+|                        NUM_RECORDS (32)                       |
++---------------------------------------------------------------+
+|                        NUM_SHIFTED_OFF (32)                   |
++---------------------------------------------------------------+
+|                        HEAD (32)                              |
++---------------------------------------------------------------+
+|                        OFFSETS (32)                           |
++---------------------------------------------------------------+
+|                        RECORD_ENDS (*MAX_RECORDS)           ...
++---------------------------------------------------------------+
+|                        RECORDS (*MAX_RECORDS)               ...
++---------------------------------------------------------------+
+ */
+
+/* eslint-disable @typescript-eslint/no-use-before-define */
+
+((window) => {
+  const MAX_RECORDS = 100;
+  const INDEX_NUM_RECORDS = 0;
+  const INDEX_NUM_SHIFTED_OFF = 1;
+  const INDEX_HEAD = 2;
+  const INDEX_OFFSETS = 3;
+  const INDEX_RECORDS = INDEX_OFFSETS + 2 * MAX_RECORDS;
+  const HEAD_INIT = 4 * INDEX_RECORDS;
+
+  // Available on start due to bindings.
+  const core = window.Deno.core;
+  const { recv, send } = core;
+
+  let sharedBytes;
+  let shared32;
+
+  let asyncHandlers;
+
+  let initialized = false;
+
+  function maybeInit() {
+    if (!initialized) {
+      init();
+      initialized = true;
+    }
+  }
+
+  function init() {
+    const shared = core.shared;
+    assert(shared.byteLength > 0);
+    assert(sharedBytes == null);
+    assert(shared32 == null);
+    sharedBytes = new Uint8Array(shared);
+    shared32 = new Int32Array(shared);
+    asyncHandlers = [];
+    // Callers should not call core.recv, use setAsyncHandler.
+    recv(handleAsyncMsgFromRust);
+  }
+
+  function ops() {
+    // op id 0 is a special value to retrieve the map of registered ops.
+    const opsMapBytes = send(0, new Uint8Array([]), null);
+    const opsMapJson = String.fromCharCode.apply(null, opsMapBytes);
+    return JSON.parse(opsMapJson);
+  }
+
+  function assert(cond) {
+    if (!cond) {
+      throw Error("assert");
+    }
+  }
+
+  function reset() {
+    maybeInit();
+    shared32[INDEX_NUM_RECORDS] = 0;
+    shared32[INDEX_NUM_SHIFTED_OFF] = 0;
+    shared32[INDEX_HEAD] = HEAD_INIT;
+  }
+
+  function head() {
+    maybeInit();
+    return shared32[INDEX_HEAD];
+  }
+
+  function numRecords() {
+    return shared32[INDEX_NUM_RECORDS];
+  }
+
+  function size() {
+    return shared32[INDEX_NUM_RECORDS] - shared32[INDEX_NUM_SHIFTED_OFF];
+  }
+
+  function setMeta(index, end, opId) {
+    shared32[INDEX_OFFSETS + 2 * index] = end;
+    shared32[INDEX_OFFSETS + 2 * index + 1] = opId;
+  }
+
+  function getMeta(index) {
+    if (index < numRecords()) {
+      const buf = shared32[INDEX_OFFSETS + 2 * index];
+      const opId = shared32[INDEX_OFFSETS + 2 * index + 1];
+      return [opId, buf];
+    } else {
+      return null;
+    }
+  }
+
+  function getOffset(index) {
+    if (index < numRecords()) {
+      if (index == 0) {
+        return HEAD_INIT;
+      } else {
+        const prevEnd = shared32[INDEX_OFFSETS + 2 * (index - 1)];
+        return (prevEnd + 3) & ~3;
+      }
+    } else {
+      return null;
+    }
+  }
+
+  function push(opId, buf) {
+    const off = head();
+    const end = off + buf.byteLength;
+    const alignedEnd = (end + 3) & ~3;
+    const index = numRecords();
+    if (alignedEnd > shared32.byteLength || index >= MAX_RECORDS) {
+      // console.log("shared_queue.js push fail");
+      return false;
+    }
+    setMeta(index, end, opId);
+    assert(alignedEnd % 4 === 0);
+    assert(end - off == buf.byteLength);
+    sharedBytes.set(buf, off);
+    shared32[INDEX_NUM_RECORDS] += 1;
+    shared32[INDEX_HEAD] = alignedEnd;
+    return true;
+  }
+
+  /// Returns null if empty.
+  function shift() {
+    const i = shared32[INDEX_NUM_SHIFTED_OFF];
+    if (size() == 0) {
+      assert(i == 0);
+      return null;
+    }
+
+    const off = getOffset(i);
+    const [opId, end] = getMeta(i);
+
+    if (size() > 1) {
+      shared32[INDEX_NUM_SHIFTED_OFF] += 1;
+    } else {
+      reset();
+    }
+
+    assert(off != null);
+    assert(end != null);
+    const buf = sharedBytes.subarray(off, end);
+    return [opId, buf];
+  }
+
+  function setAsyncHandler(opId, cb) {
+    maybeInit();
+    assert(opId != null);
+    asyncHandlers[opId] = cb;
+  }
+
+  function handleAsyncMsgFromRust(opId, buf) {
+    if (buf) {
+      // This is the overflow_response case of deno::Isolate::poll().
+      asyncHandlers[opId](buf);
+    } else {
+      while (true) {
+        const opIdBuf = shift();
+        if (opIdBuf == null) {
+          break;
+        }
+        assert(asyncHandlers[opIdBuf[0]] != null);
+        asyncHandlers[opIdBuf[0]](opIdBuf[1]);
+      }
+    }
+  }
+
+  function dispatch(opId, control, zeroCopy = null) {
+    return send(opId, control, zeroCopy);
+  }
+
+  Object.assign(window.Deno.core, {
+    setAsyncHandler,
+    dispatch,
+    ops,
+    // sharedQueue is private but exposed for testing.
+    sharedQueue: {
+      MAX_RECORDS,
+      head,
+      numRecords,
+      size,
+      push,
+      reset,
+      shift,
+    },
+  });
+})(this);