cluster-bench.js

const cluster = require('cluster');
const Benchmark = require('benchmark');
const chance = require('chance').Chance();
const smeans = require('../src/smeans');

let suite = new Benchmark.Suite();

// Generate dataset
let data;
if (cluster.isMaster) {
  data = GenerateData(500);

  // normal(data);
  // clustering(data);
  // normal(data);
  suite
  .add('Normal', function () {
    normal(data);
  })
  .add('Clustering', function () {
    clustering(data);
  })
  .on('cycle', function(event) {
    console.log(String(event.target));
  })
  .on('complete', function() {
    console.log('Fastest is ' + this.filter('fastest').map('name'));
  })
  .run();
  // .run({ 'async': true });
}
else if (cluster.isWorker) {
  process.on('message', (cmd) => {
    if (cmd === 'shutdown') {
      console.log('Worker shutting down!');
      process.exit();
      return;
    }

    let variance = findVariance(cmd);
    // console.log('Worker has variance', variance);
    process.send(variance);

    process.exit(0);
  });
}

function GenerateData(c = 10000, d = 2) {
  let data = [];

  for (let i = 1; i <= c; i++) {
    let p = [];

    for (let j = 1; j <= d; j++) {
      p.push(chance.integer({min: 1, max: 500}));
    }

    data.push(p);
  }

  return data;
}

function normal(data) {
  let variance = findVariance(data);
  // console.log('NORMAL Variance', Math.floor(variance));
}

function clustering(data) {
  if (cluster.isMaster) {
    const numCPUs = require('os').cpus().length;

    var statusInterval = setInterval(_ => {
        // console.log('Waiting for workers...');

        if (allWorkersDead()) {
          clearInterval(statusInterval);
        }
    }, 500);

    function allWorkersDead() {
      return Object.keys(cluster.workers).length === 0;
    };

    let dataSlices = [];
    let dataSliceLen = Math.ceil(data.length / numCPUs);
    for (let i = 0; i < numCPUs; i++) {
      dataSlices[i] = data.slice(dataSliceLen * i, (dataSliceLen * i) + dataSliceLen);
    }

    // console.log(`Created ${dataSlices.length} data slices of ${dataSliceLen} length each`);

    let currentWorker = 0;
    var sendWork = function(worker) {
      // console.log(`Starting worker ${worker.id} - ${currentWorker}`);
      worker.send(dataSlices[currentWorker++]);
    };

    // Split data into <numCPUs> forks
    cluster.on('online', sendWork);

    cluster.on('exit', workerExited);

    // console.log(`Starting ${numCPUs} workers`);
    for (let i = 0; i < numCPUs; i++) {
      cluster.fork();
    }

    let variance = 0;

    function messageHandler(message) {
      variance += message;
    }

    function workerExited(worker, code, signal) {
      console.log(`Removing listener on worker ${worker.id}`);
      cluster.workers[worker.id].removeListener('message', messageHandler);
    }

    Object.keys(cluster.workers).forEach((id) => {
      cluster.workers[id].on('message', messageHandler);
    });

    process.on('exit', () => {
      cluster.removeListener('online', sendWork);
      cluster.removeListener('exit', workerExited);
    });
  }
}

function findVariance(data) {
  let variance = 0;
  for (let d of data) {
    let dist = distance(d, [0, 0]);
    variance += dist;
  }
  return variance;
}

function distance(d1, d2) {
  // Build an array of distance squares
  var tot = 0;
  for (var i = 0, l = d1.length; i < l; i++) {
    var dist = d1[i] - d2[i];
    tot += dist * dist;
  }

  return Math.floor(Math.sqrt(tot));
}