master.js

#!/usr/bin/env node

//////////////////////////////////////////////////////////////////////////////////////////////////////
//
// TO START IN CONSOLE: `./scripts/actionHeroCluster`
// TO DAMEONIZE: `forever start scripts/actionHeroCluster`
//
// ** Producton-ready actionHero cluster example **
// - workers which die will be restarted
// - maser/manager specific logging
// - pidfile for master
// - USR2 restarts (graceful reload of workers while handling requets)
//   -- Note, socket/websocket clients will be disconnected, but there will always be a worker to handle them
//   -- HTTP, HTTPS, and TCP clients will be allowed to finish the action they are working on before the server goes down
// - TTOU and TTIN signals to subtract/add workers
// - WINCH to stop all workers
// - TCP, HTTP(s), and Web-socket clients will all be shared across the cluster
// - Can be run as a daemon or in-console
//   -- Lazy Dameon: `nohup ./scripts/actionHeroCluster &`
//   -- you may want to explore `forever` as a dameonizing option
//
// * Setting process titles does not work on windows or OSX
//
// This example was heavily inspired by Ruby Unicorns [[ http://unicorn.bogomips.org/ ]]
//
//////////////////////////////////////////////////////////////////////////////////////////////////////

//////////////
// Includes //
//////////////

var fs = require('fs');
var cluster = require('cluster');
var colors = require('colors');

var numCPUs = require('os').cpus().length
var numWorkers = numCPUs - 2;
if (numWorkers < 2){ numWorkers = 2};

////////////
// config //
////////////

var config = {
  // script for workers to run (You probably will be changing this)
  exec: __dirname + "/actionHero",
  workers: numWorkers,
  pidfile: "./cluster_pidfile",
  log: process.cwd() + "/log/cluster.log",
  title: "actionHero-master",
  workerTitlePrefix: " actionHero-worker",
  silent: true, // don't pass stdout/err to the master
};

/////////
// Log //
/////////

var logHandle = fs.createWriteStream(config.log, {flags:"a"});
var log = function(msg, col){

var sqlDateTime = function(time){
  if(time == null){ time = new Date(); }
  var dateStr =
  padDateDoubleStr(time.getFullYear()) +
  "-" + padDateDoubleStr(1 + time.getMonth()) +
  "-" + padDateDoubleStr(time.getDate()) +
  " " + padDateDoubleStr(time.getHours()) +
  ":" + padDateDoubleStr(time.getMinutes()) +
  ":" + padDateDoubleStr(time.getSeconds());
  return dateStr;
};

var padDateDoubleStr = function(i){
  return (i < 10) ? "0" + i : "" + i;   };  
  msg = sqlDateTime() + " | " + msg;  
  logHandle.write(msg + "\r\n");  
  if(typeof col == "string"){col = [col];}  
  for(var i in col){    
    msg = colors[col[i]](msg);  
  }   
  console.log(msg); 
} 

////////// // Main // ////////// 
log(" - STARTING CLUSTER -", ["bold", "green"]); 
// set pidFile 
if(config.pidfile != null){  
  fs.writeFileSync(config.pidfile, process.pid.toString()); 
} 
process.stdin.resume(); 
process.title = config.title; 
var workerRestartArray = []; 

// used to trask rolling restarts of workers 
var workersExpected = 0; 

// signals 
process.on('SIGINT', function(){  
  log("Signal: SIGINT"); 
  workersExpected = 0;  
  setupShutdown(); 
}); 

process.on('SIGTERM', function(){  
  log("Signal: SIGTERM");   
  workersExpected = 0;  
  setupShutdown(); 
}); 

process.on('SIGKILL', function(){
  log("Signal: SIGKILL");   
  workersExpected = 0;  
  setupShutdown();
}); 

process.on('SIGUSR2', function(){  
  log("Signal: SIGUSR2");   
  log("swap out new workers one-by-one");   
  workerRestartArray = []; 
  for(var i in cluster.workers){    
    workerRestartArray.push(cluster.workers[i]);  
  }   
  reloadAWorker(); 
}); 

process.on('SIGHUP', function(){   
  log("Signal: SIGHUP");  
  log("reload all workers now");  
  for (var i in cluster.workers){    
    var worker = cluster.workers[i];   
    worker.send("restart");   
  } 
}); 

process.on('SIGWINCH', function(){  
  log("Signal: SIGWINCH");  
  log("stop all workers");  
  workersExpected = 0;  
  for (var i in cluster.workers){     
    var worker = cluster.workers[i];    
    worker.send("stop");  
  } 
}); 

process.on('SIGTTIN', function(){   
  log("Signal: SIGTTIN");   
  log("add a worker");  
  workersExpected++;  
  startAWorker(); 
}); 

process.on('SIGTTOU', function(){   
  log("Signal: SIGTTOU");   
  log("remove a worker");   
  workersExpected--;  
  for (var i in cluster.workers){     
    var worker = cluster.workers[i];    
    worker.send("stop");    
    break;  
  } 
}); 

process.on("exit", function(){  
  workersExpected = 0;  
  log("Bye!") 
}); 

// signal helpers 
var startAWorker = function(){  
  worker = cluster.fork();  
  log("starting worker #" + worker.id);  
  worker.on('message', function(message){     
    if(worker.state != "none"){       
      log("Message ["+worker.process.pid+"]: " + message);
    }  
  });
} 

var setupShutdown = function(){   
  log("Cluster manager quitting", "red");   
  log("Stopping each worker...");   
  for(var i in cluster.workers){    
    cluster.workers[i].send('stop');  
  }   
  setTimeout(loopUntilNoWorkers, 1000); 
} 

var loopUntilNoWorkers = function(){  
  if(cluster.workers.length > 0){
    log("there are still " + cluster.workers.length + " workers...");
    setTimeout(loopUntilNoWorkers, 1000);
  }else{
    log("all workers gone");
    if(config.pidfile != null){
      fs.unlinkSync(config.pidfile);
    }
    process.exit();
  }
}

var reloadAWorker = function(next){
  var count = 0;
  for (var i in cluster.workers){ count++; }
  if(workersExpected > count){
    startAWorker();
  }
  if(workerRestartArray.length > 0){
    var worker = workerRestartArray.pop();
    worker.send("stop");
  }
}

// Fork it.
cluster.setupMaster({
  exec : config.exec,
  args : process.argv.slice(2),
  silent : config.silent
});
for (var i = 0; i < config.workers; i++) {
  workersExpected++;
  startAWorker();
}
cluster.on('fork', function(worker) {
  log("worker " + worker.process.pid + " (#"+worker.id+") has spawned");
});
cluster.on('listening', function(worker, address) {

});
cluster.on('exit', function(worker, code, signal) {
  log("worker " + worker.process.pid + " (#"+worker.id+") has exited");
  setTimeout(reloadAWorker, 1000) // to prevent CPU-splsions if crashing too fast
});