2019-02-13 21:19:01 +01:00
|
|
|
<?php
|
|
|
|
|
|
|
|
namespace Amp\Parallel\Context;
|
|
|
|
|
|
|
|
use Amp\Loop;
|
|
|
|
use Amp\Parallel\Sync\ChannelledSocket;
|
2019-05-01 20:30:57 +02:00
|
|
|
use Amp\Parallel\Sync\ExitFailure;
|
2019-02-13 21:19:01 +01:00
|
|
|
use Amp\Parallel\Sync\ExitResult;
|
2019-05-01 20:30:57 +02:00
|
|
|
use Amp\Parallel\Sync\ExitSuccess;
|
|
|
|
use Amp\Parallel\Sync\SerializationException;
|
2019-02-13 21:19:01 +01:00
|
|
|
use Amp\Parallel\Sync\SynchronizationError;
|
|
|
|
use Amp\Promise;
|
|
|
|
use parallel\Runtime;
|
|
|
|
use function Amp\call;
|
|
|
|
|
|
|
|
/**
|
2019-02-19 01:43:11 +01:00
|
|
|
* Implements an execution context using native threads provided by the parallel extension.
|
2019-02-13 21:19:01 +01:00
|
|
|
*/
|
|
|
|
final class Parallel implements Context
|
|
|
|
{
|
2019-02-18 18:48:13 +01:00
|
|
|
const EXIT_CHECK_FREQUENCY = 250;
|
2019-02-13 21:19:01 +01:00
|
|
|
const KEY_LENGTH = 32;
|
|
|
|
|
|
|
|
/** @var string|null */
|
|
|
|
private static $autoloadPath;
|
|
|
|
|
2019-02-19 01:43:11 +01:00
|
|
|
/** @var int Next thread ID. */
|
2019-02-18 18:48:13 +01:00
|
|
|
private static $nextId = 1;
|
|
|
|
|
2019-02-13 21:19:01 +01:00
|
|
|
/** @var Internal\ProcessHub */
|
|
|
|
private $hub;
|
|
|
|
|
2019-02-18 18:48:13 +01:00
|
|
|
/** @var int|null */
|
|
|
|
private $id;
|
|
|
|
|
2019-02-13 23:30:06 +01:00
|
|
|
/** @var Runtime|null */
|
2019-02-13 21:19:01 +01:00
|
|
|
private $runtime;
|
|
|
|
|
2019-02-13 23:30:06 +01:00
|
|
|
/** @var ChannelledSocket|null A channel for communicating with the parallel thread. */
|
2019-02-16 01:25:29 +01:00
|
|
|
private $channel;
|
2019-02-13 21:19:01 +01:00
|
|
|
|
2019-02-13 23:30:06 +01:00
|
|
|
/** @var string Script path. */
|
2019-02-13 21:19:01 +01:00
|
|
|
private $script;
|
|
|
|
|
2019-02-18 18:48:13 +01:00
|
|
|
/** @var string[] */
|
2019-02-18 06:50:41 +01:00
|
|
|
private $args = [];
|
2019-02-13 21:19:01 +01:00
|
|
|
|
|
|
|
/** @var int */
|
|
|
|
private $oid = 0;
|
|
|
|
|
2019-02-15 01:12:51 +01:00
|
|
|
/** @var bool */
|
|
|
|
private $killed = false;
|
|
|
|
|
2019-02-13 21:19:01 +01:00
|
|
|
/**
|
|
|
|
* Checks if threading is enabled.
|
|
|
|
*
|
|
|
|
* @return bool True if threading is enabled, otherwise false.
|
|
|
|
*/
|
|
|
|
public static function isSupported(): bool
|
|
|
|
{
|
|
|
|
return \extension_loaded('parallel');
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Creates and starts a new thread.
|
|
|
|
*
|
2019-02-18 06:50:41 +01:00
|
|
|
* @param string|array $script Path to PHP script or array with first element as path and following elements options
|
|
|
|
* to the PHP script (e.g.: ['bin/worker', 'Option1Value', 'Option2Value'].
|
2019-02-13 21:19:01 +01:00
|
|
|
* @param mixed ...$args Additional arguments to pass to the given callable.
|
|
|
|
*
|
|
|
|
* @return Promise<Thread> The thread object that was spawned.
|
|
|
|
*/
|
2019-02-18 06:50:41 +01:00
|
|
|
public static function run($script): Promise
|
2019-02-13 21:19:01 +01:00
|
|
|
{
|
2019-02-18 06:50:41 +01:00
|
|
|
$thread = new self($script);
|
2019-02-13 21:19:01 +01:00
|
|
|
return call(function () use ($thread) {
|
|
|
|
yield $thread->start();
|
|
|
|
return $thread;
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
2019-02-18 06:50:41 +01:00
|
|
|
* @param string|array $script Path to PHP script or array with first element as path and following elements options
|
|
|
|
* to the PHP script (e.g.: ['bin/worker', 'Option1Value', 'Option2Value'].
|
2019-02-13 21:19:01 +01:00
|
|
|
*
|
|
|
|
* @throws \Error Thrown if the pthreads extension is not available.
|
|
|
|
*/
|
2019-02-18 06:50:41 +01:00
|
|
|
public function __construct($script)
|
2019-02-13 21:19:01 +01:00
|
|
|
{
|
|
|
|
if (!self::isSupported()) {
|
|
|
|
throw new \Error("The parallel extension is required to create parallel threads.");
|
|
|
|
}
|
|
|
|
|
2019-04-29 17:19:32 +02:00
|
|
|
$this->hub = Loop::getState(self::class);
|
|
|
|
if (!$this->hub instanceof Internal\ParallelHub) {
|
2019-05-11 16:48:24 +02:00
|
|
|
$this->hub = new Internal\ParallelHub;
|
2019-04-29 17:19:32 +02:00
|
|
|
Loop::setState(self::class, $this->hub);
|
|
|
|
}
|
|
|
|
|
2019-02-18 06:50:41 +01:00
|
|
|
if (\is_array($script)) {
|
|
|
|
$this->script = (string) \array_shift($script);
|
2019-02-19 01:43:11 +01:00
|
|
|
$this->args = \array_values(\array_map("strval", $script));
|
2019-02-18 06:50:41 +01:00
|
|
|
} else {
|
|
|
|
$this->script = (string) $script;
|
|
|
|
}
|
2019-02-13 21:19:01 +01:00
|
|
|
|
|
|
|
if (self::$autoloadPath === null) {
|
|
|
|
$paths = [
|
|
|
|
\dirname(__DIR__, 2) . \DIRECTORY_SEPARATOR . "vendor" . \DIRECTORY_SEPARATOR . "autoload.php",
|
|
|
|
\dirname(__DIR__, 4) . \DIRECTORY_SEPARATOR . "autoload.php",
|
|
|
|
];
|
|
|
|
|
|
|
|
foreach ($paths as $path) {
|
|
|
|
if (\file_exists($path)) {
|
|
|
|
self::$autoloadPath = $path;
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if (self::$autoloadPath === null) {
|
|
|
|
throw new \Error("Could not locate autoload.php");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Returns the thread to the condition before starting. The new thread can be started and run independently of the
|
|
|
|
* first thread.
|
|
|
|
*/
|
|
|
|
public function __clone()
|
|
|
|
{
|
|
|
|
$this->runtime = null;
|
2019-02-16 01:25:29 +01:00
|
|
|
$this->channel = null;
|
2019-02-18 18:48:13 +01:00
|
|
|
$this->id = null;
|
2019-02-13 21:19:01 +01:00
|
|
|
$this->oid = 0;
|
2019-02-15 01:12:51 +01:00
|
|
|
$this->killed = false;
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Kills the thread if it is still running.
|
|
|
|
*
|
|
|
|
* @throws \Amp\Parallel\Context\ContextException
|
|
|
|
*/
|
|
|
|
public function __destruct()
|
|
|
|
{
|
|
|
|
if (\getmypid() === $this->oid) {
|
|
|
|
$this->kill();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Checks if the context is running.
|
|
|
|
*
|
|
|
|
* @return bool True if the context is running, otherwise false.
|
|
|
|
*/
|
|
|
|
public function isRunning(): bool
|
|
|
|
{
|
2019-02-16 01:25:29 +01:00
|
|
|
return $this->channel !== null;
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Spawns the thread and begins the thread's execution.
|
|
|
|
*
|
2019-02-21 00:31:22 +01:00
|
|
|
* @return Promise<int> Resolved once the thread has started.
|
2019-02-13 21:19:01 +01:00
|
|
|
*
|
|
|
|
* @throws \Amp\Parallel\Context\StatusError If the thread has already been started.
|
|
|
|
* @throws \Amp\Parallel\Context\ContextException If starting the thread was unsuccessful.
|
|
|
|
*/
|
|
|
|
public function start(): Promise
|
|
|
|
{
|
|
|
|
if ($this->oid !== 0) {
|
|
|
|
throw new StatusError('The thread has already been started.');
|
|
|
|
}
|
|
|
|
|
|
|
|
$this->oid = \getmypid();
|
|
|
|
|
|
|
|
$this->runtime = new Runtime(self::$autoloadPath);
|
|
|
|
|
2019-02-18 18:48:13 +01:00
|
|
|
$this->id = self::$nextId++;
|
2019-02-13 21:19:01 +01:00
|
|
|
|
2019-02-21 00:31:22 +01:00
|
|
|
$future = $this->runtime->run(static function (int $id, string $uri, string $key, string $path, array $argv): int {
|
2019-03-08 20:37:52 +01:00
|
|
|
// @codeCoverageIgnoreStart
|
|
|
|
// Only executed in thread.
|
2019-02-13 21:19:01 +01:00
|
|
|
\define("AMP_CONTEXT", "parallel");
|
2019-02-21 00:31:22 +01:00
|
|
|
\define("AMP_CONTEXT_ID", $id);
|
2019-02-13 21:19:01 +01:00
|
|
|
|
|
|
|
if (!$socket = \stream_socket_client($uri, $errno, $errstr, 5, \STREAM_CLIENT_CONNECT)) {
|
|
|
|
\trigger_error("Could not connect to IPC socket", E_USER_ERROR);
|
|
|
|
return 1;
|
|
|
|
}
|
|
|
|
|
2019-02-18 17:32:16 +01:00
|
|
|
$channel = new ChannelledSocket($socket, $socket);
|
2019-02-15 01:12:51 +01:00
|
|
|
|
|
|
|
try {
|
2019-02-18 17:32:16 +01:00
|
|
|
Promise\wait($channel->send($key));
|
2019-02-13 21:19:01 +01:00
|
|
|
} catch (\Throwable $exception) {
|
|
|
|
\trigger_error("Could not send key to parent", E_USER_ERROR);
|
|
|
|
return 1;
|
|
|
|
}
|
|
|
|
|
2019-02-18 17:32:16 +01:00
|
|
|
try {
|
2019-05-01 20:30:57 +02:00
|
|
|
Loop::unreference(Loop::repeat(self::EXIT_CHECK_FREQUENCY, function () {
|
|
|
|
// Timer to give the chance for the PHP VM to be interrupted by Runtime::kill(), since system calls such as
|
|
|
|
// select() will not be interrupted.
|
|
|
|
}));
|
|
|
|
|
|
|
|
try {
|
|
|
|
if (!\is_file($path)) {
|
|
|
|
throw new \Error(\sprintf("No script found at '%s' (be sure to provide the full path to the script)", $path));
|
|
|
|
}
|
|
|
|
|
|
|
|
$argc = \array_unshift($argv, $path);
|
|
|
|
|
|
|
|
try {
|
|
|
|
// Protect current scope by requiring script within another function.
|
|
|
|
$callable = (function () use ($argc, $argv): callable { // Using $argc so it is available to the required script.
|
|
|
|
return require $argv[0];
|
2019-05-11 16:48:24 +02:00
|
|
|
})->bindTo(null, null)();
|
2019-05-01 20:30:57 +02:00
|
|
|
} catch (\TypeError $exception) {
|
|
|
|
throw new \Error(\sprintf("Script '%s' did not return a callable function", $path), 0, $exception);
|
|
|
|
} catch (\ParseError $exception) {
|
|
|
|
throw new \Error(\sprintf("Script '%s' contains a parse error", $path), 0, $exception);
|
|
|
|
}
|
|
|
|
|
|
|
|
$result = new ExitSuccess(Promise\wait(call($callable, $channel)));
|
|
|
|
} catch (\Throwable $exception) {
|
|
|
|
$result = new ExitFailure($exception);
|
|
|
|
}
|
|
|
|
|
|
|
|
Promise\wait(call(function () use ($channel, $result) {
|
|
|
|
try {
|
|
|
|
yield $channel->send($result);
|
|
|
|
} catch (SerializationException $exception) {
|
|
|
|
// Serializing the result failed. Send the reason why.
|
|
|
|
yield $channel->send(new ExitFailure($exception));
|
|
|
|
}
|
|
|
|
}));
|
2019-02-21 00:31:22 +01:00
|
|
|
} catch (\Throwable $exception) {
|
|
|
|
\trigger_error("Could not send result to parent; be sure to shutdown the child before ending the parent", E_USER_ERROR);
|
|
|
|
return 1;
|
2019-02-18 17:32:16 +01:00
|
|
|
} finally {
|
|
|
|
$channel->close();
|
|
|
|
}
|
|
|
|
|
2019-02-21 00:31:22 +01:00
|
|
|
return 0;
|
2019-03-08 20:37:52 +01:00
|
|
|
// @codeCoverageIgnoreEnd
|
2019-02-13 21:19:01 +01:00
|
|
|
}, [
|
2019-02-21 00:31:22 +01:00
|
|
|
$this->id,
|
2019-02-13 21:19:01 +01:00
|
|
|
$this->hub->getUri(),
|
2019-02-18 18:48:13 +01:00
|
|
|
$this->hub->generateKey($this->id, self::KEY_LENGTH),
|
2019-02-13 21:19:01 +01:00
|
|
|
$this->script,
|
2019-02-18 06:50:41 +01:00
|
|
|
$this->args
|
2019-02-13 21:19:01 +01:00
|
|
|
]);
|
|
|
|
|
2019-02-18 18:48:13 +01:00
|
|
|
return call(function () use ($future) {
|
2019-02-13 21:19:01 +01:00
|
|
|
try {
|
2019-03-08 18:12:10 +01:00
|
|
|
$this->channel = yield $this->hub->accept($this->id);
|
2019-04-29 17:19:32 +02:00
|
|
|
$this->hub->add($this->id, $this->channel, $future);
|
2019-02-13 21:19:01 +01:00
|
|
|
} catch (\Throwable $exception) {
|
2019-02-13 23:30:06 +01:00
|
|
|
$this->kill();
|
2019-02-13 21:19:01 +01:00
|
|
|
throw new ContextException("Starting the parallel runtime failed", 0, $exception);
|
|
|
|
}
|
2019-02-15 01:12:51 +01:00
|
|
|
|
|
|
|
if ($this->killed) {
|
|
|
|
$this->kill();
|
|
|
|
}
|
2019-02-21 00:31:22 +01:00
|
|
|
|
|
|
|
return $this->id;
|
2019-02-13 21:19:01 +01:00
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Immediately kills the context.
|
|
|
|
*/
|
|
|
|
public function kill()
|
|
|
|
{
|
2019-02-15 01:12:51 +01:00
|
|
|
$this->killed = true;
|
|
|
|
|
2019-02-16 01:25:29 +01:00
|
|
|
if ($this->runtime !== null) {
|
2019-02-13 23:30:06 +01:00
|
|
|
try {
|
2019-02-18 06:50:41 +01:00
|
|
|
$this->runtime->kill();
|
2019-02-13 23:30:06 +01:00
|
|
|
} finally {
|
|
|
|
$this->close();
|
|
|
|
}
|
|
|
|
}
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Closes channel and socket if still open.
|
|
|
|
*/
|
|
|
|
private function close()
|
|
|
|
{
|
2019-02-16 01:25:29 +01:00
|
|
|
$this->runtime = null;
|
2019-02-15 01:12:51 +01:00
|
|
|
|
2019-02-16 01:25:29 +01:00
|
|
|
if ($this->channel !== null) {
|
|
|
|
$this->channel->close();
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|
|
|
|
|
2019-02-16 01:25:29 +01:00
|
|
|
$this->channel = null;
|
2019-03-08 18:12:10 +01:00
|
|
|
|
2019-04-29 17:19:32 +02:00
|
|
|
$this->hub->remove($this->id);
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Gets a promise that resolves when the context ends and joins with the
|
|
|
|
* parent context.
|
|
|
|
*
|
|
|
|
* @return \Amp\Promise<mixed>
|
|
|
|
*
|
|
|
|
* @throws StatusError Thrown if the context has not been started.
|
|
|
|
* @throws SynchronizationError Thrown if an exit status object is not received.
|
|
|
|
* @throws ContextException If the context stops responding.
|
|
|
|
*/
|
|
|
|
public function join(): Promise
|
|
|
|
{
|
2019-02-16 01:25:29 +01:00
|
|
|
if ($this->channel === null) {
|
2019-02-13 21:19:01 +01:00
|
|
|
throw new StatusError('The thread has not been started or has already finished.');
|
|
|
|
}
|
|
|
|
|
|
|
|
return call(function () {
|
|
|
|
try {
|
2019-02-16 01:25:29 +01:00
|
|
|
$response = yield $this->channel->receive();
|
2019-02-18 16:38:42 +01:00
|
|
|
$this->close();
|
2019-02-13 21:19:01 +01:00
|
|
|
} catch (\Throwable $exception) {
|
|
|
|
$this->kill();
|
2019-02-18 06:50:41 +01:00
|
|
|
throw new ContextException("Failed to receive result from thread", 0, $exception);
|
2019-02-18 16:38:42 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
if (!$response instanceof ExitResult) {
|
|
|
|
$this->kill();
|
|
|
|
throw new SynchronizationError('Did not receive an exit result from thread.');
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
return $response->getResult();
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
|
|
|
public function receive(): Promise
|
|
|
|
{
|
2019-02-16 01:25:29 +01:00
|
|
|
if ($this->channel === null) {
|
2019-02-13 21:19:01 +01:00
|
|
|
throw new StatusError('The process has not been started.');
|
|
|
|
}
|
|
|
|
|
|
|
|
return call(function () {
|
2019-02-16 01:25:29 +01:00
|
|
|
$data = yield $this->channel->receive();
|
2019-02-13 21:19:01 +01:00
|
|
|
|
|
|
|
if ($data instanceof ExitResult) {
|
|
|
|
$data = $data->getResult();
|
|
|
|
throw new SynchronizationError(\sprintf(
|
|
|
|
'Thread process unexpectedly exited with result of type: %s',
|
|
|
|
\is_object($data) ? \get_class($data) : \gettype($data)
|
|
|
|
));
|
|
|
|
}
|
|
|
|
|
|
|
|
return $data;
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
|
|
|
public function send($data): Promise
|
|
|
|
{
|
2019-02-16 01:25:29 +01:00
|
|
|
if ($this->channel === null) {
|
2019-02-13 21:19:01 +01:00
|
|
|
throw new StatusError('The thread has not been started or has already finished.');
|
|
|
|
}
|
|
|
|
|
|
|
|
if ($data instanceof ExitResult) {
|
|
|
|
throw new \Error('Cannot send exit result objects.');
|
|
|
|
}
|
|
|
|
|
2019-02-16 01:25:29 +01:00
|
|
|
return $this->channel->send($data);
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|
2019-02-19 01:43:11 +01:00
|
|
|
|
|
|
|
/**
|
|
|
|
* Returns the ID of the thread. This ID will be unique to this process.
|
|
|
|
*
|
|
|
|
* @return int
|
|
|
|
*
|
|
|
|
* @throws \Amp\Process\StatusError
|
|
|
|
*/
|
|
|
|
public function getId(): int
|
|
|
|
{
|
|
|
|
if ($this->id === null) {
|
|
|
|
throw new StatusError('The thread has not been started');
|
|
|
|
}
|
|
|
|
|
|
|
|
return $this->id;
|
|
|
|
}
|
2019-02-13 21:19:01 +01:00
|
|
|
}
|