danog/parallel
1
0
Fork 0
mirror of https://github.com/danog/parallel.git synced 2024-11-30 04:39:01 +01:00
Code Issues Projects Releases Wiki Activity

Convert ChannelException to ContextException

Browse Source
This commit is contained in:
Aaron Piotrowski 2017-03-09 16:15:30 -06:00
parent dd496568ea
commit c16a015562
5 changed files with 115 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
lib/Context.php
View File

@ -22,6 +22,9 @@ interface Context {
/**
* @return \AsyncInterop\Promise<mixed> Resolves with the returned from the context.
*
* @throws \Amp\Parallel\ContextException If the context dies unexpectedly.
* @throws \Amp\Parallel\PanicError If the context throws an uncaught exception.
*/
public function join(): Promise;
}

29
lib/Forking/Fork.php
View File

@ -308,6 +308,10 @@ class Fork implements Process, Strand {
\is_object($response) ? \get_class($response) : \gettype($response)
));
}
} catch (ChannelException $exception) {
throw new ContextException(
"The context stopped responding, potentially due to a fatal error or calling exit", 0, $exception
);
} finally {
$this->kill();
}
@ -323,7 +327,13 @@ class Fork implements Process, Strand {
throw new StatusError('The process has not been started.');
}
return \Amp\pipe($this->channel->receive(), static function ($data) {
return new Coroutine($this->doReceive());
}
private function doReceive() {
try {
$data = yield $this->channel->receive();
if ($data instanceof ExitResult) {
$data = $data->getResult();
throw new SynchronizationError(\sprintf(
@ -331,12 +341,15 @@ class Fork implements Process, Strand {
\is_object($data) ? \get_class($data) : \gettype($data)
));
}
return $data;
});
} catch (ChannelException $exception) {
throw new ContextException(
"The context stopped responding, potentially due to a fatal error or calling exit", 0, $exception
);
}
return $data;
}
/**
* {@inheritdoc}
*/
@ -349,6 +362,10 @@ class Fork implements Process, Strand {
throw new \Error('Cannot send exit result objects.');
}
return $this->channel->send($data);
return \Amp\capture($this->channel->send($data), ChannelException::class, function (ChannelException $exception) {
throw new ContextException(
"The context went away, potentially due to a fatal error or calling exit", 0, $exception
);
});
}
}

40
lib/Process/ChannelledProcess.php
View File

@ -3,7 +3,14 @@
namespace Amp\Parallel\Process;
use Amp\Coroutine;
use Amp\Parallel\{ ContextException, Process as ProcessContext, StatusError, Strand, SynchronizationError };
use Amp\Parallel\{
ChannelException,
ContextException,
Process as ProcessContext,
StatusError,
Strand,
SynchronizationError
};
use Amp\Parallel\Sync\{ ChannelledSocket, Internal\ExitResult };
use Amp\Process\Process;
use AsyncInterop\Promise;
@ -59,17 +66,27 @@ class ChannelledProcess implements ProcessContext, Strand {
throw new StatusError("The process has not been started");
}
return \Amp\pipe($this->channel->receive(), static function ($data) {
return new Coroutine($this->doReceive());
}
private function doReceive() {
try {
$data = yield $this->channel->receive();
if ($data instanceof ExitResult) {
$data = $data->getResult();
throw new SynchronizationError(\sprintf(
"Process unexpectedly exited with result of type: %s",
'Process unexpectedly exited with result of type: %s',
\is_object($data) ? \get_class($data) : \gettype($data)
));
}
return $data;
});
} catch (ChannelException $exception) {
throw new ContextException(
"The context stopped responding, potentially due to a fatal error or calling exit", 0, $exception
);
}
return $data;
}
/**
@ -84,7 +101,11 @@ class ChannelledProcess implements ProcessContext, Strand {
throw new \Error("Cannot send exit result objects");
}
return $this->channel->send($data);
return \Amp\capture($this->channel->send($data), ChannelException::class, function (ChannelException $exception) {
throw new ContextException(
"The context went away, potentially due to a fatal error or calling exit", 0, $exception
);
});
}
/**
@ -104,6 +125,11 @@ class ChannelledProcess implements ProcessContext, Strand {
if (!$data instanceof ExitResult) {
throw new SynchronizationError("Did not receive an exit result from process");
}
} catch (ChannelException $exception) {
$this->kill();
throw new ContextException(
"The context stopped responding, potentially due to a fatal error or calling exit", 0, $exception
);
} catch (\Throwable $exception) {
$this->kill();
throw $exception;

33
lib/Threading/Thread.php
View File

@ -3,7 +3,7 @@
namespace Amp\Parallel\Threading;
use Amp\Coroutine;
use Amp\Parallel\{ ContextException, StatusError, SynchronizationError, Strand };
use Amp\Parallel\{ ChannelException, ContextException, StatusError, SynchronizationError, Strand };
use Amp\Parallel\Sync\{ ChannelledSocket, Internal\ExitResult };
use AsyncInterop\Promise;
@ -204,6 +204,11 @@ class Thread implements Strand {
if (!$response instanceof ExitResult) {
throw new SynchronizationError('Did not receive an exit result from thread.');
}
} catch (ChannelException $exception) {
$this->kill();
throw new ContextException(
"The context stopped responding, potentially due to a fatal error or calling exit", 0, $exception
);
} catch (\Throwable $exception) {
$this->kill();
throw $exception;
@ -222,17 +227,27 @@ class Thread implements Strand {
throw new StatusError('The process has not been started.');
}
return \Amp\pipe($this->channel->receive(), static function ($data) {
return new Coroutine($this->doReceive());
}
private function doReceive() {
try {
$data = yield $this->channel->receive();
if ($data instanceof ExitResult) {
$data = $data->getResult();
throw new SynchronizationError(\sprintf(
'Thread unexpectedly exited with result of type: %s',
'Thread process unexpectedly exited with result of type: %s',
\is_object($data) ? \get_class($data) : \gettype($data)
));
}
return $data;
});
} catch (ChannelException $exception) {
throw new ContextException(
"The context stopped responding, potentially due to a fatal error or calling exit", 0, $exception
);
}
return $data;
}
/**
@ -247,6 +262,10 @@ class Thread implements Strand {
throw new \Error('Cannot send exit result objects.');
}
return $this->channel->send($data);
return \Amp\capture($this->channel->send($data), ChannelException::class, function (ChannelException $exception) {
throw new ContextException(
"The context went away, potentially due to a fatal error or calling exit", 0, $exception
);
});
}
}

30
test/AbstractContextTest.php
View File

@ -237,4 +237,34 @@ abstract class AbstractContextTest extends TestCase {
$value = yield $context->join();
}));
}
/**
* @expectedException \Amp\Parallel\ContextException
* @expectedExceptionMessage The context stopped responding
*/
public function testExitingContextOnJoin() {
Loop::execute(\Amp\wrap(function () {
$context = $this->createContext(function () {
exit;
});
$context->start();
$value = yield $context->join();
}));
}
/**
* @expectedException \Amp\Parallel\ContextException
* @expectedExceptionMessage The context stopped responding
*/
public function testExitingContextOnReceive() {
Loop::execute(\Amp\wrap(function () {
$context = $this->createContext(function () {
exit;
});
$context->start();
$value = yield $context->receive();
}));
}
}