Fix assigning workers to waiting tasks

Fixes #177.
amphp · May 12, 2023 · 37850ff · 37850ff
1 parent f91c44b
commit 37850ff
Show file tree

Hide file tree

Showing 3 changed files with 70 additions and 33 deletions.
diff --git a/src/Worker/ContextWorkerPool.php b/src/Worker/ContextWorkerPool.php
@@ -35,7 +35,7 @@ final class ContextWorkerPool implements WorkerPool
     private readonly \SplQueue $idleWorkers;
 
     /** @var \SplQueue<DeferredFuture<Worker|null>> Task submissions awaiting an available worker. */
-    private \SplQueue $waiting;
+    private readonly \SplQueue $waiting;
 
     /** @var \Closure(Worker):void */
     private readonly \Closure $push;
@@ -58,33 +58,26 @@ public function __construct(
         private readonly int $limit = self::DEFAULT_WORKER_LIMIT,
         private readonly ?WorkerFactory $factory = null,
     ) {
-        if ($limit < 0) {
-            throw new \Error("Maximum size must be a non-negative integer");
+        if ($limit <= 0) {
+            throw new \ValueError("Maximum size must be a positive integer");
         }
 
-        $this->workers = new \SplObjectStorage();
-        $this->idleWorkers = new \SplQueue();
-        $this->waiting = new \SplQueue();
+        $this->workers = $workers = new \SplObjectStorage();
+        $this->idleWorkers = $idleWorkers = new \SplQueue();
+        $this->waiting = $waiting = new \SplQueue();
 
         $this->deferredCancellation = new DeferredCancellation();
 
-        $workers = $this->workers;
-        $idleWorkers = $this->idleWorkers;
-        $waiting = $this->waiting;
-
+        /** @var \SplObjectStorage<Worker, int> $workers Needed for Psalm. */
         $this->push = static function (Worker $worker) use ($waiting, $workers, $idleWorkers): void {
             if (!$workers->contains($worker)) {
+                // Pool was shutdown, do not re-insert worker into collection.
                 return;
             }
 
-            if ($worker->isRunning()) {
+            if ($waiting->isEmpty()) {
                 $idleWorkers->push($worker);
             } else {
-                $workers->detach($worker);
-                $worker = null;
-            }
-
-            if (!$waiting->isEmpty()) {
                 $waiting->dequeue()->complete($worker);
             }
         };

diff --git a/test/Worker/AbstractPoolTest.php b/test/Worker/AbstractPoolTest.php
@@ -4,8 +4,10 @@
 
 use Amp\Future;
 use Amp\Parallel\Context\StatusError;
+use Amp\Parallel\Test\Worker\Fixtures\TestTask;
 use Amp\Parallel\Worker\ContextWorkerFactory;
 use Amp\Parallel\Worker\ContextWorkerPool;
+use Amp\Parallel\Worker\Execution;
 use Amp\Parallel\Worker\Task;
 use Amp\Parallel\Worker\Worker;
 use Amp\Parallel\Worker\WorkerPool;
@@ -81,25 +83,25 @@ public function testBusyPool(): void
             return new Fixtures\TestTask($value);
         }, $values);
 
-        $promises = \array_map(function (Task $task) use ($pool): Future {
+        $futures = \array_map(function (Task $task) use ($pool): Future {
             return $pool->submit($task)->getFuture();
         }, $tasks);
 
-        self::assertEquals($values, Future\await($promises));
+        self::assertEquals($values, Future\await($futures));
 
-        $promises = \array_map(function (Task $task) use ($pool): Future {
+        $futures = \array_map(function (Task $task) use ($pool): Future {
             return $pool->submit($task)->getFuture();
         }, $tasks);
 
-        self::assertEquals($values, Future\await($promises));
+        self::assertEquals($values, Future\await($futures));
 
         $pool->shutdown();
     }
 
     public function testCreatePoolShouldThrowError(): void
     {
         $this->expectException(\Error::class);
-        $this->expectExceptionMessage('Maximum size must be a non-negative integer');
+        $this->expectExceptionMessage('Maximum size must be a positive integer');
 
         $this->createPool(-1);
     }
@@ -112,28 +114,60 @@ public function testCleanGarbageCollection(): void
 
             $values = \range(1, 50);
 
-            $promises = \array_map(static function (int $value) use ($pool): Future {
+            $futures = \array_map(static function (int $value) use ($pool): Future {
                 return $pool->submit(new Fixtures\TestTask($value))->getFuture();
             }, $values);
 
-            self::assertEquals($values, Future\await($promises));
+            self::assertEquals($values, Future\await($futures));
         }
     }
 
+    /**
+     * @see https://github.com/amphp/parallel/issues/66
+     */
     public function testPooledKill(): void
     {
         $this->setTimeout(10);
+        \set_error_handler(static function (int $errno, string $errstr) use (&$error): void {
+            $error = $errstr;
+        });
+
+        try {
+            $pool = $this->createPool(1);
+            $worker1 = $pool->getWorker();
+            $worker1->kill();
+            self::assertFalse($worker1->isRunning());
+
+            unset($worker1); // Destroying the worker will trigger the pool to recognize it has been killed.
+
+            $worker2 = $pool->getWorker();
+            self::assertTrue($worker2->isRunning());
+
+            self::assertStringContainsString('Worker in pool crashed', $error);
+        } finally {
+            \restore_error_handler();
+        }
+    }
+
+    /**
+     * @see https://github.com/amphp/parallel/issues/177
+     */
+    public function testWaitingForAvailableWorker(): void
+    {
+        $count = 4;
+        $delay = 0.1;
+
+        $this->setMinimumRuntime($delay * $count);
+        $this->setTimeout($delay * $count + $delay);
 
-        // See https://github.com/amphp/parallel/issues/66
         $pool = $this->createPool(1);
-        $worker1 = $pool->getWorker();
-        $worker1->kill();
-        self::assertFalse($worker1->isRunning());
 
-        unset($worker1); // Destroying the worker will trigger the pool to recognize it has been killed.
+        $executions = [];
+        for ($i = 0; $i < $count; $i++) {
+            $executions[] = $pool->submit(new TestTask($i, $delay));
+        }
 
-        $worker2 = $pool->getWorker();
-        self::assertTrue($worker2->isRunning());
+        Future\await(\array_map(fn (Execution $e) => $e->getFuture(), $executions));
     }
 
     protected function createWorker(?string $autoloadPath = null): Worker

diff --git a/test/Worker/DefaultPoolTest.php b/test/Worker/DefaultPoolTest.php
@@ -53,10 +53,20 @@ public function testCrashedWorker(): void
                 return $worker;
             });
 
-        $pool = new ContextWorkerPool(32, $factory);
+        \set_error_handler(static function (int $errno, string $errstr) use (&$error): void {
+            $error = $errstr;
+        });
+
+        try {
+            $pool = new ContextWorkerPool(32, $factory);
+
+            $pool->submit($this->createMock(Task::class))->await();
 
-        $pool->submit($this->createMock(Task::class))->await();
+            $pool->submit($this->createMock(Task::class))->await();
 
-        $pool->submit($this->createMock(Task::class))->await();
+            self::assertStringContainsString('Worker in pool crashed', $error);
+        } finally {
+            \restore_error_handler();
+        }
     }
 }