<commit_before>var processParse = require('./processParse');
var r = require('./redis');
var redis = r.client;
var jobs = r.jobs;
var cluster = require('cluster');
var numCPUs = require('os').cpus().length;
var config = require('./config');
start();

function start() {
    redis.get("parsers", function(err, result) {
        if (err || !result) {
            console.log("no parsers in redis!");
            return setTimeout(start, 10000);
        }
        var parsers = JSON.parse(result);
        var capacity = parsers.length;
        if (cluster.isMaster && config.NODE_ENV !== "test") {
            console.log("[PARSEMANAGER] starting master");
            var urls = {};
            cluster.on('fork', function(worker) {
                worker.on('message', function(msg) {
                    console.log(msg);
                    //a new worker is running, keep track of what url it's using
                    urls[msg.pid] = msg.url;
                    console.log(urls);
                });
            });
            for (var i = 0; i < capacity; i++) {
                //fork a worker for each available parse core
                cluster.fork({
                    PARSER_URL: parsers[i]
                });
            }
            //TODO if we submitted a job with filename it must be parsed by localhost (on master)!
            cluster.on("exit", function(worker, code) {
                if (code !== 0) {
                    console.log("Worker crashed! Spawning a replacement of worker %s", worker.process.pid);
                    //give this new worker the parser url of the one that crashed
                    cluster.fork({
                        PARSER_URL: urls[worker.process.pid]
                    });
                }
            });
        }
        else {
            console.log("[PARSEMANAGER] starting worker with pid %s", process.pid);
            if (process.send) {
                process.send({
                    pid: process.pid,
                    url: process.env.PARSER_URL
                });
            }
            //process requests
            jobs.process('request_parse', function(job, cb) {
                getParserUrl(job, function() {
                    processParse(job, cb);
                });
            });
            //process regular parses
            jobs.process('parse', function(job, cb) {
                console.log("starting job: %s", job.id);
                getParserUrl(job, function() {
                    processParse(job, cb);
                });
            });<commit_msg>added debug info to request_parse<commit_after>var processParse = require('./processParse');
var r = require('./redis');
var redis = r.client;
var jobs = r.jobs;
var cluster = require('cluster');
var numCPUs = require('os').cpus().length;
var config = require('./config');
start();

function start() {
    redis.get("parsers", function(err, result) {
        if (err || !result) {
            console.log("no parsers in redis!");
            return setTimeout(start, 10000);
        }
        var parsers = JSON.parse(result);
        var capacity = parsers.length;
        if (cluster.isMaster && config.NODE_ENV !== "test") {
            console.log("[PARSEMANAGER] starting master");
            var urls = {};
            cluster.on('fork', function(worker) {
                worker.on('message', function(msg) {
                    console.log(msg);
                    //a new worker is running, keep track of what url it's using
                    urls[msg.pid] = msg.url;
                    console.log(urls);
                });
            });
            for (var i = 0; i < capacity; i++) {
                //fork a worker for each available parse core
                cluster.fork({
                    PARSER_URL: parsers[i]
                });
            }
            //TODO if we submitted a job with filename it must be parsed by localhost (on master)!
            cluster.on("exit", function(worker, code) {
                if (code !== 0) {
                    console.log("Worker crashed! Spawning a replacement of worker %s", worker.process.pid);
                    //give this new worker the parser url of the one that crashed
                    cluster.fork({
                        PARSER_URL: urls[worker.process.pid]
                    });
                }
            });
        }
        else {
            console.log("[PARSEMANAGER] starting worker with pid %s", process.pid);
            if (process.send) {
                process.send({
                    pid: process.pid,
                    url: process.env.PARSER_URL
                });
            }
            //process regular parses
            jobs.process('parse', function(job, cb) {
                console.log("starting parse job: %s", job.id);
                getParserUrl(job, function() {
                    processParse(job, cb);
                });
            });
            //process requests
            jobs.process('request_parse', function(job, cb) {
                console.log("starting request_parse job: %s", job.id);
                getParserUrl(job, function() {
                    processParse(job, cb);
                });
            });
{'fraction_non_alphanumeric': 0.12211410036252623, 'fraction_numerical': 0.0026712459454302615, 'mean_word_length': 1.385980883022303, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, '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': '6461944', 'n_tokens_mistral': 1383, 'n_tokens_neox': 1294, 'n_words': 358}