<commit_before>/**
 * 管理实例运行状态，处理进程信号
 * 支持分布式的实例启动
 * 监听scheduler的调度事件，用户更新实例的运行状态
 * 触发scheduler的事件
 */

var winton = require('winton');
var redis = require('redis');
var os = require('os');
var crypto = require('crypto');

try{
    /* 必要的初始化操作 */
    var settings = require('../settings.js');
    winton.loggers.add('core', settings.logger);
    var logger = winton.loggers.get('core');
    var store = redis.createClient(settings.redis.port, settings.redis.host);
}catch(e){
    console.log(e);
    process.exit(1);
}

/**
 * redis格式：统计信息
 * {
 *  $instance_name:{//每个instance拥有一个hkey,哈希表
 *      download:下载的总次数
 *      pipe:抓取信息总次数
 *      current_download:当前时间分片下载的次数
 *      current_pipe:当前时间分片的抓取总次数
 *      current_start_time:当前时间分片的启动时间
 *      current_run_seconds:当前分片的实际运行时间,所有进程的运行时间总和
 *      current_init_length:初始化队列长度
 *      init_time:第一次运行时间
 *  }
 *  $instance_time:[//每次分片的运行情况，列表
 *      {//json字符串
 *          download:下载的总次数
 *          pipe:抓取信息的总次数
 *          start_time:启动时间
 *          run_seconds:运行时间
 *          init_length:初始化队列长度
 *          end_time:结束事件
 *      }
 *  ]
 *  $instance_$process:{//每个运行中的process都拥有一个hkey，哈希表，设定有效期
 *      stats:0,1,2,3//当前进程状态,0:启动中 1:运行中，2关闭中，3暂停中，4队列为空停止中，5初始化队列中，6可重新执行
 *      start_time:启动时间
 *      download:该进程的下载次数
 *      pipe:该进程的抓取信息次数
 *      queue:当前的下载队列状态，几个在运行中
 *      last_heart_time:最后次心跳检测时间
 *      host_info:运行的主机信息（hostname,pid）
 *  }
 *  proxy:(//总的代理列表，有序集合
 *      ip:port score://根据score的正序获取一个代理，每调用一次则减少部分分值，如果失败则降低该代理的分值，如果成功则提高该代理的分值，对于分值为负数的则移除
 *  )
 * }
 * @param instance_name
 * @returns {{start: Function, stop: Function, status: Function}}
 */

