JavaScriptのStreams API、もっと使いやすくできるはず！「より良い設計」を考える

現在のJavaScriptにおけるStreams APIは、より直感的でパワフルな設計が可能であるという提案です。既存のAPIの使いにくさを解消し、エンジニアにとってより扱いやすい「理想的なストリーム操作」の実現に向けた議論が交わされています。

たまたまなんだけど、この記事が提案しているのよりもっと良いAPIを思いついちゃった！

彼らは単にUInt8Arrayのasyncイテレータを使うことを提案してる。このアイデア、嫌いじゃないんだけど、ちょっと惜しいんだよね。

彼らの提案はこれ：

  type Stream<T> = {
    next(): Promise<{ done, value: UInt8Array<T> }>
  }

俺の提案はこれ。「ストリーム・イテレータ」って呼んでる！

  type Stream<T> = {
    next(): { done, value: T } | Promise<{ done, value: T }>
  }

自画自賛になっちゃうけど、客観的に見ても俺のバージョンのほうが優れてると思うんだよね：

• 彼らの実装から俺のを作るのは簡単。

• 彼らのは「ストリーム」の概念がイテレータのイテレータで定義されてるから、中身を見るのにforループの二重奏が必要になる。俺のはただのイテレータだから、forループ一つで済むんだ。

• 彼らは整数（integer）のストリームに限定されてるけど、俺のはそうじゃない。

• 俺のやり方だと、同期入力に対して同期変換を定義すれば、イテレーション全体を同期にできる。だから同期関数の中でも結果を使えるんだ。これってめちゃくちゃ重要で、そうじゃないと同期用（forループ）と非同期用（for awaitループ）でコードを二回書く羽目になるからね。

• 入力を単語に分割する時にPromiseを乱発する問題も解決する。asyncイテレータだと、2つの単語を作るのにPromiseが2つ必要になる。ストリーム・イテレータなら、データがあるならPromiseは不要で、そのままyieldすればいい。

• ストリーム・イテレータは並行処理（concurrency）の管理にも役立つ。これはasyncイテレータには不可能な超重要なポイント。asyncイテレータはPromiseを見つけると「絶対に」待機しちゃうからね。それってつまり「並行処理があっても、常にそれが潰されちゃう」ってことなんだ。

ずっと前に、Repeaterっていう抽象化レイヤーを書いたんだ。基本的には「Promiseコンストラクタをasync iterableに変換したらどうなるか」っていうアイデアが元になってる。

  import { Repeater } from "@repeaterjs/repeater";
  
  const keys = new Repeater(async (push, stop) => {
    const listener = (ev) => {
      if (ev.key === "Escape") {
        stop();
      } else {
        push(ev.key);
      }
    };
    window.addEventListener("keyup", listener);
    await stop;
    window.removeEventListener("keyup", listener);
  });
  const konami = ["ArrowUp", "ArrowUp", "ArrowDown", "ArrowDown", "ArrowLeft", "ArrowRight", "ArrowLeft", "ArrowRight", "b", "a"];
  (async function() {
    let i = 0;
    for await (const key of keys) {
      if (key === konami[i]) {
        i++;
      } else {
        i = 0;
      }
      if (i >= konami.length) {
        console.log("KONAMI!!!");
        break; // keyupリスナーを削除する
      }
    }
  })();

https://github.com/repeaterjs/repeater

機能的には完成してて安定してるライブラリの一つなんだけど、NPMを見たらなぜか週に650万回以上もダウンロードされてるみたい。

最近の俺は著者とは逆の意見で、特に fetch とかの提案に組み込まれてることを考えると、素直にStreamsを使うべきじゃないかと思ってたんだ。でも、あの tee に関する批判はかなり効くね。もしかしたら著者が正しいのかも。みんながまだこの問題について考えてるのを見るのはワクワクするよ。デフォルトの抽象化としては、やっぱりasync iterableがいいんだろうな。

Async iterableも、Promiseやスタックスイッチのオーバーヘッドがあるから、必ずしも万能な解決策ってわけじゃないんだ。同期イテレータと比べるとその差はかなり大きい。

SSRの最中にタグ名、属性、バインディングみたいな細かいオブジェクトを生成側で扱うなら、各文字列をそのまま write() するのが自然だよね。でもそうすると同期イテレータに比べてパフォーマンスがボロボロになる。そこで選択を迫られるんだ：

1. バッファリングして大きなチャンクを作り、スタックスイッチを減らす。これは結局Streamsでやらなきゃいけないことと同じ。
2. 同期イテレータを使い、非同期コンポーネントのサポートを諦める。

この記事では同期ストリームでこれを回避しようとしてるけど、データ探索の中でどこで非同期処理が発生するか事前には分からないって問題がある。非同期コンポーネントにぶつかった時に初めて必要になるんだよね。本当に欲しいのは、非同期が必要なデータ「だけ」を非同期にできる方法なんだ。

Lit-SSRでこの問題に直面した時の俺たちの解決策は、サンク（thunk）を含めることができる同期イテレータに移行することだった。生成側で非同期処理が必要ならサンクを送り、消費側はサンクを受け取ったら次の値を取得する前にそれを呼び出してawaitしなきゃいけない。もし消費側が非同期をサポートしてない（同期の renderToString() とか）なら、サンクを受け取った時点でエラーを投げればいい。

