diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7bde850f..1339b0ec 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/src/IoPollLoop.php b/src/IoPollLoop.php new file mode 100644 index 00000000..9604a33a --- /dev/null +++ b/src/IoPollLoop.php @@ -0,0 +1,205 @@ +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); + } +} diff --git a/src/Loop.php b/src/Loop.php index 732c5d5e..da50f442 100644 --- a/src/Loop.php +++ b/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(); diff --git a/tests/IoPollLoopTest.php b/tests/IoPollLoopTest.php new file mode 100644 index 00000000..e8ce4476 --- /dev/null +++ b/tests/IoPollLoopTest.php @@ -0,0 +1,17 @@ +markTestSkipped('IOPollLoop tests skipped because IO Poll is not available.'); + } + + return new IoPollLoop(); + } +}