PHP で複数のプロセスを並列して起動

React の真似事です。

PHP にスレッドはありませんが、stream_select() と stream_set_blocking() を使用することで、IO を多重化させ、並列処理を行うことができます。

初めに 1 秒置きに標準出力に出力し、最後に標準エラー出力に出力するだけの簡単なプログラムを用意します。
引数に渡した数値の数だけループします。

slow.rb

ARGV[0].to_i.times do |n|
  STDOUT.puts n
  STDOUT.flush
  sleep 1
end
STDERR.puts "Bye"
STDERR.flush

次にこれを並列に起動するプログラムを PHP で実装します。
かなり汚いですが。

parallel_proc.php

<?php
class ProcLoop
{
    private $procs = array();
    private $pipes = array();
    private $stdoutCallbacks = array();
    private $stderrCallbacks = array();
    private static $fdSpecs = array(
        1 => array('pipe', 'w'),
        2 => array('pipe', 'w'),
    );

    public function addProc($cmd, $stdoutCallback, $stderrCallback)
    {
        $this->procs[] = proc_open($cmd, static::$fdSpecs, $pipes);
        stream_set_blocking($pipes[1], 0);
        stream_set_blocking($pipes[2], 0);
        $this->pipes[] = array($pipes[1], $pipes[2]);
        $this->stdoutCallbacks[] = $stdoutCallback;
        $this->stderrCallbacks[] = $stderrCallback;
    }

    public function run()
    {
        $count = count($this->procs);
        while ($this->isAnyPipeAvailable()) {
            for ($i = 0; $i < $count; $i++) {
                $ret = stream_select($tmp = $this->pipes[$i], $write = NULL, $except = NULL, 1);
                if ($ret === false) {
                    throw new RuntimeException;
                } else if ($ret !== 0) {
                    foreach ($this->pipes[$i] as $sock) {
                        if ($buf = fread($sock, 4096)) {
                            if ($buf) {
                                if ($sock === $this->pipes[$i][0]) {
                                    call_user_func($this->stdoutCallbacks[$i], $buf);
                                } else {
                                    call_user_func($this->stderrCallbacks[$i], $buf);
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    private function isAnyPipeAvailable()
    {
        foreach ($this->pipes as $pipe) {
            if (feof($pipe[0]) === false || feof($pipe[1]) === false) {
                return true;
            }
        }
        return false;
    }
}

$loop = new ProcLoop;

$echoStdout = function ($buf) {
    echo "\033[34m[STDOUT]\033[0m " . $buf;
};

$echoStderr = function ($buf) {
    echo "\033[31m[STDERR]\033[0m " . $buf;
};

$loop->addProc('ruby slow.rb 3', $echoStdout, $echoStderr);
$loop->addProc('ruby slow.rb 6', $echoStdout, $echoStderr);
$loop->addProc('ruby slow.rb 9', $echoStdout, $echoStderr);
$loop->run();

これを実行すると、以下のような出力が得られます。

3 つのプロセスが同時に起動し、起動時間の短いものから順番に終了していることがわかります。
全てのプロセスが終了した時点 (というかパイプが終端に到達した時点) でこの PHP スクリプトも終了します。

React でこういうのをやる場合、どんな書き方をするのがキレイなのか知りたい (というかまだ用意されていない?)

See also

php-uv