Implement Encoder.

src/WW.js

@@ -23,33 +23,8 @@ export class WW extends Observable {
     );
   }
 
-  async start() {
-
-    if (this.#init) {
-      return this.#init;
-    }
-
-    await new Promise(resolve => {
-
-      if (this.#init || !this.#healthy) {
-        resolve();
-      }
-
-      let sub = this.subscribe(e => {
-        if (e == COMMANDS.INIT_DONE) {
-          this.#init = true;
-        } else if (e == COMMANDS.INIT_FAIL) {
-          this.#init = false;
-          this.#healthy = false;
-        }
-        sub.unsubscribe();
-
-        resolve();
-      });
-
-      this.#ww.postMessage("Initialize");
-    });
-
+  async start(initWorks) {
+    this.#init = await initWorks(this);
     return this.#init;
   }
 

src/WWGroup.js

@@ -10,7 +10,7 @@ import { OVERRIDE_IS_REQUIRE } from './utils.js';
  * how to communicate with such WWs. */
 export class WWGroup {
 
-  start() {
+  async start() {
     OVERRIDE_IS_REQUIRE();
   }
 
@@ -18,24 +18,28 @@ export class WWGroup {
     OVERRIDE_IS_REQUIRE();
   }
 
-  dispatch() {
+  async dispatch() {
+    OVERRIDE_IS_REQUIRE();
+  }
+
+  async isReady() {
     OVERRIDE_IS_REQUIRE();
   }
 
   /* If none of WWs is in working then the
    * WWGroup is idel otherwise not. */
-  isIdle() {
+  async isIdle() {
     OVERRIDE_IS_REQUIRE();
   }
 
   /* If each of WWs reside in WWGroup is in working
    * then it's busy */
-  isBusy() {
+  async isBusy() {
       OVERRIDE_IS_REQUIRE();
   }
 
   /* WWGroup meet it's gole and all exits. */
-  isDone() {
+  async isDone() {
     OVERRIDE_IS_REQUIRE();
   }
 }

src/channel.js

@@ -86,24 +86,34 @@ export class Channel {
     return this.#view.getUint32(4);
   }
 
+  metaSize() {
+    return this.#metaSize;
+  }
+
   readPriv() {
-    return [
-      this.#view.getUint8(8),
-      this.#view.getUint8(9),
-      this.#view.getUint8(10),
-      this.#view.getUint8(11),
-    ];
+    return this.#view.getUint32(8);
   }
 
   writePriv(privData) {
-    if (!(privData instanceof Array) ||
-        privData.length > this.#priFieldLen) {
-      throw new Error("Invalid Private datas");
+    try {
+      this.#view.setUint32(8, privData);
+    } catch (error) {
+      if (error instanceof RangeError)
+        return false;
     }
 
-    for (let i = 8; i < 12; ++i) {
-      this.#view.setUint8(i);
+    return true;
   }
+
+  setPriv(flag) {
+    let old = this.readPriv();
+    this.writePriv(flag | old);
+  }
+
+  unsetPriv(flag) {
+    let old = this.readPriv();
+    flag = flag ^ 0x11111111;
+    this.writePriv(flag & old);
   }
 
   /* Semantic: Is able to write 'size' of datas
@@ -164,7 +174,7 @@ export class Channel {
   }
 
   isEmpty() {
-    return this.#writePointerCache == this.#readPointerCache;
+    return this.#getWritePointer() == this.#getReadPointer();
   }
 
   // This method is for testing purposes.
@@ -179,7 +189,7 @@ export class Channel {
       readBuffer = this.#buffer.slice(this.#readPointerCache, readTo);
       this.#readPointerUpdate(readTo);
     } else {
-      // Read two times
+      // To make sure
       let firstRSize = Math.min(size, this.#buffer.byteLength-this.#readPointerCache);
       let secondRSize = firstRSize < size ? size - firstRSize : 0;
       secondRSize = Math.min(secondRSize, writePos-this.#metaSize);

src/encGroup.js

@@ -2,29 +2,29 @@
 import { assert } from './utils.js';
 import { WWGroup } from './WWGroup.js';
 import { WW } from './WW.js';
+import { Channel } from './channel.js';
+import { MESSAGE_TYPE, makeMsg, isDataMsg, makeInitMsg } from './encGrooupMsg.js';
 
-///////////////////////////////////////////////////////////////////////////////
-//                            Message Definitions                            //
-///////////////////////////////////////////////////////////////////////////////
-
-const MESSAGE_TYPE = Object.freeze({
-  /* Once Muxer receive a 'CONNECT_PREPARE' message
-   * It will use information along with the message to
-   * change it's status to get ready to be connected by
-   * Encoder. */
-  CONNECT_PREPARE    : 0,
-  /* Similar to 'CONNECT_PREPARE' */
-  DISCONNECT_PREPARE : 1,
-  DATA               : 2,
-});
-
-function makeMessage(msgType, datas) {
-  return {type:msgType, data: datas};
-}
+async function muxInit(chnls, ww) {
+  let channel = new Channel(Math.pow(2, 20));
+  ww.postMessage(
+    makeInitMsg(channel.getShMem(), Math.pow(2, 20)));
+  chnls[ww.ident()] = channel;
 
+  return true;
+}
 
-async function connectPrepare(WW) {
+async function encInit(chnls, ww) {
+  return await muxInit(chnls, ww);
+}
 
+async function encMuxBridge(bridge, enc, mux) {
+  let channel = new Channel(Math.pow(2, 20));
+  enc.postMessage(
+    makeMsg(MESSAGE_TYPE.CONNECT_PREPARE, channel.getShMem()));
+  mux.postMessage(
+    makeMsg(MESSAGE_TYPE.CONNECT_PREPARE, channel.getShMem()));
+  bridge[enc.ident()] = channel;
 }
 
 
@@ -38,10 +38,19 @@ export class H264EncWWGroup extends WWGroup {
   #encWorkers = [];
 
   #channels = {};
+  #bridges  = {};
+
+  /* Options */
 
   /* Number of frames WWGroup can handle at a time */
   #maxPayload = 0;
 
+  /* Muxer Channel Size */
+  #muxChannelSize = Math.pow(2, 20);
+
+  /* Encoder Channel Size */
+  #encChannelSize = Math.pow(2, 20);
+
   /* Precondition: 2 <= options.numOfWW */
   constructor(name, options) {
 
@@ -54,8 +63,11 @@ export class H264EncWWGroup extends WWGroup {
 
       if ('max' in options)
         this.#maxPayload = options.max;
+      if ('muxchnlsize' in options)
+        this.#muxChannelSize = options.muxchnlsize;
+      if ('encchnlsize' in options)
+        this.#encChannelSize = options.encChanlsize;
       if ('numOfWW' in options) {
-
         // Precondition not meet:
         // At least two web workers is required, one encode worker
         // and one mux worker.
@@ -93,20 +105,23 @@ export class H264EncWWGroup extends WWGroup {
 
   async start() {
     // Start Muxer
-    await this.#muxWorker.start();
+    await this.#muxWorker.start(async ww => {
+      return await muxInit(this.#channels, ww);
+    });
 
     for (let i = 0; i < this.#numOfEncWorker; ++i) {
-      await this.#encWorkers[i].start();
+      await this.#encWorkers[i].start(async ww => {
+        return await encInit(this.#channels, ww);
+      });
     }
 
     // Connect Encoders to the Muxer
     for (let i = 0; i < this.#numOfEncWorker; ++i) {
-      this.#encWorkers[i].connect(
+      let enc = this.#encWorkers[i];
+      enc.connect(
         this.#muxWorker,
-        connectPrepare,
-        async msg => {
-          return msg.type == MESSAGE_TYPE.DATA
-        });
+        async ww => { encMuxBridge(this.#bridges, ww, this.#muxWorker) },
+        async msg => { return isDataMsg(msg); });
     }
   }
 
@@ -114,7 +129,7 @@ export class H264EncWWGroup extends WWGroup {
     return this.#numOfWorkers;
   }
 
-  dispatch(rgbFrame) {
+  async dispatch(rgbFrame) {
 
   }
 

src/workers/encWW.js

 
 
-let init = false;
-const COMMANDS = Object.freeze({
-  INIT: 0,
-  INIT_DONE: 1,
-  INIT_FAIL: 2,
-});
+import { sleep } from '../src/utils.js';
+import * as MSG from '../src/encGrooupMsg.js';
+import { Channel } from '../src/channel.js';
 
+let isInited  = false;
+let isBridged = false;
+let src       = null;
+let bridge    = null;
 
-class ChannelReader {
+// Parameters
 
-  #rFieldPosLen = 4;
-  #wFieldPosLen = 4;
-  #privFieldLen = 4;
-
-  #metaSize = this.#rFieldPosLen +
-    this.#wFieldPosLen +
-    this.#privFieldLen;
-
-  #readPos = 0;
-  #buffer = null;
-  #view   = null;
-  #shMem  = null;
+// Number of bytes for each read
+// from channel
+let READ_SIZE = (1920*1080*4) * 10 ;
+const RETRY_COUNT = 5;
+const SLEEP_INTERVAL = 100;
 
+// Init
+onmessage = async e => {
+  let msg = e.data;
+  await main(msg);
+}
 
-  /* Buffer should follow the specification
-   * of Channel in src/encGroup.js */
-  constructor(shMem) {
-    this.#shMem  = shMem;
-    this.#buffer = new Uint8Array(this.#shMem);
-    this.#view   = new DataView(this.#shMem);
+async function main(msg) {
 
-    this.#readPos = this.#view.getUint32(0);
+  if (!MSG.isEncGrpMsg(msg)) {
+    return;
   }
 
-  readPriv() {
-    return [
-      this.#view.getUint8(8),
-      this.#view.getUint8(9),
-      this.#view.getUint8(10),
-      this.#view.getUint8(11),
-    ];
+  switch (MSG.typeOfMsg(msg)) {
+    case MSG.MESSAGE_TYPE.INIT:
+      init(msg);
+      break;
+    case MSG.MESSAGE_TYPE.DATA:
+      await steps();
+      break;
+    case MSG.MESSAGE_TYPE.DESTROY:
+      break;
+    case MSG.MESSAGE_TYPE.ERROR:
+      break;
   }
+}
 
-  writePriv(privData) {
-    if (!(privData instanceof Array) ||
-        privData.length > this.#privFieldLen) {
-      throw new Error("Invalid Private datas");
-    }
+function init(msg) {
+  let info = MSG.getInfoFromMsg(msg);
+  src = new Channel(info.size, SharedArrayBuffer, info.shm);
+  isInited = true;
+}
 
-    for (let i = 8; i < 12; ++i) {
-      this.#view.setUint8(i);
-    }
+// Return value:
+// True  :  Success
+// False : Fail
+async function step() {
+  // Read RGB Frame from Channel.
+  let data = src.readData(READ_SIZE);
 
-  }
+  /* RGB Frame Processing */
+  return await RGBProcessing(data);
+}
 
-  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);
-      }
-    }
+async function steps() {
 
-    return readBuffer;
-  }
+  let retryCount = RETRY_COUNT;
 
-  #writePos() {
-    return this.#view.getUint32(4);
-  }
+  assert(retryCount > 0);
 
-  #readPosUpdate(pos) {
-    this.#readPos = pos;
-    this.#view.setUint32(0, pos);
-  }
-};
+  src.setPriv(MSG.PRIV_FLAGS.EXECUTING);
 
+  while (true) {
+    if (await step() === false && retryCount >= 0) {
 
-// Init
-onmessage = e => {
-
-  if (init == false) {
-    postMessage(COMMANDS.INIT_DONE);
-    init = true;
-  } else {
-    // Echo
-    postMessage(e.data);
+      if (retryCount == 0)
+        break;
+
+      await sleep(SLEEP_INTERVAL);
+      --retryCount;
     }
+  }
+
+  src.unsetPriv(MSG.PRIV_FLAGS.EXECUTING);
+}
+
+///////////////////////////////////////////////////////////////////////////////
+//                         RGB Processing Definitions                        //
+///////////////////////////////////////////////////////////////////////////////
+
+async function RGBProcessing(frame) {
 
 }

src/workers/muxWW.js

 
-const COMMANDS = Object.freeze({
-  INIT: 0,
-  INIT_DONE: 1,
-  INIT_FAIL: 2,
-});
-
-let init = false;
 
 // Init
 onmessage = e => {
-
-  if (init == false) {
-    postMessage(COMMANDS.INIT_DONE);
-    init = true;
-  } else {
-    // Echo
-    postMessage(e.data);
-  }
-
 }

tests/WW.spec.js

@@ -7,7 +7,8 @@ describe("WW Spec", () => {
   it("Instantiation", async () => {
     let ww = new WW('core', url);
 
-    let success = await ww.start();
+    let success = await ww.start(
+      async w => { return true });
 
     expect(success).toBe(true);
     expect(ww.isReady()).toBe(true);
@@ -16,7 +17,7 @@ describe("WW Spec", () => {
   it("EchoWorker", async () => {
     let ww = new WW('echo', url);
 
-    await ww.start();
+    await ww.start(async w => { return true });
     expect(ww.isReady()).toBe(true);
 
     let count = 0;

tests/WWGroup.spec.js

@@ -82,7 +82,7 @@ describe("Channel Spec", () => {
       .withContext("Data Mismatch")
       .toBe(true);
     expect(channel.isEmpty()).toBe(true);
-  })
+  });
 
   it("32 KB Datas Write and Read", () => {
     let ret = false;
@@ -131,7 +131,7 @@ describe("Channel Spec", () => {
   it("Transfer 64 MB to WebWorker", async () => {
     let url = new URL('./workers/channelWW.js', import.meta.url),
         ret = true;
-    let channel = new Channel(Math.pow(2,20));
+    let channel = new Channel(Math.pow(2,16));
     let worker = new Worker(url, { type: 'module' });
 
     let sended = 0;
@@ -139,9 +139,9 @@ describe("Channel Spec", () => {
       [...Array(getRandomInt(Math.pow(2,10))).keys()]);
 
     let cur = 0;
-    let size = Math.pow(2, 26);
-    let rBuffer = new Uint8Array(Math.pow(2, 27));
-    let sBuffer = new Uint8Array(Math.pow(2, 27));
+    let size = Math.pow(2, 16);
+    let rBuffer = new Uint8Array(Math.pow(2, 17));
+    let sBuffer = new Uint8Array(Math.pow(2, 17));
 
     worker.postMessage(channel.getShMem());
 
@@ -191,7 +191,11 @@ describe("H264EncWWGroup Spec", () => {
   it("Instantiation", async () => {
     let wg = new H264EncWWGroup("h264enc", { numOfWW: 3 });
     await wg.start();
+
+    await sleep(1000);
+
     expect(wg.numOfWorker()).toBe(3);
+
   });
 
 })

tests/workers/channelWW.js

 import { Channel } from '../src/channel.js';
 
-
 async function sleep(ms) {
   await new Promise(r => setTimeout(() => r(), ms));
 }