在 PHP7 pthreads 扩展中使用 Pool 类

我做了 pthreads PHP7 扩展的最基本的演示，它使用Pool类（这个演示https://github.com/krakjoe/pthreads#polyfill）并稍微扩展它，这样我就可以从线程中获取结果（或者至少我认为我可以）：

$pool = new Pool(4);

foreach (range(1, 8) as $i) {
    $pool->submit(new class($i) extends Threaded
    {
        public $i;
        private $garbage = false;

        public function __construct($i)
        {
            $this->i = $i;
        }

        public function run()
        {
            echo "Hello World\n";
            $this->result = $this->i * 2;
            $this->garbage = true;
        }

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

while ($pool->collect(function(Collectable $task) {
    if ($task->isGarbage()) {
        echo $task->i . ' ' . $task->result . "\n";
    }
    return $task->isGarbage();
})) continue;

$pool->shutdown();

让我感到困惑的是，它有时无法获得所有任务的结果：

Hello World
Hello World
Hello World
Hello World
Hello World
1 2
2 4
Hello World
Hello World
3 6
Hello World
7 14
4 8
8 16

现在有两行5 10 and 6 12失踪了，但我不明白为什么。这种情况仅有时发生（可能是 1/10 次运行）。

看起来原始演示是针对旧版本的pthreads因为有Collectable现在自动实现的接口Threaded如果我没错的话。

然后自述文件说：

Pool::collect 机制从 Pool 转移到 Worker，以获得更健壮的 Worker 和更简单的 Pool 继承。

所以我想我做错了什么。

Edit: 我举的例子是Pool::collect 是如何工作的？并更新它以使用最新的 pthreads 和当前的 PHP7，但结果是相同的。看起来它无法从最后执行的线程收集结果。

$pool = new Pool(4);

while (@$i++<10) {
    $pool->submit(new class($i) extends Thread implements Collectable {
        public $id;
        private $garbage;

        public function __construct($id) {
            $this->id = $id;
        }

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

        public function setGarbage() {
            $this->garbage = true;
        }

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

    });
}

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

$pool->shutdown();

这输出以下内容，显然没有收集所有线程：

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

正如您非常正确地指出的那样，您复制的代码针对 pthreads v2（适用于 PHP 5.x）。

问题归结为 pthread 中的垃圾收集器不是确定性的。这意味着它的行为无法预测，因此无法可靠地使用它来从池已执行的任务中获取数据。

获取此数据的一种方法是传入Threaded对象放入正在提交到池中的任务中：

<?php

$pool = new Pool(4);
$data = [];

foreach (range(1, 8) as $i) {
    $dataN = new Threaded();
    $dataN->i = $i;

    $data[] = $dataN;

    $pool->submit(new class($dataN) extends Threaded {
        public $data;

        public function __construct($data)
        {
            $this->data = $data;
        }

        public function run()
        {
            echo "Hello World\n";
            $this->data->i *= 2;
        }
    });
}

while ($pool->collect());

$pool->shutdown();

foreach ($data as $dataN) {
    var_dump($dataN->i);
}

上面的代码有几点需要注意：

• Collectable（现在是 pthreads v3 中的一个接口）由Threaded类已经存在，所以不需要自己实现。
• 一旦任务被提交到池中，它就已经被认为是垃圾，因此不需要自己处理这部分。虽然您仍然可以覆盖默认的垃圾收集器，但在绝大多数情况下（包括您的情况）不需要这样做。
• 我仍然调用collect方法（在阻塞主线程的循环中，直到所有任务执行完毕），以便在池执行任务时可以对任务进行垃圾收集（使用 pthread 的默认收集器）以释放内存。