highWaterMarkから探るNode.jsのStreamの仕組み

こんにちは。Node.js 言語サポートチームの柄澤史也 (@fmy) です。

12/6のエントリ「今年のうちに対応したい、Node.jsのBufferに潜む危険性」に引き続き、社内での Node.js Core API 勉強会の内容を紹介します。

1. Stream API の重要性

Node.js 界隈の有名な言葉に次のようなものがあります。

「Stream を制するものは、 Node.js を制す」

Node.js 最大の特徴は何と言っても非同期 I/O です。
そしてその非同期 I/O の実装こそが stream モジュールなのです。

• HTTP サーバのリクエスト・レスポンス
• gulp のようなタスクランナー
• TCP ソケット
• 標準入出力

などなど、Stream は数多のモジュールの裏側に存在し、Node.js の根幹をなしています。

Stream を使わない Node.js エンジニアは存在しないと言っていいでしょう。
だからこそ「Stream を制するものは、 Node.js を制す」と言われるのです。

しかし、Stream をきちんと理解して利用しているという人は少ないのではないでしょうか。
かくいう私も曖昧なまま利用していました。

今回 Node.js Core API 勉強会で Stream API を担当し、ドキュメントとソースコードを読み込むことでかなり理解をすることができました。

Stream は複雑な実装となっており、すべてを一度に理解しようとするのは難しいと思います。
ここでは、Stream のバッファリングとバックプレッシャーの仕組みを highWaterMark をキーとして紐解いていこうと思います。

2. highWaterMark とは

Node.js v0.10 から Stream に内部バッファの仕組みが導入されました。ストリームインスタンスの入力データは、内部バッファとしてメモリ上に保持されます。

インスタンス作成時のオプションで highWaterMark を与えると、内部バッファの閾値を指定することができます。
単位は Byte で、デフォルト値は 16KB となっています。

highWaterMark とは河川の氾濫などの目印となる高水位線を意味する言葉です。

highWaterMark は Node.js の Stream の性能を決める重要な要素です。

先に述べたように、stream モジュールは多くのモジュールにおいて内部的に利用されています。
コアモジュールの幾つかは、highWaterMark を適した値に変更して stream を利用しています。

例えば、fs モジュールの fs.createReadStream(), fs.createWriteStream() で作られるストリームインスタンスはデフォルト値を 64KB に増やしています。

逆に process モジュールの stdin は 0B に減らして、内部バッファなし状態に設定してあります。

highWaterMark の値を変えることでどのような変化があるのでしょうか？

3. highWaterMark を変化させてみる

1MB のファイルを読み込みそのまま別ファイルに書き込む処理速度と発生する drain イベントの回数を、highWaterMark を変化させながら計測してみましょう。
drain イベントは highWaterMark まで一杯になった内部バッファが highWaterMark を下回ると発生します。

// hwm.js
const fs = require('fs');

const hwm = parseInt(process.argv[2], 10) * 1024; // highWaterMark を指定
let drain_counter = 0;
const reader = fs.createReadStream('./1M.file', {highWaterMark: hwm});
const writer = fs.createWriteStream('./out.file', {highWaterMark: hwm});

console.time('hwm'); // 計測開始
reader.pipe(writer);
writer.on('drain', () => {
  drain_counter++;
});
writer.on('finish', () => { // 書き込み終了
  console.log('drain: ' + drain_counter); // drain 回数表示
  console.timeEnd('hwm'); // 計測終了
});

highWaterMark を 1024KB, 16KB, 1KB にして計測した結果が以下となります。

$ node hwm.js 1024
drain: 0
hwm: 7.552ms

$ node hwm.js 16
drain: 62
hwm: 11.188ms

$ node hwm.js 1
drain: 1000
hwm: 63.679ms

highWaterMark を小さくするほど、drain の回数が増え、処理速度が遅くなることがわかります。
drain イベントの回数はおよそのイベントループの回数に相当します。

