Merge pull request #9 from reactphp-parallel/add-metrics

Add Metrics

Author: WyriHaximus

README.md

@@ -42,6 +42,25 @@ $infinite->run(function () {
 $loop->run();
 ```
 
+## Metrics
+
+This package supports metrics through [`wyrihaximus/metrics`](https://github.com/wyrihaximus/php-metrics):
+
+```php
+use React\EventLoop\Factory;
+use ReactParallel\EventLoop\EventLoopBridge;
+use ReactParallel\EventLoop\Metrics as EventLoopMetrics;
+use ReactParallel\Pool\Infinite\Infinite;
+use ReactParallel\Pool\Infinite\Metrics;
+use WyriHaximus\Metrics\Configuration;
+use WyriHaximus\Metrics\InMemory\Registry;
+
+$loop = Factory::create();
+$registry = new Registry(Configuration::create());
+$eventLoopBridge = (new EventLoopBridge($loop))->withMetrics(EventLoopMetrics::create($registry));
+$finite = (new Infinite($loop, $eventLoopBridge, 1.3))->withMetrics(Metrics::create($registry));
+```
+
 ## License ##
 
 Copyright 2020 [Cees-Jan Kiewiet](http://wyrihaximus.net/)

composer.json

@@ -20,8 +20,8 @@
         "react/promise": "^2.7",
         "wyrihaximus/async-test-utilities": "^3.2",
         "wyrihaximus/constants": "^1.5",
-        "wyrihaximus/pool-info": "^1.0",
-        "wyrihaximus/ticking-promise": "^1.6"
+        "wyrihaximus/metrics": "^1.0",
+        "wyrihaximus/pool-info": "^1.0"
     },
     "require-dev": {
         "bruli/php-value-objects": "^2.0",

composer.lock

@@ -2,6 +2,7 @@ parameters:
 	checkMissingIterableValueType: false
 	ignoreErrors:
 	    - '#Call to an undefined method React\\Promise\\PromiseInterface::always\(\).#'
+	    - '#Only numeric types are allowed in -, float\|int\|null given on the right side.#'
 	ergebnis:
 		classesAllowedToBeExtended:
 			- ReactParallel\Tests\AbstractPoolTest

etc/qa/phpstan.neon

@@ -14,6 +14,7 @@
 use ReactParallel\Contracts\LowLevelPoolInterface;
 use ReactParallel\EventLoop\EventLoopBridge;
 use ReactParallel\Runtime\Runtime;
+use WyriHaximus\Metrics\Label;
 use WyriHaximus\PoolInfo\Info;
 
 use function array_key_exists;
@@ -22,6 +23,7 @@
 use function count;
 use function dirname;
 use function file_exists;
+use function hrtime;
 use function is_string;
 use function React\Promise\reject;
 use function spl_object_hash;
@@ -51,6 +53,8 @@ final class Infinite implements LowLevelPoolInterface
 
     private float $ttl;
 
+    private ?Metrics $metrics = null;
+
     /** @var GroupInterface[] */
     private array $groups = [];
 
@@ -71,6 +75,14 @@ public function __construct(LoopInterface $loop, EventLoopBridge $eventLoopBridg
         $this->eventLoopBridge = $eventLoopBridge;
     }
 
+    public function withMetrics(Metrics $metrics): self
+    {
+        $self          = clone $this;
+        $self->metrics = $metrics;
+
+        return $self;
+    }
+
     /**
      * @param mixed[] $args
      */
@@ -89,8 +101,21 @@ public function run(Closure $callable, array $args = []): PromiseInterface
 
             $resolve($this->getIdleRuntime());
         }))->then(function (Runtime $runtime) use ($callable, $args): PromiseInterface {
+            $time = null;
+            if ($this->metrics instanceof Metrics) {
+                $this->metrics->threads()->gauge(new Label('state', 'busy'))->incr();
+                $this->metrics->threads()->gauge(new Label('state', 'idle'))->dcr();
+                $time = hrtime(true);
+            }
+
             /** @psalm-suppress UndefinedInterfaceMethod */
-            return $runtime->run($callable, $args)->always(function () use ($runtime): void {
+            return $runtime->run($callable, $args)->always(function () use ($runtime, $time): void {
+                if ($this->metrics instanceof Metrics) {
+                    $this->metrics->executionTime()->summary()->observe((hrtime(true) - $time) / 1e+9);
+                    $this->metrics->threads()->gauge(new Label('state', 'idle'))->incr();
+                    $this->metrics->threads()->gauge(new Label('state', 'busy'))->dcr();
+                }
+
                 if ($this->ttl >= 0.1) {
                     $this->addRuntimeToIdleList($runtime);
                     $this->startTtlTimer($runtime);
@@ -182,6 +207,10 @@ private function spawnRuntime(): Runtime
         $runtime                                   = new Runtime($this->eventLoopBridge, $this->autoload);
         $this->runtimes[spl_object_hash($runtime)] = $runtime;
 
+        if ($this->metrics instanceof Metrics) {
+            $this->metrics->threads()->gauge(new Label('state', 'idle'))->incr();
+        }
+
         return $runtime;
     }
 
@@ -205,6 +234,10 @@ private function closeRuntime(string $hash): void
             unset($this->idleRuntimes[$hash]);
         }
 
+        if ($this->metrics instanceof Metrics) {
+            $this->metrics->threads()->gauge(new Label('state', 'idle'))->dcr();
+        }
+
         if (! array_key_exists($hash, $this->ttlTimers)) {
             return;
         }

src/Infinite.php

@@ -0,0 +1,49 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\Pool\Infinite;
+
+use WyriHaximus\Metrics\Factory as MetricsFactory;
+use WyriHaximus\Metrics\Label\Name;
+use WyriHaximus\Metrics\Registry;
+
+final class Metrics
+{
+    private Registry\Gauges $threads;
+    private Registry\Summaries $executionTime;
+
+    public function __construct(
+        Registry\Gauges $threads,
+        Registry\Summaries $executionTime
+    ) {
+        $this->threads       = $threads;
+        $this->executionTime = $executionTime;
+    }
+
+    public static function create(Registry $registry): self
+    {
+        return new self(
+            $registry->gauge(
+                'react_parallel_pool_infinite_threads',
+                'Currently active or idle thread count',
+                new Name('state')
+            ),
+            $registry->summary(
+                'react_parallel_pool_infinite_execution_time',
+                'Thread call execution time',
+                MetricsFactory::defaultQuantiles()
+            ),
+        );
+    }
+
+    public function threads(): Registry\Gauges
+    {
+        return $this->threads;
+    }
+
+    public function executionTime(): Registry\Summaries
+    {
+        return $this->executionTime;
+    }
+}

src/Metrics.php

@@ -2,12 +2,14 @@
 
 namespace ReactParallel\Tests\Pool\Infinite;
 
+use WyriHaximus\Metrics\Factory as MetricsFactory;
 use React\EventLoop\Factory;
 use React\EventLoop\LoopInterface;
 use React\Promise\PromiseInterface;
 use ReactParallel\Contracts\PoolInterface;
 use ReactParallel\EventLoop\EventLoopBridge;
 use ReactParallel\Pool\Infinite\Infinite;
+use ReactParallel\Pool\Infinite\Metrics;
 use ReactParallel\Tests\AbstractPoolTest;
 use WyriHaximus\PoolInfo\Info;
 use WyriHaximus\PoolInfo\PoolInfoInterface;
@@ -28,7 +30,7 @@ final class InfiniteTest extends AbstractPoolTest
     public function withAZeroTTLThreadsShouldBeKilledOffImmidetally(): void
     {
         $loop = Factory::create();
-        $pool = new Infinite($loop, new EventLoopBridge($loop), 0.0);
+        $pool = $pool = (new Infinite($loop, new EventLoopBridge($loop), 0.0))->withMetrics(Metrics::create(MetricsFactory::create()));
 
         self::assertSame([
             Info::TOTAL => 0,
@@ -74,7 +76,7 @@ public function withAZeroTTLThreadsShouldBeKilledOffImmidetally(): void
     public function withAnAlmostZeroTTLThreadsShouldNotBeKilledOffImmidetally(): void
     {
         $loop = Factory::create();
-        $pool = new Infinite($loop, new EventLoopBridge($loop), 5);
+        $pool = $pool = (new Infinite($loop, new EventLoopBridge($loop), 5))->withMetrics(Metrics::create(MetricsFactory::create()));
 
         self::assertSame([
             Info::TOTAL => 0,
@@ -141,12 +143,12 @@ public function withAnAlmostZeroTTLThreadsShouldNotBeKilledOffImmidetally(): voi
     private function poolFactory(): PoolInfoInterface
     {
         $loop = Factory::create();
-        return new Infinite($loop, new EventLoopBridge($loop), 5);
+        return $pool = (new Infinite($loop, new EventLoopBridge($loop), 5))->withMetrics(Metrics::create(MetricsFactory::create()));
     }
 
     protected function createPool(LoopInterface $loop): PoolInterface
     {
-        return new Infinite($loop, new EventLoopBridge($loop), 5);
+        return $pool = (new Infinite($loop, new EventLoopBridge($loop), 5))->withMetrics(Metrics::create(MetricsFactory::create()));
     }
 
     /**
@@ -155,7 +157,7 @@ protected function createPool(LoopInterface $loop): PoolInterface
     public function aquireLock(): void
     {
         $loop = Factory::create();
-        $pool = new Infinite($loop, new EventLoopBridge($loop), 5);
+        $pool = (new Infinite($loop, new EventLoopBridge($loop), 5))->withMetrics(Metrics::create(MetricsFactory::create()));
 
         $group = $pool->acquireGroup();
         self::assertFalse($pool->close());