1
0
mirror of https://github.com/danog/parallel.git synced 2024-12-12 09:09:35 +01:00
parallel/lib/Worker/AbstractWorker.php

184 lines
4.5 KiB
PHP
Raw Normal View History

2016-12-30 02:16:04 +01:00
<?php
2015-12-05 06:50:32 +01:00
2016-08-23 23:47:40 +02:00
namespace Amp\Parallel\Worker;
2016-08-18 18:04:48 +02:00
2017-05-18 09:51:31 +02:00
use Amp\Coroutine;
use Amp\Deferred;
2017-06-18 06:57:12 +02:00
use Amp\Parallel\ContextException;
2017-05-18 09:51:31 +02:00
use Amp\Parallel\StatusError;
use Amp\Parallel\Strand;
use Amp\Promise;
2015-12-05 06:50:32 +01:00
/**
* Base class for most common types of task workers.
*/
2016-08-18 18:04:48 +02:00
abstract class AbstractWorker implements Worker {
2016-08-23 23:47:40 +02:00
/** @var \Amp\Parallel\Strand */
2015-12-05 06:50:32 +01:00
private $context;
/** @var bool */
2015-12-05 06:50:32 +01:00
private $shutdown = false;
2017-05-18 09:51:31 +02:00
/** @var \Amp\Deferred[] */
private $jobQueue = [];
2017-05-18 09:51:31 +02:00
/** @var callable */
2017-06-18 06:57:12 +02:00
private $onResolve;
2015-12-05 06:50:32 +01:00
/**
2016-08-23 23:47:40 +02:00
* @param \Amp\Parallel\Strand $strand
2015-12-05 06:50:32 +01:00
*/
2016-08-18 18:04:48 +02:00
public function __construct(Strand $strand) {
2015-12-21 18:04:51 +01:00
$this->context = $strand;
2017-05-18 09:51:31 +02:00
2017-06-18 06:57:12 +02:00
$this->onResolve = function ($exception, $data) {
if ($exception) {
2017-06-18 06:57:12 +02:00
$this->cancel($exception);
return;
}
2017-05-18 09:51:31 +02:00
2017-06-18 06:57:12 +02:00
if (!$data instanceof Internal\TaskResult) {
$this->cancel(new ContextException("Context did not return a task result"));
return;
}
2017-05-18 09:51:31 +02:00
$id = $data->getId();
2017-05-18 09:51:31 +02:00
if (!isset($this->jobQueue[$id])) {
2017-06-18 06:57:12 +02:00
$this->cancel(new ContextException("Job ID returned by context does not exist"));
return;
}
2017-05-18 09:51:31 +02:00
$deferred = $this->jobQueue[$id];
unset($this->jobQueue[$id]);
$empty = empty($this->jobQueue);
2017-05-18 09:51:31 +02:00
$deferred->resolve($data->promise());
if (!$empty) {
2017-06-18 06:57:12 +02:00
$this->context->receive()->onResolve($this->onResolve);
}
};
2015-12-05 06:50:32 +01:00
}
/**
* {@inheritdoc}
*/
2016-08-18 18:04:48 +02:00
public function isRunning(): bool {
2015-12-05 06:50:32 +01:00
return $this->context->isRunning();
}
/**
* {@inheritdoc}
*/
2016-08-18 18:04:48 +02:00
public function isIdle(): bool {
return empty($this->jobQueue);
2015-12-05 06:50:32 +01:00
}
/**
* {@inheritdoc}
*/
2016-08-18 18:04:48 +02:00
public function start() {
2015-12-05 06:50:32 +01:00
$this->context->start();
}
/**
* {@inheritdoc}
*/
2016-11-15 00:43:44 +01:00
public function enqueue(Task $task): Promise {
2015-12-05 06:50:32 +01:00
if (!$this->context->isRunning()) {
2017-06-18 06:57:12 +02:00
throw new StatusError("The worker has not been started");
2015-12-05 06:50:32 +01:00
}
2017-05-18 09:51:31 +02:00
2015-12-05 06:50:32 +01:00
if ($this->shutdown) {
2017-06-18 06:57:12 +02:00
throw new StatusError("The worker has been shut down");
}
2017-05-18 09:51:31 +02:00
2016-08-18 18:04:48 +02:00
return new Coroutine($this->doEnqueue($task));
}
2016-08-18 18:04:48 +02:00
/**
* @coroutine
*
2016-08-23 23:47:40 +02:00
* @param \Amp\Parallel\Worker\Task $task
2016-08-18 18:04:48 +02:00
*
* @return \Generator
2016-08-23 23:47:40 +02:00
* @throws \Amp\Parallel\StatusError
* @throws \Amp\Parallel\Worker\TaskException
* @throws \Amp\Parallel\Worker\TaskError
* @throws \Amp\Parallel\Worker\WorkerException
2016-08-18 18:04:48 +02:00
*/
private function doEnqueue(Task $task): \Generator {
$empty = empty($this->jobQueue);
2017-05-18 09:51:31 +02:00
2015-12-16 18:13:06 +01:00
try {
2017-06-18 06:57:12 +02:00
$job = new Internal\Job($task);
$this->jobQueue[$job->getId()] = $deferred = new Deferred;
if ($empty) {
$this->context->receive()->onResolve($this->onResolve);
}
yield $this->context->send($job);
2016-01-26 06:02:22 +01:00
} catch (\Throwable $exception) {
2017-06-18 06:57:12 +02:00
$exception = new WorkerException("Sending the task to the worker failed", $exception);
$this->cancel($exception);
throw $exception;
2015-12-05 06:50:32 +01:00
}
2017-05-18 09:51:31 +02:00
2016-11-15 00:43:44 +01:00
return yield $deferred->promise();
2016-08-18 18:04:48 +02:00
}
/**
* {@inheritdoc}
*/
2016-11-15 00:43:44 +01:00
public function shutdown(): Promise {
if (!$this->context->isRunning() || $this->shutdown) {
2017-06-18 06:57:12 +02:00
throw new StatusError("The worker is not running");
}
2017-05-18 09:51:31 +02:00
2016-08-18 18:04:48 +02:00
return new Coroutine($this->doShutdown());
}
2015-12-05 06:50:32 +01:00
/**
* {@inheritdoc}
*/
2016-08-18 18:04:48 +02:00
private function doShutdown(): \Generator {
2015-12-05 06:50:32 +01:00
$this->shutdown = true;
// If a task is currently running, wait for it to finish.
2017-05-18 06:28:18 +02:00
yield Promise\any(\array_map(function (Deferred $deferred): Promise {
return $deferred->promise();
}, $this->jobQueue));
2016-08-18 18:04:48 +02:00
yield $this->context->send(0);
return yield $this->context->join();
2015-12-05 06:50:32 +01:00
}
/**
* {@inheritdoc}
*/
2016-08-18 18:04:48 +02:00
public function kill() {
2017-06-18 06:57:12 +02:00
$this->cancel();
}
/**
2017-06-18 06:57:12 +02:00
* Cancels all pending tasks and kills the context.
*
* @param \Throwable|null $exception Optional exception to be used as the previous exception.
*/
2017-06-18 06:57:12 +02:00
protected function cancel(\Throwable $exception = null) {
if (!empty($this->jobQueue)) {
2017-06-18 06:57:12 +02:00
$exception = new WorkerException('Worker was shut down', $exception);
2017-05-18 09:51:31 +02:00
foreach ($this->jobQueue as $job) {
$job->fail($exception);
}
2017-05-18 09:51:31 +02:00
$this->jobQueue = [];
}
2017-06-18 06:57:12 +02:00
$this->context->kill();
2015-12-05 06:50:32 +01:00
}
}