zentao/module/cron/control.php

<?php
/**
 * The control file of cron of ZenTaoPMS.
 *
 * @copyright   Copyright 2009-2023 禅道软件（青岛）有限公司(ZenTao Software (Qingdao) Co., Ltd. www.cnezsoft.com)
 * @license     ZPL(http://zpl.pub/page/zplv12.html) or AGPL(https://www.gnu.org/licenses/agpl-3.0.en.html)
 * @author      Yidong Wang <yidong@cnezsoft.com>
 * @package     cron
 * @version     $Id$
 * @link        https://www.zentao.net
 */
class cron extends control
{
    /**
     * 定时任务首页。
     * Index page.
     *
     * @access public
     * @return void
     */
    public function index()
    {
        $this->view->title = $this->lang->cron->common;
        $this->view->crons = $this->cron->getCrons();
        $this->display();
    }

    /**
     * 开启/关闭定时任务功能。
     * Turnon cron.
     *
     * @access public
     * @return void
     */
    public function turnon()
    {
        $turnon = empty($this->config->global->cron) ? '1' : '0';
        $this->loadModel('setting')->setItem('system.common.global.cron', $turnon);
        return $this->sendSuccess(array('load' => inlink('index')));
    }

    /**
     * 开启进程。
     * Open cron process.
     *
     * @access public
     * @return void
     */
    public function openProcess()
    {
        $this->display();
    }

    /**
     * 创建一个定时任务。
     * Create cron.
     *
     * @access public
     * @return void
     */
    public function create()
    {
        if($_POST)
        {
            $this->cron->create();
            if(dao::isError()) return $this->send(array('result' => 'fail', 'message' => dao::getError()));
            return $this->sendSuccess(array('load' => inlink('index'), 'closeModal' => true));
        }

        $this->view->title = $this->lang->cron->create . $this->lang->cron->common;
        $this->display();
    }

    /**
     * 编辑一个定时任务。
     * Edit cron.
     *
     * @param  int    $cronID
     * @access public
     * @return void
     */
    public function edit($cronID)
    {
        if($_POST)
        {
            $this->cron->update($cronID);
            if(dao::isError()) return $this->send(array('result' => 'fail', 'message' => dao::getError()));
            return $this->sendSuccess(array('load' => inlink('index'), 'closeModal' => true));
        }

        $this->view->title = $this->lang->cron->edit . $this->lang->cron->common;
        $this->view->cron  = $this->cron->getById($cronID);
        $this->display();
    }

    /**
     * 修改一个定时任务状态。
     * Toggle run cron.
     *
     * @param  int    $cronID
     * @param  string $status
     * @access public
     * @return void
     */
    public function toggle($cronID, $status)
    {
        $this->cron->changeStatus($cronID, $status);
        return $this->send(array('result' => 'success', 'load' => true));
    }

    /**
     * 删除一个定时任务。
     * Delete cron.
     *
     * @param  int    $cronID
     * @access public
     * @return void
     */
    public function delete($cronID)
    {
        $this->dao->delete()->from(TABLE_CRON)->where('id')->eq($cronID)->exec();
        return $this->sendSuccess(array('load' => true));
    }

    /**
     * 使用Ajax请求执行定时任务.
     * Ajax execute cron.
     *
     * @param  bool $restart
     * @access public
     * @return void
     */
    public function ajaxExec($restart = false)
    {
        if(empty($this->config->global->cron)) return;

        /* Run as daemon. */
        ignore_user_abort(true);
        set_time_limit(0);
        session_write_close();

        $execId = mt_rand();
        if($restart)
        {
            $this->cron->restartCron($execId);
            return $this->sendSuccess(array('load' => true));
        }

        while(true)
        {
            /* Only one scheduler and max 4 consumers. */
            $roles = $this->applyExecRoles($execId);
            if(empty($roles))
            {
                ignore_user_abort(false);
                return;
            }

            try
            {
                if(in_array('scheduler', $roles)) $this->schedule($execId);
                if(in_array('consumer', $roles))  $this->consumeTasks($execId);
                $this->cron->logCron($execId . " finished job at " . date("Y-m-d H:i:s") . "\n", true);
            }
            catch(Exception $e)
            {
                $this->cron->logCron($execId . " error: " . $e->getMessage());
            }

            sleep(20);
        }
    }

