multilevel-cluster

gistfile1.js

var cluster = require('cluster');
var cpus = require('os').cpus().length;
var levelup = require('levelup');
var multilevel = require('multilevel');
var net = require('net');

if (cluster.isMaster) {
  var db = levelup(__dirname + '/db');

  for (var i = 0; i < cpus; i++) {
    cluster.fork();
  }

  cluster.on('exit', function (worker, code, signal) {
    console.log('worker %s died', worker.process.pid);
    cluster.fork();
  });

  Object.keys(cluster.workers).forEach(function (id) {
    cluster.workers[id].on('message', onMessage(cluster.workers[id]));
  });

  function onMessage (worker) {
    return function (msg) {
      //console.log('master', JSON.parse(msg));
      var op = msg;
      var method = op[0];
      var args = op[1];
      var id = op[2];

      args.push(function cb (err, va) {
        args = [].slice.call(arguments);
        worker.send([id, args]);
      });

      db[op[0]].apply(db, args);
    }
  };

  return;
}

var callbacks = {};
var nextId = 0;

var db = {
  get : function (key, cb) {
    //console.log('client get', arguments);
    process.send(['get', [key], nextId]);
    callbacks[nextId] = cb;
    nextId++;
  },
  put : function (key, value, cb) {
    //console.log('client put', arguments);
    process.send(['put', [key, value], nextId]);
    callbacks[nextId] = cb;
    nextId++;
  }
}

process.on('message', function (msg) {
  var res = msg;
  var id = res[0];
  var args = res[1];

  callbacks[id].apply(null, args);
  delete callbacks[id];
});

var server = net.createServer(function (con) {
  con.pipe(multilevel.server(db)).pipe(con);
});

server.listen(function () {
  console.log('listening on port %s', server.address().port);
});