count-lines.js

var DEBUG = false;
var DEFAULT_NB_WORKERS = require('os').cpus().length;

var fork = require('child_process').fork;
var fs = require('fs');
var path = require('path');

if (process.argv.length == 2) console.error('Arguments: <file> [<nb-workers>]'), process.exit(1);

var file = path.resolve(process.argv[2]);

if (typeof process.connected == 'undefined') {

  // Master

  fs.stat(file, function (err, stat) {
    if (err) throw err;

    if (stat.size == 0) throw new Error('Empty file');

    // Init workers
    var nb_workers = parseInt(process.argv[3]) || DEFAULT_NB_WORKERS;
    var workers = [];
    for (var i=0; i<nb_workers; i++) {
      workers.push((function () {
        var worker = fork(__filename, [file]);
        worker.on('exit', function () {
          console.error('Worker', worker.pid, 'exited');
        });
        return worker;
      })());
    }

    // Distribute work
    var max_sent = -1;

    // Listen to results
    var total_lf = 0;
    var total_cr = 0;
    var total_done = 0;
    workers.forEach(function (worker) {
      worker.on('message', function (message) {
        if (message.length == 5 && String(message) == 'ready') {
          // Message received as soon as worker is ready
          // Nothing specific to do, just keep going
          // A new job will be dispatched to this worker
        } else if (message.length == 0) {
          // Ignore character
          total_done++;
        } else if (message.length == 2) {
          // CR or LF
          message = message.toString();
          if (message == 'LF') {
            total_done++;
            total_lf++;
          } else if (message == 'CR') {
            total_done++;
            total_cr++;
          }
        } else {
          console.error('UNKNOWN MESSAGE', String(message));
          return;
        }
        // Finished?
        if (max_sent < stat.size - 1) {
          // Not finished yet, make this worker work again!
          worker.send(++max_sent);
          if (DEBUG) console.log('Master.Job(re)', worker.pid, max_sent);
        }
        if (total_done == stat.size) {
          // Finished
          if (total_cr == 0) {
            console.log(path.basename(file), total_lf, 'LF');
          } else if (total_lf == 0) {
            console.log(path.basename(file), total_cr, 'CR');
          } else if (total_lf == total_cr) {
            console.log(path.basename(file), total_lf, 'CRLF');
          } else {
            console.log(path.basename(file), total_cr, total_lf, '???');
          }
          workers.forEach(function (w) { w.kill() });
          process.exit(0);
        }
      });
    });

    // Final operation
    function test_finished () {
      if (DEBUG) console.log('Master.Done', total_done, '/', stat.size);
      if (total_done < stat.size) return;
    }
  });

} else {

  // Child

  process.send('ready');
  process.on('message', function (message) {
    fs.open(file, 'r', function (err, fd) {
      if (err) throw err;
      if (DEBUG) console.log('Child.Job', process.pid, String(message));
      // Read character
      var pos = parseInt(message.toString());
      fs.read(fd, new Buffer(1), 0, 1, pos, function (err, bytesRead, char) {
        if (err) throw err;
        var code = char.readUInt8(0);
        if (DEBUG) console.log('Child.Done', process.pid, pos, code, char.toString('binary'));
        if      (code == 10) process.send('LF');
        else if (code == 13) process.send('CR');
        else                 process.send('');
      });
    });
  });

}