Merge pull request #10 from reactphp-parallel/scale-depending-on-demand

Scale depending on demand

Author: WyriHaximus

etc/qa/phpcs.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0"?>
 <ruleset>
-    <arg name="basepath" value="." />
+    <arg name="basepath" value="../../" />
     <arg name="extensions" value="php" /> <!-- which extensions to look for -->
     <arg name="colors" />
     <arg name="cache" value=".phpcs.cache" /> <!-- cache the results and don't commit them -->

src/EventLoopBridge.php

@@ -24,6 +24,14 @@
 
 final class EventLoopBridge
 {
+    private const DEFAULT_SCALE_RANGE = [
+        0.1,
+        0.01,
+        0.001,
+    ];
+
+    private const DEFAULT_SCALE_POSITION = 2;
+
     private LoopInterface $loop;
 
     private ?Metrics $metrics = null;
@@ -41,6 +49,11 @@ final class EventLoopBridge
 
     private bool $timerActive = FALSE_;
 
+    /** @var array<float> */
+    private array $scaleRange      = self::DEFAULT_SCALE_RANGE;
+    private int $scalePosition     = self::DEFAULT_SCALE_POSITION;
+    private int $scaleNoItemsCount = 0;
+
     public function __construct(LoopInterface $loop)
     {
         $this->loop   = $loop;
@@ -96,8 +109,30 @@ private function startTimer(): void
             $this->metrics->timer()->counter(new Label('event', 'start'))->incr();
         }
 
-        // Call 1K times per second
-        $this->timer       = $this->loop->addPeriodicTimer(0.001, function (): void {
+        $this->runTimer();
+
+        $this->timerActive = TRUE_;
+    }
+
+    private function stopTimer(): void
+    {
+        if (count($this->channels) !== ZERO || count($this->futures) !== ZERO) {
+            return;
+        }
+
+        $this->loop->cancelTimer($this->timer);
+        $this->timerActive = FALSE_;
+
+        if (! ($this->metrics instanceof Metrics)) {
+            return;
+        }
+
+        $this->metrics->timer()->counter(new Label('event', 'stop'))->incr();
+    }
+
+    private function runTimer(): void
+    {
+        $this->timer = $this->loop->addPeriodicTimer($this->scaleRange[$this->scalePosition], function (): void {
             $items = 0;
 
             try {
@@ -131,30 +166,37 @@ private function startTimer(): void
 
             $this->stopTimer();
 
+            /** @phpstan-ignore-next-line */
+            if ($items > 0 && isset($this->scaleRange[$this->scalePosition + 1])) {
+                $this->loop->cancelTimer($this->timer);
+
+                $this->scalePosition++;
+                $this->runTimer();
+
+                $this->scaleNoItemsCount = 0;
+            }
+
+            if ($items === 0) {
+                $this->scaleNoItemsCount++;
+
+                /** @phpstan-ignore-next-line */
+                if ($this->scaleNoItemsCount > 10 && isset($this->scaleRange[$this->scalePosition - 1])) {
+                    $this->loop->cancelTimer($this->timer);
+
+                    $this->scalePosition--;
+                    $this->runTimer();
+
+                    $this->scaleNoItemsCount = 0;
+                }
+            }
+
             if (! ($this->metrics instanceof Metrics)) {
                 return;
             }
 
             $this->metrics->timer()->counter(new Label('event', 'tick'))->incr();
             $this->metrics->timerItems()->counter(new Label('count', (string) $items))->incr();
         });
-        $this->timerActive = TRUE_;
-    }
-
-    private function stopTimer(): void
-    {
-        if (count($this->channels) !== ZERO || count($this->futures) !== ZERO) {
-            return;
-        }
-
-        $this->loop->cancelTimer($this->timer);
-        $this->timerActive = FALSE_;
-
-        if (! ($this->metrics instanceof Metrics)) {
-            return;
-        }
-
-        $this->metrics->timer()->counter(new Label('event', 'stop'))->incr();
     }
 
     private function handleReadEvent(Events\Event $event): void