<commit_before>var util = require('util');
var EventEmitter = require('events').EventEmitter;

var debug = require('debug')('qp:Worker');

var redis = require('./redis');

var Worker = module.exports = function(queue, opts, red) {
  var self = this;

  this.opts = opts;

  this.redis = red || redis.createClient();
  this.queue = queue;

  this.processed = 0;
  this.processing = 0;
  this.start = Date.now();

  this.minRate = opts.minRate || 0;
  this.maxRate = opts.maxRate || Infinity;
  this.maxProcessing = opts.maxProcessing || this.maxRate * 2;
  this.rateInterval = opts.rateInterval || 1000;
  this.checkInterval = opts.checkInterval || this.rateInterval / 10;

  setInterval(function() {
    self.start = Date.now();
    self.processed = 0;
  }, this.rateInterval);

  self.checkRate();


};

util.inherits(Worker, EventEmitter);

Worker.prototype.checkRate = function() {
  var self = this;

  if (self.stopped) return;

  debug('checking rate');

  var minBound = self.minRate * ((Date.now() - self.start) / self.rateInterval);
  var maxBound = self.maxRate * ((Date.now() - self.start) / self.rateInterval);

  if (self.processed > maxBound) {
    debug('too fast - pausing');
    var ahead = self.processed - maxBound;
    self.pause();

    // check again when we'll be back behind the maxbound
    setTimeout(self.checkRate.bind(self), (self.processed - maxBound) * (self.maxRate / self.rateInterval));
    return;
  }

  // needs to go faster!
  if (self.processed < minBound) {
    debug('too slow');
    var behind = minBound - self.processed;
    self.paused = false;

    if (self.processing > self.maxProcessing) {
      debug('already spawned enough processes');
    } else {
      debug('spawning workers');
      for (var i = 0; i < behind; i++) self.process();
    }
  }

  setTimeout(self.checkRate.bind(self), self.checkInterval);

};

Worker.prototype.getBlockingJob = function(cb) {
  debug('getting blocking job');

  this.redis.blpop('qp:' + this.queue.name + ':jobs', 0, function(e, job) {
    cb(e, job && job[1]);
  });
};

Worker.prototype.getNonBlockingJob = function(cb) {
  var self = this;

  debug('getting non-blocking job');

  this.redis.lpop('qp:' + this.queue.name + ':jobs', function(err, job) {
    if (err || !job) {
      debug('no job, retry');
      setTimeout(function() {
        self.process();
      }, self.queue.qp.opts.checkInterval || 200);
      return;
    }
    cb(err, job);
  });
};

Worker.prototype.getJob = function(cb) {
  if (!this.queue.qp.opts.noBlock) {
    this.getBlockingJob(cb);
  } else {
    this.getNonBlockingJob(cb);
  }
};

Worker.prototype.process = function() {
  var self = this;

  if (self.stopped) {
    debug('worker stopped, not processing');
    if (!self.processing) self.emit('stopped');
    return;
  }

  if (self.paused) {
    debug('worker paused, not processing');
    self.emit('paused');
    return;
  }

  self.processing++;

  self.getJob(function(err, jobID) {
    if (err) {
      debug('error getting job');
      self.emit('error', err);
      return;
    }
    debug('got job');

    var job = self.queue.create();
    job.id = jobID;
    job.worker = self;
    job._saved = true;

    self.working = true;

    self.emit('job', job);

    job.once('done', function() {
      debug('job complete');<commit_msg>some rates sanity and improvements<commit_after>var util = require('util');
var EventEmitter = require('events').EventEmitter;

var debug = require('debug')('qp:Worker');

var redis = require('./redis');

var Worker = module.exports = function(queue, opts, red) {
  var self = this;

  this.opts = opts;

  this.redis = red || redis.createClient();
  this.queue = queue;

  this.processed = 0;
  this.processing = 0;
  this.start = Date.now();

  this.minRate = opts.minRate || 0;
  this.maxRate = opts.maxRate || Infinity;
  this.maxProcessing = opts.maxProcessing || this.maxRate * 2;
  this.rateInterval = opts.rateInterval || 1000;
  this.checkInterval = opts.checkInterval || this.rateInterval / 10;

  setInterval(function() {
    self.start = Date.now();
    self.processed = 0;
  }, this.rateInterval);

  self.checkRate();


};

util.inherits(Worker, EventEmitter);

Worker.prototype.checkRate = function() {
  var self = this;

  if (self.stopped) return;

  debug('checking rate');

  var diff = Date.now() - self.start;
  var minBound = self.minRate * (diff / self.rateInterval);
  var maxBound = self.maxRate * (diff / self.rateInterval);

  if (self.processed > maxBound) {
    var ahead = self.processed - maxBound;
    self.pause();

    var timeout = ahead * (self.rateInterval / self.maxRate);

    // sanity for small maxRates
    if (timeout > self.rateInterval) timeout = self.rateInterval - diff;

    debug('too fast - pausing for ' + timeout + 'ms');
    debug(self.processed + ' / ' + maxBound + ' done');

    setTimeout(self.checkRate.bind(self), timeout);
    return;
  }

  // needs to go faster!
  if (self.processed < minBound) {
    var behind = minBound - self.processed;

    debug('too slow, ' + behind +' jobs behind');

    self.paused = false;

    if (self.processing > self.maxProcessing) {
      debug('already spawned enough processes');
    } else {
      var numNew = Math.ceil(behind);
      debug('spawning ' + numNew + ' workers');
      for (var i = 0; i < numNew; i++) self.process();
    }
  }

  setTimeout(self.checkRate.bind(self), self.checkInterval);

};

Worker.prototype.getBlockingJob = function(cb) {
  debug('getting blocking job');

  this.redis.blpop('qp:' + this.queue.name + ':jobs', 0, function(e, job) {
    cb(e, job && job[1]);
  });
};

Worker.prototype.getNonBlockingJob = function(cb) {
  var self = this;

  debug('getting non-blocking job');

  this.redis.lpop('qp:' + this.queue.name + ':jobs', function(err, job) {
    if (err || !job) {
      debug('no job, retry');
      setTimeout(function() {
        self.process();
      }, self.queue.qp.opts.checkInterval || 200);
      return;
    }
    cb(err, job);
  });
};

Worker.prototype.getJob = function(cb) {
  if (!this.queue.qp.opts.noBlock) {
    this.getBlockingJob(cb);
  } else {
    this.getNonBlockingJob(cb);
  }
};

Worker.prototype.process = function() {
  var self = this;

  if (self.stopped) {
    debug('worker stopped, not processing');
    if (!self.processing) self.emit('stopped');
    return;
  }

  if (self.paused) {
    debug('worker paused, not processing');
    self.emit('paused');
    return;
  }

  if (self.processing > self.maxProcessing) {
    debug('too many processing, stopping');
    return;
  }

  self.processing++;

  self.getJob(function(err, jobID) {
    if (err) {
      debug('error getting job');
      self.emit('error', err);
      return;
    }
    debug('got job');

    var job = self.queue.create();
    job.id = jobID;
    job.worker = self;
    job._saved = true;

    self.working = true;

    self.emit('job', job);

    job.once('done', function() {
      debug('job complete');
{'fraction_non_alphanumeric': 0.16275071633237823, 'fraction_numerical': 0.004871060171919771, 'mean_word_length': 3.058720930232558, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, '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': '5780427', 'n_tokens_mistral': 2399, 'n_tokens_neox': 2276, 'n_words': 515}