emscriptenでnode.jsのstreamに対応

ukyo/lz4.jsでnode.jsのTransform streamを扱えるようにしたのでメモ。実装はC言語。今回の実装によって

• jsオブジェクトとcの構造体の紐付け
• 複数のstreamを同時に実行(多分)

が実現できた感じです。

実装メモ

c: 常に構造体のポインタを介するコードにする。

src/lz4js.h#15

LZ4JS_compressionContext_t* LZ4JS_createCompressionContext(LZ4F_blockSizeID_t blockSizeID, LZ4F_blockMode_t blockMode, LZ4F_contentChecksum_t contentChecksum, unsigned int compressionLevel);
LZ4F_decompressionContext_t* LZ4JS_createDecompressionContext();
void LZ4JS_freeCompressionContext(LZ4JS_compressionContext_t* cctxPtr);
void LZ4JS_freeDecompressionContext(LZ4F_decompressionContext_t* dctxPtr);
int LZ4JS_compressBegin(LZ4JS_compressionContext_t* cctxPtr);
int LZ4JS_compressUpdate(LZ4JS_compressionContext_t* cctxPtr);
int LZ4JS_compressEnd(LZ4JS_compressionContext_t* cctxPtr);
int LZ4JS_decompress(LZ4F_decompressionContext_t* dctxPtr);

js: 返ってきたポインタをキーにしてオブジェクトにインスタンスを保存。

src/post.js#69

function BaseCompressor(options) {
  this.options = assign({}, defaultCompressOptions, options);
  this.cctxPtr = _LZ4JS_createCompressionContext(/* ...中略... */); // ポインタ返ってくる
  if (!this.cctxPtr) throw new Error('LZ4JS_createCompressionContext');
  LZ4JS_instances[this.cctxPtr] = this; // ポインタをキーにしてインスタンスを保存
  this.$error = null;
}

js: 必要なリソース、今回はバイト列srcと長さsrcSizeを用意してC側の関数を呼び出す。

src/post.js#146

CompressStream.prototype['_transform'] = function(chunk, encoding, callback) {
  try {
    /* ...中略... */
    for (offset = 0; offset < chunk.length; offset += BUF_SIZE) {
      this.srcSize = Math.min(chunk.length - offset, BUF_SIZE);
      this.src = chunk.slice(offset, offset + this.srcSize);
      this.compressUpdate(); // 以下参照
    }
    callback();
  } catch (error) {
    callback(error);
  }
};

src/post.js#86

BaseCompressor.prototype.compressUpdate = function(first_argument) {
  _LZ4JS_compressUpdate(this.cctxPtr) || this.cleanup();
};

c: LZ4JS_readでjsから入力を行う(自作fread的なやつ、以下参照)。

src/lz4js.c#69

int LZ4JS_compressUpdate(LZ4JS_compressionContext_t* cctxPtr) {
  size_t n = LZ4F_compressUpdate(cctxPtr->cctx, dst, dstMaxSize, src, LZ4JS_read(cctxPtr, src, BUF_SIZE), NULL);
  if (LZ4JS_validate(cctxPtr, n)) {
    LZ4JS_write(cctxPtr, dst, n);
    return 1;
  }
  return 0;
}

c: EM_ASM_INTを使ってjs側のコードにキーptrとバイト列の先頭位置bufと最大サイズsizeを送る。

src/lz4js.c#14

size_t LZ4JS_read(void* ptr, char* buf, size_t size) {
  return EM_ASM_INT({return LZ4JS_read($0, $1, $2)}, ptr, buf, size);
}

js: キーでインスタンスを参照して、実装されている$readメソッドを呼び出す。

src/post.js#3

function LZ4JS_read(id, srcPtr, size) {
  return LZ4JS_instances[id].$read(srcPtr, size);
}

js: emscripten上のメモリHEAPU8に直接書き込む。

src/post.js#136

CompressStream.prototype.$read = function(srcPtr, size) {
  HEAPU8.set(new Uint8Array(this.src.buffer, this.src.byteOffset, this.srcSize), srcPtr);
  return this.srcSize;
};

c: すると、LZ4F_compressUpdate呼び出し時点でデータが入っているという。LZ4JS_validate、LZ4JS_writeも同様にjs側とやりとりをする。

src/lz4js.c#69

int LZ4JS_compressUpdate(LZ4JS_compressionContext_t* cctxPtr) {
  size_t n = LZ4F_compressUpdate(cctxPtr->cctx, dst, dstMaxSize, src, LZ4JS_read(cctxPtr, src, BUF_SIZE), NULL);
  if (LZ4JS_validate(cctxPtr, n)) {
    LZ4JS_write(cctxPtr, dst, n);
    return 1;
  }
  return 0;
}

まとめ

EM_ASM、HEAPU8や構造体のポインタを直接利用してアクロバティックにjsとやり取りする方法を紹介しました。ちなみに、jsはシングルスレッドなのでc側の配列はstreamごとに作らないで共有しても大丈夫です。