php - 在 PHP7 pthreads 扩展中使用 Pool 类

Question

我采用了使用 Pool 类的 pthreads PHP7 扩展的最基本演示(此演示 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 和 6 12 但我不明白为什么。这种情况只是偶尔发生(可能有 1/10 次运行)。

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

然后自述文件说:

The Pool::collect mechanism was moved from Pool to Worker for a more robust Worker and simpler Pool inheritance.

所以我想我做错了什么。

编辑:我以How does Pool::collect works? 为例并更新它以使用最新的 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

Answer 1

您已经非常正确地指出，您复制的代码以 pthreads v2(适用于 PHP 5.x)为目标。

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

获取此数据的一种方法是将 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 中的接口(interface))已经由 Threaded 类实现，因此无需自己实现。
• 一旦一个任务被提交到池中，它就已经被认为是垃圾，所以没有必要自己处理这部分。虽然您仍然可以覆盖默认的垃圾收集器，但在绝大多数情况下(包括您的情况)不需要这样做。
• 我仍然调用 collect 方法(在一个循环中阻塞主线程，直到所有任务都完成执行)以便任务可以被垃圾收集(使用 pthreads 的默认收集器)以释放池执行任务时的内存。