backup

src/channel.js

+
+import { WW } from './WW.js';
+import { assert } from './utils.js';
+
+class CpySchedule {
+  first = { pos: 0, size: 0 }
+  second = { pos: 0, size: 0 }
+}
+
+/* An abstraction that help to
+ * easily to transfer datas to target worker
+ * in real time manner */
+export class Channel {
+
+  /* Size of memory area that help to
+   * maintain Channel. Types of meta info
+   * are shown below:
+   *
+   * 1.Write Position
+   * 2.Read Position
+   *
+   * Figure:
+   *  --------------------------------------------------------------------------------------------
+   * | ReadPointer (4 bytes) | WritePointer (4 Bytes) | Private (4 Bytes) | Data Area (N bytes)  |
+   * --------------------------------------------------------------------------------------------
+   * where N >= 2
+   *
+   * */
+  #rFieldPosLen   = 4;
+  #wFieldPosLen   = 4;
+  #priFieldLen    = 4;
+  #numOfMetaField = 3
+  #fieldSize      = 0;
+
+  #metaSize  = 0;
+  #size      = 0;
+  #totalSize = 0;
+
+  #WW     = undefined;
+  #shMem  = undefined;
+  #view   = undefined;
+  #buffer = undefined;
+
+  #writeValCache = 0;
+  #readValCache  = 0;
+  #endPos        = 0;
+
+  // size's unit is byte
+  // bufferType parameter is mainly for testability.
+  constructor(size, bufferType = SharedArrayBuffer) {
+
+    assert(size >= 2, `Channel require its data area has at least 2 Bytes.`)
+
+    this.#WW = WW;
+    this.#size = size;
+
+    // Init shared memory
+    this.#metaSize      = this.#rFieldPosLen + this.#wFieldPosLen + this.#priFieldLen;
+    this.#shMem         = new bufferType(size + this.#metaSize);
+    this.#view          = new DataView(this.#shMem);
+    this.#buffer        = new Uint8Array(this.#shMem);
+    this.#writeValCache = this.#metaSize;
+    this.#readValCache  = this.#metaSize;
+    this.#size          = size;
+    this.#totalSize     = this.#metaSize + this.#size;
+    this.#endPos        = this.#metaSize + this.#size;
+
+    // Init readPointer and writePointer to
+    // the first bytes of data area.
+    this.#view.setUint32(0, this.#writeValCache);
+    this.#view.setUint32(4, this.#readValCache);
+  }
+
+  #getReadPointer() {
+    return this.#view.getUint32(0);
+  }
+
+  #getWritePointer() {
+    return this.#view.getUint32(4);
+  }
+
+  readPriv() {
+    return [
+      this.#view.getUint8(8),
+      this.#view.getUint8(9),
+      this.#view.getUint8(10),
+      this.#view.getUint8(11),
+    ];
+  }
+
+  writePriv(privData) {
+    if (!(privData instanceof Array) ||
+        privData.length > this.#priFieldLen) {
+      throw new Error("Invalid Private datas");
+    }
+
+    for (let i = 8; i < 12; ++i) {
+      this.#view.setUint8(i);
+    }
+
+  }
+
+  /* Semantic: Is able to write 'size' of datas
+   * into #shMem. */
+  #isAbleToWrite(size) {
+    return this.#remain() >= size;
+  }
+
+  getShMem() {
+    return this.#shMem;
+  }
+
+  #remain() {
+    let readPos = this.#readPointer();
+
+    if (this.#writePos == readPos) {
+      return this.#size - 1;
+    } else if (this.#writePos > readPos) {
+      return this.#size - (this.#writePos - readPos) - 1;
+    } else {
+      return readPos - this.#writePos - 1;
+    }
+  }
+
+  /* Channel use an array buffer as
+   * a circular buffer, so some datas
+   * may unable to be done in one copy.
+   * Two times of copy is required to handle
+   * such kind of situation.
+   *
+   * Caution: Before making a copy schedule you must
+   *          make sure there has enough spaces.*/
+  #cpySchedule(size) {
+    let firstCpySize = 0, secondCpySize = 0, spaceToTail = 0;
+    let schedule = new CpySchedule();
+    let readPos = this.#readPos();
+
+    if (this.#writePos >= readPos) {
+      spaceToTail = this.#endPos - this.#writePos;
+      firstCpySize = Math.min(size, spaceToTail);
+      secondCpySize = firstCpySize < size ? size - firstCpySize : 0;
+      secondCpySize = Math.min(secondCpySize, readPos - this.#metaSize);
+
+      schedule.first.pos = this.#writePos;
+      schedule.first.size = firstCpySize;
+      schedule.second.pos = secondCpySize > 0 ? this.#metaSize : 0;
+      schedule.second.size = secondCpySize;
+
+    } else {
+      schedule.first.pos = this.#writePos;
+      schedule.first.size = Math.min(readPos - this.#writePos - 1, size);
+    }
+
+    return schedule;
+  }
+
+  dataView() {
+    return this.#buffer;
+  }
+
+  isEmpty() {
+    return this.#writePos == this.#readPos();
+  }
+
+  // This method is for testing purposes.
+  readData(size) {
+    let writePos = this.#writePos();
+    let readTo = 0, readBuffer = null;
+
+
+    if (this.#readPos == writePos) {
+      return new Uint8Array(0);
+    } else if (this.#readPos < writePos) {
+      readTo = this.#readPos + Math.min(size, writePos - this.#readPos);
+      readBuffer = this.#buffer.slice(this.#readPos, readTo);
+      this.#readPosUpdate(readTo);
+    } else {
+      // Read two times
+      let firstRSize = Math.min(size, this.#buffer.byteLength-this.#readPos);
+      let secondRSize = firstRSize < size ? size - firstRSize : 0;
+      secondRSize = Math.min(secondRSize, writePos-this.#metaSize);
+
+      readBuffer = new Uint8Array(firstRSize+secondRSize);
+
+      // First read
+      readBuffer.set(this.#buffer.slice(
+        this.#readPos, this.#readPos+firstRSize), 0);
+
+      // Second Read
+      if (secondRSize > 0) {
+        readBuffer.set(
+          this.#buffer.slice(this.#metaSize, this.#metaSize+secondRSize),
+          firstRSize);
+        this.#readPosUpdate(this.#metaSize+secondRSize);
+      } else {
+        let newPos = this.#readPos+firstRSize;
+        newPos = newPos == this.#buffer.byteLength ? this.#metaSize : newPos;
+        this.#readPosUpdate(newPos);
+      }
+    }
+
+    return readBuffer;
+  }
+
+  push(data /* Uint8Array */) {
+
+    let writePos = this.#writePos;
+
+    if (!this.#isAbleToWrite(data.byteLength)) {
+      return false;
+    }
+
+    /* Write to shared Memory */
+
+    // Get a copy schedule, more details please read
+    // comment of #cpySchedule.
+    let schedule = this.#cpySchedule(data.byteLength);
+
+    // Perfrom write schedule
+    let srcPos = 0, plan;
+
+    for (let key in schedule) {
+      plan = schedule[key];
+
+      if (plan.size == 0)
+        continue;
+
+      this.#buffer.set(
+        data.slice(srcPos, srcPos+plan.size), plan.pos)
+      srcPos += plan.size;
+      writePos = plan.pos+plan.size;
+    }
+
+    // Caution: 'Write Pointer' must be updated after
+    //          all datas are writed but not before or
+    //          at intermediate of some writes otherwise
+    //          oppsite side may read invalid datas.
+    this.#writePosUpdate(writePos);
+
+    return true;
+  }
+
+  #writePointerUpdate(pos) {
+    this.#writePos = pos;
+    this.#view.setUint32(4, pos);
+  }
+
+  #readPointerUpdate(pos) {
+    this.#readPos = pos;
+    this.#view.setUint32(0, pos);
+  }
+}

tests/workers/moduleWW.js

+import {COMMANDS} from '../src/WWOpts.js';
+
+onmessage = e => {
+  console.log(COMMANDS.INIT_DONE);
+  console.log(e);
+}