Pool::collect 是如何工作的？

Question

帮助我了解 Pool::collect 的具体工作原理。

Pool::collect — 收集对已完成任务的引用

public void Pool::collect ( Callable $collector )

我的假设是：Pool::collect 注册了一个函数，该函数将在每个 \Threaded $task 完成后调用。所以，我做到了：

<?php
$pool = new Pool(4);
$pool->collect($collector);
$pool->submit(new Task);

没用。但以下是：

<?php
$pool = new Pool(4);
$pool->submit(new Task);
$pool->collect($collector);

所以，我猜Pool::collect 的作用是：将$collector 附加到之前提交的每个\Threaded $task。

现在，究竟什么时候调用了$collector？我假设在Threaded::run() 完成后调用。又错了。

<?php
class Task extends Threaded {
    public function run () { echo "Task complete\n"; }
}

$collector = function (\Task $task) {
    echo "Collect task\n";
    return true;
};

$pool = new Pool(4);
$pool->submit(new Task);
$pool->collect($collector);
$pool->shutdown();

输出：

Collect task
Task complete

$collector 在Threaded::run() 完成之前被调用。

---

文档没有说太多。事件不是说$collector 必须返回一个布尔值。 I didn't know that.

我试图在每个 $task 完成后使用 Pool::collect 作为一种回调。我想我走错了路。

编辑 1.What about this attempt?

Answer 1

Pool::collect 遍历一个对象列表，将每个对象传递给$collector。

当引擎可以销毁对工作列表中Threaded 对象的引用时，$collector 函数应该返回true。

在 PHP7 中

::collect 功能已移至 Worker，尽管它仍由 Pool 公开以用于实用程序。

有两个列表，一个是准备执行的项目列表，另一个是已执行（或当前正在执行）的项目。

Pool::collect 遍历第二个列表，即已执行（或当前正在执行）的项目。

Pool::collect 为Pool 中的所有Worker 对象返回垃圾列表中剩余的项目数，以帮助安排垃圾回收。

PHP7 代码

<?php
$pool = new Pool(4);

while (@$i++<10) {
    $pool->submit(new class($i) extends Threaded {
        public function __construct($id) {
            $this->id = $id;
        }

        public function run() {
            printf(
                "Hello World from %d\n", $this->id);
        }

        public $id;
    });
}

while ($pool->collect(function(Collectable $work){
    printf(
        "Collecting %d\n", $work->id);
    return $work->isGarbage();
})) continue;

$pool->shutdown();
?>

会产生类似的东西：

Hello World from 1
Hello World from 2
Hello World from 3
Hello World from 4
Hello World from 8
Collecting 1
Hello World from 7
Hello World from 5
Collecting 5
Hello World from 6
Collecting 9
Collecting 2
Collecting 6
Collecting 10
Hello World from 9
Collecting 3
Collecting 7
Collecting 4
Hello World from 10
Collecting 8
Collecting 5
Collecting 10

Answer 2

您错过了审核周期。 任务不能立即完成，所以起初收集功能没有收到肯定的结果。需要检查几次，需要一个循环。

在示例中，如果无限为真，则始终有 4 个任务在工作。否则，如果队列大小等于池大小（4），则结束，可以退出循环。

<?php

class Task extends \Threaded
{

    protected $result;
    protected $completed = false;

    public function run()
    {
        echo "Task complete\n";
        $this->result = md5(rand(1, 2000));
        sleep(rand(1, 5));
        $this->completed = true;
    }

    public function isCompleted()
    {
        return $this->completed;
    }

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

$infinite = false;
$poolSize = 4;
$queue = array();

$pool = new \Pool($poolSize);
$pool->submit(new Task);
$pool->submit(new Task);
$pool->submit(new Task);
$pool->submit(new Task);


do {
    if($infinite === true){
        $queue = array();
    }

    $pool->collect(function (\Task $task) use (&$queue) {
        if ($task->isCompleted()) {
            echo "Collect task\n";
            $queue[] = $task->getResult();

            return true;
        } else {
            echo "task not complete\n";

            return false;
        }
    });

    $size = sizeof($queue);
    if ($size > 0) {

        echo $size . " result\n";
        print_r($queue);

        if($infinite === true) {
            for ($m = 0; $m < $size; $m++) {
                $pool->submit(new Task);
            }
        } else{
            if($size == $poolSize){
                break;
            }
        }
    }

    usleep(100000);

} while (true);
$pool->shutdown();