ClusterとSocket.IO について

[ナベ]

ワタナベと申します。

現在、Node.js + Express + Socket.IO でアプリケーションを作成しているのですが、
あるメソッドで時間のかかる処理、具体的には、他のサイトのWeb APIをループで叩く処理をします。
そのため、そのメソッドが終わらないと、別のリクエストを受け付けても、処理が始まらないので、
Cluster化をしました。

通常のシングルスレッドでの起動の場合はうまく動作するのですが、
Cluster化をしたところ、Socket.IOのsocket.onが fire されなくなり、
そもそもExpressを使った場合のCluster化がおかしいのか、
Cluster化した場合のSocket.IOの書き方が間違っているのか
わからず、その点に関して教えを願いたいと思ってメールしました。

具体的には

cluster.js
----------------------------------------------
var cluster = require('cluster');
var http = require('http');
var numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
// Fork workers.
for (var i = 0; i < numCPUs; i++) {
cluster.fork();
}

cluster.on('death', function(worker) {
console.log('worker ' + worker.pid + ' died');
});
} else {
app = require('./app');
}
--------------------------------------------------

Experss のapp.jsのSocket.IO部分
-----------------------------------------------
var io = require('socket.io');
io = io.listen(app);

io.sockets.on('connection', function (socket) {

console.log('connected --- server');

　　 socket.on('send_ex_id', function(data) {
^^^^^^^^^^^^^ ここが起動しない
console.log('---------experiment_id---------------',
data.experiment_id);
// other to do
});

console.log("----------io.sockets.on", new Date());

});
-----------------------------------------------

このような形になっています。
Node V0.6.4
Express 2.5.2
Socket.IO 0.8.7

それでは、よろしくお願いします。

[Koichi Kobayashi]

小林 (koichik) です．

> あるメソッドで時間のかかる処理、具体的には、他のサイトのWeb APIをループで叩く処理をします。
> そのため、そのメソッドが終わらないと、別のリクエストを受け付けても、処理が始まらないので、

本題ではないのかもしれませんが，ここが根本的に間違ってるのでは．
というか，fs モジュールなら ～Sync() があるけれど http/net で
ブロックするのは至難の業のような気も．．．

Socket.IO や Express が Node v0.6 の cluster をサポート
しているかどうかは他の人よろしくですー．

--
{
name: "Koichi Kobayashi",
mail: "koi...@improvement.jp",
blog: "http://d.hatena.ne.jp/koichik/",
twitter: "@koichik"
}

[Tsutomu Watanabe]

小林さん、レスありがとうございます。 

本題ではないのかもしれませんが，ここが根本的に間違ってるのでは．
というか，fs モジュールなら ～Sync() があるけれど http/net で
ブロックするのは至難の業のような気も．．． 

Node.jsから他のWeb APIをたたくところは当然非同期になっていています。

ただ、DBから取得したレコード数、1000 ~ 10000に対してhttpリクエストを発行するので、

簡単に書くと

