Node.js cluster + collective

I have submitted the issue to the github repo, so as to track it there!

I'm running a clustered app that could be on a machine with N cores. Let's say I am running 2 of the app instances locally for testing, really emulating 2 different boxes. So N cores on N machines using the cluster module (in reality, the N machines is static, e.g. just 2 behind an AWS Load Balancer).

  1. How do I properly configure the collective.js "all_hosts" option for this? Would I use process.id somehow along with IP?

Running the code snippets would be something along the lines of 2 bash terminals:

terminal 1:

coffee cluster1

terminal 2:

coffee cluster2

Note: the code below works, but doesn't really work, as I can't quite figure out the configuration; each time I log data it's specific to the process.

cluster1.coffee:

cluster = require 'cluster'
numCPUs = require('os').cpus().length

if cluster.isMaster

  i = 0 
  cluster.setupMaster 
    exec: './server1'

  console.log "App 1 clustering with: #{numCPUs} clusters"

  while i < numCPUs
    cluster.fork()
    i++

  cluster.on 'fork', (worker) ->
    console.log 'Forked App 1 server worker ' + worker.process.pid

server1.coffee:

Collective = require 'collective'

all_hosts = [
    host: 'localhost', port: 8124 # Wrong
]

collective = new Collective(
  host: 'localhost'
  port: 8124
, all_hosts, (collective) ->

)

collectiveUpsert = () ->

  num = Math.floor((Math.random()*10000)+1)

  data = 
    num: num

  console.log process.pid + ' sees current num as: ' + JSON.stringify(collective.get('foo.bar'))
  console.log process.pid + ' setting num to: ' + JSON.stringify(data)

  collective.set 'foo.bar', data

setInterval (->
  collectiveUpsert()
), 5 * 1000

cluster2.coffee:

cluster = require 'cluster'
numCPUs = require('os').cpus().length

if cluster.isMaster

  i = 0 
  cluster.setupMaster 
    exec: './server2'

  console.log "App 2 clustering with: #{numCPUs} clusters"

  while i < numCPUs
    cluster.fork()
    i++

  cluster.on 'fork', (worker) ->
    console.log 'Forked App 2 server worker ' + worker.process.pid

server2.coffee:

Collective = require 'collective'

all_hosts = [
    host: 'localhost', port: 8124 # Wrong
]

collective = new Collective(
  host: 'localhost'
  port: 8124
, all_hosts, (collective) ->

)

collectiveUpsert = () ->

  num = Math.floor((Math.random()*10000)+1)

  data = 
    num: num

  console.log process.pid + ' sees current num as: ' + JSON.stringify(collective.get('foo.bar'))
  console.log process.pid + ' setting num to: ' + JSON.stringify(data)

  collective.set 'foo.bar', data

setInterval (->
  collectiveUpsert()
), 5 * 1000

In order to use collective.js with cluster and/or multiple servers you need to start it on every Node.js child process. Think of it as a http module, where you have to create the listener on every child/slave, not the master (http://nodejs.org/api/cluster.html#cluster_cluster). Following similar logic, for collective.js, you should do something like this (single server):

if (cluster.isMaster) {
    // fork n children
} else {
    var current_host = {host: "localhost", port: 10000};
    current_host.port += cluster.worker.id; // this is incremented for every new process.

    var all_hosts = [
        {"host": "localhost", "port": 10001},
        {"host": "localhost", "port": 10002},
        {"host": "localhost", "port": 10003},
        {"host": "localhost", "port": 10004},
        {"host": "localhost", "port": 10005},
        {"host": "localhost", "port": 10006}
        // must be the same amount as is the child process count.
    ];

    var collective = new modules.collective(current_host, all_hosts, function (collective) {
        // Do your usual stuff. Start http listener, etc...
    });
}

You should modify localhost to your ip addresses and make sure ports increment properly, if you want to use this on different servers.

For any additional information you can check crude tests at test/index.js

Hope that helps! If you need any further assistance - please ask.

P.S. Admittedly, this way is to cumbersome and needs clearer explanation. I hope to figure out a cleaner and easier initialization process in the near future. In addition to that, clarify the readme and provide some full examples.