    /**
     * RoadRunner的调度进程。
     * Schedule cron task by RoadRunner.
     *
     * @access public
     * @return void
     */
    public function rrSchedule()
    {
        if('cli' !== PHP_SAPI) return;

        set_time_limit(0);
        session_write_close();

        $execId = mt_rand();
        $this->cron->restartCron($execId);

        $this->loadModel('common');

        while(true)
        {
            if(empty($this->config->global->cron))
            {
                sleep(60);
                continue;
            }

            if($this->canSchedule($execId)) $this->schedule($execId);

            sleep(20);
        }
    }

    /**
     * RoadRunner的消费进程。
     * Consume cron task by RoadRunner.
     *
     * @access public
     * @return void
     */
    public function rrConsume()
    {
        if('cli' !== PHP_SAPI) return;

        set_time_limit(0);
        session_write_close();

        $this->loadModel('common');

        $execId = mt_rand();
        while(true)
        {
            if(empty($this->config->global->cron))
            {
                sleep(60);
                continue;
            }

            $this->consumeTasks($execId);
            sleep(20);
        }
    }

    /**
     * 检查该execId是否可以进行调度(最近1分钟没有其他进程在调度).
     * Check the execId can schedule(No other execId scheduled 1 minutes ago).
     *
     * @param  int    $execId
     * @access protected
     * @return bool
     */
    protected function canSchedule($execId)
    {
        $settings = $this->dao->select('`key`,`value`')->from(TABLE_CONFIG)->where('owner')->eq('system')->andWhere('module')->eq('cron')->andWhere('section')->eq('scheduler')->fetchPairs();
        if(!isset($settings['execId']) || $settings['execId'] == $execId) return true;
        if(!isset($settings['lastTime']) || $settings['lastTime'] < date('Y-m-d H:i:s', strtotime('-1 minute'))) return true;

        return false;
    }

    /**
     * 检查该execId是否可以执行任务(最多允许1个调度进程，4个执行进程).
     * Check the execId can exec(1 scheduler, 4 consumers).
     *
     * @param  int    $execId
     * @access protected
     * @return array
     */
    protected function applyExecRoles($execId)
    {
        $roles = array();

        $settings = $this->dao->select('id,section,`key`,value')->from(TABLE_CONFIG)->where('owner')->eq('system')->andWhere('module')->eq('cron')->fetchAll();

        $scheduler = array('execId' => 0, 'lastTime' => '');
        $consumers = [];

        /* 生成scheduler和consumer的配置信息。 Set scheduler config and consumers. */
        $expirDate = date('Y-m-d H:i:s', strtotime('-1 minute'));
        foreach($settings as $setting)
        {
            if($setting->section == 'scheduler')
            {
                if($setting->key == 'lastTime')
                {
                    if($setting->value < $expirDate) continue;
                    $scheduler['lastTime'] = $setting->value;
                }
                if($setting->key == 'execId') $scheduler['execId'] = $setting->value;
            }
            elseif($setting->section == 'consumer')
            {
                if($setting->value < $expirDate)
                {
                    $this->dao->delete()->from(TABLE_CONFIG)->where('id')->eq($setting->id)->exec();
                    continue;
                }
                $consumers[$setting->key] = $setting->key;
            }
        }

        /* 如果当前进程ID是scheduler，或者没有活动的scheduler，该进程抢占为scheduler. */
        if($scheduler['execId'] == $execId || $scheduler['lastTime'] < $expirDate) $roles[] = 'scheduler';

        /* 如果当前进程ID是consumer，或者consumer数量不足，该进程抢占为consumer. */
        if(in_array($execId, $consumers) || count($consumers) < $this->config->cron->maxConsumer) $roles[] = 'consumer';

        $this->cron->logCron("$execId apply roles: [" . implode(',', $roles). "]\n", true);
        return $roles;
    }

