Add parallel benchmark using AMP async framework

Author: GromNaN

.github/workflows/benchmark.yml

@@ -64,14 +64,15 @@ jobs:
       - name: "Install dependencies with Composer"
         uses: "ramsey/[email protected]"
         with:
-          composer-options: "--no-suggest"
+          composer-options: "--no-suggest --working-dir=./benchmark"
 
       - name: "Run phpbench"
+        working-directory: "./benchmark"
         run: "vendor/bin/phpbench run --report=aggregate --report=bar_chart_time --report=env --output html"
 
       - name: Upload HTML report
         uses: actions/upload-artifact@v3
         with:
           name: phpbench-${{ github.sha }}.html
-          path: .phpbench/html/index.html
+          path: ./benchmark/.phpbench/html/index.html
           retention-days: 3

benchmark/DriverBench/ParallelBench.php

@@ -2,6 +2,7 @@
 
 namespace MongoDB\Benchmark\DriverBench;
 
+use Amp\Parallel\Worker\DefaultPool;
 use Generator;
 use MongoDB\Benchmark\Fixtures\Data;
 use MongoDB\Benchmark\Utils;
@@ -14,7 +15,8 @@
 use PhpBench\Attributes\Revs;
 use RuntimeException;
 
-use function array_chunk;
+use function Amp\ParallelFunctions\parallelMap;
+use function Amp\Promise\wait;
 use function array_map;
 use function ceil;
 use function count;
@@ -68,8 +70,6 @@ public static function afterClass(): void
     /**
      * Parallel: LDJSON multi-file import
      * Using single thread
-     *
-     * @see https://github.com/mongodb/specifications/blob/ddfc8b583d49aaf8c4c19fa01255afb66b36b92e/source/benchmarking/benchmarking.rst#ldjson-multi-file-import
      */
     #[BeforeMethods('beforeMultiFileImport')]
     #[Revs(1)]
@@ -85,16 +85,15 @@ public function benchMultiFileImport(): void
      * Parallel: LDJSON multi-file import
      * Using multiple forked threads
      *
-     * @see https://github.com/mongodb/specifications/blob/ddfc8b583d49aaf8c4c19fa01255afb66b36b92e/source/benchmarking/benchmarking.rst#ldjson-multi-file-import
      * @param array{processes:int, files:string[], batchSize:int} $params
      */
     #[BeforeMethods('beforeMultiFileImport')]
-    #[ParamProviders(['provideProcessesParameter', 'provideMultiFileImportParameters'])]
+    #[ParamProviders(['provideProcessesParameter'])]
     #[Revs(1)]
     public function benchMultiFileImportFork(array $params): void
     {
         $pids = [];
-        foreach ($params['files'] as $files) {
+        foreach (self::getFileNames() as $file) {
             // Wait for a child process to finish if we have reached the maximum number of processes
             if (count($pids) >= $params['processes']) {
                 $pid = pcntl_waitpid(-1, $status);
@@ -107,11 +106,7 @@ public function benchMultiFileImportFork(array $params): void
                 // If we don't reset, we will get the same manager client_zval in the child process
                 // and share the libmongoc client.
                 Utils::reset();
-                $collection = Utils::getCollection();
-
-                foreach ($files as $file) {
-                    self::importFile($file, $collection);
-                }
+                self::importFile($file, Utils::getCollection());
 
                 // Exit the child process
                 exit(0);
@@ -132,21 +127,31 @@ public function benchMultiFileImportFork(array $params): void
         }
     }
 
-    public static function provideProcessesParameter(): Generator
+    /**
+     * Parallel: LDJSON multi-file import
+     * Using amphp/parallel-functions with worker pool
+     *
+     * @param array{processes:int, files:string[], batchSize:int} $params
+     */
+    #[BeforeMethods('beforeMultiFileImport')]
+    #[ParamProviders(['provideProcessesParameter'])]
+    #[Revs(1)]
+    public function benchMultiFileImportAmp(array $params): void
     {
-        // Max number of forked processes
-        for ($i = 1; $i <= 30; $i = (int) ceil($i * 1.25)) {
-            yield $i . 'fork' => ['processes' => $i];
-        }
+        wait(parallelMap(
+            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']),
+        ));
     }
 
-    public static function provideMultiFileImportParameters(): Generator
+    public static function provideProcessesParameter(): Generator
     {
-        $files = self::getFileNames();
-
-        // Chunk of file names to be handled by each processes
-        for ($i = 1; $i <= 10; $i += 3) {
-            yield 'by ' . $i => ['files' => array_chunk($files, $i)];
+        // Max number of forked processes
+        for ($i = 1; $i <= 30; $i = (int) ceil($i * 1.25)) {
+            yield $i . ' proc' => ['processes' => $i];
         }
     }
 
@@ -166,8 +171,10 @@ public function afterMultiFileImport(): void
         unset($this->files);
     }
 
-    private static function importFile(string $file, Collection $collection): void
+    public static function importFile(string $file, ?Collection $collection = null): void
     {
+        $collection ??= Utils::getCollection();
+
         // Read file contents into BSON documents
         $docs = array_map(
             static fn (string $line) => Document::fromJSON($line),
@@ -186,8 +193,7 @@ private static function getFileNames(): array
 
         return array_map(
             static fn (int $i) => sprintf('%s/%03d.txt', $tempDir, $i),
-            //range(0, 99),
-            range(0, 5),
+            range(0, 99),
         );
     }
 }

benchmark/composer.json

@@ -0,0 +1,24 @@
+{
+  "name": "mongodb/mongodb-benchmark",
+  "type": "project",
+  "repositories": [
+    {
+      "type": "path",
+        "url": "../"
+    }
+  ],
+  "require": {
+    "php": ">=8.1",
+    "amphp/parallel-functions": "^1.1",
+    "mongodb/mongodb": "@dev",
+    "phpbench/phpbench": "^1.2"
+  },
+  "autoload": {
+    "psr-4": {
+      "MongoDB\\Benchmark\\": "./"
+    }
+  },
+  "scripts": {
+    "benchmark": "phpbench run --report=aggregate"
+  }
+}

phpbench.json.dist renamed to benchmark/phpbench.json.dist

@@ -21,7 +21,6 @@
     },
     "require-dev": {
         "doctrine/coding-standard": "^11.1",
-        "phpbench/phpbench": "^1.2",
         "rector/rector": "^0.16.0",
         "squizlabs/php_codesniffer": "^3.7",
         "symfony/phpunit-bridge": "^5.2",
@@ -39,7 +38,6 @@
         "files": [ "tests/PHPUnit/Functions.php" ]
     },
     "scripts": {
-        "benchmark": "phpbench run --report=aggregate",
         "checks": [
             "@check:cs",
             "@check:psalm",