これで、実際のWebサイトから抽出したコンポーネントのSSRベンチマークで12〜18倍の高速化を実現できたよ。

Streams APIにこんな壊れやすい規約（ next() を早く呼びすぎると壊れる、みたいな）を採用するのは無理だと思うけど、消費側が1つのマイクロタスクで可能な限り多くの値をプルして、非同期な値に遭遇した時だけawaitするような仕組みがあれば、すごく価値があると思う。 write() と writeAsync() みたいな感じでね。

残念なのは、木構造のデータを扱うストリーミングAPIにはジェネレータが本来最適なんだけど、ジェネレータはとにかく遅すぎるんだよね。

Node.jsでWeb Streamsを使う際の実用的な悩みは、あれがブラウザ用途を最優先に設計されて、サーバーに後付けされたように感じることだね。大きなファイルを処理したりサービス間でデータをパイプしたりするたびに、仕事をこなすことよりAPIと格闘することに時間を取られちゃう。

async iterableのアプローチの方がずっと理にかなってるよ。 for-await-of で自然に構成できるし、他のasync/awaitエコシステムとも相性がいい。今のWeb Streams APIには変なインピーダンス不整合があって、単純な操作をしたいだけなのに全部トランスフォーム・ストリームでラップしなきゃいけなかったりするしね。

Nodeのオリジナルのストリーム実装も問題はあったけど、少なくとも .pipe() は直感的だった。スペックを読み込まなくても、操作を繋げたりバックプレッシャーについて考えたりできたから。Web Streamsのスペックは、複雑な問題の解決策はいつだって「さらなる抽象化」だと信じてる人が書いたような気がするよ。

JavaのOKIOの設計にかなり近い気がするね。最終的な目標も似ているし。内部の詳細や設計上の決定事項についてのプレゼンがここにあるよ。

このパターンによって、undici（Node.js内蔵のfetch実装）を使用しているNode.jsアプリケーションでコネクションプールの枯渇が発生しており、他のランタイムでも同様の問題が見られます。

これってガベージコレクションがある言語特有の欠陥だよね。リソースを明示的に閉じなきゃいけないのは、まるでCを書いてる気分だよ。そうしないとメモリリークやリソース枯渇が起きる。GCがいつリソースを解放してくれるか分からないからね。リファレンスカウントを採用してるC++の方が、この点に関してはまだマシだよ。

バリューストリーム（値のストリーム）が便利なユースケースはたくさんあるよね。ただ、それとは別に、よりシンプルなバイト専用のストリームがあった方がいいっていうのは同意。今のWeb Streamsの機能は維持しつつ、バイトストリームを最適化するためにIOStreamを追加するのがいいんじゃないかな。

理想を言えば、ユースケースを切り分けることで両方の実装をシンプルにできるんだろうけど、もう手遅れ（時すでに遅し）って感じもするね。

数ヶ月前、ネイティブのストリームが最悪な挙動をしてパフォーマンスの問題にぶつかったんだけど、どうやらバックプレッシャーの実装がまずかったみたいなんだ。

色んな実装を試したり設定をいじったりしたけど、結局解決できなかった。消費側のキャパシティに余裕があるのに、なぜか処理がガクッと落ちる不気味な現象が起きたりしてね。

彼らが言及してる別の問題、つまりPromiseのコストも関係してたのかも。俺のストリームは山ほどPromiseを生成してたから。大量のデータを扱うとき、そのコストはバカにならないんだ。

最終的に、バッチ処理でPromiseの数を減らす複雑なロジックを組んで、手動でバックプレッシャーを管理する賢い並行処理戦略を考えなきゃいけなかった。結果的にはうまくいったけどね。

満足いくものができたところで、DenoからGoに移植してみたんだけど、結果が違いすぎて衝撃を受けたよ。パフォーマンスが数桁レベルで向上したんだ。

あと、Effectライブラリを使ってカスタム/ネイティブなソリューションも作ってみた。非効率で遅いって言う人もいるけど、特に追加のチューニングもせずそのままの状態で、俺の自作コードより15%くらい速かったんだよね。最初からこれを使えばよかったよ。

この違いは、おそらく実行レイヤーでPromiseじゃなくてファイバーベースのモデルを使ってるからだと思うけど、詳しくは分からないな。

SocketClusterは、少なくとも2019年からこういうストリームの実装とバックプレッシャー管理を取り入れてるよ。

これがWritableConsumableStreamモジュール：
https://github.com/SocketCluster/writable-consumable-stream

SocketClusterは、非同期処理でメッセージの順序を維持するっていう問題を解決してるんだ。

この機能は今のLLMブームでさらに役立つよ。メッセージ順をバラバラにするリスクなく、AIでストリームをリアルタイムに変換できるからね。

バックプレッシャー付きで for-await-of ループをこんな風に使ったのは、少なくともオープンソースプロジェクトでは俺が最初だったんじゃないかな。

自分は.NETでしか使ったことがないけど、実践的な面でもAPI設計の抽象的な観点でも、ストリームの使用に関する強い意見があればぜひ読んでみたいね。