Using a server cluster in an Express.js app

Question

I'm doing a little OJT on my first node project and, while I can stand up a simple server, the app is going to get hammered so using cluster seems like a good idea. I've cobbled together some code snippets that I've found in various searches (including SO), but the server won't start. I'm sure my inexperience with Node has me doing something stupid, but I don't see it.

var express = require( 'express' );
var cluster = require( 'cluster' );
var path    = require( 'path' );

var cCPUs   = require( 'os' ).cpus().length;
var port    = 3000;
var root    = path.dirname( __dirname );

if( cluster.isMaster ) {
    for( var i = 0; i < cCPUs; i++ ) {
      cluster.fork();
    }

    cluster.on( 'death', function( worker ) {
      console.log( 'Worker ' + worker.pid + ' died.' );
    });
}
else {
    // eyes.inspect( process.env );
    console.log( 'Worker: %s', process.env.NODE_WORKER_ID );

    var app = express();
    var routes  = require( './routes' )( app );
    app
      .use( cluster.repl( root + 'cluster.repl' ) )
      .use( cluster.stats({ connections: true, requests: true }) )
      .use( cluster.reload( root ) )
      .listen( port );
}

RESULT:

TypeError: Object #<Cluster> has no method 'repl'

If I remove the use calls, the workers start up correctly, but process.env.NODE_WORKER_ID is undefined. Inspecting process.env shows me that it's definitely not defined. Maybe the snippet I used was from an old version, but I'm not sure how to identify the worker thread in any other way.

How can I unscramble this?

Lop

The repl, stats and reload methods you're calling don't exist on cluster. Start with the canonical example from the cluster docs and go from there instead. – Selfabsorption

Hmmm. Did the cluster module get added to core sometime before 0.8.16 (the version I'm using)? Maybe I've just been looking in exactly the wrong places all along. Thanks. – Lop

I think cluster has been around since 0.6, but it was reworked a bit in 0.8. – Selfabsorption

Cluster had major changes – Inflationary

Answer 1

For anyone searching later, here's what I ended up with:

const cluster = require('cluster');
const express = require('express');
const path = require('path');

const port = 3000;
const root = path.dirname(__dirname);
const cCPUs = require('os').cpus().length;

if (cluster.isMaster) {
    // Create a worker for each CPU
    for (let i = 0; i < cCPUs; i++) {
        cluster.fork();
    }
    cluster.on('online', function (worker) {
        console.log('Worker ' + worker.process.pid + ' is online.');
    });
    cluster.on('exit', function (worker, code, signal) {
        console.log('worker ' + worker.process.pid + ' died.');
    });
} else {
    const app = express();
    const routes = require('./routes')(app);
    app.use(express.bodyParser()).listen(port);
}

I'm still very early in the node learning curve, but the server starts and appears to have a working running on each core. Thanks to JohnnyH for getting me on the right track.

Lop

can i use forky instead of cluster if yes then can u please give me a example for that , it would be very helpfull... – Goldman

Hi Rob, Do you use any Websocket tools like Socket.io or SocketJS or PubSub? Have you seen any duplicated logs or messages or odd behavior in connecting or disconnecting users? Tnx. – Necking

@Necking - No, I haven't needed websockets for anything I'm doing. Sorry. – Lop

Answer 2

Also take a look at cluster2. It's used by eBay and has an express example

var Cluster = require('cluster2'),
    express = require('express');

var app = express.createServer();

app.get('/', function(req, res) {
  res.send('hello');
});

var c = new Cluster({
  port: 3000,
});

c.listen(function(cb) {
  cb(app);
});

Ciliolate

Wont work for me. running on windows 7, the 'require' call returns a json object and when I try to 'new Cluster' I receive 'object #Object is not a function'. – Carolyncarolyne

Answer 3

Here is my draft of Cluster.js class. Note that we should catch port conflict when you start master process.

/*jslint indent: 2, node: true, nomen: true, vars: true */

'use strict';

module.exports = function Cluster(options, resources, logger) {
  var start = function () {
    var cluster = require('cluster');

    if (cluster.isMaster) {
      require('portscanner').checkPortStatus(options.express.port, '127.0.0.1', function (error, status) {
        if (status === 'open') {
          logger.log.error('Master server failed to start on port %d due to port conflict', options.express.port);
          process.exit(1);
        }
      });

      // Each core to run a single process.
      // Running more than one process in a core does not add to the performance.
      require('os').cpus().forEach(function () {
        cluster.fork();
      });

      cluster.on('exit', function (worker, code, signal) {
        logger.log.warn('Worker server died (ID: %d, PID: %d)', worker.id, worker.process.pid);
        cluster.fork();
      });
    } else if (cluster.isWorker) {
      var _ = require('underscore');
      var express = require('express');
      var resource = require('express-resource');

      // Init App

      var app = express();

      // App Property

      app.set('port', process.env.PORT || options.express.port);
      app.set('views', options.viewPath);
      app.set('view engine', 'jade');
      app.set('case sensitive routing', true);
      app.set('strict routing', false);

      // App Middleware

      app.use(express.favicon(options.faviconPath));
      app.use(express.logger({ stream: logger.stream() }));
      app.use(express.bodyParser());
      app.use(express.methodOverride());
      app.use(express.responseTime());
      app.use(app.router);
      app.use(require('stylus').middleware(options.publicPath));
      app.use(express['static'](options.publicPath));

      if (options.express.displayError) {
        app.use(express.errorHandler());
      }

      // App Format

      app.locals.pretty = options.express.prettyHTML;

      // App Route Handler

      if (!_.isUndefined(resources) && _.isArray(resources)) {
        _.each(resources, function (item) {
          if (!_.isUndefined(item.name) && !_.isUndefined(item.path)) {
            app.resource(item.name, require(item.path));
          }
        });
      }

      // Start Server

      var domain = require('domain').create();

      domain.run(function () {
        require('http').createServer(app).listen(app.get('port'), function () {
          logger.log.info('Worker server started on port %d (ID: %d, PID: %d)', app.get('port'), cluster.worker.id, cluster.worker.process.pid);
        });
      });

      domain.on('error', function (error) {
        logger.log.error(error.stack);
      });
    }
  };

  return {
    start: start
  };
};

Lucy

I think this is a better answer. 1) It modularizes the needed use case into an NPM module. 2) The integrated use of the Domain class. One addition I would make is to fire up a new worker on the domain error event. – Dominik