danog/parallel
1
0
Fork 0
mirror of https://github.com/danog/parallel.git synced 2025-01-22 14:01:14 +01:00
Code Issues Projects Releases Wiki Activity

Improve cancellation of pending tasks; remove worker from pool if stopped

Browse Source
This commit is contained in:
Aaron Piotrowski 2016-01-14 23:37:02 -06:00
parent aac99ed603
commit dc41c173b9
2 changed files with 12 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/Worker/AbstractWorker.php
View File

@ -152,10 +152,12 @@ abstract class AbstractWorker implements Worker
*/
private function cancelPending()
{
$exception = new WorkerException('Worker was shut down.');
if (!$this->busyQueue->isEmpty()) {
$exception = new WorkerException('Worker was shut down.');
while (!$this->busyQueue->isEmpty()) {
$this->busyQueue->dequeue()->cancel($exception);
do {
$this->busyQueue->dequeue()->cancel($exception);
} while (!$this->busyQueue->isEmpty());
}
}
}

10
src/Worker/DefaultPool.php
View File

@ -2,10 +2,8 @@
namespace Icicle\Concurrent\Worker;
use Icicle\Awaitable;
use Icicle\Awaitable\Delayed;
use Icicle\Exception\InvalidArgumentError;
use Icicle\Concurrent\Exception\StatusError;
use Icicle\Concurrent\Exception\WorkerException;
use Icicle\Coroutine\Coroutine;
/**
@ -269,7 +267,13 @@ class DefaultPool implements Pool
// Shift a worker off the idle queue.
$worker = $this->idleWorkers->shift();
}
} while (!$worker->isRunning());
if ($worker->isRunning()) {
break;
}
$this->workers->detach($worker);
} while (true);
$this->busyQueue->push($worker);
$this->workers[$worker] += 1;