AMPHP driver

.github/workflows/benchmark.yml

@@ -0,0 +1,39 @@
+name: Benchmark
+
+on: [push]
+
+jobs:
+  test:
+    runs-on: ${{ matrix.os }}
+
+    strategy:
+      fail-fast: true
+      matrix:
+        # os: [ubuntu-latest, windows-latest]
+        os: [ubuntu-latest]
+        php: [8.1, 8.2]
+
+    name: PHP ${{ matrix.php }} / ${{ matrix.os }}
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Cache dependencies
+        uses: actions/cache@v3
+        with:
+          path: ~/.composer/cache/files
+          key: dependencies-php-${{ matrix.php }}-composer-${{ hashFiles('composer.json') }}-os-${{ matrix.os }}
+
+      - name: Setup PHP
+        uses: shivammathur/setup-php@v2
+        with:
+          php-version: ${{ matrix.php }}
+          coverage: xdebug
+
+      - name: Install dependencies
+        run: |
+          composer install --prefer-dist --no-interaction --no-suggest
+
+      - name: Create benchmarks
+        run: vendor/bin/phpbench run --report=aggregate

README.md

@@ -9,7 +9,7 @@
 
 A simple and efficient solution for concurrently sending HTTP requests using PSR-18 client implementations.
 
-Peak is a library that enables concurrent request sending using a request pool. It leverages the event loop of [ReactPHP](https://github.com/reactphp) or [PSL](https://github.com/azjezz/psl) to handle and manage the requests concurrently.
+Peak is a library that enables concurrent request sending using a request pool. It leverages the event loop of [AMPHP](https://github.com/amphp), [ReactPHP](https://github.com/reactphp) or [PSL](https://github.com/azjezz/psl) to handle and manage the requests concurrently.
 
 ## Requirements
 
@@ -26,6 +26,12 @@ composer require fansipan/peak
 
 Additionally, depending on your choice of driver, these packages may also need to be installed.
 
+### AMPHP
+
+```bash
+composer require amphp/pipeline
+```
+
 ### PSL
 
 ```bash
@@ -60,9 +66,13 @@ The `Fansipan\Peak\PoolFactory` provides a configured request pool based on the
 First, you need to create your desired driver:
 
 ```php
+use Fansipan\Peak\Concurrency\AmpDeferred;
 use Fansipan\Peak\Concurrency\PslDeferred;
 use Fansipan\Peak\Concurrency\ReactDeferred;
 
+// AMPHP
+$defer = new AmpDeferred();
+
 // PSL
 $defer = new PslDeferred();
 

composer.json

@@ -27,6 +27,7 @@
         "psr/http-factory": "^1.0"
     },
     "require-dev": {
+        "amphp/pipeline": "^1.0",
         "azjezz/psl": "^2.7",
         "clue/mq-react": "^1.6",
         "fansipan/mock-client": "^1.0",
@@ -55,7 +56,7 @@
             "vendor/bin/phpstan analyse"
         ],
         "bench": [
-            "./vendor/bin/phpbench run --report=aggregate"
+            "./vendor/bin/phpbench run --report=all --progress=blinken"
         ],
         "cs": [
             "vendor/bin/php-cs-fixer fix"
@@ -73,6 +74,7 @@
         ]
     },
     "suggest": {
+        "php-http/discovery": "Provides a convenient solution for auto-discovery and auto-installation of well-known implementations of PSR-17 (HTTP Factories), PSR-18 (HTTP Clients)",
         "azjezz/psl": "Required for the PSL driver",
         "clue/mq-react": "Required for the ReactPHP driver",
         "react/async": "Required for the ReactPHP driver"

phpbench.json

@@ -2,5 +2,36 @@
     "$schema":"./vendor/phpbench/phpbench/phpbench.schema.json",
     "runner.bootstrap": "vendor/autoload.php",
     "runner.path": "tests/Benchmark",
-    "runner.file_pattern": "*Bench.php"
+    "runner.file_pattern": "*Bench.php",
+    "report.generators": {
+        "chart": {
+            "generator": "component",
+            "components": [
+                {
+                    "title": "Request average time comparison",
+                    "component": "bar_chart_aggregate",
+                    "x_partition": "variant_params['limit'] ~ ' requests'",
+                    "bar_partition": "subject_name",
+                    "y_expr": "mode(partition['result_time_avg']) as time",
+                    "y_axes_label": "yValue as time"
+                },
+                {
+                    "title": "Peak memory usage comparison",
+                    "component": "bar_chart_aggregate",
+                    "x_partition": "variant_params['limit'] ~ ' requests'",
+                    "bar_partition": "subject_name",
+                    "y_expr": "mode(partition['result_mem_peak']) as memory",
+                    "y_axes_label": "yValue as memory"
+                }
+            ]
+        },
+        "all": {
+            "generator": "composite",
+            "reports": [
+                "default",
+                "aggregate",
+                "chart"
+            ]
+        }
+    }
 }

