It is a common task for me to need to run many nearly identical child processes. Not often but sometimes I need those child processes to communicate their results back to the parent process. I have created the following implementation to avoid boiler plate pcntl_fork() calls, and standardize the client code.
I will in line my questions with the relative components.
<?php namespace robbmj;
/* The MIT License (MIT) */
interface ChildWorker {
/**
* produce must return a string
*/
function produce();
}
interface ParentWorker {
/**
* $input will always be a string
*/
function consume($input);
}
1) Is the SocketPair class over engineering, I could easily just use an array?
2) Is there a preferred way to initialize class members in PHP? default arguments to the constructor, initialize the variables when they are declared or initialize in the constructor?
class SocketPair {
private $clientSock, $serverSock, $createTime;
function __construct($clientSock, $serverSock, $createTime = null) {
$this->clientSock = $clientSock;
$this->serverSock = $serverSock;
$this->createTime = isset($createTime) ? $createTime : time();
}
function passedAllotedTime($allotedTime) {
return $this->createTime + $allotedTime <= time();
}
function clientSock() {
return $this->clientSock;
}
function serverSock() {
return $this->serverSock;
}
function closeClient() {
socket_close($this->clientSock);
}
function closeServer() {
socket_close($this->serverSock);
}
}
3) Should I be using bean style method names setMaxChildren instead maxChildren?
class IPC {
// indented properly in the file, just trying to avoid horizontal scroll
private $pWorker, $cWorkers, $maxChildren, $maxWaitTime;
function __construct(ParentWorker $pWorker, array /* ChildWorker */ $cWorkers) {
$this->pWorker = $pWorker;
$this->cWorkers = $cWorkers;
$this->maxChildren = 0;
$this->maxWaitTime = 0;
}
/**
* Sets the maximum number of Child Processes that can be running at any one time.
* If set to 0, There is no limit.
*
* If $max is not a integer of is less than 0 an InvalidArgumentException is thrown
*/
public function maxChildren($max) {
if (!is_int($max) || $max < 0) {
throw new \InvalidArgumentException("max must be greater than or equal to 0");
}
$this->maxChildren = $max;
return $this;
}
/**
* Sets the maximum amount of time a child process can run for before the process is terminated.
* If set to 0, There is no limit.
*
* If $seconds is not a integer of is less than 0 an InvalidArgumentException is thrown
*/
public function maxWaitTime($seconds) {
if (!is_int($seconds) || $seconds < 0) {
throw new \InvalidArgumentException("seconds must be greater than or equal to 0");
}
$this->maxWaitTime = $seconds;
return $this;
}
4) Should the start method be reporting errors to the caller? Or would that be an example of a leaked implementation detail?
function start() {
$pids = array();
$sockets = array();
foreach ($this->cWorkers as $i => $cWorker) {
$socketPair = $this->makeSocketPair();
if (!$socketPair) {
continue;
}
$pid = pcntl_fork();
if ($pid === 0) {
$this->childProcess($socketPair, $cWorker);
exit(0);
}
else if ($pid > 0) {
$sockets[$pid] = $socketPair;
if ($this->maxChildren > 0 && (count($sockets) >= $this->maxChildren)) {
$this->reduceProcessCount($sockets, $this->maxChildren - 1);
}
}
}
$this->reduceProcessCount($sockets, 0);
}
5) Can anyone foresee an edge cases I am not accounting for?
6) Should I be trimming the output of $cWorker->produce()
protected function childProcess(SocketPair $socketPair, ChildWorker $cWorker) {
$socketPair->closeClient();
$output = $cWorker->produce();
socket_set_nonblock($socketPair->serverSock());
$output = ($output) ? trim($output) : '';
while ((strlen($output) > 0) && ($wrote = socket_write($socketPair->serverSock(), $output))) {
$output = substr($output, $wrote);
}
$socketPair->closeServer();
}
protected function parentProcess(SocketPair $socketPair) {
$socketPair->closeServer();
$content = '';
while ($line = socket_read($socketPair->clientSock(), 1129)) {
$len = strlen($content);
$content .= $line;
}
$socketPair->closeClient();
$this->pWorker->consume($content);
}
7) Should this makeSocketPair be static, should it be moved to the SocketPair?
8) Should the $sockets array created in start() be a class member? It would save me having to pass by reference to reduceProcessCount and killExpiredProcesses.
protected function makeSocketPair() {
$pair = array();
if (socket_create_pair(AF_UNIX, SOCK_STREAM, 0, $pair) === false) {
// TODO: install a logger, echoing is not cool
echo "socket_create_pair failed. Reason: " . socket_strerror(socket_last_error());
return null;
}
return new SocketPair($pair[0], $pair[1]);
}
protected function reduceProcessCount(array &$sockets, $to) {
while (count($sockets) > $to) {
$pid = pcntl_wait($status, WNOHANG);
if ($pid > 0) {
$this->parentProcess($sockets[$pid]);
unset($sockets[$pid]);
}
else {
$this->killExpiredProcesses($sockets);
var_dump(count($sockets));
usleep(200000);
}
}
}
protected function killExpiredProcesses(array &$sockets) {
if ($this->maxWaitTime) {
foreach ($sockets as $pid => $pair) {
if ($pair->passedAllotedTime($this->maxWaitTime)) {
$pair->closeServer();
$pair->closeClient();
unset($sockets[$pid]);
// TODO: install a logger, echoing is not cool
echo "PID: $pid took to long\n";
posix_kill($pid, SIGINT);
}
}
}
}} // closes the class
As a reference this is one example of using the library
<?php
require_once 'ipc.php';
class CurlWorker implements robbmj\ChildWorker {
private $url;
public function __construct($url) {
$this->url = $url;
}
public function produce() {
$ch = curl_init($this->url);
curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, 0);
curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, 0);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
$content = curl_exec($ch);
curl_close($ch);
return $content;
}
}
class PWorker implements robbmj\ParentWorker {
private $content = array();
public function consume($input) {
$this->content[] = $input;
}
public function getContent() {
foreach ($this->content as $value) {
$r[] = strlen($value);
}
return $r;
}
}
$cWorkers = [new CurlWorker('https://www.google.ca/'), new CurlWorker('http://php.net/')];
$p = new PWorker();
$ipc = (new robbmj\IPC($p, $cWorkers))
->maxWaitTime(0)
->maxChildren(1);
$ipc->start();
var_dump($p->getContent());
All comments, criticisms, suggestions for improvement, sarcastic comments, pictures of unicorns, etc. are welcome.
