2016-08-22 06:40:48 +02:00
|
|
|
<?php declare(strict_types = 1);
|
2015-07-14 00:30:59 +02:00
|
|
|
|
2016-08-23 23:47:40 +02:00
|
|
|
namespace Amp\Parallel\Threading;
|
2016-08-18 18:04:48 +02:00
|
|
|
|
|
|
|
use Amp\Coroutine;
|
2016-08-23 23:47:40 +02:00
|
|
|
use Amp\Parallel\{ ContextException, StatusError, SynchronizationError, Strand };
|
|
|
|
use Amp\Parallel\Sync\{ ChannelledStream, Internal\ExitStatus };
|
2016-08-18 18:04:48 +02:00
|
|
|
use Amp\Socket\Socket;
|
|
|
|
use Interop\Async\Awaitable;
|
2015-07-27 00:53:00 +02:00
|
|
|
|
2015-07-15 19:36:32 +02:00
|
|
|
/**
|
2015-08-22 23:27:44 +02:00
|
|
|
* Implements an execution context using native multi-threading.
|
2015-08-11 00:38:58 +02:00
|
|
|
*
|
2015-08-22 23:27:44 +02:00
|
|
|
* The thread context is not itself threaded. A local instance of the context is
|
|
|
|
* maintained both in the context that creates the thread and in the thread
|
|
|
|
* itself.
|
2015-07-15 19:36:32 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
class Thread implements Strand {
|
2015-07-27 00:53:00 +02:00
|
|
|
/**
|
2015-08-31 00:52:00 +02:00
|
|
|
* @var Internal\Thread An internal thread instance.
|
2015-07-27 00:53:00 +02:00
|
|
|
*/
|
2015-08-22 23:27:44 +02:00
|
|
|
private $thread;
|
2015-07-27 00:53:00 +02:00
|
|
|
|
2015-08-07 06:25:04 +02:00
|
|
|
/**
|
2016-08-23 23:47:40 +02:00
|
|
|
* @var \Amp\Parallel\Sync\Channel A channel for communicating with the thread.
|
2015-08-07 06:25:04 +02:00
|
|
|
*/
|
2015-08-22 23:27:44 +02:00
|
|
|
private $channel;
|
2015-08-07 06:25:04 +02:00
|
|
|
|
2015-10-18 08:54:09 +02:00
|
|
|
/**
|
2016-08-18 18:04:48 +02:00
|
|
|
* @var \Amp\Socket\Socket
|
2015-10-18 08:54:09 +02:00
|
|
|
*/
|
|
|
|
private $pipe;
|
|
|
|
|
2015-08-25 02:35:42 +02:00
|
|
|
/**
|
|
|
|
* @var resource
|
|
|
|
*/
|
|
|
|
private $socket;
|
|
|
|
|
2015-08-27 20:06:39 +02:00
|
|
|
/**
|
2015-09-04 23:22:41 +02:00
|
|
|
* @var callable
|
2015-08-27 20:06:39 +02:00
|
|
|
*/
|
2015-09-04 23:22:41 +02:00
|
|
|
private $function;
|
|
|
|
|
|
|
|
/**
|
|
|
|
* @var mixed[]
|
|
|
|
*/
|
|
|
|
private $args;
|
2015-08-27 20:06:39 +02:00
|
|
|
|
2015-09-08 19:55:29 +02:00
|
|
|
/**
|
2015-09-19 05:20:35 +02:00
|
|
|
* @var int
|
2015-09-08 19:55:29 +02:00
|
|
|
*/
|
2015-09-19 05:20:35 +02:00
|
|
|
private $oid = 0;
|
2015-09-08 19:55:29 +02:00
|
|
|
|
2015-11-11 08:07:59 +01:00
|
|
|
/**
|
|
|
|
* Checks if threading is enabled.
|
|
|
|
*
|
|
|
|
* @return bool True if threading is enabled, otherwise false.
|
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public static function supported(): bool {
|
|
|
|
return \extension_loaded('pthreads');
|
2015-11-11 08:07:59 +01:00
|
|
|
}
|
|
|
|
|
2015-08-07 01:59:25 +02:00
|
|
|
/**
|
2015-08-22 23:27:44 +02:00
|
|
|
* Spawns a new thread and runs it.
|
|
|
|
*
|
2015-08-31 00:52:00 +02:00
|
|
|
* @param callable $function The callable to invoke in the thread.
|
2015-08-22 23:27:44 +02:00
|
|
|
*
|
2015-08-31 00:52:00 +02:00
|
|
|
* @return Thread The thread object that was spawned.
|
2015-08-07 01:59:25 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public static function spawn(callable $function, ...$args) {
|
2016-01-23 07:00:56 +01:00
|
|
|
$thread = new self($function, ...$args);
|
2015-08-22 23:27:44 +02:00
|
|
|
$thread->start();
|
|
|
|
return $thread;
|
|
|
|
}
|
2015-08-05 09:48:43 +02:00
|
|
|
|
2015-08-18 17:12:06 +02:00
|
|
|
/**
|
2015-08-31 00:52:00 +02:00
|
|
|
* Creates a new thread.
|
2015-08-22 23:27:44 +02:00
|
|
|
*
|
2015-08-31 00:52:00 +02:00
|
|
|
* @param callable $function The callable to invoke in the thread when run.
|
|
|
|
*
|
2016-08-18 18:04:48 +02:00
|
|
|
* @throws \Error Thrown if the pthreads extension is not available.
|
2015-08-18 17:12:06 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function __construct(callable $function, ...$args) {
|
|
|
|
if (!self::supported()) {
|
|
|
|
throw new \Error("The pthreads extension is required to create threads.");
|
2015-11-11 08:07:59 +01:00
|
|
|
}
|
|
|
|
|
2015-09-04 23:22:41 +02:00
|
|
|
$this->function = $function;
|
|
|
|
$this->args = $args;
|
|
|
|
}
|
2015-08-22 23:27:44 +02:00
|
|
|
|
2015-09-04 23:22:41 +02:00
|
|
|
/**
|
|
|
|
* Returns the thread to the condition before starting. The new thread can be started and run independently of the
|
|
|
|
* first thread.
|
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function __clone() {
|
2015-09-04 23:22:41 +02:00
|
|
|
$this->thread = null;
|
|
|
|
$this->socket = null;
|
2015-10-18 08:54:09 +02:00
|
|
|
$this->pipe = null;
|
2015-09-04 23:22:41 +02:00
|
|
|
$this->channel = null;
|
2015-09-19 05:20:35 +02:00
|
|
|
$this->oid = 0;
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Kills the thread if it is still running.
|
|
|
|
*
|
2016-08-23 23:47:40 +02:00
|
|
|
* @throws \Amp\Parallel\ContextException
|
2015-09-19 05:20:35 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function __destruct() {
|
|
|
|
if (\getmypid() === $this->oid) {
|
2015-09-19 05:20:35 +02:00
|
|
|
$this->kill();
|
|
|
|
}
|
2015-08-22 23:27:44 +02:00
|
|
|
}
|
2015-08-18 17:12:06 +02:00
|
|
|
|
2015-07-27 00:53:00 +02:00
|
|
|
/**
|
2015-08-22 23:27:44 +02:00
|
|
|
* Checks if the context is running.
|
2015-07-27 00:53:00 +02:00
|
|
|
*
|
2015-08-22 23:27:44 +02:00
|
|
|
* @return bool True if the context is running, otherwise false.
|
2015-07-27 00:53:00 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function isRunning(): bool {
|
|
|
|
return null !== $this->pipe && $this->pipe->isReadable();
|
2015-08-07 01:59:25 +02:00
|
|
|
}
|
|
|
|
|
2015-08-05 09:48:43 +02:00
|
|
|
/**
|
2015-08-31 00:52:00 +02:00
|
|
|
* Spawns the thread and begins the thread's execution.
|
|
|
|
*
|
2016-08-23 23:47:40 +02:00
|
|
|
* @throws \Amp\Parallel\StatusError If the thread has already been started.
|
|
|
|
* @throws \Amp\Parallel\ContextException If starting the thread was unsuccessful.
|
2015-08-05 09:48:43 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function start() {
|
2016-08-23 01:25:19 +02:00
|
|
|
if ($this->oid !== 0) {
|
2015-08-27 20:06:39 +02:00
|
|
|
throw new StatusError('The thread has already been started.');
|
2015-08-05 09:48:43 +02:00
|
|
|
}
|
|
|
|
|
2016-08-18 18:04:48 +02:00
|
|
|
$this->oid = \getmypid();
|
2015-09-08 19:55:29 +02:00
|
|
|
|
2016-08-18 18:04:48 +02:00
|
|
|
list($channel, $this->socket) = \Amp\Socket\pair();
|
2015-09-04 23:22:41 +02:00
|
|
|
|
|
|
|
$this->thread = new Internal\Thread($this->socket, $this->function, $this->args);
|
|
|
|
|
2016-08-18 18:04:48 +02:00
|
|
|
if (!$this->thread->start(PTHREADS_INHERIT_INI)) {
|
|
|
|
throw new ContextException('Failed to start the thread.');
|
2015-08-31 01:25:44 +02:00
|
|
|
}
|
2015-08-27 20:06:39 +02:00
|
|
|
|
2016-08-18 18:04:48 +02:00
|
|
|
$this->channel = new ChannelledStream($this->pipe = new Socket($channel));
|
2015-08-22 23:27:44 +02:00
|
|
|
}
|
2015-08-11 00:38:58 +02:00
|
|
|
|
2015-08-22 23:27:44 +02:00
|
|
|
/**
|
|
|
|
* Immediately kills the context.
|
2015-08-31 00:52:00 +02:00
|
|
|
*
|
2016-08-18 18:04:48 +02:00
|
|
|
* @throws ContextException If killing the thread was unsuccessful.
|
2015-08-22 23:27:44 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function kill() {
|
2016-08-23 01:25:19 +02:00
|
|
|
if ($this->thread !== null) {
|
2015-09-08 19:55:29 +02:00
|
|
|
try {
|
|
|
|
if ($this->thread->isRunning() && !$this->thread->kill()) {
|
2016-08-18 18:04:48 +02:00
|
|
|
throw new ContextException('Could not kill thread.');
|
2015-09-08 19:55:29 +02:00
|
|
|
}
|
|
|
|
} finally {
|
|
|
|
$this->close();
|
2015-09-06 21:59:24 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2015-09-03 00:24:01 +02:00
|
|
|
|
|
|
|
/**
|
|
|
|
* Closes channel and socket if still open.
|
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
private function close() {
|
2016-08-23 01:25:19 +02:00
|
|
|
if ($this->pipe !== null && $this->pipe->isReadable()) {
|
2015-10-18 08:54:09 +02:00
|
|
|
$this->pipe->close();
|
2015-09-03 00:24:01 +02:00
|
|
|
}
|
|
|
|
|
2016-08-18 18:04:48 +02:00
|
|
|
if (\is_resource($this->socket)) {
|
|
|
|
@\fclose($this->socket);
|
2015-08-28 23:18:02 +02:00
|
|
|
}
|
2015-09-08 19:55:29 +02:00
|
|
|
|
|
|
|
$this->thread = null;
|
2015-10-18 08:54:09 +02:00
|
|
|
$this->channel = null;
|
2015-08-05 09:48:43 +02:00
|
|
|
}
|
|
|
|
|
2015-08-18 17:12:06 +02:00
|
|
|
/**
|
2015-08-22 23:27:44 +02:00
|
|
|
* Gets a promise that resolves when the context ends and joins with the
|
|
|
|
* parent context.
|
2015-08-18 17:12:06 +02:00
|
|
|
*
|
2016-08-18 18:04:48 +02:00
|
|
|
* @return \Interop\Async\Awaitable<mixed>
|
2015-08-25 02:35:42 +02:00
|
|
|
*
|
2015-09-03 00:24:01 +02:00
|
|
|
* @throws StatusError Thrown if the context has not been started.
|
2015-08-31 00:52:00 +02:00
|
|
|
* @throws SynchronizationError Thrown if an exit status object is not received.
|
2015-08-18 17:12:06 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function join(): Awaitable {
|
2016-08-23 01:25:19 +02:00
|
|
|
if ($this->channel == null || $this->thread === null) {
|
2015-09-08 19:55:29 +02:00
|
|
|
throw new StatusError('The thread has not been started or has already finished.');
|
2015-08-27 20:06:39 +02:00
|
|
|
}
|
2016-08-18 18:04:48 +02:00
|
|
|
|
|
|
|
return new Coroutine($this->doJoin());
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* @coroutine
|
|
|
|
*
|
|
|
|
* @return \Generator
|
|
|
|
*
|
2016-08-23 23:47:40 +02:00
|
|
|
* @throws \Amp\Parallel\SynchronizationError If the thread does not send an exit status.
|
2016-08-18 18:04:48 +02:00
|
|
|
*/
|
|
|
|
private function doJoin(): \Generator {
|
2015-08-18 17:12:06 +02:00
|
|
|
try {
|
2016-08-18 18:04:48 +02:00
|
|
|
$response = yield $this->channel->receive();
|
2015-08-22 23:27:44 +02:00
|
|
|
|
2015-12-05 06:50:32 +01:00
|
|
|
if (!$response instanceof ExitStatus) {
|
2015-08-22 23:27:44 +02:00
|
|
|
throw new SynchronizationError('Did not receive an exit status from thread.');
|
2015-08-18 17:12:06 +02:00
|
|
|
}
|
2015-08-22 23:27:44 +02:00
|
|
|
|
2016-01-23 07:00:56 +01:00
|
|
|
$result = $response->getResult();
|
2015-09-03 00:24:01 +02:00
|
|
|
|
2015-08-22 23:27:44 +02:00
|
|
|
$this->thread->join();
|
2016-01-23 07:00:56 +01:00
|
|
|
} catch (\Throwable $exception) {
|
2015-09-03 00:24:01 +02:00
|
|
|
$this->kill();
|
|
|
|
throw $exception;
|
2015-08-18 17:12:06 +02:00
|
|
|
}
|
2015-09-03 00:24:01 +02:00
|
|
|
|
|
|
|
$this->close();
|
2016-01-23 07:00:56 +01:00
|
|
|
|
|
|
|
return $result;
|
2015-08-18 17:12:06 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
2015-08-22 23:27:44 +02:00
|
|
|
* {@inheritdoc}
|
2015-08-18 17:12:06 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function receive(): Awaitable {
|
2016-08-23 01:25:19 +02:00
|
|
|
if ($this->channel === null) {
|
2016-08-18 18:04:48 +02:00
|
|
|
throw new StatusError('The process has not been started.');
|
2015-08-22 23:27:44 +02:00
|
|
|
}
|
2016-08-18 18:04:48 +02:00
|
|
|
|
|
|
|
return \Amp\pipe($this->channel->receive(), static function ($data) {
|
|
|
|
if ($data instanceof ExitStatus) {
|
|
|
|
$data = $data->getResult();
|
|
|
|
throw new SynchronizationError(\sprintf(
|
|
|
|
'Thread unexpectedly exited with result of type: %s',
|
|
|
|
\is_object($data) ? \get_class($data) : \gettype($data)
|
|
|
|
));
|
|
|
|
}
|
|
|
|
|
|
|
|
return $data;
|
|
|
|
});
|
2015-08-18 17:12:06 +02:00
|
|
|
}
|
|
|
|
|
2015-08-07 01:59:25 +02:00
|
|
|
/**
|
2015-08-22 23:27:44 +02:00
|
|
|
* {@inheritdoc}
|
2015-08-07 01:59:25 +02:00
|
|
|
*/
|
2016-08-18 18:04:48 +02:00
|
|
|
public function send($data): Awaitable {
|
2016-08-23 01:25:19 +02:00
|
|
|
if ($this->channel === null) {
|
2015-09-08 19:55:29 +02:00
|
|
|
throw new StatusError('The thread has not been started or has already finished.');
|
2015-08-27 20:06:39 +02:00
|
|
|
}
|
|
|
|
|
2015-12-05 06:50:32 +01:00
|
|
|
if ($data instanceof ExitStatus) {
|
2016-08-18 18:04:48 +02:00
|
|
|
throw new \Error('Cannot send exit status objects.');
|
2015-07-27 00:53:00 +02:00
|
|
|
}
|
2015-07-14 00:30:59 +02:00
|
|
|
|
2016-08-18 18:04:48 +02:00
|
|
|
return $this->channel->send($data);
|
2015-08-22 23:27:44 +02:00
|
|
|
}
|
2015-07-14 00:30:59 +02:00
|
|
|
}
|