src/Client/AsyncClientFactory.php

@@ -4,6 +4,7 @@
 
 namespace Fansipan\Peak\Client;
 
+use Fansipan\Peak\Concurrency\AmpDeferred;
 use Fansipan\Peak\Concurrency\Driver;
 use Fansipan\Peak\Concurrency\DriverDiscovery;
 use Fansipan\Peak\Concurrency\PslDeferred;
@@ -36,6 +37,21 @@
 
         $driver = DriverDiscovery::find();
 
+        if ($driver === Driver::AMP) {
+            if ($client instanceof GuzzleClientInterface) {
+                return new GuzzleClient(new AmpDeferred(), $client);
+            }
+
+            if ($client instanceof Psr18Client) {
+                return new SymfonyClient(new AmpDeferred(), self::getUnderlyingSymfonyHttpClient($client));
+            }
+
+            throw new UnsupportedClientException(\sprintf(
+                'The client %s is not supported. The PSL Pool only supports "guzzlehttp/guzzle" and "symfony/http-client".',
+                \get_debug_type($client)
+            ));
+        }
+
         if ($driver === Driver::PSL) {
             if ($client instanceof GuzzleClientInterface) {
                 return new GuzzleClient(new PslDeferred(), $client);

src/Client/AsyncClientTrait.php

@@ -4,6 +4,7 @@
 
 namespace Fansipan\Peak\Client;
 
+use Fansipan\Peak\Concurrency\AmpDeferred;
 use Fansipan\Peak\Concurrency\Deferrable;
 use Fansipan\Peak\Concurrency\Driver;
 use Fansipan\Peak\Concurrency\PslDeferred;
@@ -18,6 +19,7 @@ public function driver(): ?Driver
         $deferrable = $this->getDeferrable();
 
         return match (true) {
+            $deferrable instanceof AmpDeferred => Driver::AMP,
             $deferrable instanceof PslDeferred => Driver::PSL,
             $deferrable instanceof ReactDeferred => Driver::REACT,
             default => null,

src/Client/DelayTrait.php

@@ -16,7 +16,9 @@ trait DelayTrait
     public function delay(int $milliseconds): void
     {
         if ($milliseconds < 0) {
+            // @codeCoverageIgnoreStart
             throw new \ValueError('Delay must be positive, got '.$milliseconds);
+            // @codeCoverageIgnoreEnd
         }
 
         $this->delay = $milliseconds;

src/Client/GuzzleClient.php

@@ -36,12 +36,9 @@ public function sendRequest(RequestInterface $request): ResponseInterface
 
         $this->delay = 0;
 
-        return $this->deferred->defer(static function (\Closure $resolve, \Closure $reject) use ($promise) {
-            $promise->then(
-                static fn (ResponseInterface $response) => $resolve($response),
-                static fn (\Throwable $e) => $reject($e)
-            )->wait();
-        }, $delay);
+        return $this->deferred->defer(
+            static fn () => $promise->wait(), $delay
+        );
     }
 
     private function getDeferrable(): Deferrable

src/Client/SymfonyClient.php

@@ -80,13 +80,9 @@ private function createResponse(SymfonyResponseInterface $response): mixed
 
         $this->delay = 0;
 
-        return $this->deferred->defer(function (\Closure $resolve, \Closure $reject) use ($response) {
-            try {
-                $resolve($this->convertToPsrResponse($response));
-            } catch (\Throwable $e) {
-                $reject($e);
-            }
-        }, $delay);
+        return $this->deferred->defer(
+            fn () => $this->convertToPsrResponse($response), $delay
+        );
     }
 
     private function convertToPsrResponse(SymfonyResponseInterface $response): ResponseInterface

src/ClientPool.php

@@ -23,6 +23,7 @@ public function __construct(private readonly AsyncClientInterface $client)
     public function send(iterable $requests): array
     {
         $promises = static function (AsyncClientInterface $client, iterable $requests) {
+            /** @var array-key $key */
             foreach ($requests as $key => $request) {
                 if ($request instanceof RequestInterface) {
                     yield $key => static fn (): ResponseInterface => $client->sendRequest($request);

src/Concurrency/AmpDeferred.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Fansipan\Peak\Concurrency;
+
+use Amp;
+
+final class AmpDeferred implements Deferrable
+{
+    public function defer(callable $callback, float $delay = 0): mixed
+    {
+        return Amp\async(static function (callable $callback, float $delay) {
+            if ($delay > 0) {
+                Amp\delay($delay);
+            }
+
+            return $callback();
+        }, $callback, $delay)->await();
+    }
+}

src/Concurrency/AmpTask.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Fansipan\Peak\Concurrency;
+
+/**
+ * @template T
+ */
+final class AmpTask
+{
+    /**
+     * @var T
+     */
+    private mixed $value = null;
+
+    /**
+     * @param  \Closure(): T $task
+     */
+    public function __construct(
+        private readonly string|int $key,
+        private readonly \Closure $task
+    ) {
+    }
+
+    public function key(): string|int
+    {
+        return $this->key;
+    }
+
+    /**
+     * @return T
+     */
+    public function value(): mixed
+    {
+        return $this->value;
+    }
+
+    public function __invoke(): self
+    {
+        $this->value = ($this->task)();
+
+        return $this;
+    }
+}

src/Concurrency/AmpWorker.php

@@ -0,0 +1,49 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Fansipan\Peak\Concurrency;
+
+use Amp;
+use Amp\Future;
+use Amp\Pipeline\Pipeline;
+
+final class AmpWorker implements Worker
+{
+    /**
+     * @param  int<1, max> $limit
+     */
+    public function __construct(private readonly int $limit = 10)
+    {
+        if ($limit < 1) {
+            // @codeCoverageIgnoreStart
+            throw new \ValueError('Argument #1 ($limit) must be positive, got '.$limit);
+            // @codeCoverageIgnoreEnd
+        }
+    }
+
+    public function run(iterable $tasks): array
+    {
+        $promises = Pipeline::fromIterable(static function () use ($tasks): \Generator {
+            foreach ($tasks as $key => $task) {
+                yield new AmpTask($key, $task);
+            }
+        })
+            ->concurrent($this->limit)
+            ->unordered()
+            ->map(static fn (AmpTask $task) => Amp\async(\Closure::fromCallable($task)));
+
+        $results = [];
+
+        foreach (Future::iterate($promises) as $promise) {
+            try {
+                /** @var AmpTask $t */
+                $t = $promise->await();
+                $results[$t->key()] = $t->value();
+            } catch (\Throwable) {
+            }
+        }
+
+        return $results;
+    }
+}

src/Concurrency/Deferrable.php

@@ -4,16 +4,14 @@
 
 namespace Fansipan\Peak\Concurrency;
 
-/**
- * @template-covariant T
- */
 interface Deferrable
 {
     /**
      * Defer the operation with optional delay in seconds.
      *
-     * @param  callable(\Closure(T): void, \Closure(\Throwable): void): void $callback
-     * @parma  float<0, max> $delay
+     * @template T
+     *
+     * @param  callable(): T $callback
      * @return T
      */
     public function defer(callable $callback, float $delay = 0): mixed;

src/Concurrency/Driver.php

@@ -6,6 +6,7 @@
 
 enum Driver: string
 {
+    case AMP = 'amphp/pipeline';
     case PSL = 'azjezz/psl';
     case REACT = 'react/async clue/mq-react';
 }