<commit_before>'use strict';

const _ = require('lodash');
const Promise = require('bluebird');
const pubsub = require('pubsub-js');
const cluster = require('cluster');
const moment = require('moment');
const cpus = require('os').cpus();
const CONST = require('./lib/constants');
const logger = require('./lib/logger');
const config = require('./lib/config');
const errors = require('./lib/errors');
const maintenanceManager = require('./lib/maintenance').maintenanceManager;
const serviceFabrikClient = require('./lib/cf').serviceFabrikClient;
require('./lib/fabrik');

let cpuCount = cpus.length;
let maxWorkers = 0;

//Assuming broker/jobscheduler are running on same VM, keeping the # jobs 1 less than cpu count. 
cpuCount = cpuCount > 1 ? cpuCount - 1 : 1;
if (config.scheduler.max_workers && config.scheduler.max_workers < cpuCount) {
  maxWorkers = config.scheduler.max_workers;
} else {
  maxWorkers = cpuCount;
}

class JobScheduler {
  constructor() {
    this.ready = Promise.try(() => {
      this.jobWorkers = [];
      this.workerCount = 0;
      this.workerType = '';
      this.serviceFabrikInMaintenance = true;
      this.intervalTimer = undefined;
      this.maintenanceStartTime = undefined;
      this.shutDownHook = () => this.shutDown();
      process.on('SIGTERM', this.shutDownHook);
      process.on('SIGINT', this.shutDownHook);
      this.unhandleRejectionHook = (reason, p) => this.processUnhandledRejection(reason, p);
      process.on('unhandledRejection', this.unhandleRejectionHook);
      if (cluster.isMaster) {
        cluster.on('exit', (worker, code, signal) => this.workerExitHandler(worker, code, signal));
        //This delay is added to ensure that DBManager is initialized prior to scheduler 
        //checking for maintenance status. Retry is anyways part of this check, but this 
        //delay ensures we dont have exception always on first try. 
        logger.info(`Scheduler will now sleep for ${config.scheduler.start_delay} (ms) before initializing...`);
        return Promise
          .delay(config.scheduler.start_delay)
          .then(() => this.ensureSystemNotInMainenanceThenInitMaster());
      } else {
        return this.initWorker();
      }
    });
  }

  initMaster() {
    this.serviceFabrikInMaintenance = false;
    logger.info(`Configured number of workers ${config.scheduler.max_workers} - No. of CPUs : ${cpus.length} - job workers : ${maxWorkers}`);
    this.workerType = `MASTER - ${process.pid}`;
    this.workerCount = 0;
    this.jobWorkers = [];
    // Create a worker for each CPU
    for (var i = 0, delay = 0; i < maxWorkers; i += 1, delay += CONST.JOB_SCHEDULER.WORKER_CREATE_DELAY) {
      logger.debug(`Set scheduled job worker ${i} with delay ${delay} - ${this.workerCount}`);
      //Agenda Scheduler for each worker will be configured to processJobQueue once every ${maxWorkers} minutes.
      //So each worker job is started with a delay of 1 minute, as it creates a constant sliding window of 1 min
      setTimeout(() => this.addJobWorker(), delay);
    }
  }

  initWorker() {
    this.workerType = `Worker - ${cluster.worker.id} - ${process.pid}`;
    process.on('message', (msg) => this.handleMessage(msg));
    logger.info(`Starting Service Fabrik Batch Job worker: ${cluster.worker.id} - ${process.pid}  @${new Date()}`);
    require('./lib/jobs');
  }<commit_msg>Initializing event listener in broker/JobScheduler.js workers<commit_after>'use strict';

const _ = require('lodash');
const Promise = require('bluebird');
const pubsub = require('pubsub-js');
const cluster = require('cluster');
const moment = require('moment');
const cpus = require('os').cpus();
const CONST = require('./lib/constants');
const logger = require('./lib/logger');
const config = require('./lib/config');
const utils = require('./lib/utils');
const errors = require('./lib/errors');
const maintenanceManager = require('./lib/maintenance').maintenanceManager;
const serviceFabrikClient = require('./lib/cf').serviceFabrikClient;
require('./lib/fabrik');

let cpuCount = cpus.length;
let maxWorkers = 0;

//Assuming broker/jobscheduler are running on same VM, keeping the # jobs 1 less than cpu count. 
cpuCount = cpuCount > 1 ? cpuCount - 1 : 1;
if (config.scheduler.max_workers && config.scheduler.max_workers < cpuCount) {
  maxWorkers = config.scheduler.max_workers;
} else {
  maxWorkers = cpuCount;
}

class JobScheduler {
  constructor() {
    this.ready = Promise.try(() => {
      this.jobWorkers = [];
      this.workerCount = 0;
      this.workerType = '';
      this.serviceFabrikInMaintenance = true;
      this.intervalTimer = undefined;
      this.maintenanceStartTime = undefined;
      this.shutDownHook = () => this.shutDown();
      process.on('SIGTERM', this.shutDownHook);
      process.on('SIGINT', this.shutDownHook);
      this.unhandleRejectionHook = (reason, p) => this.processUnhandledRejection(reason, p);
      process.on('unhandledRejection', this.unhandleRejectionHook);
      if (cluster.isMaster) {
        cluster.on('exit', (worker, code, signal) => this.workerExitHandler(worker, code, signal));
        //This delay is added to ensure that DBManager is initialized prior to scheduler 
        //checking for maintenance status. Retry is anyways part of this check, but this 
        //delay ensures we dont have exception always on first try. 
        logger.info(`Scheduler will now sleep for ${config.scheduler.start_delay} (ms) before initializing...`);
        return Promise
          .delay(config.scheduler.start_delay)
          .then(() => this.ensureSystemNotInMainenanceThenInitMaster());
      } else {
        return this.initWorker();
      }
    });
  }

  initMaster() {
    this.serviceFabrikInMaintenance = false;
    logger.info(`Configured number of workers ${config.scheduler.max_workers} - No. of CPUs : ${cpus.length} - job workers : ${maxWorkers}`);
    this.workerType = `MASTER - ${process.pid}`;
    this.workerCount = 0;
    this.jobWorkers = [];
    // Create a worker for each CPU
    for (var i = 0, delay = 0; i < maxWorkers; i += 1, delay += CONST.JOB_SCHEDULER.WORKER_CREATE_DELAY) {
      logger.debug(`Set scheduled job worker ${i} with delay ${delay} - ${this.workerCount}`);
      //Agenda Scheduler for each worker will be configured to processJobQueue once every ${maxWorkers} minutes.
      //So each worker job is started with a delay of 1 minute, as it creates a constant sliding window of 1 min
      setTimeout(() => this.addJobWorker(), delay);
    }
  }

  initWorker() {
    this.workerType = `Worker - ${cluster.worker.id} - ${process.pid}`;
    process.on('message', (msg) => this.handleMessage(msg));
    logger.info(`Starting Service Fabrik Batch Job worker: ${cluster.worker.id} - ${process.pid}  @${new Date()}`);
    require('./lib/jobs');
    utils.initializeEventListener(config.external, 'external');
  }
{'fraction_non_alphanumeric': 0.1395112016293279, 'fraction_numerical': 0.0034914169333721268, 'mean_word_length': 3.57418496340652, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 19, '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': '3181204', 'n_tokens_mistral': 2039, 'n_tokens_neox': 1992, 'n_words': 550}