1
0
mirror of https://github.com/danog/parallel.git synced 2024-12-11 16:49:51 +01:00
parallel/lib/Worker/DefaultPool.php

316 lines
8.7 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\CallableMaker;
use Amp\Coroutine;
2016-08-23 23:47:40 +02:00
use Amp\Parallel\StatusError;
2017-05-18 09:51:31 +02:00
use Amp\Promise;
2015-12-05 06:50:32 +01:00
/**
* Provides a pool of workers that can be used to execute multiple tasks asynchronously.
*
* A worker pool is a collection of worker threads that can perform multiple
* tasks simultaneously. The load on each worker is balanced such that tasks
* are completed as soon as possible and workers are used efficiently.
*/
2016-08-18 18:04:48 +02:00
class DefaultPool implements Pool {
2016-08-31 00:36:21 +02:00
use CallableMaker;
2017-05-18 09:51:31 +02:00
2016-08-26 17:10:03 +02:00
/** @var bool Indicates if the pool is currently running. */
2015-12-05 06:50:32 +01:00
private $running = false;
2016-08-26 17:10:03 +02:00
/** @var int The minimum number of workers the pool should spawn. */
2015-12-05 06:50:32 +01:00
private $minSize;
2016-08-26 17:10:03 +02:00
/** @var int The maximum number of workers the pool should spawn. */
2015-12-05 06:50:32 +01:00
private $maxSize;
2016-08-26 17:10:03 +02:00
/** @var WorkerFactory A worker factory to be used to create new workers. */
2015-12-05 06:50:32 +01:00
private $factory;
2016-08-26 17:10:03 +02:00
/** @var \SplObjectStorage A collection of all workers in the pool. */
private $workers;
2015-12-05 06:50:32 +01:00
2016-08-26 17:10:03 +02:00
/** @var \SplQueue A collection of idle workers. */
2015-12-05 06:50:32 +01:00
private $idleWorkers;
2016-08-26 17:10:03 +02:00
/** @var \SplQueue A queue of workers that have been assigned to tasks. */
2015-12-05 06:50:32 +01:00
private $busyQueue;
2016-08-26 17:10:03 +02:00
/** @var \Closure */
private $push;
2015-12-05 06:50:32 +01:00
/**
* Creates a new worker pool.
*
2015-12-16 23:39:25 +01:00
* @param int|null $minSize The minimum number of workers the pool should spawn.
* Defaults to `Pool::DEFAULT_MIN_SIZE`.
* @param int|null $maxSize The maximum number of workers the pool should spawn.
* Defaults to `Pool::DEFAULT_MAX_SIZE`.
2016-08-23 23:47:40 +02:00
* @param \Amp\Parallel\Worker\WorkerFactory|null $factory A worker factory to be used to create
2015-12-05 06:50:32 +01:00
* new workers.
*
2016-08-18 18:04:48 +02:00
* @throws \Error
2015-12-05 06:50:32 +01:00
*/
2016-08-18 18:04:48 +02:00
public function __construct(int $minSize = null, int $maxSize = null, WorkerFactory $factory = null) {
2015-12-05 06:50:32 +01:00
$minSize = $minSize ?: self::DEFAULT_MIN_SIZE;
$maxSize = $maxSize ?: self::DEFAULT_MAX_SIZE;
2016-08-18 18:04:48 +02:00
if ($minSize < 0) {
throw new \Error('Minimum size must be a non-negative integer.');
2015-12-05 06:50:32 +01:00
}
2016-08-18 18:04:48 +02:00
if ($maxSize < 0 || $maxSize < $minSize) {
throw new \Error('Maximum size must be a non-negative integer at least '.$minSize.'.');
2015-12-05 06:50:32 +01:00
}
$this->maxSize = $maxSize;
$this->minSize = $minSize;
2016-01-11 16:32:06 +01:00
// Use the global factory if none is given.
$this->factory = $factory ?: factory();
2015-12-05 06:50:32 +01:00
2016-08-18 18:04:48 +02:00
$this->workers = new \SplObjectStorage;
$this->idleWorkers = new \SplQueue;
$this->busyQueue = new \SplQueue;
2016-08-31 00:36:21 +02:00
$this->push = $this->callableFromInstanceMethod("push");
2015-12-05 06:50:32 +01:00
}
/**
* Checks if the pool is running.
*
* @return bool True if the pool is running, otherwise false.
*/
2016-08-18 18:04:48 +02:00
public function isRunning(): bool {
2015-12-05 06:50:32 +01:00
return $this->running;
}
/**
* Checks if the pool has any idle workers.
*
* @return bool True if the pool has at least one idle worker, otherwise false.
*/
2016-08-18 18:04:48 +02:00
public function isIdle(): bool {
2015-12-05 06:50:32 +01:00
return $this->idleWorkers->count() > 0;
}
/**
2015-12-16 22:53:53 +01:00
* {@inheritdoc}
2015-12-05 06:50:32 +01:00
*/
2016-08-18 18:04:48 +02:00
public function getMinSize(): int {
2015-12-05 06:50:32 +01:00
return $this->minSize;
}
/**
2015-12-16 22:53:53 +01:00
* {@inheritdoc}
2015-12-05 06:50:32 +01:00
*/
2016-08-18 18:04:48 +02:00
public function getMaxSize(): int {
2015-12-05 06:50:32 +01:00
return $this->maxSize;
}
/**
* {@inheritdoc}
*/
2016-08-18 18:04:48 +02:00
public function getWorkerCount(): int {
2015-12-16 22:53:53 +01:00
return $this->workers->count();
2015-12-05 06:50:32 +01:00
}
/**
* {@inheritdoc}
*/
2016-08-19 00:36:58 +02:00
public function getIdleWorkerCount(): int {
2015-12-05 06:50:32 +01:00
return $this->idleWorkers->count();
}
/**
* Starts the worker pool execution.
*
* When the worker pool starts up, the minimum number of workers will be created. This adds some overhead to
* starting the pool, but allows for greater performance during runtime.
*/
2016-08-18 18:04:48 +02:00
public function start() {
2015-12-05 06:50:32 +01:00
if ($this->isRunning()) {
throw new StatusError('The worker pool has already been started.');
}
// Start up the pool with the minimum number of workers.
$count = $this->minSize;
while (--$count >= 0) {
$worker = $this->createWorker();
$this->idleWorkers->enqueue($worker);
2015-12-05 06:50:32 +01:00
}
$this->running = true;
}
/**
* Enqueues a task to be executed by the worker pool.
*
* @param Task $task The task to enqueue.
*
* @return \Amp\Promise<mixed> The return value of Task::run().
2015-12-05 06:50:32 +01:00
*
2016-08-23 23:47:40 +02:00
* @throws \Amp\Parallel\StatusError If the pool has not been started.
2017-07-18 05:53:19 +02:00
* @throws \Amp\Parallel\Worker\TaskException If the task throws an exception.
2015-12-05 06:50:32 +01:00
*/
2016-11-15 00:43:44 +01:00
public function enqueue(Task $task): Promise {
2016-09-02 01:10:52 +02:00
return new Coroutine($this->doEnqueue($this->pull(), $task));
2016-08-31 00:36:21 +02:00
}
2017-05-18 09:51:31 +02:00
2016-08-31 00:36:21 +02:00
/**
* @coroutine
*
2016-09-02 01:10:52 +02:00
* Keeps the worker marked as busy until the task has completed.
*
* @param \Amp\Parallel\Worker\Worker $worker
2016-08-31 00:36:21 +02:00
* @param \Amp\Parallel\Worker\Task $task
*
* @return \Generator
*/
2016-09-02 01:10:52 +02:00
public function doEnqueue(Worker $worker, Task $task): \Generator {
2016-08-31 00:36:21 +02:00
try {
$result = yield $worker->enqueue($task);
} finally {
$this->push($worker);
}
2017-05-18 09:51:31 +02:00
2016-08-31 00:36:21 +02:00
return $result;
2015-12-05 06:50:32 +01:00
}
/**
* Shuts down the pool and all workers in it.
*
* @coroutine
*
* @return \Amp\Promise<int[]> Array of exit status from all workers.
2015-12-05 06:50:32 +01:00
*
2016-08-23 23:47:40 +02:00
* @throws \Amp\Parallel\StatusError If the pool has not been started.
2015-12-05 06:50:32 +01:00
*/
2016-11-15 00:43:44 +01:00
public function shutdown(): Promise {
2015-12-05 06:50:32 +01:00
if (!$this->isRunning()) {
throw new StatusError('The pool 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
2016-08-18 18:04:48 +02:00
/**
* Shuts down the pool and all workers in it.
*
* @coroutine
*
* @return \Generator
*
2016-08-23 23:47:40 +02:00
* @throws \Amp\Parallel\StatusError If the pool has not been started.
2016-08-18 18:04:48 +02:00
*/
private function doShutdown(): \Generator {
$this->running = false;
2015-12-05 06:50:32 +01:00
$shutdowns = [];
foreach ($this->workers as $worker) {
2015-12-16 22:53:53 +01:00
if ($worker->isRunning()) {
2016-08-18 18:04:48 +02:00
$shutdowns[] = $worker->shutdown();
2015-12-16 22:53:53 +01:00
}
}
2015-12-05 06:50:32 +01:00
return yield Promise\all($shutdowns);
2015-12-05 06:50:32 +01:00
}
/**
* Kills all workers in the pool and halts the worker pool.
*/
2016-08-18 18:04:48 +02:00
public function kill() {
$this->running = false;
2015-12-05 06:50:32 +01:00
foreach ($this->workers as $worker) {
$worker->kill();
}
}
/**
* Creates a worker and adds them to the pool.
*
* @return Worker The worker created.
*/
2016-08-18 18:04:48 +02:00
private function createWorker() {
2015-12-05 06:50:32 +01:00
$worker = $this->factory->create();
$worker->start();
$this->workers->attach($worker, 0);
2015-12-05 06:50:32 +01:00
return $worker;
}
/**
* {@inheritdoc}
*/
2016-08-18 18:04:48 +02:00
public function get(): Worker {
2016-08-31 00:36:21 +02:00
return new Internal\PooledWorker($this->pull(), $this->push);
}
2017-05-18 09:51:31 +02:00
2016-08-31 00:36:21 +02:00
/**
* Pulls a worker from the pool. The worker should be put back into the pool with push() to be marked as idle.
*
* @return \Amp\Parallel\Worker\Worker
* @throws \Amp\Parallel\StatusError
*/
protected function pull(): Worker {
if (!$this->isRunning()) {
2016-08-18 18:04:48 +02:00
throw new StatusError("The queue is not running");
}
do {
if ($this->idleWorkers->isEmpty()) {
if ($this->getWorkerCount() >= $this->maxSize) {
// All possible workers busy, so shift from head (will be pushed back onto tail below).
$worker = $this->busyQueue->shift();
} else {
// Max worker count has not been reached, so create another worker.
$worker = $this->createWorker();
}
} else {
// Shift a worker off the idle queue.
$worker = $this->idleWorkers->shift();
}
if ($worker->isRunning()) {
break;
}
$this->workers->detach($worker);
} while (true);
$this->busyQueue->push($worker);
$this->workers[$worker] += 1;
2017-05-18 09:51:31 +02:00
2016-08-31 00:36:21 +02:00
return $worker;
}
/**
* Pushes the worker back into the queue.
*
2016-08-23 23:47:40 +02:00
* @param \Amp\Parallel\Worker\Worker $worker
*
2016-08-18 18:04:48 +02:00
* @throws \Error If the worker was not part of this queue.
*/
2016-08-31 00:36:21 +02:00
protected function push(Worker $worker) {
if (!$this->workers->contains($worker)) {
2016-08-18 18:04:48 +02:00
throw new \Error("The provided worker was not part of this queue");
}
2016-08-23 01:25:19 +02:00
if (($this->workers[$worker] -= 1) === 0) {
// Worker is completely idle, remove from busy queue and add to idle queue.
foreach ($this->busyQueue as $key => $busy) {
if ($busy === $worker) {
unset($this->busyQueue[$key]);
break;
}
}
$this->idleWorkers->push($worker);
}
}
2015-12-05 06:50:32 +01:00
}