从 cron 运行时 PHP pthreads 失败答案

【问题标题】：PHP pthreads failing when run from cron从 cron 运行时 PHP pthreads 失败
【发布时间】：2016-10-29 20:38:38
【问题描述】：

好的，让我们慢慢开始吧……

我有一个 pthreads 脚本正在运行并为我工作，当我通过 ssh 从命令行手动运行它时，100% 的时间都经过测试和工作。脚本如下，主线程进程代码调整为模拟随机进程的运行时间。

class ProcessingPool extends Worker {
    public function run(){}
}
class LongRunningProcess extends Threaded implements Collectable {
    public function __construct($id,$data) {
        $this->id = $id;
        $this->data = $data;
    }

    public function run() {
        $data = $this->data;
        $this->garbage = true;

        $this->result = 'START TIME:'.time().PHP_EOL;

        // Here is our actual logic which will be handled within a single thread (obviously simulated here instead of the real functionality)
        sleep(rand(1,100));

        $this->result .= 'ID:'.$this->id.' RESULT: '.print_r($this->data,true).PHP_EOL;
        $this->result .= 'END TIME:'.time().PHP_EOL;

        $this->finished = time();
    }
    public function __destruct () {
        $Finished = 'EXITED WITHOUT FINISHING';
        if($this->finished > 0) {
            $Finished = 'FINISHED';
        }

        if ($this->id === null) {
            print_r("nullified thread $Finished!");
        } else {
            print_r("Thread w/ ID {$this->id} $Finished!");
        }
    }

    public function isGarbage() : bool { return $this->garbage; }

    public function getData() {
        return $this->data;
    }
    public function getResult() {
        return $this->result;
    }

    protected $id;
    protected $data;
    protected $result;
    private $garbage = false;
    private $finished = 0;
}

$LoopDelay = 500000; // microseconds
$MinimumRunTime = 300; // seconds (5 minutes)

// So we setup our pthreads pool which will hold our collection of threads
$pool = new Pool(4, ProcessingPool::class, []);

$Count = 0;

$StillCollecting = true;
$CountCollection = 0;
do {

    // Grab all items from the conversion_queue which have not been processed
    $result = $DB->prepare("SELECT * FROM `processing_queue` WHERE `processed` = 0 ORDER BY `queue_id` ASC");
    $result->execute();
    $rows = $result->fetchAll(PDO::FETCH_ASSOC);

    if(!empty($rows)) {

        // for each of the rows returned from the queue, and allow the workers to run and return
        foreach($rows as $id => $row) {
            $update = $DB->prepare("UPDATE `processing_queue` SET `processed` = 1 WHERE `queue_id` = ?");
            $update->execute([$row['queue_id']]);

            $pool->submit(new LongRunningProcess($row['fqueue_id'],$row));

            $Count++;
        }
    } else {
        // 0 Rows To Add To Pool From The Queue, Do Nothing...
    }


    // Before we allow the loop to move on to the next part, lets try and collect anything that finished
    $pool->collect(function ($Processed) use(&$CountCollection) {
        global $DB;

        $data = $Processed->getData();
        $result = $Processed->getResult();


        $update = $DB->prepare("UPDATE `processing_queue` SET `processed` = 2 WHERE `queue_id` = ?");
        $update->execute([$data['queue_id']]);

        $CountCollection++;

        return $Processed->isGarbage();
    });
    print_r('Collecting Loop...'.$CountCollection.'/'.$Count);


    // If we have collected the same total amount as we have processed then we can consider ourselves done collecting everything that has been added to the database during the time this script started and was running
    if($CountCollection == $Count) {
        $StillCollecting = false;
        print_r('Done Collecting Everything...');
    }

    // If we have not reached the full MinimumRunTime that this cron should run for, then lets continue to loop
    $EndTime = microtime(true);
    $TimeElapsed = ($EndTime - $StartTime);
    if(($TimeElapsed/($LoopDelay/1000000)) < ($MinimumRunTime/($LoopDelay/1000000))) {
        $StillCollecting = true;
        print_r('Ended To Early, Lets Force Another Loop...');
    }

    usleep($LoopDelay);

} while($StillCollecting);

$pool->shutdown();

所以虽然上面的脚本会通过命令行运行（已经调整为基本示例，详细处理代码已在上面示例中模拟） , 以下命令在每 5 分钟从 cron 设置运行时会给出不同的结果...

/opt/php7zts/bin/php -q /home/account/cron-entry.php file=every-5-minutes/processing-queue.php

上面的脚本，当使用上面的命令行调用时，会在脚本运行期间反复循环，并从数据库队列中收集任何新项目，并将它们插入池中，这样可以同时允许 4 个进程运行和完成的时间，然后收集并在另一个循环发生之前更新队列，从数据库中提取任何新项目。该脚本将一直运行，直到我们在脚本执行期间处理并收集了队列中的所有进程。如果脚本没有运行完整的 5 分钟预期时间，则强制循环继续检查队列，如果脚本运行超过 5 分钟标记，则它允许任何当前线程完成并在关闭之前被收集。请注意，上面的代码还包括一个基于代码的“flock”功能，它使这个空闲循环的未来 crons 并在锁定解除后退出或启动，确保队列和线程不会相互碰撞。同样，所有这些都通过 SSH 从命令行运行。

一旦我接受上述命令，并将其放入 cron 中以每 5 分钟运行一次，本质上给了我一个永无止境的循环，同时保持内存，我得到不同的结果...

该结果描述如下......脚本启动，检查flock，如果锁不存在则继续，它创建锁，并运行上述脚本。项目从数据库中的队列中取出，并插入到池中，池按预期一次触发 4 个线程。但是 意外的结果是 run() 命令没有似乎已执行，而是运行 __destruct 函数，并显示“线程 w/ ID 2 已完成！”消息类型返回到输出。这反过来意味着事物的收集端不收集任何东西，而启动脚本（cron 脚本本身 /home/account/cron-entry.php file=every-5-minutes/processing-queue.php ) 在所有东西都放入池中并销毁后完成。这过早地“完成”了 cron 作业，因为除了循环并从队列中提取任何新内容之外别无他法，因为在队列中处理 == 1 时它们被视为“正在处理”。

问题最终变成了...... 我如何让 cron 的脚本知道产生并运行（）它们的线程，而不会在它们可以做任何事情之前关闭池？

（注意...如果您复制/粘贴提供的脚本，请注意我在删除详细逻辑后没有对其进行测试，因此可能需要一些简单的修复...请不要挑剔说代码，因为这里的关键是，如果从命令行执行脚本，pthreads 可以工作，但当从 CRON 执行脚本时无法正常运行。如果您打算用非建设性的批评进行评论，请动动手指做别的事！）

乔·沃特金斯！我需要你的才华！提前致谢！

【问题讨论】：

标签： php linux cron pthreads

【解决方案1】：

毕竟，问题似乎与用户权限有关。我在 cpanel 中设置了这个特定的 cron，当手动运行命令时，我以 root 身份登录。

在根 crontab 中设置此命令后，我能够让它成功运行池中的线程。我现在唯一的问题是一些线程永远不会完成，有时我无法关闭池。但这是一个不同的问题，所以我将在别处提出另一个问题。

对于遇到此问题的人，请确保您知道 cron 的所有者是谁，因为这对 php 的 pthreads 很重要。

【讨论】：