本 日 の h o g e

■ [JS][メモ] Node.js の cluster がどう動くのか

本家ドキュメント に書いてあるけれど，実際何をどうしてるのか気になったので． ドキュメントにも "This causes potentially surprising behavior" とある通りだいぶサプライジングだけど，確かに書いてある通りの挙動をする．

if (cluster.isMaster) {
    cluster.fork();
}
else {
    http.createServer(function(req, res) { ... }).listen(8080);
}

こんなコードを書いた時，一体何がどうなってマルチプロセシングになるのか．

まずは親側．cluster を見てみる．

cluster.fork = function(env) {
  ...
  return (new cluster.Worker(env));
};

Worker インスタンスを返しているだけ． じゃあ Worker のコンストラクタはどうなっているか．

function Worker(customEnv) {
  ...
  if (cluster.isMaster) {
    ...
    this.process = fork(settings.exec, settings.args, {
      'env': envCopy,
      'silent': settings.silent,
      'execArgv': settings.execArgv
    });
  } else {
  ...
}

Worker 内の fork は child_process.fork そのもの．

では子供の方はどうか．子供の方は cluster を呼ばず http を呼んでいるだけなので，ここが最大の謎． http のコードを見てみる．

util.inherits(Server, net.Server);

exports.Server = Server;

exports.createServer = function(requestListener) {
  return new Server(requestListener);
};

どうやら net を見れということらしい．listen がどうなっているのかを見る．

Server.prototype.listen = function() {
  ...
  } else if (typeof arguments[1] == 'undefined' ||
             typeof arguments[1] == 'function' ||
             typeof arguments[1] == 'number') {
    // The first argument is the port, no IP given.
    listen(self, '0.0.0.0', port, 4, backlog);
  ...
};

何か別の listen 関数を呼んでいる．これは何者か．

function listen(self, address, port, addressType, backlog, fd) {
  if (!cluster) cluster = require('cluster');

  if (cluster.isWorker) {
    cluster._getServer(self, address, port, addressType, fd, function(handle) {
      self._handle = handle;
      self._listen2(address, port, addressType, backlog, fd);
    });
  } else {
    self._listen2(address, port, addressType, backlog, fd);
  }
}

なんと net 内で cluster.isWorker を評価して振る舞いを変えている! なんだこれは!

子供なら cluster._getServer というものを呼んで self._handle を初期化するらしい． cluster._getServer ってなんじゃい．

cluster._getServer = function(tcpSelf, address, port, addressType, fd, cb) {
  ...
  // Request the fd handler from the master process
  var message = {
    cmd: 'queryServer',
    address: address,
    port: port,
    addressType: addressType,
    fd: fd
  };

  // The callback will be stored until the master has responded
  sendInternalMessage(cluster.worker, message, function(msg, handle) {
    cb(handle);
  });

};

sendInternalMessage という関数を呼んで，その callback の引数で self._handle を初期化するらしい． ここで起こしているのは 'queryServer' というメッセージ． sendInternalMessage は child.send() の wrapper のようなので割愛． 要はプロセス間通信を発生させていて，ここで親側とネゴるらしい．

では親の 'queryServer' のハンドラはどうなっているのかと言うと，

if (cluster.isMaster) {
  ...
  messageHandler.queryServer = function(message, worker, send) {

    // This sequence of information is unique to the connection
    // but not to the worker
    var args = [message.address,
                message.port,
                message.addressType,
                message.fd];
    var key = args.join(':');
    var handler;

    if (serverHandlers.hasOwnProperty(key)) {
      handler = serverHandlers[key];
    } else {
      handler = serverHandlers[key] = net._createServerHandle.apply(net, args);
    }

    // echo callback with the fd handler associated with it
    send({}, handler);
  };
...

要求に対応する handler がなければ作って，あればそれを子に返す． この send というのが何者かというと，先ほど出てきた sendInternalMessage の wrapper． つまり子に handler を送りつける．

子に戻り，net の listen をもう一度見てみると，_listen2 なるメソッドを呼んでいる．

Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
  ...
  if (!self._handle) {
    self._handle = createServerHandle(address, port, addressType, fd);
  ...

親から貰った handle で既に self._handle は初期化されているので，新しく handle を生成することなしに本来の処理を行うということのようだ．

さてこれを実際にシステムコールレベルで見てみると，こんな感じ．

$ strace -f -s 8192 -e sendmsg,recvmsg,write,read node hoge.js
...
[pid 17615] write(3, "{\"cmd\":\"NODE_CLUSTER_online\"}\n", 30) = 30
[pid 17613] recvmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"{\"cmd\":\"NODE_CLUSTER_online\"}\n", 65536}], msg_controllen=0, msg_flags=0}, 0) = 30
[pid 17615] write(3, "{\"addressType\":4,\"port\":8080,\"address\":\"0.0.0.0\",\"cmd\":\"NODE_CLUSTER_queryServer\",\"_requestEcho\":\"1:1\"}\n", 104) = 104
[pid 17613] recvmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"{\"addressType\":4,\"port\":8080,\"address\":\"0.0.0.0\",\"cmd\":\"NODE_CLUSTER_queryServer\",\"_requestEcho\":\"1:1\"}\n", 65536}], msg_controllen=0, msg_flags=0}, 0) = 104
[pid 17613] sendmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"{\"cmd\":\"NODE_HANDLE\",\"type\":\"net.Native\",\"msg\":{\"cmd\":\"NODE_CLUSTER_\",\"_queryEcho\":\"1:1\"}}\n", 91}], msg_controllen=20, {cmsg_len=20, cmsg_level=SOL_SOCKET, cmsg_type=SCM_RIGHTS, {8}}, msg_flags=0}, 0) = 91
[pid 17615] recvmsg(3, {msg_name(0)=NULL, msg_iov(1)=[{"{\"cmd\":\"NODE_HANDLE\",\"type\":\"net.Native\",\"msg\":{\"cmd\":\"NODE_CLUSTER_\",\"_queryEcho\":\"1:1\"}}\n", 65536}], msg_controllen=24, {cmsg_len=20, cmsg_level=SOL_SOCKET, cmsg_type=SCM_RIGHTS, {8}}, msg_flags=0}, 0) = 91
[pid 17615] write(3, "{\"addressType\":4,\"port\":8080,\"address\":\"0.0.0.0\",\"cmd\":\"NODE_CLUSTER_listening\"}\n", 81) = 81
[pid 17613] recvmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"{\"addressType\":4,\"port\":8080,\"address\":\"0.0.0.0\",\"cmd\":\"NODE_CLUSTER_listening\"}\n", 65536}], msg_controllen=0, msg_flags=0}, 0) = 81

ここでは pid 17615 が子，pid 17613 が親となっている．