<?php

//error_reporting();

use Swoole\Process;



include_once __DIR__."/../vendor/autoload.php";


function start(){

// 删除停止运行的值
    redis()->delete(RUNNING_REDIS_KEY);

    /** 创建一个表 **/
    $table = new Swoole\Table(128);// 128 行
    $table->column('val', Swoole\Table::TYPE_INT);
    $table->create();

    // 初始时，进行一次统计
    $table->set('etotal',['val'=> db()->count(\Model\email::count())]);

    // 进程管理器
    $pm = new Process\Manager();

    // 启动一个进程来管理定时
    $pm->add(function (Process\Pool $pool, int $workerId)use (&$table){
        _echo("进程（{$workerId}）启动成功");
        // 每10分钟统计一次邮箱数量
        \Swoole\Timer::tick(600000,function () use (&$table){
            $table->set('etotal',['val'=> db()->count(\Model\email::count())]);
        });

        // 每2秒执行一次
        \Swoole\Timer::tick(2000,function () use (&$table,&$pool){
            // 是否停止脚本
            $table->set('stop',['val'=> redis()->get(RUNNING_REDIS_KEY) === 'stop' ? 1 : 0]);
//            _echo('定时器');
            // 检查是否结束了所有的协程同步代码
            if ($table->get('stop','val')) {
                $stop_num = 0;
                foreach (range(0, WORKER_NUM) as $i) {
                    if ($table->exists('ps' . $i)) {
                        $stop_num++;
                    }
                }
                if($stop_num >= WORKER_NUM){
//                退出进程
                    $pool->shutdown();
                }
            }

            // 邮件总数
            $total = redis()->get('email_total',0);
            if($total > $table->get('etotal','val')){
                $table->set('etotal',['val'=> $total]);
            }

        });

        // 进行阻塞，否则定时器无法运行
        while (true){
            co::sleep(9999);
        }
    },true);


// 协程配置
    \co::set([
        'max_coroutine'=>COROUTINE_MAX_NUM, // 最大携程数量
        'hook_flags'=>SWOOLE_HOOK_TCP,  //  redis需要的配置
    ]);

    // 启动业务进程
    $pm->addBatch(WORKER_NUM,function (Process\Pool $pool, int $worker_id) use (&$table){
        _echo("进程（{$worker_id}）启动成功");

        // 协程id集
        $cid = [];

        $i = 0;
        $email_total = $table->get('etotal','val');//邮件总数量
        $isRunMaxCNum = 1; // 允许最大协程数量,如果为0则停止所有协程工作，相当于停止脚本
        // 是否退出进程
        while (true){
            $i++;
            // 每10秒 验证一次邮箱数量，好控制协程的数量
            if($i>=10){
                $email_total = $table->get('etotal','val');
                $i = 0;
            }

            if(!$email_total){
                break;
            }

            // 每个协程 分配 10个邮箱任务
            $cnum = ceil($email_total/40);
            // 当前协程的数量
            $nowCnum = count($cid);
            // 说明 需要新的协程了
            if($cnum > $nowCnum){
                // 开启所需要的协程数量
                foreach (range(0,$cnum-$nowCnum) as $v){
                    // 启动一个协程
                    create_coroutine($cid,$isRunMaxCNum,$worker_id);
                }
            }
            // 暂时没有实现 减少协程数量操作
//            else if ($cnum < $nowCnum){
//                // 说明 协程数量过多，小于了1个协程处理10个邮箱的，资源闲置情况
//                // 销毁多余协程
//                $isRunMaxCNum = $nowCnum - $cnum;
//            }

            // 每3秒检查一次是否要停止 协程
            if($i%3 === 0){
//                _echo('是否收到退出信号:'.$table->get('stop','val'));
                if ($table->get('stop','val')){
                    // 停止
                    $isRunMaxCNum = 0;
                }
            }

            // 这个是检查 cid的如果协程全部退出，则退出进程
            co::sleep(1);
            // 跳出无限循环了
            if(!$cid){
                _echo('正常关闭进程（'.$worker_id.'）');
                break;
            }
        }

        // 是否停止，这里进行阻塞
        if ($table->get('stop','val')){
            // 某个进程退出了
            $table->set('ps'.$worker_id,['val'=>1]);
            // 阻塞直到 主进程 kill掉所有子进程
            while (true){
                co::sleep(5);
            }
        }


    },true);


    $pm->start();

}

/**
 * 创建协程
 * @param array $cid
 * @param int $isRunMaxCNum
 * @param $worker_id
 * @author：dc
 * @time 2023/2/14 17:04
 */
function create_coroutine(array &$cid,int &$isRunMaxCNum,$worker_id){
    go(function () use (&$cid,&$isRunMaxCNum,$worker_id){
        // 协程id
        $cid[co::getCid()] = co::getCid();

        // 同步操作
        while (true){
            // 是否退出协程
            if(!$isRunMaxCNum){
//                _echo('协程（'.co::getCid().'）: stop '.$isRunMaxCNum);
                break;
            }

            // 开始同步
            try {
                sync();
            }catch (\Throwable $e){
                _echo($e->getMessage());
                logs($e->getMessage().PHP_EOL.$e->getTraceAsString(),LOG_PATH.'/'.$worker_id.'_'.co::getCid().'.log');
            }

            // 阻塞1秒
            co::sleep(1);
        }

        // 协程完成后执行的函数
        co::defer(function () use (&$cid,$worker_id){
            _echo('正常关闭进程（'.$worker_id.'）下的协程（'.co::getCid().'）');
            unset($cid[co::getCid()]);
        });

    });
}


/**
 * 开始同步
 * @author：dc
 * @time 2023/2/13 9:42
 */
function sync(){
    co::sleep(1);
}




switch ($argv[1]){
    case 'start':{
        start();
        break;
    }
    case 'stop':{
        \Co\run(function (){
            echo "正在退出程序...\n";
            redis()->set(RUNNING_REDIS_KEY,'stop');
            while (true){
                $num = exec("ps -ef | grep \"sync_email.php start\" | grep -v grep | wc -l");
                if(!$num){
                    break;
                }
                co::sleep(0.5);
            }
            echo "已退出程序\n";
        });
        break;
    }
    default:{
        break;
    }
}