フォークの処理 – コードの断片

<!–

–>

function async(Process $process) : Process 
    socket_create_pair(AF_UNIX, SOCK_STREAM, 0, $sockets);
    [$parentSocket, $childSocket] = $sockets;

    if (($pid = pcntl_fork()) == 0) 
        socket_close($childSocket);
        socket_write($parentSocket, serialize($process->execute()));
        socket_close($parentSocket);
        exit;
    

    socket_close($parentSocket);

    return $process
        ->setStartTime(time())
        ->setPid($pid)
        ->setSocket($childSocket);


function wait(array $processes) : array {
    $output = [];

    while (count($processes)) {
        foreach ($processes as $key => $process) {
            $processStatus = pcntl_waitpid($process->getPid(), $status, WNOHANG | WUNTRACED);

            if ($processStatus == $process->getPid()) 
                $output[] = unserialize(socket_read($process->getSocket(), 4096));
                socket_close($process->getSocket());
                $process->triggerSuccess();

                unset($processes[$key]);
             else if ($processStatus == 0)  else 
                throw new \Exception("Could not reliably manage process $process->getPid()");
            
        }
        
        if (!count($processes)) 
            break;
        

        usleep(100000);
    }

    return $output;
}

の Process 定義された方法でデータを渡すために使用されるクラス。

abstract class Process

    protected $pid;
    protected $name;
    protected $socket;
    protected $successCallback;
    protected $startTime;
    protected $maxRunTime = 300;
    
    public abstract function execute();

    public function onSuccess(callable $callback) : Process 
        $this->successCallback = $callback;

        return $this;
    

    public function triggerSuccess() 
        if (!$this->successCallback) 
            return null;
        

        return call_user_func_array($this->successCallback, [$this]);
    

    public function setPid($pid) : Process 
        $this->pid = $pid;

        return $this;
    

    public function getPid() 
        return $this->pid;
    

    public function setSocket($socket) : Process 
        $this->socket = $socket;

        return $this;
    

    public function getSocket() 
        return $this->socket;
    

    public function setName(string $name) : Process 
        $this->name = $name;

        return $this;
    

    public function getName() : string 
        return $this->name;
    

    public function setStartTime($startTime) 
        $this->startTime = $startTime;

        return $this;
    

    public function getStartTime() 
        return $this->startTime;
    

    public function setMaxRunTime(int $maxRunTime) : Process 
        $this->maxRunTime = $maxRunTime;

        return $this;
    

    public function getMaxRunTime() : int 
        return $this->maxRunTime;
    

具体的なプロセスの実装。

class MyProcess extends Process

    public function execute() 
        sleep(1);
        
        return true;
    

そして、すべてをまとめます。

$processA = async(new MyProcess());
$processB = async(new MyProcess());

$output = wait([$processA, $processB]);

print_r($output);
die('Done!');