danog/parallel
1
0
Fork 0
mirror of https://github.com/danog/parallel.git synced 2024-11-26 20:34:40 +01:00
Code Issues Projects Releases Wiki Activity

Remove dependency on amphp/socket

Browse Source
This commit is contained in:
Aaron Piotrowski 2016-08-30 18:27:14 -05:00
parent 046f7defb8
commit fa8985d7e3
11 changed files with 503 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
bin/worker
View File

@ -2,9 +2,8 @@
<?php declare(strict_types = 1);
use Amp\Parallel\{ ChannelException, SerializationException} ;
use Amp\Parallel\Sync\{ ChannelledStream, Internal\ExitFailure, Internal\ExitSuccess };
use Amp\Parallel\Sync\{ ChannelledSocket, Internal\ExitFailure, Internal\ExitSuccess };
use Amp\Parallel\Worker\{ BasicEnvironment, Internal\TaskRunner };
use Amp\Socket\Socket;
@cli_set_process_title("amp-worker");
error_reporting(E_ALL);
@ -38,7 +37,7 @@ ob_start(function ($data) {
})();
Amp\execute(function () {
$channel = new ChannelledStream(new Socket(STDIN, false), new Socket(STDOUT, false));
$channel = new ChannelledSocket(STDIN, STDOUT, false);
$environment = new BasicEnvironment;
$runner = new TaskRunner($channel, $environment);

4
composer.json
View File

@ -21,11 +21,11 @@
}
],
"require": {
"amphp/amp": "^2.0",
"amphp/socket": "dev-amp_v2 as 0.2"
"amphp/amp": "^2.0"
},
"require-dev": {
"amphp/loop": "dev-master",
"amphp/stream": "dev-master as 0.1",
"phpunit/phpunit": "^5.0"
},
"minimum-stability": "dev",

30
lib/Forking/Fork.php
View File

