<commit_before>/*!
 * node.io
 * Copyright(c) 2010 <NAME> <<EMAIL>>
 * MIT Licensed
 */

var fs = require('fs'),
    JobClass = require('./job').JobClass,
    multi = require('./multi_node'),
    v = require('validator'),
    status = require('./utils').status,
    crc32 = require('./utils').crc32,
    exec = require('child_process').exec,
    cwd = process.cwd(),
    path = require('path');

var id, 
    isMaster,
    isSlave,
    isWorker = true,
    fork,
    distribute;
    
//Keep track of child processes and slaves
var workers = [],
    worker_count = 0;

//Keep track of the master process
var master;

var ready_to_request_input = true;

//Start the processor and run the specified job
exports.start = function(job, options, callback, capture_output) {
    if (typeof options === 'function') {
        capture_output = callback;
        callback = options;
        options = {};
    }
    
    callback = callback || function(err) {
        if (err) status.error(err);
        process.exit(1);
    }
    
	var p = new Processor(options);
        
    //Callback for once we've loaded the job
    var onload = function(job, job_obj) {
    
        var output;
        if (capture_output) {
            output = [];
            job_obj.output = function(out) {
                if (out instanceof Array) {
                    out.forEach(function(line) {
                        output.push(line);
                    });
                } else {
                    output.push(out);
                }
            }
        }

        var oncomplete_callback = function(err) {
            callback(err, output);
        }
        
        //Initialise the processor and start the job
        fork = job_obj.options.fork;
        p.init(function() {
            p.startJob(job, job_obj, oncomplete_callback);
        });
    }
    
    //Load the job
	var job_obj;
    if (typeof job !== 'string') {
    
        job_obj = job;
        job = crc32(JSON.stringify(job));    
        onload(job, job_obj);
        
    } else {
        
        if (!~job.indexOf('/')) job = cwd + '/' + job;
        
        var joblen = job.length;
        if (path.extname(job) === '.coffee') {
            
            var basename = path.basename(job, '.coffee');
            var compiled_filename = cwd + '/' + basename + '_compiled.js';
            
            if (!process.env._CHILD_ID_) {
                
                //If we're the master, compile and load the .coffee file
                exec('coffee -p -c "'+job+'"', {cwd:cwd}, function(err, stdout, stderr) {
                    if (err || stderr) {
                        status.error(err || stderr);
                        process.exit(1);
                    } else {
                        
                        fs.writeFileSync(compiled_filename, stdout);
                        
                        var oldcallback = callback;
                        callback = function (err, output) {
                            fs.unlinkSync(compiled_filename);
                            oldcallback(err, output);
                        };
                        
                        job = compiled_filename;
                        job_obj = p.loadJob(job);
                        onload(job, job_obj);
                    }
                });
            
            } else {
                
                //If we're a child, the file is already compiled
                job = compiled_filename;
                job_obj = p.loadJob(job);
                onload(job, job_obj);
                
            }
            
        } else {
            
            //Load the .js file
            job_obj = p.loadJob(job);
            onload(job, job_obj);
            
        }
    }    
}

//Start the processor as a slave. It will sit and wait til it receives input
exports.startSlave = function(options) {
	var p = new Processor(options);
    isSlave = true;
	p.init(function() {
        //TODO: Implement distributed processing
        //Setup a server that accepts connections on options.port
    });
}

var Processor = exports.Processor = function(options) {
    this.options = options || {};
    this.jobs = {};<commit_msg>Improved readability<commit_after>/*!
 * node.io
 * Copyright(c) 2010 <NAME> <<EMAIL>>
 * MIT Licensed
 */

var fs = require('fs'),
    JobClass = require('./job').JobClass,
    multi = require('./multi_node'),
    v = require('validator'),
    status = require('./utils').status,
    crc32 = require('./utils').crc32,
    exec = require('child_process').exec,
    cwd = process.cwd(),
    path = require('path');

var id, 
    isMaster,
    isSlave,
    isWorker = true,
    fork,
    distribute;
    
//Keep track of child processes and slaves
var workers = [],
    worker_count = 0;

//Keep track of the master process
var master;

var ready_to_request_input = true;

//Start the processor and run the specified job
exports.start = function(job, options, callback, capture_output) {
    if (typeof options === 'function') {
        capture_output = callback;
        callback = options;
        options = {};
    }
    
    callback = callback || function(err) {
        if (err) status.error(err);
        process.exit(1);
    }
    
	var proc = new Processor(options);
        
    //Callback for once we've loaded the job
    var onload = function(job, job_obj) {
    
        var output;
        if (capture_output) {
            output = [];
            job_obj.output = function(out) {
                if (out instanceof Array) {
                    out.forEach(function(line) {
                        output.push(line);
                    });
                } else {
                    output.push(out);
                }
            }
        }

        var oncomplete_callback = function(err) {
            callback(err, output);
        }
        
        //Initialise the processor and start the job
        fork = job_obj.options.fork;
        proc.init(function() {
            proc.startJob(job, job_obj, oncomplete_callback);
        });
    }
    
    //Load the job
	var job_obj;
    if (typeof job !== 'string') {
    
        job_obj = job;
        job = crc32(JSON.stringify(job));    
        onload(job, job_obj);
        
    } else {
        
        //Make it a full path
        if (!~job.indexOf('/')) job = cwd + '/' + job;
        
        //Are we dealing with a CoffeeScript job?
        if (path.extname(job) === '.coffee') {
            
            var basename = path.basename(job, '.coffee');
            var compiled_js = cwd + '/' + basename + '_compiled.js';
            
            //If we're the master, compile and load the .coffee file
            if (!process.env._CHILD_ID_) {
                
                exec('coffee -p -c "'+job+'"', {cwd:cwd}, function(err, stdout, stderr) {
                    if (err || stderr) {
                        status.error(err || stderr);
                        process.exit(1);
                    } else {
                        
                        fs.writeFileSync(compiled_js, stdout);
                        
                        var oldcallback = callback;
                        callback = function (err, output) {
                            fs.unlinkSync(compiled_js);
                            oldcallback(err, output);
                        };
                        
                        onload(compiled_js, proc.loadJob(compiled_js));
                    }
                });
            
            } else {
                
                //If we're a child, the file is already compiled
                onload(compiled_js, proc.loadJob(compiled_js));
                
            }
            
        } else {
            
            //We're dealing with a plain old .js file
            onload(job, proc.loadJob(job));
            
        }
    }    
}

//Start the processor as a slave. It will sit and wait til it receives input
exports.startSlave = function(options) {
	var proc = new Processor(options);
    isSlave = true;
	proc.init(function() {
        //TODO: Implement distributed processing
        //Setup a server that accepts connections on options.port
    });
}

var Processor = exports.Processor = function(options) {
    this.options = options || {};
    this.jobs = {};
{'fraction_non_alphanumeric': 0.11289742051589682, 'fraction_numerical': 0.003119376124775045, 'mean_word_length': 1.3428892636312535, 'pattern_counts': {'":': 0, '<': 9, '<?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': '1040147', 'n_tokens_mistral': 2308, 'n_tokens_neox': 2210, 'n_words': 570}