Upgrade benchmark to AMPHP with fibers

Author: GromNaN

benchmark/composer.json

@@ -15,7 +15,7 @@
     "require": {
         "php": ">=8.1",
         "ext-pcntl": "*",
-        "amphp/parallel-functions": "^1.1",
+        "amphp/parallel": "^2.2",
         "mongodb/mongodb": "@dev",
         "phpbench/phpbench": "^1.2"
     },
@@ -26,5 +26,8 @@
     },
     "scripts": {
         "benchmark": "phpbench run --report=aggregate"
+    },
+    "config": {
+        "sort-packages": true
     }
 }

benchmark/src/DriverBench/Amp/ImportFileTask.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace MongoDB\Benchmark\DriverBench\Amp;
+
+use Amp\Cancellation;
+use Amp\Parallel\Worker\Task;
+use Amp\Sync\Channel;
+use MongoDB\Benchmark\DriverBench\ParallelMultiFileImportBench;
+
+final class ImportFileTask implements Task
+{
+    public function __construct(
+        private string $file,
+    ) {
+    }
+
+    public function run(Channel $channel, Cancellation $cancellation): mixed
+    {
+        ParallelMultiFileImportBench::importFile($this->file);
+
+        return null;
+    }
+}

benchmark/src/DriverBench/ParallelMultiFileImportBench.php

@@ -2,8 +2,11 @@
 
 namespace MongoDB\Benchmark\DriverBench;
 
-use Amp\Parallel\Worker\DefaultPool;
+use Amp\Future;
+use Amp\Parallel\Worker\ContextWorkerFactory;
+use Amp\Parallel\Worker\ContextWorkerPool;
 use Generator;
+use MongoDB\Benchmark\DriverBench\Amp\ImportFileTask;
 use MongoDB\Benchmark\Fixtures\Data;
 use MongoDB\Benchmark\Utils;
 use MongoDB\BSON\Document;
@@ -16,8 +19,6 @@
 use PhpBench\Attributes\Revs;
 use RuntimeException;
 
-use function Amp\ParallelFunctions\parallelMap;
-use function Amp\Promise\wait;
 use function array_map;
 use function count;
 use function fclose;
@@ -107,7 +108,7 @@ public function benchMultiFileImportInsertMany(): void
     /**
      * Using multiple forked threads
      *
-     * @param array{processes:int, files:string[], batchSize:int} $params
+     * @param array{processes:int} $params
      */
     #[ParamProviders(['provideProcessesParameter'])]
     public function benchMultiFileImportFork(array $params): void
@@ -148,20 +149,25 @@ public function benchMultiFileImportFork(array $params): void
     }
 
     /**
-     * Using amphp/parallel-functions with worker pool
+     * Using amphp/parallel with worker pool
      *
-     * @param array{processes:int, files:string[], batchSize:int} $params
+     * @param array{processes:int} $params
      */
     #[ParamProviders(['provideProcessesParameter'])]
     public function benchMultiFileImportAmp(array $params): void
     {
-        wait(parallelMap(
+        $workerPool = new ContextWorkerPool($params['processes'], new ContextWorkerFactory());
+
+        $futures = array_map(
+            fn ($file) => $workerPool->submit(new ImportFileTask($file))->getFuture(),
             self::getFileNames(),
-            // Uses array callable instead of closure to skip complex serialization
-            [self::class, 'importFile'],
-            // The pool size is the number of processes
-            new DefaultPool($params['processes']),
-        ));
+        );
+
+        foreach (Future::iterate($futures) as $future) {
+            $future->await();
+        }
+
+        $workerPool->shutdown();
     }
 
     public static function provideProcessesParameter(): Generator
@@ -172,7 +178,6 @@ public static function provideProcessesParameter(): Generator
         yield '8 proc' => ['processes' => 8]; // 13 sequences
         yield '13 proc' => ['processes' => 13]; // 8 sequences
         yield '20 proc' => ['processes' => 20]; // 5 sequences
-        yield '34 proc' => ['processes' => 34]; // 3 sequences
     }
 
     /**