@ -12,21 +12,17 @@ use Amp\Parallel\{
Strand,
SynchronizationError
};
use Amp\Parallel\Sync\{ Channel, ChannelledStream };
use Amp\Parallel\Sync\{ Channel, ChannelledSocket };
use Amp\Parallel\Sync\Internal\{ ExitFailure, ExitStatus, ExitSuccess };
use Amp\Socket\Socket;
use Interop\Async\Awaitable;
/**
* Implements a UNIX-compatible context using forked processes.
*/
class Fork implements Process, Strand {
/** @var \Amp\Parallel\Sync\Channel A channel for communicating with the child. */
/** @var \Amp\Parallel\Sync\ChannelledSocket A channel for communicating with the child. */
private $channel;
/** @var \Amp\Socket\Socket */
private $pipe;
/** @var int */
private $pid = 0;
@ -73,7 +69,6 @@ class Fork implements Process, Strand {
public function __clone() {
$this->pid = 0;
$this->oid = 0;
$this->pipe = null;
$this->channel = null;
}
@ -152,14 +147,23 @@ class Fork implements Process, Strand {
* Starts the context execution.
*
* @throws \Amp\Parallel\ContextException If forking fails.
* @throws \Amp\Socket\SocketException If creating a socket pair fails.
*/
public function start() {
if (0 !== $this->oid) {
throw new StatusError('The context has already been started.');
}
list($parent, $child) = \Amp\Socket\pair();
$sockets = @\stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
if ($sockets === false) {
$message = "Failed to create socket pair";
if ($error = \error_get_last()) {
$message .= \sprintf(" Errno: %d; %s", $error["type"], $error["message"]);
}
throw new ContextException($message);
}
list($parent, $child) = $sockets;
switch ($pid = \pcntl_fork()) {
case -1: // Failure
@ -171,7 +175,7 @@ class Fork implements Process, Strand {
try {
\Amp\execute(function () use ($parent) {
$channel = new ChannelledStream(new Socket($parent));
$channel = new ChannelledSocket($parent, $parent);
return $this->execute($channel);
});
$code = 0;
@ -184,7 +188,7 @@ class Fork implements Process, Strand {
default: // Parent
$this->pid = $pid;
$this->oid = \posix_getpid();
$this->channel = new ChannelledStream($this->pipe = new Socket($child));
$this->channel = new ChannelledSocket($child, $child);
\fclose($parent);
}
}
@ -248,8 +252,8 @@ class Fork implements Process, Strand {
\posix_kill($this->pid, SIGKILL);
}
if (null !== $this->pipe && $this->pipe->isReadable()) {
$this->pipe->close();
if ($this->channel !== null) {
$this->channel->close();
}
// "Detach" from the process and let it die asynchronously.

4
lib/Process/ChannelledProcess.php
View File

@ -3,7 +3,7 @@
namespace Amp\Parallel\Process;
use Amp\Parallel\{ Process as ProcessContext, StatusError, Strand, SynchronizationError };
use Amp\Parallel\Sync\{ ChannelledStream, Internal\ExitStatus };
use Amp\Parallel\Sync\{ ChannelledSocket, Internal\ExitStatus };
use Interop\Async\Awaitable;
class ChannelledProcess implements ProcessContext, Strand {
@ -36,7 +36,7 @@ class ChannelledProcess implements ProcessContext, Strand {
*/
public function start() {
$this->process->start();
$this->channel = new ChannelledStream($this->process->getStdOut(), $this->process->getStdIn());
$this->channel = new ChannelledSocket($this->process->getStdOut(), $this->process->getStdIn(), false);
}
/**

56
lib/Process/Process.php
View File

@ -4,8 +4,6 @@ namespace Amp\Parallel\Process;
use Amp\Deferred;
use Amp\Parallel\{ ContextException, Process as ProcessContext, StatusError };
use Amp\Socket\Socket;
use Amp\Stream\Stream;
use Interop\Async\{ Awaitable, Loop };
class Process implements ProcessContext {
@ -23,14 +21,14 @@ class Process implements ProcessContext {
/** @var array */
private $options;
/** @var \Amp\Stream\Stream|null */
/** @var resource|null */
private $stdin;
/** @var \Amp\Stream\Stream|null */
/** @var resource|null */
private $stdout;
/** @var \Amp\Stream\Stream|null */
/** @var resource|null */
private $stderr;
/** @var int */
@ -75,16 +73,16 @@ class Process implements ProcessContext {
if (\getmypid() === $this->oid) {
$this->kill(); // Will only terminate if the process is still running.
if ($this->stdin !== null) {
$this->stdin->close();
if (\is_resource($this->stdin)) {
\fclose($this->stdin);
}
if ($this->stdout !== null) {
$this->stdout->close();
if (\is_resource($this->stdout)) {
\fclose($this->stdout);
}
if ($this->stderr !== null) {
$this->stderr->close();
if (\is_resource($this->stderr)) {
\fclose($this->stderr);
}
}
}
@ -146,9 +144,9 @@ class Process implements ProcessContext {
$this->pid = $status["pid"];
$this->stdin = $stdin = new Socket($pipes[0]);
$this->stdout = new Socket($pipes[1]);
$this->stderr = new Socket($pipes[2]);
$this->stdin = $stdin = $pipes[0];
$this->stdout = $pipes[1];
$this->stderr = $pipes[2];
$stream = $pipes[3];
\stream_set_blocking($stream, false);
@ -175,7 +173,9 @@ class Process implements ProcessContext {
\proc_close($process);
$process = null;
}
$stdin->close();
if (\is_resource($stdin)) {
\fclose($stdin);
}
Loop::cancel($watcher);
}
@ -200,14 +200,10 @@ class Process implements ProcessContext {
Loop::enable($this->watcher);
$awaitable = $this->deferred->getAwaitable();
$awaitable->when(function () {
$this->stdout->close();
$this->stderr->close();
});
return $awaitable;
\fclose($this->stdout);
\fclose($this->stderr);
return $this->deferred->getAwaitable();
}
/**
@ -304,11 +300,11 @@ class Process implements ProcessContext {
/**
* Gets the process input stream (STDIN).
*
* @return \Amp\Stream\Stream
* @return resource
*
* @throws \Amp\Parallel\StatusError If the process is not running.
*/
public function getStdIn(): Stream {
public function getStdIn() {
if ($this->stdin === null) {
throw new StatusError("The process has not been started");
}
@ -319,11 +315,11 @@ class Process implements ProcessContext {
/**
* Gets the process output stream (STDOUT).
*
* @return \Amp\Stream\Stream
* @return resource
*
* @throws \Amp\Parallel\StatusError If the process is not running.
*/
public function getStdOut(): Stream {
public function getStdOut() {
if ($this->stdout === null) {
throw new StatusError("The process has not been started");
}
@ -334,11 +330,11 @@ class Process implements ProcessContext {
/**
* Gets the process error stream (STDERR).
*
* @return \Amp\Stream\Stream
* @return resource
*
* @throws \Amp\Parallel\StatusError If the process is not running.
*/
public function getStdErr(): Stream {
public function getStdErr() {
if ($this->stderr === null) {
throw new StatusError("The process has not been started");
}

309
lib/Sync/ChannelledSocket.php Normal file
View File

@ -0,0 +1,309 @@
<?php declare(strict_types = 1);
namespace Amp\Parallel\Sync;
use Amp\{ Coroutine, Deferred, Failure, Success };
use Amp\Parallel\{ ChannelException, SerializationException };
use Interop\Async\{ Awaitable, Loop };
class ChannelledSocket implements Channel {
const HEADER_LENGTH = 5;
/** @var resource Stream resource. */
private $readResource;
/** @var resource Stream resource. */
private $writeResource;
/** @var string onReadable loop watcher. */
private $readWatcher;
/** @var string onWritable loop watcher. */
private $writeWatcher;
/** @var \SplQueue Queue of pending reads. */
private $reads;
/** @var \SplQueue Queue of pending writes. */
private $writes;
/** @var bool */
private $open = true;
/** @var bool */
private $autoClose = true;
/**
* @param resource $resource Stream resource.
* @param bool $autoClose True to close the stream resource when this object is destroyed, false to leave open.
*
* @throws \Error If a stream resource is not given for $resource.
*/
public function __construct($read, $write, bool $autoClose = true) {
if (!\is_resource($read) ||\get_resource_type($read) !== 'stream') {
throw new \Error('Invalid resource given to constructor!');
}
if (!\is_resource($write) ||\get_resource_type($write) !== 'stream') {
throw new \Error('Invalid resource given to constructor!');
}
$this->readResource = $read;
$this->writeResource = $write;
$this->autoClose = $autoClose;
\stream_set_blocking($this->readResource, false);
\stream_set_read_buffer($this->readResource, 0);
\stream_set_write_buffer($this->readResource, 0);
if ($this->readResource !== $this->writeResource) {
\stream_set_blocking($this->writeResource, false);
\stream_set_read_buffer($this->writeResource, 0);
\stream_set_write_buffer($this->writeResource, 0);
}
$this->reads = $reads = new \SplQueue;
$this->writes = $writes = new \SplQueue;
$errorHandler = static function ($errno, $errstr) {
throw new ChannelException(\sprintf('Received corrupted data. Errno: %d; %s', $errno, $errstr));
};
$this->readWatcher = Loop::onReadable($this->readResource, static function ($watcher, $stream) use ($reads, $errorHandler) {
while (!$reads->isEmpty()) {
/** @var \Amp\Deferred $deferred */
list($buffer, $length, $deferred) = $reads->shift();
if ($length === 0) {
// Error reporting suppressed since fread() produces a warning if the stream unexpectedly closes.
$data = @\fread($stream, self::HEADER_LENGTH - \strlen($buffer));
if ($data === false || ($data === '' && (\feof($stream) || !\is_resource($stream)))) {
$deferred->fail(new ChannelException("The socket unexpectedly closed"));
return;
}
$buffer .= $data;
if (\strlen($buffer) !== self::HEADER_LENGTH) {
// Not enough data available.
$reads->unshift([$buffer, 0, $deferred]);
return;
}
$data = \unpack("Cprefix/Llength", $data);
if ($data["prefix"] !== 0) {
$deferred->fail(new ChannelException("Invalid header received"));
return;
}
$length = $data["length"];
$buffer = '';
}
// Error reporting suppressed since fread() produces a warning if the stream unexpectedly closes.
$data = @\fread($stream, $length - \strlen($buffer));
if ($data === false || ($data === '' && (\feof($stream) || !\is_resource($stream)))) {
$deferred->fail(new ChannelException("The socket unexpectedly closed"));
return;
}
$buffer .= $data;
if (\strlen($buffer) < $length) {
// Not enough data available.
$reads->unshift([$buffer, $length, $deferred]);
return;
}
\set_error_handler($errorHandler);
// Attempt to unserialize the received data.
try {
$data = \unserialize($buffer);
} catch (\Throwable $exception) {
$deferred->fail(new SerializationException("Exception thrown when unserializing data", $exception));
continue;
} finally {
\restore_error_handler();
}
$deferred->resolve($data);
}
Loop::disable($watcher);
});
$this->writeWatcher = Loop::onWritable($this->writeResource, static function ($watcher, $stream) use ($writes) {
while (!$writes->isEmpty()) {
/** @var \Amp\Deferred $deferred */
list($data, $previous, $deferred) = $writes->shift();
$length = \strlen($data);
if ($length === 0) {
$deferred->resolve(0);
continue;
}
// Error reporting suppressed since fwrite() emits E_WARNING if the pipe is broken or the buffer is full.
$written = @\fwrite($stream, $data);
if ($written === false || $written === 0) {
$message = "Failed to write to socket";
if ($error = \error_get_last()) {
$message .= \sprintf(" Errno: %d; %s", $error["type"], $error["message"]);
}
$deferred->fail(new ChannelException($message));
return;
}
if ($length <= $written) {
$deferred->resolve($written + $previous);
continue;
}
$data = \substr($data, $written);
$writes->unshift([$data, $written + $previous, $deferred]);
return;
}
});
Loop::disable($this->readWatcher);
Loop::disable($this->writeWatcher);
}
public function __destruct() {
if ($this->readResource !== null) {
$this->close();
}
}
/**
* {@inheritdoc}
*/
public function close() {
if (\is_resource($this->readResource)) {
if ($this->autoClose) {
@\fclose($this->readResource);
if ($this->readResource !== $this->writeResource) {
@\fclose($this->writeResource);
}
}
$this->readResource = null;
$this->writeResource = null;
}
$this->open = false;
if (!$this->reads->isEmpty()) {
$exception = new ChannelException("The connection was unexpectedly closed before reading completed");
do {
/** @var \Amp\Deferred $deferred */
list( , , $deferred) = $this->reads->shift();
$deferred->fail($exception);
} while (!$this->reads->isEmpty());
}
if (!$this->writes->isEmpty()) {
$exception = new ChannelException("The connection was unexpectedly writing completed");
do {
/** @var \Amp\Deferred $deferred */
list( , , $deferred) = $this->writes->shift();
$deferred->fail($exception);
} while (!$this->writes->isEmpty());
}
// defer this, else the Loop::disable() may be invalid
$read = $this->readWatcher;
$write = $this->writeWatcher;
Loop::defer(static function () use ($read, $write) {
Loop::cancel($read);
Loop::cancel($write);
});
}
/**
* {@inheritdoc}
*/
public function receive(): Awaitable {
if (!$this->open) {
return new Failure(new ChannelException("The channel is has been closed"));
}
$deferred = new Deferred;
$this->reads->push(["", 0, $deferred]);
Loop::enable($this->readWatcher);
return $deferred->getAwaitable();
}
/**
* @param string $data
* @param bool $end
*
* @return \Interop\Async\Awaitable
*/
public function send($data): Awaitable {
if (!$this->open) {
return new Failure(new ChannelException("The channel is has been closed"));
}
// Serialize the data to send into the channel.
try {
$data = \serialize($data);
} catch (\Throwable $exception) {
throw new SerializationException(
"The given data cannot be sent because it is not serializable.", $exception
);
}
$length = \strlen($data);
$data = \pack("CL", 0, $length) . $data;
$written = 0;
if ($this->writes->isEmpty()) {
// Error reporting suppressed since fwrite() emits E_WARNING if the pipe is broken or the buffer is full.
$written = @\fwrite($this->writeResource, $data);
if ($written === false) {
$message = "Failed to write to stream";
if ($error = \error_get_last()) {
$message .= \sprintf(" Errno: %d; %s", $error["type"], $error["message"]);
}
return new Failure(new ChannelException($message));
}
if ($length <= $written) {
return new Success($written);
}
$data = \substr($data, $written);
}
return new Coroutine($this->doSend($data, $written));
}
private function doSend(string $data, int $written): \Generator {
$deferred = new Deferred;
$this->writes->push([$data, $written, $deferred]);
Loop::enable($this->writeWatcher);
try {
$written = yield $deferred->getAwaitable();
} catch (\Throwable $exception) {
$this->close();
throw $exception;
} finally {
if ($this->writes->isEmpty()) {
Loop::disable($this->writeWatcher);
}
}
return $written;
}
}

5
lib/Threading/Internal/Thread.php
View File

@ -4,8 +4,7 @@ namespace Amp\Parallel\Threading\Internal;
use Amp\Coroutine;
use Amp\Parallel\{ ChannelException, SerializationException };
use Amp\Parallel\Sync\{ Channel, ChannelledStream, Internal\ExitFailure, Internal\ExitSuccess };
use Amp\Socket\Socket;
use Amp\Parallel\Sync\{ Channel, ChannelledSocket, Internal\ExitFailure, Internal\ExitSuccess };
use Interop\Async\Awaitable;
/**
@ -76,7 +75,7 @@ class Thread extends \Thread {
try {
\Amp\execute(function () {
$channel = new ChannelledStream(new Socket($this->socket, false));
$channel = new ChannelledSocket($this->socket, $this->socket, false);
$watcher = \Amp\repeat(self::KILL_CHECK_FREQUENCY, function () {
if ($this->killed) {

33
lib/Threading/Thread.php
View File

@ -4,8 +4,7 @@ namespace Amp\Parallel\Threading;
use Amp\Coroutine;
use Amp\Parallel\{ ContextException, StatusError, SynchronizationError, Strand };
use Amp\Parallel\Sync\{ ChannelledStream, Internal\ExitStatus };
use Amp\Socket\Socket;
use Amp\Parallel\Sync\{ ChannelledSocket, Internal\ExitStatus };
use Interop\Async\Awaitable;
/**
@ -19,12 +18,9 @@ class Thread implements Strand {
/** @var Internal\Thread An internal thread instance. */
private $thread;
/** @var \Amp\Parallel\Sync\Channel A channel for communicating with the thread. */
/** @var \Amp\Parallel\Sync\ChannelledSocket A channel for communicating with the thread. */
private $channel;
/** @var \Amp\Socket\Socket */
private $pipe;
/** @var resource */
private $socket;
@ -82,7 +78,6 @@ class Thread implements Strand {
public function __clone() {
$this->thread = null;
$this->socket = null;
$this->pipe = null;
$this->channel = null;
$this->oid = 0;
}
@ -104,7 +99,7 @@ class Thread implements Strand {
* @return bool True if the context is running, otherwise false.
*/
public function isRunning(): bool {
return null !== $this->pipe && $this->pipe->isReadable();
return $this->channel !== null;
}
/**
@ -120,7 +115,21 @@ class Thread implements Strand {
$this->oid = \getmypid();
list($channel, $this->socket) = \Amp\Socket\pair();
$sockets = @\stream_socket_pair(
\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX,
STREAM_SOCK_STREAM,
STREAM_IPPROTO_IP
);
if ($sockets === false) {
$message = "Failed to create socket pair";
if ($error = \error_get_last()) {
$message .= \sprintf(" Errno: %d; %s", $error["type"], $error["message"]);
}
throw new ContextException($message);
}
list($channel, $this->socket) = $sockets;
$this->thread = new Internal\Thread($this->socket, $this->function, $this->args);
@ -128,7 +137,7 @@ class Thread implements Strand {
throw new ContextException('Failed to start the thread.');
}
$this->channel = new ChannelledStream($this->pipe = new Socket($channel));
$this->channel = new ChannelledSocket($channel, $channel);
}
/**
@ -152,8 +161,8 @@ class Thread implements Strand {
* Closes channel and socket if still open.
*/
private function close() {
if ($this->pipe !== null && $this->pipe->isReadable()) {
$this->pipe->close();
if ($this->channel !== null) {
$this->channel->close();
}
if (\is_resource($this->socket)) {

1
test/Sync/AbstractSemaphoreTest.php
View File

@ -2,7 +2,6 @@
namespace Amp\Parallel\Test\Sync;
use Amp\Parallel\Sync\Lock;
use Amp\Parallel\Test\TestCase;
use Amp\Pause;

121
test/Sync/ChannelledSocketTest.php Normal file
View File

@ -0,0 +1,121 @@
<?php declare(strict_types = 1);
namespace Amp\Parallel\Test\Sync;
use Amp\Parallel\Sync\ChannelledSocket;
use Amp\Parallel\Test\TestCase;
class ChannelledSocketTest extends TestCase {
/**
* @return resource[]
*/
protected function createSockets() {
if (($sockets = @\stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP)) === false) {
$message = "Failed to create socket pair";
if ($error = \error_get_last()) {
$message .= \sprintf(" Errno: %d; %s", $error["type"], $error["message"]);
}
$this->fail($message);
}
return $sockets;
}
public function testSendReceive() {
\Amp\execute(function () {
list($left, $right) = $this->createSockets();
$a = new ChannelledSocket($left, $left);
$b = new ChannelledSocket($right, $right);
$message = 'hello';
yield $a->send($message);
$data = yield $b->receive();
$this->assertSame($message, $data);
});
}
/**
* @depends testSendReceive
*/
public function testSendReceiveLongData() {
\Amp\execute(function () {
list($left, $right) = $this->createSockets();
$a = new ChannelledSocket($left, $left);
$b = new ChannelledSocket($right, $right);
$length = 0xffff;
$message = '';
for ($i = 0; $i < $length; ++$i) {
$message .= chr(mt_rand(0, 255));
}
$a->send($message);
$data = yield $b->receive();
$this->assertSame($message, $data);
});
}
/**
* @depends testSendReceive
* @expectedException \Amp\Parallel\ChannelException
*/
public function testInvalidDataReceived() {
\Amp\execute(function () {
list($left, $right) = $this->createSockets();
$a = new ChannelledSocket($left, $left);
$b = new ChannelledSocket($right, $right);
fwrite($left, pack('L', 10) . '1234567890');
$data = yield $b->receive();
});
}
/**
* @depends testSendReceive
* @expectedException \Amp\Parallel\ChannelException
*/
public function testSendUnserializableData() {
\Amp\execute(function () {
list($left, $right) = $this->createSockets();
$a = new ChannelledSocket($left, $left);
$b = new ChannelledSocket($right, $right);
// Close $a. $b should close on next read...
yield $a->send(function () {});
$data = yield $b->receive();
});
}
/**
* @depends testSendReceive
* @expectedException \Amp\Parallel\ChannelException
*/
public function testSendAfterClose() {
\Amp\execute(function () {
list($left, $right) = $this->createSockets();
$a = new ChannelledSocket($left, $left);
$a->close();
yield $a->send('hello');
});
}
/**
* @depends testSendReceive
* @expectedException \Amp\Parallel\ChannelException
*/
public function testReceiveAfterClose() {
\Amp\execute(function () {
list($left, $right) = $this->createSockets();
$a = new ChannelledSocket($left, $left);
$a->close();
$data = yield $a->receive();
});
}
}

2
test/Sync/PosixSemaphoreTest.php
View File

@ -3,7 +3,7 @@
namespace Amp\Parallel\Test\Sync;
use Amp\Parallel\Forking\Fork;
use Amp\Parallel\Sync\{PosixSemaphore, Semaphore};
use Amp\Parallel\Sync\{ PosixSemaphore, Semaphore };
/**
* @group posix