时间戳

首页 > 解决方案 > php pthreads中的动态任务调度？

php - php pthreads中的动态任务调度？

问题描述

我是多线程 php pthreads 的新手。我已经通过建立一个池、设置一些工人、使用 for 循环提交任务和使用池来收集结果来实现我的目标。

但我认为在提交到池时任务与工作人员绑定。而且我相信内部逻辑是让每个worker获得相等或尽可能相等的任务数量。

在实际实现中，这样的逻辑可能会导致效率问题。给每个任务不同的数据，可能会产生很大的时间差异。或者如果worker需要与服务器通信来发送/获取数据，它在数据传输甚至服务器超时和重新连接方面也有很大差异，这导致处理不同任务的进一步差异。

举个简单的例子，4 个任务被提交到一个有 2 个工作人员的池中。让我们假设每个任务都需要一定的时间来处理，不管它是由哪个工作人员处理的。

任务 1 2 3 4

处理时间 1s 7s 2s 4s

我的理解是，在 php pthreads 中，每个工作人员堆叠 2 个要收集的任务。并且当一个任务被提交到池中时，池总是将它堆叠给任务较少的工作者。假设没有其他开销，worker #1 在 3 秒内处理任务 #1 和 #3，而 worker #2 在 11 秒内处理任务 #2 和 #4。因此，总池执行时间为 11 秒。由于顺序运行所有任务需要 14 秒，因此效率非常低。

另一个值得注意的副作用是工人 #1 将空闲 8 秒等待工人 #2 完成。在实际应用中可能会导致服务器超时。在我的应用程序中，我反复重用同一个池。我必须在每次迭代时关闭空闲工作人员的连接，并在它移动到下一次迭代时重新建立连接。

如果可以进行动态任务调度，则工人 #1 会接手任务 #1，而工人 #2 会接手任务 #2。1 秒后，工人 #1 拿起任务 #3，而工人 #2 仍在处理任务 #2。再过 2 秒后，工人 #1 拿起任务 #4，而工人 #2 仍在处理任务 #2。又过了 4 秒，它们都完成了，池中没有其他任务，然后它们被关闭。所以池执行时间 = worker #1 (1s + 2s + 4s) = worker #2 (7s) = 7s。这是完美的 2 倍加速和 100% 的线程效率。同时，任何一个工人都不会闲着造成潜在的问题。

上面的例子是为了演示而编造的。现实世界会复杂得多。但是我碰巧有一个应用程序，最慢的工作人员比最快的工作人员运行时间长 5 倍，这让我很头疼。

实际上，我更熟悉 OpenMP for C++。它提供了多种循环调度方法。我认为php线程中的调度方法是静态的。我希望有一种方法可以完成动态方法。

标签： phpmultithreadingpthreads

解决方案

我遇到了一个问题，即当有其他空闲的工作人员可以使用时，工作人员上的单个长时间运行的任务会导致任务堆叠在该工作人员上。

Pool::submit()根据其文档的行为是：

将任务提交给池中的下一个 Worker

对于我的用例，我扩展了 Pool 类：

namespace Task\Runner;

class Pool extends \Pool
{
    public function getIdleWorker() : ?int
    {
        if (empty($this->workers)) {
            return null;
        }

        $idleWorkerIndexes = [];

        foreach ($this->workers as $i => $worker) {
            if ($worker->getStacked() == 0)
                $idleWorkerIndexes[] = $i;
        }

        if (empty($idleWorkerIndexes))
            return null;

        return $idleWorkerIndexes[mt_rand(0, count($idleWorkerIndexes)-1)];
    }
}

我的任务运行器中的用法：

$workerIndex = $this->pool->getIdleWorker();
if (is_null($workerIndex))
    $this->pool->submit($taskWorkUnit);
else
    $this->pool->submitTo($workerIndex, $taskWorkUnit);

需要调用Pool::submit()来创建不超过您的池大小的新工作线程，然后可以使用Pool::submitTo()将任务分配给特定的工作线程。

这种方法可以很容易地适应您自己的调度需求。

推荐阅读