[WIP][3.x] Add Io\Poll Event Loop

.github/workflows/ci.yml

@@ -11,6 +11,7 @@ jobs:
     strategy:
       matrix:
         php:
+          - 8.6
           - 8.5
           - 8.4
           - 8.3
@@ -45,6 +46,7 @@ jobs:
     strategy:
       matrix:
         php:
+          - 8.6
           - 8.5
           - 8.4
           - 8.3
@@ -81,6 +83,7 @@ jobs:
     strategy:
       matrix:
         php:
+          - 8.6
           - 8.5
           - 8.4
           - 8.3

src/IoPollLoop.php

@@ -0,0 +1,205 @@
+<?php
+
+namespace React\EventLoop;
+
+use React\EventLoop\Tick\FutureTickQueue;
+use React\EventLoop\Timer\Timer;
+use React\EventLoop\Timer\Timers;
+use SplObjectStorage;
+
+final class IoPollLoop implements LoopInterface
+{
+    /** @internal */
+    const MICROSECONDS_PER_SECOND = 1000000;
+
+    private $running = false;
+    private $context;
+    private $futureTickQueue;
+    private $timers;
+    private $pcntl = false;
+    private $pcntlPoll = false;
+    private $signals;
+    private $watchers = [];
+    private $readListeners = [];
+    private $writeListeners = [];
+
+    public function __construct()
+    {
+        $this->context = new \Io\Poll\Context();
+        $this->futureTickQueue = new FutureTickQueue();
+        $this->timers = new Timers();
+        $this->pcntl = \function_exists('pcntl_signal') && \function_exists('pcntl_signal_dispatch');
+        $this->pcntlPoll = $this->pcntl && !\function_exists('pcntl_async_signals');
+        $this->signals = new SignalsHandler();
+
+        // prefer async signals if available (PHP 7.1+) or fall back to dispatching on each tick
+        if ($this->pcntl && !$this->pcntlPoll) {
+            \pcntl_async_signals(true);
+        }
+    }
+
+    public function addReadStream($stream, $listener)
+    {
+        $key = (int) $stream;
+        if (!isset($this->readListeners[$key])) {
+            $this->readListeners[$key] = $listener;
+        }
+        $this->manageStream($key, $stream, \Io\Poll\Event::Read, true);
+    }
+
+    public function addWriteStream($stream, $listener)
+    {
+        $key = (int) $stream;
+        if (!isset($this->writeListeners[$key])) {
+            $this->writeListeners[$key] = $listener;
+        }
+        $this->manageStream($key, $stream, \Io\Poll\Event::Write, true);
+    }
+
+    public function removeReadStream($stream)
+    {
+        $key = (int) $stream;
+        unset($this->readListeners[$key]);
+        $this->manageStream($key, $stream, \Io\Poll\Event::Read, false);
+    }
+
+    public function removeWriteStream($stream)
+    {
+        $key = (int) $stream;
+        unset($this->writeListeners[$key]);
+        $this->manageStream($key, $stream, \Io\Poll\Event::Write, false);
+    }
+
+    public function addTimer($interval, $callback)
+    {
+        $timer = new Timer($interval, $callback, false);
+
+        $this->timers->add($timer);
+
+        return $timer;
+    }
+
+    public function addPeriodicTimer($interval, $callback)
+    {
+        $timer = new Timer($interval, $callback, true);
+
+        $this->timers->add($timer);
+
+        return $timer;
+    }
+
+    public function cancelTimer(TimerInterface $timer)
+    {
+        $this->timers->cancel($timer);
+    }
+
+    public function futureTick($listener)
+    {
+        $this->futureTickQueue->add($listener);
+    }
+
+    public function addSignal($signal, $listener)
+    {
+        if ($this->pcntl === false) {
+            throw new \BadMethodCallException('Event loop feature "signals" isn\'t supported by the "StreamSelectLoop"');
+        }
+
+        $first = $this->signals->count($signal) === 0;
+        $this->signals->add($signal, $listener);
+
+        if ($first) {
+            \pcntl_signal($signal, [$this->signals, 'call']);
+        }
+    }
+
+    public function removeSignal($signal, $listener)
+    {
+        if (!$this->signals->count($signal)) {
+            return;
+        }
+
+        $this->signals->remove($signal, $listener);
+
+        if ($this->signals->count($signal) === 0) {
+            \pcntl_signal($signal, \SIG_DFL);
+        }
+    }
+
+    public function run()
+    {
+        $this->running = true;
+
+        while ($this->running) {
+            $this->futureTickQueue->tick();
+
+            $this->timers->tick();
+
+            // Future-tick queue has pending callbacks ...
+            if (!$this->futureTickQueue->isEmpty()) {
+                $timeout = 0;
+
+                // There is a pending timer, only block until it is due ...
+            } elseif ($scheduledAt = $this->timers->getFirst()) {
+                $timeout = $scheduledAt - $this->timers->getTime();
+                if ($timeout < 0) {
+                    $timeout = 0;
+                } else {
+                    // Convert float seconds to int microseconds.
+                    // Ensure we do not exceed maximum integer size, which may
+                    // cause the loop to tick once every ~35min on 32bit systems.
+                    $timeout *= self::MICROSECONDS_PER_SECOND;
+                    $timeout = $timeout > \PHP_INT_MAX ? \PHP_INT_MAX : (int)$timeout;
+                }
+
+                // The only possible event is stream or signal activity, so wait forever ...
+            } elseif ($this->readListeners || $this->writeListeners || !$this->signals->isEmpty()) {
+                $timeout = null;
+
+                // There's nothing left to do ...
+            } else {
+                break;
+            }
+
+            foreach ($this->context->wait(\Time\Duration::fromMicroseconds((float) $timeout)) as $watcher) {
+                $stream = $watcher->getHandle()->getStream();
+                $key = (int) $stream;
+
+                if (in_array(\Io\Poll\Event::Read, $watcher->getTriggeredEvents()) && array_key_exists($key, $this->readListeners)) {
+                    \call_user_func($this->readListeners[$key], $stream);
+                }
+
+                if (in_array(\Io\Poll\Event::Write, $watcher->getTriggeredEvents()) && array_key_exists($key, $this->writeListeners)) {
+                    \call_user_func($this->writeListeners[$key], $stream);
+                }
+            }
+        }
+    }
+
+    public function stop()
+    {
+        $this->running = false;
+    }
+
+    private function manageStream($key, $stream, \Io\Poll\Event $event, bool $add)
+    {
+        if (!array_key_exists($key, $this->watchers)) {
+            if (!$add) {
+                return;
+            }
+
+            $handle = new \StreamPollHandle($stream);
+            $this->watchers[$key] = $this->context->add($handle, [$event]);
+
+            return;
+        }
+
+        $events = $this->watchers[$key]->getWatchedEvents();
+        $events = array_filter($events, function (\Io\Poll\Event $e) use ($event) {
+            return $e === $event;
+        });
+        if ($add) {
+            $events[] = $event;
+        }
+        $this->watchers[$key]->modifyEvents($events);
+    }
+}

