<commit_before>'use strict';

/**!
 * [square]
 * @copyright (c) 2012 observe.it (observe.it) <<EMAIL>>
 * MIT Licensed
 */

/**
 * Native modules.
 */
var spawn = require('child_process').spawn
  , cluster = require('cluster')
  , zlib = require('zlib')
  , path = require('path')
  , os = require('os');

/**
 * Third party modules.
 */
var canihaz = require('canihaz')('square')
  , request = require('request')
  , async = require('async')
  , _ = require('lodash');

/**
 * A fork received a new task to process.
 *
 * Task:
 *
 * task.engines: Comma seperated list of compressors that need to be used
 * task.extension: The file extension of content
 * task.content: The actual content that needs to be processed
 * task.gzip: Calculate the size of content after gzipping it
 * task.id: The id of this task (private)
 *
 * @param {Object} task
 */
if (!cluster.isMaster) process.on('message', function message(task) {
  var engines = exports[task.extension]
    , started = Date.now()
    , durations = {};

  async.reduce(
      task.engines.split(/\s?\,\s?/).filter(Boolean)
    , task
    , function reduce(memo, engine, done) {
        var backup = memo.content;

        // Compile the shizzle.
        if (engine in engines) {
          durations[engine] = Date.now();

          return engines[engine](memo, function crush(err, content) {
            durations[engine] = Date.now() - durations[engine];

            if (err) {
              memo.content = backup;
              return done(err, memo);
            }

            // Update the content and process all the things again, and again and
            // again.
            memo.content = content;
            done(err, memo);
          });
        }

        // The engine does not exist, send an error response
        process.nextTick(function nexTick() {
          done(new Error('The engine '+ engine +' does not exist'), memo);
        });
      }
    , function done(err, result) {
        result = result || task || {};

        // Add some metrics about this task like:
        // - The total time it took to process this task
        // - The time each individual compiler took to compress the content
        result.duration = Date.now() - started;
        result.individual = durations;

        // Transform the Error to something that can be serialized
        if (err) result.err = { message: err.message, stack: err.stack };
        if (!result.gzip || err) return process.send(result);

        // We want to calculate the size of the generated code once it has been
        // gzipped as that might be more important to users than the actual file
        // size after minification.
        result.gzip = 0;
        zlib.gzip(result.content, function gzip(err, buff) {
          if (err) result.err = { message: err.message, stack: err.stack };

          result.gzip = buff.length;
          process.send(result);
        });
      }
  );
});

/**
 * Send a message to the workers that they need to start processing something.
 *
 *
 * @param {Object} task work for the workers
 * @param {Function} cb callback
 * @api public
 */
exports.id = 0; // Date.now() is not unique enough
exports.send = function send(task, cb) {
  if (!exports.initialized) exports.initialize();

  var worker = exports.workers.pop();

  task.id = task.id || ++exports.id;
  worker.queue[task.id] = cb || function noop(){};
  worker.send(task);

  // Add it back at the end of the array, so we implement a round robin load
  // balancing technique for our workers.
  exports.workers.push(worker);
};

/**<commit_msg>[fix] Do a proper round robin, so we actually spread load across the cores<commit_after>'use strict';

/**!
 * [square]
 * @copyright (c) 2012 observe.it (observe.it) <<EMAIL>>
 * MIT Licensed
 */

/**
 * Native modules.
 */
var spawn = require('child_process').spawn
  , cluster = require('cluster')
  , zlib = require('zlib')
  , path = require('path')
  , os = require('os');

/**
 * Third party modules.
 */
var canihaz = require('canihaz')('square')
  , request = require('request')
  , async = require('async')
  , _ = require('lodash');

/**
 * A fork received a new task to process.
 *
 * Task:
 *
 * task.engines: Comma seperated list of compressors that need to be used
 * task.extension: The file extension of content
 * task.content: The actual content that needs to be processed
 * task.gzip: Calculate the size of content after gzipping it
 * task.id: The id of this task (private)
 *
 * @param {Object} task
 */
if (!cluster.isMaster) process.on('message', function message(task) {
  var engines = exports[task.extension]
    , started = Date.now()
    , durations = {};

  async.reduce(
      task.engines.split(/\s?\,\s?/).filter(Boolean)
    , task
    , function reduce(memo, engine, done) {
        var backup = memo.content;

        // Compile the shizzle.
        if (engine in engines) {
          durations[engine] = Date.now();

          return engines[engine](memo, function crush(err, content) {
            durations[engine] = Date.now() - durations[engine];

            if (err) {
              memo.content = backup;
              return done(err, memo);
            }

            // Update the content and process all the things again, and again and
            // again.
            memo.content = content;
            done(err, memo);
          });
        }

        // The engine does not exist, send an error response
        process.nextTick(function nexTick() {
          done(new Error('The engine '+ engine +' does not exist'), memo);
        });
      }
    , function done(err, result) {
        result = result || task || {};

        // Add some metrics about this task like:
        // - The total time it took to process this task
        // - The time each individual compiler took to compress the content
        result.duration = Date.now() - started;
        result.individual = durations;

        // Transform the Error to something that can be serialized
        if (err) result.err = { message: err.message, stack: err.stack };
        if (!result.gzip || err) return process.send(result);

        // We want to calculate the size of the generated code once it has been
        // gzipped as that might be more important to users than the actual file
        // size after minification.
        result.gzip = 0;
        zlib.gzip(result.content, function gzip(err, buff) {
          if (err) result.err = { message: err.message, stack: err.stack };

          result.gzip = buff.length;
          process.send(result);
        });
      }
  );
});

/**
 * Send a message to the workers that they need to start processing something.
 *
 *
 * @param {Object} task work for the workers
 * @param {Function} cb callback
 * @api public
 */
exports.id = 0; // Date.now() is not unique enough
exports.send = function send(task, cb) {
  if (!exports.initialized) exports.initialize();

  var worker = exports.workers.shift();

  task.id = task.id || ++exports.id;
  worker.queue[task.id] = cb || function noop(){};
  worker.send(task);

  // Add it back at the end of the array, so we implement a round robin load
  // balancing technique for our workers.
  exports.workers.push(worker);
};

/**
{'fraction_non_alphanumeric': 0.12795684050352746, 'fraction_numerical': 0.001659980633559275, 'mean_word_length': 2.3911819887429644, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '2191223', 'n_tokens_mistral': 2089, 'n_tokens_neox': 2032, 'n_words': 707}