Динамическое планирование задач в php pthreads?

Question

Я новичок в php pthreads для многопоточности.Мне удалось достичь своей цели путем создания пула, задания количества работников, использования цикла for для отправки задачи и использования пула для сбора результата.

Но я думаю, что задача связана с работником при отправкев бассейн.И я полагаю, что внутренняя логика состоит в том, чтобы сделать каждого работника равным или равным как можно большему числу задач.

В реальной реализации такая логика может вызвать проблему эффективности.Для разных данных, предоставленных для каждой задачи, это может привести к большой разнице во времени.Или, если работнику необходимо связаться с сервером для отправки / получения данных, он также сильно различается при передаче данных или даже по таймауту и ​​переподключению сервера, что приводит к дополнительной разнице в обработке другой задачи.

Для простого примера:4 задания передаются в пул, в котором есть 2 работника.Давайте предположим, что для выполнения каждой задачи требуется определенное время, независимо от того, с какой именно задачей она обрабатывается.

Задача 1 2 3 4

Время обработки 1 с 7 с 2 с 4 с

Насколько я понимаю, в php pthreads каждый работник собирает по 2 задачи.И когда задача передается в пул, пул всегда укладывает ее работнику с меньшим количеством задач.Предполагая, что нет никаких других издержек, рабочий № 1 обрабатывает задачи № 1 и № 3 за 3 секунды, а рабочий № 2 обрабатывает задачи № 2 и № 4 за 11 секунд.Таким образом, общее выполнение пула будет 11 с.Эффективность очень низкая, поскольку выполнение всех задач последовательно занимает 14 с.

Еще одним заметным побочным эффектом является то, что рабочий № 1 будет бездействовать в течение 8 секунд, ожидая завершения рабочего № 2.Это может вызвать тайм-аут сервера в реальном приложении.В моем приложении я многократно использую один и тот же пул.Мне нужно отключить соединение для работающего на холостом ходу на каждую итерацию и восстановить соединение при переходе на следующую итерацию.

Если динамическое планирование задач возможно, работник № 1 берет задачу № 1, а работник №2 поднимает задание № 2.Через 1 секунду работник # 1 берет задачу № 3, а работник № 2 продолжает обрабатывать задачу № 2.Еще через 2 секунды работник # 1 берет задачу № 4, а работник № 2 все еще обрабатывает задачу № 2.Еще через 4 секунды они оба завершили работу, и в пуле не осталось других задач, после чего они были закрыты.Таким образом, время выполнения пула = рабочий № 1 (1 с + 2 с + 4 с) = рабочий № 2 (7 с) = 7 с.Это в 2 раза быстрее и 100% эффективности потоков.Между тем, ни один из рабочих не будет простаивать, чтобы вызвать потенциальные проблемы.

Приведенный выше пример был составлен для демонстрации.Реальный мир был бы намного сложнее.Но у меня есть приложение, в котором самый медленный работник работает в 5 раз дольше, чем самый быстрый работник, что вызывает у меня сильную головную боль.

На самом деле я больше знаком с OpenMP для C ++.Он обеспечивает методы планирования с несколькими циклами .Я думаю, что метод планирования в потоках php является статическим.Хотелось бы, чтобы был способ реализовать динамический метод.

Answer 1

У меня возникла проблема, из-за которой одно длительное задание на работнике могло привести к тому, что задачи были бы наложены на этого работника, когда могли использоваться другие неработающие работники.

Поведение Pool :: submit () в соответствии с документацией:

Отправить задание следующему работнику в пуле

Для моего случая использования я расширил класс 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 () можно использовать для назначения задачи конкретный рабочий поток.

Этот метод может быть легко адаптирован к вашим потребностям в планировании.