src/Loop.php

@@ -238,16 +238,20 @@ public static function stop()
     private static function create()
     {
         // @codeCoverageIgnoreStart
-        if (\function_exists('uv_loop_new')) {
-            return new ExtUvLoop();
-        }
-
-        if (\class_exists('EvLoop', false)) {
-            return new ExtEvLoop();
-        }
+//        if (\function_exists('uv_loop_new')) {
+//            return new ExtUvLoop();
+//        }
+//
+//        if (\class_exists('EvLoop', false)) {
+//            return new ExtEvLoop();
+//        }
+//
+//        if (\class_exists('EventBase', false)) {
+//            return new ExtEventLoop();
+//        }
 
-        if (\class_exists('EventBase', false)) {
-            return new ExtEventLoop();
+        if (\class_exists('Io\Poll\Context', false)) {
+            return new IoPollLoop();
         }
 
         return new StreamSelectLoop();

tests/IoPollLoopTest.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace React\Tests\EventLoop;
+
+use React\EventLoop\IoPollLoop;
+
+class IoPollLoopTest extends \React\Tests\EventLoop\AbstractLoopTest
+{
+    public function createLoop()
+    {
+        if (\class_exists('Io\Poll\Context', false)) {
+            $this->markTestSkipped('IOPollLoop tests skipped because IO Poll is not available.');
+        }
+
+        return new IoPollLoop();
+    }
+}