この違いはどのように発生するのでしょうか？

4. pipe() の実装

その答えは pipe() にあります。pipe() は src.pipe(dest) の形で呼び出すことで、読み込んだデータをそのまま受け渡すことができます。
gulp でもよく使用される関数です。

この関数の実装を簡略化したものが以下です。

Readable.prototype.pipe = function(dest) {
  var src = this;
  src.on('data', (chunk) => {
    var ret = dest.write(chunk); // 読み込んだデータを dest に書き込む
    if (false === ret) { // highWaterMark に達していたら
      src.pause(); // 読み込み一時停止
    }
  });

  dest.on('drain', () => { // highWaterMark を下回る
    flow(src); // 読み込み再開
  });
};

読み込みごとに write() の返り値が false かチェックしています。
false になると、データの流し先が一杯になった（dest の内部バッファが highWaterMark に達した）という意味なので読み込みを一時停止します。

dest の内部バッファが highWaterMark を下回ると drain イベントが発火します。これを受けとり、データ元の読み込みを再開します。

つまり、パイプ先ストリームの内部バッファが highWaterMark を大きく上回らないように調整しつつデータを流し続けるようになっているのです。

先ほどの 1MB のファイルコピーの例では、highWaterMark に達した回数、つまり読み込みを停止した回数がそのまま速度に影響していたのです。

5. バックプレッシャー

もし highWaterMark がなかったらどうなるでしょうか。

highWaterMark を無視してデータを流し続けると、パイプ先ストリームの内部バッファが増え続けてメモリをどんどん圧迫していきます。
流れるデータ量がとても多いとメモリ不足でプロセスが落ちてしまうかもしれません。

パイプの下流の圧迫具合を感じ取り、データ流量を調整するために highWaterMark が存在しているのです。

下流の流量圧だったり、それをトリガーとした流量調整機能のことをバックプレッシャーと呼びます。

Stream API のドキュメントにはこんな一文があります。

“Stream API（特に pipe 関数）の鍵となるゴールは、許容できるレベルまでバッファリングを制限することで、データソースとデータ先の速度が違っても、メモリを圧迫しないようにすることである。”

A key goal of the stream API, particularly the stream.pipe() method, is to limit the buffering of data to acceptable levels such that sources and destinations of differing speeds will not overwhelm the available memory.

これこそがバックプレッシャーであり、それをコントロールするのが highWaterMark なのです。

5.1 Node.js 以外でも見られるバックプレッシャー

上流からの流量が下流の処理速度を上回り詰まってしまうことは自然界でも見られます。
例えば、下水の排水速度よりも雨量が上回ると下水から逆流し水が吹き上がります。

工場などでは、パイプのガス圧力や水圧の上昇を防ぐために背圧弁（バックプレッシャーバルブ）が用いられています。

また、スイッチングハブではイーサネット通信の輻輳を防ぐために同様のバックプレッシャー制御が採用されています。

6. まとめ

• highWaterMark オプションでストリームインスタンスの内部バッファの閾値を指定できる
• pipe() は送り先の内部バッファが highWaterMark を超えないようにデータの流れを止めたり再開したりする
• stream モジュールは流速の異なるストリーム間で適切にデータを受け渡すための仕組みを備えている
• このような仕組みをバックプレッシャーという

7. さいごに

いかがでしたでしょうか。
Stream API が実現したいことのイメージが掴めたでしょうか。

今回はバックプレッシャーに絞っての紹介となりました。
Stream API の関数・イベント群の説明、ストリームクラスを独自に拡張する方法などはまた機会があればご紹介したいと思います。

Yahoo! JAPAN は Node.js エンジニアを募集中です！

弊社内での Node.js 活用事例はNode学園祭2016で発表した資料をご覧ください。

勉強資料

本エントリの記述は Node.js v6.9.1 を前提に記載しています。

• GitHub nodejs/node at v6.9.1
• Stream | Node.js v6.9.1 Documentation