ストリーミング処理の落とし穴

Mongoose(MongoDBのODM)を使って、コレクションからほんの10万ほどのドキュメントを処理するスクリプトを実行した時でした。処理が固まってしまうのです。なぜでしょう？

各ドキュメント処理では、コールバックを伴う非同期的な処理を含みます。
問題のCoffeeScriptのコードは以下のようなイメージです：

Foo.find()
    .sort value:-1
    .stream()
    .on 'data', (d)->
        someTask d

someTask = (d)->
    doAwesomeJob (result)->
        console.log "Great!"

モデルFooのドキュメントを全て読み込み、各ドキュメントをストリームに流し込んでいます。
メモリを節約するためにストリームの手法を採用しています。
doAwesomeJobは、例えばAPIを叩くとか、そういう非同期な処理です。
しかし、doAwesomeJobのコールバック関数がなかなか呼ばれない。遅い。ついには全く呼ばれなくなる。

イベントループのキューが溢れている

Node.jsアプリケーションはシングルスレッドです。
下図のように、イベントループ機構が備わっていて、Event Queueに溜まった処理キューを順次消化していく仕組みになっています。

(via http://misclassblog.com/interactive-web-development/node-js/)

つまり、Node.jsはひとつずつしかキューを処理できないのです。
上記スクリプトで、ストリームのイベントのInvocationが、大量にこのイベントキューに押し寄せている事を想像してください。
doAwesomeJobのコールバックは、その非同期処理の完了時点で最後列にエンキューされます。
しかし、非同期処理の間にドキュメント読み出しイベントの処理キューが大量に溜まります。
その結果、doAwesomeJobのコールバックが呼ばれるまでに時間がかかってしまっているのでした。

似た問題に直面している人がいました。

• rabbitmq flooding nodejs app

RabbitMQのメッセージキューがいちどに大量に来た時に、イベントループのキューが溢れるようです。
問題は僕のケースより少し深刻なようです。見なかったことにしましょう。

ストリームを分割しよう

溢れるのなら、分割すればいいですね。
以下のように変更を加えました：

processNext = (offset, num, on_completion)->
    i = 0, j = 0, closed = false
    Foo.find()
        .sort value:-1
        .skip offset
        .limit num
        .stream()
        .on 'data', (d)->
            ++j
            someTask d, ->
                ++i
                if i==j && closed
                    processNext offset+i, num, on_completion
        .on 'close', ->
            closed = true
            on_completion() if j==0

someTask = (d, callback)->
    doAwesomeJob (result)->
        console.log "Great!"
        callback()

processNext 0, 100, ->
    console.log "done!"

ドキュメントを100個ずつ読み出して、処理して、終わったら次の100個・・というフローに変更しました。
今のところ、これで上手く動いています。

でもこのコードは少し汚いので、よりエレガントに書くためにasyncを使うことをお薦めします。
drainで完了をフックするといいと思います。

このストリームの分割では、ストリームを処理している間にドキュメントが追加・削除される事を想定していません。
コレクションの変更まで考慮する場合は、想定される変更内容によって変わると思います。

ご参考まで！