var core = function(instance_name){
    this.instance_name = instance_name;
    this.start_time = new Date().getMilliseconds();
    this.logger = logger;
    this.store = store;
    this.settings = settings;
    this.stats = 0;
    var engine = new (require('./engine.js'))(this);
    var self = this;
    return {
        start:function(options){
            self.host_info = os.hostname+':'+process.pid;
            self.process_name = self.instance_name+'_'+ crypto.createHash('md5').update(self.host_info).digest('hex').slice(0,5);
            async.waterfall([
                //监听初始化事件
                function(callback) {
                    engine.on('finish_init',function(){
                        callback(null);
                    });
                },
                //初始化完毕
                function(callback) {
                    engine.emit('scheduler', function(scheduler){
                        callback(null, scheduler);
                    });
                    engine.emit('downloader', function(downloader){
                        downloader.on('finish_download', function(url){
                            self.store.hincrby(self.instance_name, 'download', 1);
                            self.store.hincrby(self.process_name, 'download', 1);
                            self.store.hincrby(self.instance_name, 'current_pipe', 1);
                            engine.logger.info('[ DOWNLOAD ] %s', url);
                        });
                    });
                    engine.emit('pipeline', function(pipeline){
                        pipeline.on('finish_pipeline', function(error, url){
                            if(error){
                                engine.logger.warn('[ PIPELINE ] %s', url);
                            }else{
                                self.store.hincrby(self.instance_name, 'pipe', 1);
                                self.store.hincrby(self.process_name, 'pipe', 1);
                                self.store.hincrby(self.instance_name, 'current_pipe', 1);
                                engine.logger.info('[ PIPELINE ] %s', url);
                            }
                        });
                    });
                },
                //获取scheduler
                function(scheduler, callback) {
                    scheduler.emit('start',function(){
                        self.store.hset(self.process_name, 'stats', 1);
                        self.stats = 1;
                    });
                    //设定退出逻辑
                    process.on('exit',function(){
                        scheduler.emit('stop',function(){
                            self.store.delete('status_'+self.instance_name);
                            self.store.end();
                        });
                    });
                    scheduler.on('finish_queue', function(loop){
                        callback(null);
                    });
                },
                //队列执行为空
                function(callback){
                    self.store.hset(self.process_name, 'stats', 4);
                    self.stats = 4;
                    self.store.keys(self.instance_name+'_*', callback);
                },
                //获取所有进程列表
                function(process_list, callback){
                    async.filter(process_list, function(process_name, callback){
                        if(process_name == self.process_name) return callback(false);
                        self.store.hget(process_name, 'stats', function(error, result){<commit_msg>修正推出后删除进程信息<commit_after>/**
 * 管理实例运行状态，处理进程信号
 * 支持分布式的实例启动
 * 监听scheduler的调度事件，用户更新实例的运行状态
 * 触发scheduler的事件
 */

var winton = require('winton');
var redis = require('redis');
var os = require('os');
var crypto = require('crypto');

try{
    /* 必要的初始化操作 */
    var settings = require('../settings.js');
    winton.loggers.add('core', settings.logger);
    var logger = winton.loggers.get('core');
    var store = redis.createClient(settings.redis.port, settings.redis.host);
}catch(e){
    console.log(e);
    process.exit(1);
}

/**
 * redis格式：统计信息
 * {
 *  $instance_name:{//每个instance拥有一个hkey,哈希表
 *      download:下载的总次数
 *      pipe:抓取信息总次数
 *      current_download:当前时间分片下载的次数
 *      current_pipe:当前时间分片的抓取总次数
 *      current_start_time:当前时间分片的启动时间
 *      current_run_seconds:当前分片的实际运行时间,所有进程的运行时间总和
 *      current_init_length:初始化队列长度
 *      init_time:第一次运行时间
 *  }
 *  $instance_time:[//每次分片的运行情况，列表
 *      {//json字符串
 *          download:下载的总次数
 *          pipe:抓取信息的总次数
 *          start_time:启动时间
 *          run_seconds:运行时间
 *          init_length:初始化队列长度
 *          end_time:结束事件
 *      }
 *  ]
 *  $instance_$process:{//每个运行中的process都拥有一个hkey，哈希表，设定有效期
 *      stats:0,1,2,3//当前进程状态,0:启动中 1:运行中，2关闭中，3暂停中，4队列为空停止中，5初始化队列中，6可重新执行
 *      start_time:启动时间
 *      download:该进程的下载次数
 *      pipe:该进程的抓取信息次数
 *      queue:当前的下载队列状态，几个在运行中
 *      last_heart_time:最后次心跳检测时间
 *      host_info:运行的主机信息（hostname,pid）
 *  }
 *  proxy:(//总的代理列表，有序集合
 *      ip:port score://根据score的正序获取一个代理，每调用一次则减少部分分值，如果失败则降低该代理的分值，如果成功则提高该代理的分值，对于分值为负数的则移除
 *  )
 * }
 * @param instance_name
 * @returns {{start: Function, stop: Function, status: Function}}
 */

var core = function(instance_name){
    this.instance_name = instance_name;
    this.start_time = new Date().getMilliseconds();
    this.logger = logger;
    this.store = store;
    this.settings = settings;
    this.stats = 0;
    var engine = new (require('./engine.js'))(this);
    var self = this;
    return {
        start:function(options){
            self.host_info = os.hostname+':'+process.pid;
            self.process_name = self.instance_name+'_'+ crypto.createHash('md5').update(self.host_info).digest('hex').slice(0,5);
            async.waterfall([
                //监听初始化事件
                function(callback) {
                    engine.on('finish_init',function(){
                        callback(null);
                    });
                },
                //初始化完毕
                function(callback) {
                    engine.emit('scheduler', function(scheduler){
                        callback(null, scheduler);
                    });
                    engine.emit('downloader', function(downloader){
                        downloader.on('finish_download', function(url){
                            self.store.hincrby(self.instance_name, 'download', 1);
                            self.store.hincrby(self.process_name, 'download', 1);
                            self.store.hincrby(self.instance_name, 'current_pipe', 1);
                            engine.logger.info('[ DOWNLOAD ] %s', url);
                        });
                    });
                    engine.emit('pipeline', function(pipeline){
                        pipeline.on('finish_pipeline', function(error, url){
                            if(error){
                                engine.logger.warn('[ PIPELINE ] %s', url);
                            }else{
                                self.store.hincrby(self.instance_name, 'pipe', 1);
                                self.store.hincrby(self.process_name, 'pipe', 1);
                                self.store.hincrby(self.instance_name, 'current_pipe', 1);
                                engine.logger.info('[ PIPELINE ] %s', url);
                            }
                        });
                    });
                },
                //获取scheduler
                function(scheduler, callback) {
                    scheduler.emit('start',function(){
                        self.store.hset(self.process_name, 'stats', 1);
                        self.stats = 1;
                    });
                    //设定退出逻辑
                    process.on('exit',function(){
                        scheduler.emit('stop',function(){
                            //删除进程信息
                            self.store.del(self.process_name);
                            self.store.end();
                        });
                    });
                    scheduler.on('finish_queue', function(loop){
                        callback(null);
                    });
                },
                //队列执行为空
                function(callback){
                    self.store.hset(self.process_name, 'stats', 4);
                    self.stats = 4;
                    self.store.keys(self.instance_name+'_*', callback);
                },
                //获取所有进程列表
                function(process_list, callback){
                    async.filter(process_list, function(process_name, callback){
                        if(process_name == self.process_name) return callback(false);
                        self.store.hget(process_name, 'stats', function(error, result){
{'fraction_non_alphanumeric': 0.12469954812037304, 'fraction_numerical': 0.004999519276992597, 'mean_word_length': 1.5186440677966102, 'pattern_counts': {'":': 0, '<': 3, '<?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': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6764853', 'n_tokens_mistral': 3502, 'n_tokens_neox': 3597, 'n_words': 390}