    /**
     * 调度生成队列任务.
     * Schedule, push tasks to queue.
     *
     * @param  int    $execId
     * @access public
     * @return bool
     */
    public function schedule($execId)
    {
        if(empty($this->config->global->cron)) return;

        $this->dao->clearCache();
        $this->loadModel('common');
        $this->cron->updateTime('scheduler', $execId);

        /* Get and parse crons. */
        $tasks = $this->dao->select('cron,MAX(`createdDate`) `datetime`')->from(TABLE_QUEUE)->groupBy('cron')->fetchAll('cron');
        $crons = $this->cron->getCrons('nostop');
        foreach($crons as $cron)
        {
            /* 最后一次创建队列任务的时间作为cron的最新执行时间。 The lasttime of task in queue is the last runtime of cron. */
            $cron->datetime = isset($tasks[$cron->id]) ? $tasks[$cron->id]->datetime : '1970-01-01';
        }

        $parsedCrons = $this->cron->parseCron($crons);
        $now         = date(DT_DATETIME1);

        foreach($parsedCrons as $id => $cron)
        {
            $cronInfo = $crons[$id];

            /* Skip empty and stop cron.*/
            if(empty($cronInfo) || $cronInfo->status == 'stop') continue;

            if(!$cron['command'] || !isset($crons[$id])) continue;

            /* Check time. */
            if($now < $cron['time']->format(DT_DATETIME1)) continue;

            /* Push task into queue. */
            $task = new stdclass();
            $task->cron        = $id;
            $task->type        = $crons[$id]->type;
            $task->command     = $cron['command'];
            $task->createdDate = $now;
            $this->dao->insert(TABLE_QUEUE)->data($task)->exec();

            $log = date('G:i:s') . " schedule\ncronId: $id\nexecId: $execId\noutput: push task to queue\n\n";
            $this->cron->logCron($log);
        }
    }

    /**
     * 执行所有定时任务.
     * Execute all tasks.
     *
     * @param  int    $execId
     * @access public
     * @return bool
     */
    public function consumeTasks($execId)
    {
        while(true)
        {
            $this->cron->updateTime('consumer', $execId);

            /* Consume. */
            $task = $this->dao->select('*')->from(TABLE_QUEUE)->where('status')->eq('wait')->orderBy('createdDate')->fetch();
            if(!$task) break;

            $this->consumeTask($execId, $task);
        }
    }

    /**
     * 执行一个定时任务.
     * Execute one task.
     *
     * @param  int    $execId
     * @param  object $task
     * @access public
     * @return bool
     */
    public function consumeTask($execId, $task)
    {
        $this->dao->clearCache();

        if(!empty($task->command))
        {
            $affectedRows = $this->dao->update(TABLE_QUEUE)
                ->set('status')->eq('doing')->set('execId')->eq($execId)
                ->where('id')->eq($task->id)
                ->andWhere('status')->eq('wait')
                ->andWhere('execId')->eq('0')
                ->exec();
            if($affectedRows != 1) return;

            /* Other executor may execute the task at the same time，so we mark execId and wait 0.5s to check whether we own it. */
            usleep(500000);
            $task = $this->dao->select('*')->from(TABLE_QUEUE)->where('id')->eq($task->id)->fetch();
            if($task->execId != $execId) return;

            /* Execution command. */
            $output = '';
            $return = '';

            unset($_SESSION['company']);
            unset($this->app->company);

            /* Mark that this request was triggered by the scheduled task, not by the user. */
            $_SESSION['fromCron'] = true;

            $this->loadModel('common');
            $this->common->setCompany();
            $this->common->loadConfigFromDB();

            try
            {
                if($task->type == 'zentao')
                {
                    parse_str($task->command, $params);
                    if(isset($params['moduleName']) and isset($params['methodName']))
                    {
                        $this->viewType = 'html';
                        $moduleName     = $params['moduleName'];
                        $methodName     = $params['methodName'];
                        $this->app->loadLang($moduleName);
                        $this->app->loadConfig($moduleName);
                        unset($params['moduleName'], $params['methodName']);
                        $output = $this->fetch($moduleName, $methodName, $params);
                    }
                }
                elseif($task->type == 'system')
                {
                    exec($task->command, $out, $return);
                    if($out) $output = implode(PHP_EOL, $out);
                }
            }
            catch(EndResponseException $endResponseException)
            {
                $output = $endResponseException->getContent();
            }
            catch(Exception $e)
            {
                $output = $e;
            }
        }

        $this->dao->update(TABLE_QUEUE)->set('status')->eq('done')->where('id')->eq($task->id)->exec();
        $this->dao->update(TABLE_CRON)->set('lastTime')->eq(date(DT_DATETIME1))->where('id')->eq($task->cron)->exec();

        $log = date('G:i:s') . " execute\ncronId: {$task->cron}\nexecId: $execId\ntaskId: {$task->id}\ncommand: {$task->command}\nreturn : $return\noutput : $output\n\n";
        $this->cron->logCron($log);

        return true;
    }
}