{

/**

protected $slaves = [];

/**

protected $loop;

/**

protected $controller;

/**

protected $web;

/**

protected $tcpConnector;

/**

protected $slaveCount = 1;

/**

protected $waitForSlaves = true;

/**

protected $isRunning = false;

/**

protected $index = 0;

/**

protected $bridge;

/**

protected $appBootstrap;

/**

protected $appenv;

/**

protected $debug = false;

/**

protected $logging = true;

/**

protected $host = '127.0.0.1';

/**

protected $port = 8080;

/**

protected $inReload = false;

/**

protected $output;

/**

protected $maxRequests = 1000;

protected $filesToTrack = [];

protected $filesLastMTime = [];

/**

function __construct(OutputInterface $output, $port = 8080, $host = '127.0.0.1', $slaveCount = 8)

{

$this->output = $output;

$this->slaveCount = $slaveCount;

$this->host = $host;

$this->port = $port;

register_shutdown_function([$this, 'signal']);

}

/**

public function signal()

{

//this method is also called during startup when something crashed, so

//make sure we don't operate on nulls.

$this->output->writeln('<info>Termination received, exiting.</info>');

if ($this->controller) {

@$this->controller->shutdown();

}

if ($this->web) {

@$this->web->shutdown();

}

if ($this->loop) {

$this->loop->tick();

$this->loop->stop();

}

foreach ($this->slaves as $slave) {

if (is_resource($slave['process'])) {

proc_terminate($slave['process']);

}

if ($slave['pid']) {

//make sure its dead

posix_kill($slave['pid'], SIGKILL);

}

}

exit;

}

/**

public function setMaxRequests($maxRequests)

{

$this->maxRequests = $maxRequests;

}

/**

public function setBridge($bridge)

{

$this->bridge = $bridge;

}

/**

public function getBridge()

{

return $this->bridge;

}

/**

public function setAppBootstrap($appBootstrap)

{

$this->appBootstrap = $appBootstrap;

}

/**

public function getAppBootstrap()

{

return $this->appBootstrap;

}

/**

public function setAppEnv($appenv)

{

$this->appenv = $appenv;

}

/**

public function getAppEnv()

{

return $this->appenv;

}

/**

public function isLogging()

{

return $this->logging;

}

/**

public function setLogging($logging)

{

$this->logging = $logging;

}

/**

public function isDebug()

{

return $this->debug;

}

/**

public function setDebug($debug)

{

$this->debug = $debug;

}

/**

public function run()

{

gc_disable(); //necessary, since connections will be dropped without reasons after several hundred connections.

$this->loop = \React\EventLoop\Factory::create();

$this->controller = new \React\Socket\Server($this->loop);

$this->controller->on('connection', array($this, 'onSlaveConnection'));

$this->controller->listen(5500);

$this->web = new \React\Socket\Server($this->loop);

$this->web->on('connection', array($this, 'onWeb'));

$this->web->listen($this->port, $this->host);

$this->tcpConnector = new \React\SocketClient\TcpConnector($this->loop);

$pcntl = new \MKraemer\ReactPCNTL\PCNTL($this->loop);

$pcntl->on(SIGTERM, [$this, 'signal']);

$pcntl->on(SIGINT, [$this, 'signal']);

$this->isRunning = true;

$loopClass = (new \ReflectionClass($this->loop))->getShortName();

$this->output->writeln("<info>Starting PHP-PM with {$this->slaveCount} workers, using {$loopClass} ...</info>");

for ($i = 0; $i < $this->slaveCount; $i++) {

$this->newInstance(5501 + $i);

}

$this->loop->run();

}

/**

public function onWeb(\React\Socket\Connection $incoming)

{

if ($this->isDebug()) {

$this->checkChangedFiles();

}

// preload sent data from $incoming to $buffer, otherwise it would be lost,

// since getNextSlave is async.

$redirect = null;

$buffer = '';

$incoming->on(

'data',

function ($data) use (&$redirect, &$buffer) {

if (!$redirect) {

$buffer .= $data;

}

}

);

$this->getNextSlave(

function ($id) use ($incoming, &$buffer, &$redirect) {

$slave =& $this->slaves[$id];

$slave['busy'] = true;

$slave['connections']++;

$this->tcpConnector->create('127.0.0.1', $slave['port'])->then(

function (\React\Stream\Stream $stream) use (&$buffer, $redirect, $incoming, &$slave) {

$stream->write($buffer);

$stream->on(

'close',

function () use ($incoming, &$slave) {

$slave['busy'] = false;

$slave['connections']--;

$slave['requests']++;

$incoming->end();

if ($slave['requests'] > $this->maxRequests) {

$info['ready'] = false;

$slave['connection']->close();

}

if ($slave['closeWhenFree']) {

$slave['connection']->close();

}

}

);

$stream->on(

'data',

function ($data) use ($incoming) {

$incoming->write($data);

}

);

$incoming->on(

'data',

function ($data) use ($stream) {

$stream->write($data);

}

);

$incoming->on(

'close',

function () use ($stream) {

$stream->close();

}

);

}

);

}

);

}

/**

protected function getNextSlave($cb)

{

$that = $this;

$checkSlave = function () use ($cb, $that, &$checkSlave) {

$minConnections = null;

$minPort = null;

foreach ($this->slaves as $slave) {

if (!$slave['ready']) {

continue;

}

// we pick a slave that currently handles the fewest connections

if (null === $minConnections || $slave['connections'] < $minConnections) {

$minConnections = $slave['connections'];

$minPort = $slave['port'];

}

}

if (null !== $minPort) {

$cb($minPort);

return;

}

$this->loop->futureTick($checkSlave);

};

$checkSlave();

}

/**

public function onSlaveConnection(\React\Socket\Connection $conn)

{

$buffer = '';

$conn->on(

'data',

\Closure::bind(

function ($data) use ($conn, &$buffer) {

$buffer .= $data;

if (substr($buffer, -1) === PHP_EOL) {

foreach (explode(PHP_EOL, $buffer) as $message) {

if ($message) {

$this->processMessage($message, $conn);

}

}

$buffer = '';

}

},

$this

)

);

$conn->on(

'close',

\Closure::bind(

function () use ($conn) {

foreach ($this->slaves as $id => $slave) {

if ($slave['connection'] === $conn) {

if ($this->output->isVerbose()) {

$this->output->writeln('Worker closed '.$slave['port']);

}

$slave['ready'] = false;

$slave['stdout']->close();

$slave['stderr']->close();

if (is_resource($slave['process'])) {

proc_terminate($slave['process'], SIGKILL);

}

posix_kill($slave['pid'], SIGKILL); //make sure its really dead

$this->newInstance($slave['port']);

return;

}

}

},

$this

)

);

}

/**

public function processMessage($data, \React\Socket\Connection $conn)

{

$array = json_decode($data, true);

$method = 'command' . ucfirst($array['cmd']);

if (is_callable(array($this, $method))) {

$this->$method($array, $conn);

} else {

echo($data);

throw new \Exception(sprintf('Command %s not found', $method));

}

}

/**

protected function commandStatus(array $data, \React\Socket\Connection $conn)

{

$conn->end(json_encode('todo'));

}

/**

protected function commandRegister(array $data, \React\Socket\Connection $conn)

{

$pid = (int)$data['pid'];

$port = (int)$data['port'];

if (!isset($this->slaves[$port]) || !$this->slaves[$port]['waitForRegister']) {

throw new \LogicException('A slaves wanted to register on master which was not expected. Emergency close. port='.$port);

}

if ($this->output->isVerbose()) {

$this->output->writeln('Worker registered '.$port);

}

$this->slaves[$port]['pid'] = $pid;

$this->slaves[$port]['connection'] = $conn;

$this->slaves[$port]['ready'] = true;

$this->slaves[$port]['waitForRegister'] = false;

if ($this->waitForSlaves && $this->slaveCount === count($this->slaves)) {

$this->waitForSlaves = false; // all slaves started

$this->output->writeln(

sprintf(

"%d slaves (starting at 5501) up and ready. Application is ready at http://%s:%s/",

$this->slaveCount,

$this->host,

$this->port

)

);

}

}

/**

protected function commandLog(array $data, \React\Socket\Connection $conn)

{

$this->output->writeln($data['message']);

}

/**

protected function commandFiles(array $data, \React\Socket\Connection $conn)

{

$this->filesToTrack = array_unique(array_merge($this->filesToTrack, $data['files']));

}

/**

protected function checkChangedFiles()

{

$reload = false;

$filePath = '';

$start = microtime(true);

foreach ($this->filesToTrack as $filePath) {

$currentFileMTime = filemtime($filePath);

if (isset($this->filesLastMTime[$filePath])) {

if ($this->filesLastMTime[$filePath] !== $currentFileMTime) {

$this->filesLastMTime[$filePath] = $currentFileMTime;

$reload = true;

break;

}

} else {

$this->filesLastMTime[$filePath] = $currentFileMTime;

}

}

if ($reload) {

$this->output->writeln(

sprintf(

"<info>[%s] File changed %s (detection %f, %d). Reload workers.</info>",

date('d/M/Y:H:i:s O'),

$filePath,

microtime(true) - $start,

count($this->filesToTrack)

)

);

$this->reload();

}

}

/**

protected function reload()

{

$this->inReload = true;

foreach ($this->slaves as $pid => $info) {

$info['ready'] = false; //does not accept new connections

if ($info['busy']) {

$info['closeWhenFree'] = true;

} else {

$info['connection']->close();

}

};

$this->inReload = false;

}

/**

function newInstance($port)

{

$dir = var_export(__DIR__, true);

$this->slaves[$port] = [

'ready' => false,

'pid' => null,

'port' => $port,

'closeWhenFree' => false,

'waitForRegister' => true,

'busy' => false,

'requests' => 0,

'connections' => 0,

'connection' => null,

];

if ($this->output->isVerbose()) {

$this->output->writeln('Start new worker '.$port);

}

$bridge = var_export($this->getBridge(), true);

$bootstrap = var_export($this->getAppBootstrap(), true);

$config = [

'port' => $port,

'app-env' => $this->getAppEnv(),

'debug' => $this->isDebug(),

'logging' => $this->isLogging(),

'static' => true

];

$config = var_export($config, true);

$script = <<<EOF

EOF;

$executableFinder = new PhpExecutableFinder();

$commandline = $executableFinder->find() . '-cgi';

$file = tempnam(sys_get_temp_dir(), 'dbg');

file_put_contents($file, $script);

register_shutdown_function('unlink', $file);

$commandline .= ' -C -q ' . ProcessUtils::escapeArgument($file);

$descriptorspec = [

['pipe', 'r'], //stdin

['pipe', 'w'], //stdout

['pipe', 'w'], //stderr

];

$this->slaves[$port]['process'] = proc_open($commandline, $descriptorspec, $pipes);

$this->slaves[$port]['stdout'] = new \React\Stream\Stream($pipes[1], $this->loop);

$this->slaves[$port]['stderr'] = new \React\Stream\Stream($pipes[2], $this->loop);

$this->slaves[$port]['stdout']->on(

'data',

function ($data) {

$this->output->write($data);

}

);

$this->slaves[$port]['stderr']->on(

'data',

function ($data) {

$this->output->write("<error>$data</error>");

}

);

}