PHP - Многопоточность и пулы - PullRequest
Новая
       17

PHP - Многопоточность и пулы

0 голосов
/ 16 января 2019

Я использую Pool объект в PHP pthread, и сделал следующий тестовый скрипт, чтобы увидеть, как пул должен работать. Я подумал, что пулинг должен выполнять заданное количество задач, открывать максимальное число рабочих x и назначать им задачи, и как только работник заканчивает задачу, если выполняется больше задач доступно, назначьте этому рабочему новое задание.

Учитывая приведенный ниже пример и приведенное выше предположение: 

class Work extends Threaded {
    public $id;

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

    public function run() {
        if ($this->id == 0) {
            sleep(3);
            echo $this->id . " is ready\n";
            return;
        } else {
            echo $this->id . " is ready\n";
            return;
        }
    }
}

$pool = new Pool(2, 'Worker', []);
for ($i=0; $i<4; $i++) $pool->submit(new Work($i));
while ($pool->collect());
$pool->shutdown();

Я ожидал, что этот скрипт выдаст следующую информацию:

1 готов
2 готово
3 готово
0 готово

потому что, по существу, доступно 2 работника, а из-за операции sleep, на которую наткнулся первый работник, задача 1,2,3 должна быть выполнена вторым работником.

Вместо этого я получаю вывод:

1 готово
3 готово
0 готово
2 готово

Понятно, что работник 1 получает задание 0 и задание 2 сразу же, таким образом, работник 2, после окончания заданий 1 и 3, просто ждет, вместо того, чтобы принять работу 2 у работника 1.

Это ошибка? Или это должно работать так?

Моя версия PHP: 

PHP 7.2.14 (cli) (built: Jan  9 2019 22:23:26) ( ZTS MSVC15 (Visual C++ 2017) x64 )
Copyright (c) 1997-2018 The PHP Group
Zend Engine v3.2.0, Copyright (c) 1998-2018 Zend Technologies

Ответы [ 2 ]

0 голосов
/ 20 марта 2019

Позвольте мне ответить: из того, что я знаю о pthreads в php, пул похож на число обрабатывающих php.exe, которые могут быть запущены одновременно.

Итак, в вашем случае вы определяете два пула с помощью new Pool(2, 'Worker', []);

Итак, давайте сделаем абстрактное объяснение по этому поводу. Есть 2 пула, назовите его PoolA и PoolB.

Цикл от 0 до 3, каждый цикл отправляет задачу в Пул.

Есть 4 задачи от 0 до 3, давайте их вызывать по task0, task1, task2, task3.

Когда цикл происходит, с моей точки зрения, это должна быть очередь, подобная этой 

PoolA -> submit task0
PoolB -> submit task1
PoolA -> submit task2
PoolB -> submit task3

Но с class Work это будет задача0, ... до задачи3.

Положение / состояние

Вы определяете некоторую логику в run () =>, когда параметр (в данном случае $ id из конструктора) равен 0, тогда sleep(3).

В этой ситуации PoolA означает, что task0, содержащий параметр ($ id), имеет значение 0, PoolA будет ожидать 3 секунды. PoolA также отправьте task2.

С другой стороны, PoolB отправить task1 и task3, из этой ситуации не нужно ждать 3 секунды.

Таким образом, при запуске while($pool->collect()); возможная очередь, которая вероятнее всего произойдет 

task1    (PoolB)
task3    (PoolB)
task0    (PoolA)  ->>>> PoolA delayed because from task0 needs to sleep for 3 seconds
task2    (PoolA)

Так что я думаю, что это правильно, когда выходные данные

1 готов
3 готово
0 готово
2 готово

Есть вопросы.

Почему задерживается только PoolA, даже если задерживается PoolA, почему задача 2 не отправляется в PoolB или почему задача 1 или задача 3 не передаются в PoolA ??

Ну, я тоже не понимаю. У меня есть задача, похожая на вашу, после многих экспериментов, я не уверен, pthreads , которые используют Pool & Threaded, это multi-threading or multiprocessing.

0 голосов
/ 17 января 2019

По какой-то причине мой Docker потерял самообладание теперь, когда я обновил Windows до 1809, поэтому публикация не прошла тестирование. (Извините, нет выходных данных для атм)

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

$pool = new Pool(2);
foreach ([0,1,2,3] as $count) {
    $pool->submit(
        new class ($count) extends Threaded
        {
            private $count;

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

            public function run()
            {
                if ($this->count== 0) {
                    sleep(3);
                    echo $this->count . " is ready\n";
                } else {
                    echo $this->count . " is ready\n";
                }
            }
        }
    );
}

while ($pool->collect());

$pool->shutdown();

Я использую анонимный класс (new class ($count) extends Threaded) в качестве параметра submit().

На сервере это работает отлично, используя экземпляр Docker с PHP ZTS 7.2.13 на Alpine 3.8

...