exports.merged = function(req, res) {

    Result.find({条件}, function(results) {

        for(var i = 0; i < results.length; i++) {

            http.get(options, function(req, res) {

                res.on('data', function(chunk) {

                    // chunkに対する処理

                }.on('end') {

                    // 処理

　　　　　     }

            }

        }

        res.render('/hoge');

    }

} 

これ書き方まずいでしょうか？

 

[Koichi Kobayashi]

小林 (koichik) です．

> ただ、DBから取得したレコード数、1000 ~ 10000に対してhttpリクエストを発行するので、

(略)
> これ書き方まずいでしょうか？

http.get() というより、大量のデータを一気に処理するのが
まずいのではないでしょうか。
process.nextTick() を挟むなり、それをやってくれるフロー制御
モジュールを使うなりした方がいいでしょう。

Result.find({条件}, function(results) {
var i = 0, len = results.length;
function processRow() {
for (var j = 0; i < len && j < 10; ++i, ++j) { // 10件ずつ処理
http.get(...);
}
if (i < len) {
process.nextTick(processRow);
}
}
processRow();
}

Date: Fri, 16 Dec 2011 18:37:37 +0900
From: Tsutomu Watanabe <watan...@gmail.com>
Subject: Re: [nodejs_jp:492] ClusterとSocket.IO について

--

[Tsutomu Watanabe]

小林さん、どうもです。

 

http.get() というより、大量のデータを一気に処理するのが
まずいのではないでしょうか。
process.nextTick() を挟むなり、それをやってくれるフロー制御
モジュールを使うなりした方がいいでしょう。

process.nextTick()の存在は知りませんでしたが、調べた限り、フロー制御などを使っても

結局は、そのリクエスト（例えばhttp://localhost/mergeのようなURL）に対する処理から

は抜けれないので、つまりメモリスタック上から除かれない限り、そのリクエストに対する

処理はブロックされるのではないでしょうか？

1つのリクエストの中の処理では、フロー制御なり、ノンブロッキングIOで前の処理の終了を

待たずに、処理が進んでいきますが。

今、manualを読みつつ、child_processがプロセスのUNIXのforkみたいな感じなので、

Node.jsっぽくないですが、トライしてみます。

[Koichi Kobayashi]

小林 (koichik) です．

> 結局は、そのリクエスト（例えばhttp://localhost/mergeのようなURL）に対する処理から
> は抜けれないので、つまりメモリスタック上から除かれない限り、そのリクエストに対する
> 処理はブロックされるのではないでしょうか？
> 1つのリクエストの中の処理では、フロー制御なり、ノンブロッキングIOで前の処理の終了を
> 待たずに、処理が進んでいきますが。

ちょっと意味が分かりませんが．．．
一つのリクエストに対する処理 (http.get() を呼び出す処理) が
仮に 1 ミリ秒だった場合，10000 件一気に行うと 10 秒かかり，
その間はイベントループに戻らないので他の I/O をブロック
することになりますが，10 件ごとにイベントループに戻れば
10 ミリ秒ごとに他の I/O を処理できますよね．

そういうことが可能なフロー制御モジュールもあります．
Async.js (fjacobs/async.js) とか．
# 最近 JS アドカレで紹介された Async (caolan/async) とは
# 似た名前の別物．

https://github.com/fjakobs/async.js

Result.find({条件}, function(results) {
async.list(results)
.each(function(row) {
... // http.get() など 1 件ごとの処理
})
.delay(0, 10) // 10 件ごとにイベントループに戻る
.end();
}

みたいな．

> 今、manualを読みつつ、child_processがプロセスのUNIXのforkみたいな感じなので、
> Node.jsっぽくないですが、トライしてみます。

child_process.fork() は実際はただの fork() ではなく，
fork()+exec() なので注意．
たぶん，それをトライする必要はないはずですが．

それよりも，小さな単位でイベントループに戻るように
する方が重要だと思いますよ．
クライアントサイド (ブラウザ) でも setTimeout() を
使うなどして処理を小さく分割するのと同じです．

どうせ http.get() は同じホストに対しては (デフォルトでは)
5 接続しか同時に実行しないので，それ以上一気に呼び出しても
効果はありません．

根本的な問題の解決に向かわず，Cluster を試したり
child_process.fork() にトライするというのは結局は
遠回りになるんじゃないかと思います．

[Tsutomu Watanabe]

小林さん、どうも。

結論から言うと、processTickを使っても、同じように

１つのリクエストがデータを裁いている間は、他のリクエストは

待たされました。

forkした場合は、当たり前ですが同時に複数のリクエストを裁きました。

httpを発行する場所と、res.endの場所でconsole.logで確認してみましたが、

ループでhttpを発行している間にも、res.endが検知されると、res.endの実行をしています。

> 仮に 1 ミリ秒だった場合，10000 件一気に行うと 10 秒かかり，

> その間はイベントループに戻らないので他の I/O をブロック

ここがよく分からないのですが、他のI/Oをブロックしている

とは、ループに入るところですよね？

そこがブロックされるので、他のリクエストが、そこで待たされると

いうことだと思うのですが、

小林さんが言ってるのは、ループを細切れにして、その途中で、

そのループの中に他からのリクエストのループ処理を走らすということでしょうか？

そういうことは可能なんでしょうか？、その場合の変数の値などはどうなるのでしょうか？

 

 

[Koichi Kobayashi]

小林 (koichik) です．

> 結論から言うと、processTickを使っても、同じように
> １つのリクエストがデータを裁いている間は、他のリクエストは
> 待たされました。

nextTick() の使い方が悪いのでは？
繰り返しますが，「１つのリクエストがデータを裁いている間」の
「途中」で細かくイベントループに戻ればそんなことにはなりません．

> httpを発行する場所と、res.endの場所でconsole.logで確認してみましたが、
> ループでhttpを発行している間にも、res.endが検知されると、res.endの実行をしています。

そんなはずはありません．
res.end() はイベントループに戻ってから呼び出されます．

var http = require('http');

http.createServer(function(req, res) {
res.writeHead(200);
res.end();
}).listen(3000, function() {
processRequest('Foo');
});

function processRequest(name) {
for (var i = 0; i < 10; ++i) {
(function(i) {
http.get({port: 3000}, function(res) {
res.on('end', function() {
console.log(name, i, 'response end');
});
});
console.log(name, i, 'request end');
})(i);
}
console.log('=====', name, 'processRequest end =====');
}

実行結果 (Ctrl-C で終了)

Foo 0 request end
Foo 1 request end
Foo 2 request end
Foo 3 request end
Foo 4 request end
Foo 5 request end
Foo 6 request end
Foo 7 request end
Foo 8 request end
Foo 9 request end
===== Foo processRequest end =====
Foo 0 response end
Foo 1 response end
Foo 2 response end
Foo 3 response end
Foo 4 response end
Foo 5 response end
Foo 6 response end
Foo 7 response end
Foo 8 response end
Foo 9 response end

このように，res.end() は http.get() を繰り返し呼び出している
processRequest() のループが終了するまで，決して呼び出されません．
繰り返しをどれだけ大きくしても同様です．
この例はサーバを同じプロセスで実行しているので，
ループ中にレスポンスが返ってきているわけではありませんが，
別プロセスにしても結果は同じです．

> 小林さんが言ってるのは、ループを細切れにして、その途中で、
> そのループの中に他からのリクエストのループ処理を走らすということでしょうか？

そうです．

> そういうことは可能なんでしょうか？、

イベントループに戻れば可能です．

> その場合の変数の値などはどうなるのでしょうか？

JavaScript なのでクロージャーを多用することになります．
先の例の

}).listen(3000, function() {
processRequest('Foo');
});

を

}).listen(3000, function() {
processRequest('Foo');
processRequest('Bar');
});

と変更すると，実行結果は次のようになります．

Foo 0 request end
Foo 1 request end
Foo 2 request end
Foo 3 request end
Foo 4 request end
Foo 5 request end
Foo 6 request end
Foo 7 request end
Foo 8 request end
Foo 9 request end
===== Foo processRequest end =====
Bar 0 request end
Bar 1 request end
Bar 2 request end
Bar 3 request end
Bar 4 request end
Bar 5 request end
Bar 6 request end
Bar 7 request end
Bar 8 request end
Bar 9 request end
===== Bar processRequest end =====
Foo 0 response end
Foo 1 response end
Foo 2 response end
Foo 3 response end
Foo 4 response end
Foo 5 response end
Foo 6 response end
Foo 7 response end
Foo 8 response end
Foo 9 response end
Bar 0 response end
Bar 1 response end
Bar 2 response end
Bar 3 response end
Bar 4 response end
Bar 5 response end
Bar 6 response end
Bar 7 response end
Bar 8 response end
Bar 9 response end

Foo (これが一つのリクエストだと考えてください) を処理
している間，Bar (これは別のリクエスト) は処理されません．
これが今回の問題です．

そこで processRow() を次のようにします．

function processRequest(name) {
var i = 0, len = 10;
(function processRows() {
for (var j = 0; i < len && j < 3; ++i, ++j) { // 3件ずつ処理
(function(i) {
http.get({port: 3000}, function(res) {
res.on('end', function() {
console.log(name, i, 'response end');
});
});
console.log(name, i, 'request end');
})(i);
}
if (i < len) {
process.nextTick(processRows);
}
})();
console.log('=====', name, 'processRequest end =====');
}

結果は次のようになります．

Foo 0 request end
Foo 1 request end
Foo 2 request end
===== Foo processRequest end =====
Bar 0 request end
Bar 1 request end
Bar 2 request end
===== Bar processRequest end =====
Foo 3 request end
Foo 4 request end
Foo 5 request end
Bar 3 request end
Bar 4 request end
Bar 5 request end
Foo 6 request end
Foo 7 request end
Foo 8 request end
Bar 6 request end
Bar 7 request end
Bar 8 request end
Foo 9 request end
Bar 9 request end
Foo 0 response end
Foo 1 response end
Foo 2 response end
Bar 0 response end
Bar 1 response end
Bar 2 response end
Foo 3 response end
Foo 4 response end
Foo 5 response end
Bar 3 response end
Bar 4 response end
Bar 5 response end
Foo 6 response end
Foo 7 response end
Foo 8 response end
Bar 6 response end
Bar 7 response end
Bar 8 response end
Foo 9 response end
Bar 9 response end

Foo の処理の合間に Bar の処理が挟まりました．
Foo・Bar それぞれ 10 件ずつだと，全ての GET リクエストが
完了してからレスポンスを受信しますが，len = 100 とすれば
GET リクエストが全て完了する前にレスポンスの受信も
挟まってきます．

おそらく，Node に限らずイベント駆動のプログラミングに
慣れていないのではないかと思います．
ひとまず上記のサンプルがどのような動きをするのか，
イベントループとどのように協調動作すべきなのか，
じっくり考えてみては？