<commit_before>var _ = require('underscore'),
  cp = require('child_process'),
  EventEmitter = require('events').EventEmitter,
  util = require('util');

var numCPUs = require('os').cpus().length,
  queue = [],
  activeTasks = 0,
  totalWorkers = 0,
  isBackground;

exports = module.exports = function (module, options) {
  EventEmitter.call(this);

  this.module = module;
  this.options = _.extend({
    workers: numCPUs,
    keepAlive: 500
  }, options);
  this.workers = [];
  this.totalWorkers = 0;
  this.cullTimeout = undefined;
};
util.inherits(exports, EventEmitter);

exports.prototype.sendAll = function (message) {
  _.each(this.workers, function (worker) {
    worker.process.send(message);
  });
};
exports.prototype.send = function (message, callback) {
  var self = this;

  callback = callback || function () {};

  function fork() {
    totalWorkers++;
    self.totalWorkers++;

    var execArgv = process.execArgv;
    if (_.contains(execArgv, '--prof')) {
      execArgv = _.clone(execArgv);

      var logFile = _.find(execArgv, function (arg) {
          return /^--logfile=.*/.test(arg);
        }) || '--logfile=v8.log',
        location = execArgv.indexOf(logFile);

      if (self.options.logId) {
        logFile = logFile + '.' + self.options.logId + '.' + self.totalWorkers;
      } else {
        logFile = logFile + '.' + totalWorkers;
      }
      if (location >= 0) {
        execArgv.splice(location, 1, logFile);
      } else {
        execArgv.push(logFile);
      }
    }

    // Node 0.11 supports passing the execArgv option to fork but we don't want to break compatibility
    // so we're going to emulate the fork logic in the mean time.
    worker = {
      callback: false,
      process: cp.spawn(
        process.execPath,
        execArgv.concat([__dirname + '/worker'], execArgv), {
          stdio: ['pipe', 'pipe', 'pipe', 'ipc']
        })
    };
    if (self.options.stdout) {
      worker.process.stdout.on('data', self.options.stdout);
    }
    worker.process.send(self.module);

    function processMessage(msg) {


      var callback = worker.callback;
      worker.callback = undefined;

      if (queue.length) {
        var queued = queue.shift();
        queued.pool.send(queued.message, queued.callback);
      }

      // Dispatch
      process.nextTick(function () {
        var err;
        if (msg.err) {
          err = new Error(msg.err);
          if (msg.stack) {
            err.stack = msg.stack;
          }
        }

        // Kill the worker as we don't know what state it might be in
        if (msg.fatal) {
          worker.process.removeAllListeners('exit');
          worker.process.kill();
          self.workers = _.without(self.workers, worker);
        }
        activeTasks--;
        if (!callback) {
          self.emit('error', err || new Error('Out of band data: ' + msg.data));
        } else {
          callback(err, msg.data);
        }

      });
    }

    worker.process.on('exit', function (code, signal) {
      if (code === null) {
        //process wasn't killed properly
        var error = new Error('Worker exited unexpectedly');
        processMessage({<commit_msg>add util.inspect for out of band<commit_after>var _ = require('underscore'),
  cp = require('child_process'),
  EventEmitter = require('events').EventEmitter,
  util = require('util');

var numCPUs = require('os').cpus().length,
  queue = [],
  activeTasks = 0,
  totalWorkers = 0,
  isBackground;

exports = module.exports = function (module, options) {
  EventEmitter.call(this);

  this.module = module;
  this.options = _.extend({
    workers: numCPUs,
    keepAlive: 500
  }, options);
  this.workers = [];
  this.totalWorkers = 0;
  this.cullTimeout = undefined;
};
util.inherits(exports, EventEmitter);

exports.prototype.sendAll = function (message) {
  _.each(this.workers, function (worker) {
    worker.process.send(message);
  });
};
exports.prototype.send = function (message, callback) {
  var self = this;

  callback = callback || function () {};

  function fork() {
    totalWorkers++;
    self.totalWorkers++;

    var execArgv = process.execArgv;
    if (_.contains(execArgv, '--prof')) {
      execArgv = _.clone(execArgv);

      var logFile = _.find(execArgv, function (arg) {
          return /^--logfile=.*/.test(arg);
        }) || '--logfile=v8.log',
        location = execArgv.indexOf(logFile);

      if (self.options.logId) {
        logFile = logFile + '.' + self.options.logId + '.' + self.totalWorkers;
      } else {
        logFile = logFile + '.' + totalWorkers;
      }
      if (location >= 0) {
        execArgv.splice(location, 1, logFile);
      } else {
        execArgv.push(logFile);
      }
    }

    // Node 0.11 supports passing the execArgv option to fork but we don't want to break compatibility
    // so we're going to emulate the fork logic in the mean time.
    worker = {
      callback: false,
      process: cp.spawn(
        process.execPath,
        execArgv.concat([__dirname + '/worker'], execArgv), {
          stdio: ['pipe', 'pipe', 'pipe', 'ipc']
        })
    };
    if (self.options.stdout) {
      worker.process.stdout.on('data', self.options.stdout);
    }
    worker.process.send(self.module);

    function processMessage(msg) {


      var callback = worker.callback;
      worker.callback = undefined;

      if (queue.length) {
        var queued = queue.shift();
        queued.pool.send(queued.message, queued.callback);
      }

      // Dispatch
      process.nextTick(function () {
        var err;
        if (msg.err) {
          err = new Error(msg.err);
          if (msg.stack) {
            err.stack = msg.stack;
          }
        }

        // Kill the worker as we don't know what state it might be in
        if (msg.fatal) {
          worker.process.removeAllListeners('exit');
          worker.process.kill();
          self.workers = _.without(self.workers, worker);
        }
        activeTasks--;
        if (!callback) {
          self.emit('error', err || new Error('Out of band data: ' + util.inspect(msg.data)));
        } else {
          callback(err, msg.data);
        }

      });
    }

    worker.process.on('exit', function (code, signal) {
      if (code === null) {
        //process wasn't killed properly
        var error = new Error('Worker exited unexpectedly');
        processMessage({
{'fraction_non_alphanumeric': 0.14422323248158017, 'fraction_numerical': 0.003762345195171657, 'mean_word_length': 2.261758691206544, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2082135', 'n_tokens_mistral': 1992, 'n_tokens_neox': 1929, 'n_words': 448}