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

Store exit status promise

Browse Source 
Return this promise from subsequent calls to shutdown().
This commit is contained in:
Aaron Piotrowski 2018-10-27 09:59:10 -05:00
parent 60e0213904
commit d783e13fa4
No known key found for this signature in database
GPG Key ID: ADD1EF783EDE9EEB
1 changed files with 18 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
lib/Worker/TaskWorker.php
View File

@ -2,6 +2,7 @@
namespace Amp\Parallel\Worker; namespace Amp\Parallel\Worker;
use Amp\Failure;
use Amp\Parallel\Context\Context; use Amp\Parallel\Context\Context;
use Amp\Parallel\Context\StatusError; use Amp\Parallel\Context\StatusError;
use Amp\Parallel\Sync\ChannelException; use Amp\Parallel\Sync\ChannelException;
@ -17,12 +18,12 @@ abstract class TaskWorker implements Worker
/** @var \Amp\Parallel\Context\Context */ /** @var \Amp\Parallel\Context\Context */
private $context; private $context;
/** @var bool */
private $shutdown = false;
/** @var \Amp\Promise|null */ /** @var \Amp\Promise|null */
private $pending; private $pending;
/** @var \Amp\Promise|null */
private $exitStatus;
/** /**
* @param \Amp\Parallel\Context\Context $context A context running an instance of TaskRunner. * @param \Amp\Parallel\Context\Context $context A context running an instance of TaskRunner.
*/ */
@ -40,7 +41,7 @@ abstract class TaskWorker implements Worker
*/ */
public function isRunning(): bool public function isRunning(): bool
{ {
return !$this->shutdown; return !$this->exitStatus;
} }
/** /**
@ -56,7 +57,7 @@ abstract class TaskWorker implements Worker
*/ */
public function enqueue(Task $task): Promise public function enqueue(Task $task): Promise
{ {
if ($this->shutdown) { if ($this->exitStatus) {
throw new StatusError("The worker has been shut down"); throw new StatusError("The worker has been shut down");
} }
@ -69,7 +70,7 @@ abstract class TaskWorker implements Worker
} }
} }
if ($this->shutdown) { if ($this->exitStatus) {
throw new WorkerException("The worker was shutdown"); throw new WorkerException("The worker was shutdown");
} }
@ -113,17 +114,15 @@ abstract class TaskWorker implements Worker
*/ */
public function shutdown(): Promise public function shutdown(): Promise
{ {
if ($this->shutdown) { if ($this->exitStatus) {
return new Success(0); return $this->exitStatus;
} }
$this->shutdown = true;
if (!$this->context->isRunning()) { if (!$this->context->isRunning()) {
return new Success(0); return $this->exitStatus = new Success(0);
} }
return call(function () { return $this->exitStatus = call(function () {
if ($this->pending) { if ($this->pending) {
// If a task is currently running, wait for it to finish. // If a task is currently running, wait for it to finish.
yield Promise\any([$this->pending]); yield Promise\any([$this->pending]);
@ -139,10 +138,16 @@ abstract class TaskWorker implements Worker
*/ */
public function kill() public function kill()
{ {
$this->shutdown = true; if ($this->exitStatus) {
return;
}
if ($this->context->isRunning()) { if ($this->context->isRunning()) {
$this->context->kill(); $this->context->kill();
$this->exitStatus = new Failure(new WorkerException("The worker was killed"));
return;
} }
$this->exitStatus = new Success(0);
} }
} }