testCluster.js

/* 
Up front scenario: run "node testCluster.js upFront"
On demand scenario: run "node testCluster.js ondemand"
*/

var cluster = require("cluster")
var os = require("os")

module.exports = {
    spawnUpFront: spawnUpFront,
    spawnOnDemand: spawnOnDemand
}

var _data = []
var _startDt
if (cluster.isWorker)
{
    return runWorker(cluster.worker)
}

if (process.argv[2] == "ondemand") {
    spawnOnDemand()
}
else {
    spawnUpFront()
}

function spawnUpFront() {
    populateData(false);
    subscribe()
    for (var i = 0; i < os.cpus().length; i++) {
        cluster.fork();
    }
}

function spawnOnDemand() {
    populateData(true);
    subscribe()
    cluster.on("exit", function () {
        if (_data.length > 0) cluster.fork()
    });
    for (var i = 0; i < os.cpus().length; i++) {
        cluster.fork();
    }
}


function populateData(killSelf) {
    _startDt = Date.now()
    for (var i = 0; i < 5000; i++) {
        _data.push(killSelf);
    }
}

function subscribe() {
    /* Fix for Node.js v6, change callback signature to "function (worker, msg)" */
    cluster.on("message", function (msg) {
        var data = _data.pop();
        if (data == undefined) {
            console.log("Elapsed: %s ms", Date.now() - _startDt)
            process.exit()
        }
        console.log(_data.length)
        cluster.workers[msg.workerid].send(data)
    })
}

function runWorker(worker) {
    worker.on("message", handleMessage);
    worker.send({workerid: worker.id})
    worker.send({workerid: worker.id})
    
    
    function handleMessage(shouldExit) {
        if (shouldExit == undefined)  process.exit()
        
        setTimeout(function () {
            if (shouldExit == true) process.exit()
            
            worker.send({workerid: worker.id})
        }, 30)
    }
}