<commit_before>var processParse = require('./processParse');
var r = require('./redis');
var redis = r.client;
var jobs = r.jobs;
var kue = r.kue;
var utility = require('./utility');
var cluster = require('cluster');
var buildSets = require('./buildSets');
var config = require('./config');
start();

function start() {
    buildSets(function() {
        redis.get("parsers", function(err, result) {
            if (err || !result) {
                console.log('failed to get parsers from redis, retrying');
                return setTimeout(start, 10000);
            }
            var parsers = JSON.parse(result);
            //concurrent job processors per parse worker
            var parallelism = 4;
            var capacity = parsers.length * parallelism;
            if (cluster.isMaster && config.NODE_ENV !== "test") {
                console.log("[PARSEMANAGER] starting master");
                for (var i = 0; i < capacity; i++) {
                    if (false) {
                        //fork a worker for each available parse core
                        forkWorker(i);
                    }
                    else {
                        //run workers in parallel in a single thread (uses less memory)
                        runWorker(i);
                    }
                }
            }
            else {
                runWorker(0);
            }

            function forkWorker(i) {
                var worker = cluster.fork({
                    PARSER_URL: parsers[i]
                });
                worker.on("exit", function() {
                    console.log("Worker crashed! Spawning a replacement of worker %s", worker.id);
                    forkWorker(i);
                });
            }

            function runWorker(i) {
                console.log("[PARSEMANAGER] starting worker with pid %s", process.pid);
                jobs.process('parse', function(job, ctx, cb) {
                    console.log("starting parse job: %s", job.id);
                    job.parser_url = getParserUrl(job);
                    //TODO check if the assigned url is healthy before trying to parse?
                    //if not, use ctx to pause
                    //keep checking status and resume the worker when the parse worker is alive again
                    //current behavior will just keep retrying the url
                    return processParse(job, ctx, cb);
                });
                utility.cleanup(jobs, kue, 'parse');

                function getParserUrl(job) {
                    return config.PARSER_URL || parsers[i] || parsers[Math.floor(Math.random() * parsers.length)];
                }
            }
        });
    });
}<commit_msg>move parse job cleanup to master thread<commit_after>var processParse = require('./processParse');
var r = require('./redis');
var redis = r.client;
var jobs = r.jobs;
var kue = r.kue;
var utility = require('./utility');
var cluster = require('cluster');
var buildSets = require('./buildSets');
var config = require('./config');
start();

function start() {
    buildSets(function() {
        redis.get("parsers", function(err, result) {
            if (err || !result) {
                console.log('failed to get parsers from redis, retrying');
                return setTimeout(start, 10000);
            }
            var parsers = JSON.parse(result);
            //concurrent job processors per parse worker
            var parallelism = 4;
            var capacity = parsers.length * parallelism;
            if (cluster.isMaster && config.NODE_ENV !== "test") {
                console.log("[PARSEMANAGER] starting master");
                utility.cleanup(jobs, kue, 'parse');
                for (var i = 0; i < capacity; i++) {
                    if (false) {
                        //fork a worker for each available parse core
                        forkWorker(i);
                    }
                    else {
                        //run workers in parallel in a single thread (uses less memory)
                        runWorker(i);
                    }
                }
            }
            else {
                runWorker(0);
            }

            function forkWorker(i) {
                var worker = cluster.fork({
                    PARSER_URL: parsers[i]
                });
                worker.on("exit", function() {
                    console.log("Worker crashed! Spawning a replacement of worker %s", worker.id);
                    forkWorker(i);
                });
            }

            function runWorker(i) {
                console.log("[PARSEMANAGER] starting worker with pid %s", process.pid);
                jobs.process('parse', function(job, ctx, cb) {
                    console.log("starting parse job: %s", job.id);
                    job.parser_url = getParserUrl(job);
                    //TODO check if the assigned url is healthy before trying to parse?
                    //if not, use ctx to pause
                    //keep checking status and resume the worker when the parse worker is alive again
                    //current behavior will just keep retrying the url
                    return processParse(job, ctx, cb);
                });

                function getParserUrl(job) {
                    return config.PARSER_URL || parsers[i] || parsers[Math.floor(Math.random() * parsers.length)];
                }
            }
        });
    });
}
{'fraction_non_alphanumeric': 0.10817748441510817, 'fraction_numerical': 0.0029336266960029336, 'mean_word_length': 1.3769063180827887, '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': '6151300', 'n_tokens_mistral': 1393, 'n_tokens_neox': 1283, 'n_words': 398}