From 712ac5469329ad5fb43c9152da5063c435e18c97 Mon Sep 17 00:00:00 2001 From: Nicolas Grekas Date: Sun, 20 Sep 2026 22:05:16 +0200 Subject: [PATCH 1/2] Add an event loop driver on the Io\Poll API of PHP 8.6 IoPollDriver watches streams through Io\Poll, which is epoll on Linux, kqueue on BSD and macOS and event ports on Solaris, so the loop is no longer capped at the FD_SETSIZE of stream_select() and no longer pays its O(n) scan per tick. The driver is picked over StreamSelectDriver when the API is available, natively on PHP 8.6 and through symfony/polyfill-io-poll below it. --- .github/workflows/ci.yml | 17 + composer.json | 4 + psalm.xml | 1 + src/EventLoop/Driver/IoPollDriver.php | 378 ++++++++++++++++++ src/EventLoop/DriverFactory.php | 5 + stubs/io-poll.php | 130 ++++++ test/Driver/IoPollDriverTest.php | 205 ++++++++++ .../event_loop_destruction_order_io_poll.phpt | 56 +++ 8 files changed, 796 insertions(+) create mode 100644 src/EventLoop/Driver/IoPollDriver.php create mode 100644 stubs/io-poll.php create mode 100644 test/Driver/IoPollDriverTest.php create mode 100644 test/event_loop_destruction_order_io_poll.phpt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f0c17e0..2060709 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,6 +32,18 @@ jobs: php-version: 8.5 composer-flags: '--ignore-platform-req=php+' + - operating-system: 'ubuntu-latest' + php-version: 8.6 + composer-flags: '--ignore-platform-req=php+' + psalm: 'skip' # psalm/phar does not run on 8.6 yet + + - operating-system: 'ubuntu-latest' + php-version: 8.4 + composer-flags: '--ignore-platform-req=php+' + job-description: 'with the Io\Poll polyfill' + io-poll-polyfill: 'yes' + psalm: 'skip' + name: PHP ${{ matrix.php-version }} ${{ matrix.job-description }} runs-on: ${{ matrix.operating-system }} @@ -80,6 +92,11 @@ jobs: php -v composer info -D + # 1.x until the Io\Poll fixes the driver needs are released, see symfony/polyfill#655 and #656 + - name: Install the Io\Poll polyfill + run: composer require --dev --no-interaction --no-progress ${{ matrix.composer-flags }} symfony/polyfill-io-poll:1.x-dev symfony/polyfill-time:1.x-dev + if: matrix.io-poll-polyfill + - name: Run tests run: php vendor/bin/phpunit diff --git a/composer.json b/composer.json index 6176de5..ec65881 100644 --- a/composer.json +++ b/composer.json @@ -38,6 +38,10 @@ "jetbrains/phpstorm-stubs": "^2019.3", "psalm/phar": "6.16.*" }, + "suggest": { + "symfony/polyfill-io-poll": "To use IoPollDriver on PHP < 8.6, together with symfony/polyfill-time", + "symfony/polyfill-time": "To use IoPollDriver on PHP < 8.6, together with symfony/polyfill-io-poll" + }, "autoload": { "psr-4": { "Revolt\\": "src" diff --git a/psalm.xml b/psalm.xml index fd49f97..e7b1513 100644 --- a/psalm.xml +++ b/psalm.xml @@ -106,5 +106,6 @@ + diff --git a/src/EventLoop/Driver/IoPollDriver.php b/src/EventLoop/Driver/IoPollDriver.php new file mode 100644 index 0000000..62944da --- /dev/null +++ b/src/EventLoop/Driver/IoPollDriver.php @@ -0,0 +1,378 @@ += 80600 && \class_exists(Context::class) && \class_exists(Duration::class); + } + + private readonly Context $context; + + /** @var array One watcher per stream: the poll context keys them by file descriptor. */ + private array $watchers = []; + + /** + * Context for the streams the preferred backend refuses. epoll and kqueue only accept handles that + * implement polling, while regular files, /dev/null and the like are accepted by poll(), which + * reports them as always ready, the way stream_select() does. + */ + private ?Context $alwaysReadyContext = null; + + /** @var array Streams watched in the context above, keyed by stream id. */ + private array $alwaysReadyStreams = []; + + /** @var array> */ + private array $readCallbacks = []; + + /** @var array> */ + private array $writeCallbacks = []; + + private readonly TimerQueue $timerQueue; + + /** @var array> */ + private array $signalCallbacks = []; + + /** @var \SplQueue */ + private readonly \SplQueue $signalQueue; + + private bool $signalHandling; + + public function __construct() + { + parent::__construct(); + + $this->context = new Context(); + $this->signalQueue = new \SplQueue(); + $this->timerQueue = new TimerQueue(); + $this->signalHandling = \extension_loaded("pcntl") + && \function_exists('pcntl_signal_dispatch') + && \function_exists('pcntl_signal'); + } + + public function __destruct() + { + foreach ($this->signalCallbacks as $signalCallbacks) { + foreach ($signalCallbacks as $signalCallback) { + $this->deactivate($signalCallback); + } + } + } + + /** + * @throws UnsupportedFeatureException If the pcntl extension is not available. + */ + #[\Override] + public function onSignal(int $signal, \Closure $closure): string + { + if (!$this->signalHandling) { + throw new UnsupportedFeatureException("Signal handling requires the pcntl extension"); + } + + return parent::onSignal($signal, $closure); + } + + #[\Override] + public function getHandle(): Context + { + return $this->context; + } + + #[\Override] + protected function now(): float + { + return (float) \hrtime(true) / 1_000_000_000; + } + + /** + * @throws \Throwable + */ + #[\Override] + protected function dispatch(bool $blocking): void + { + if ($this->signalHandling) { + \pcntl_signal_dispatch(); + + while (!$this->signalQueue->isEmpty()) { + $signal = $this->signalQueue->dequeue(); + + foreach ($this->signalCallbacks[$signal] as $callback) { + $this->enqueueCallback($callback); + } + + $blocking = false; + } + } + + $this->poll($blocking ? $this->getTimeout() : 0.0); + + $now = $this->now(); + + while ($callback = $this->timerQueue->extract($now)) { + $this->enqueueCallback($callback); + } + } + + #[\Override] + protected function activate(array $callbacks): void + { + foreach ($callbacks as $callback) { + if ($callback instanceof StreamReadableCallback) { + \assert(\is_resource($callback->stream)); + + $streamId = (int) $callback->stream; + $this->readCallbacks[$streamId][$callback->id] = $callback; + $this->watch($streamId, $callback->stream); + } elseif ($callback instanceof StreamWritableCallback) { + \assert(\is_resource($callback->stream)); + + $streamId = (int) $callback->stream; + $this->writeCallbacks[$streamId][$callback->id] = $callback; + $this->watch($streamId, $callback->stream); + } elseif ($callback instanceof TimerCallback) { + $this->timerQueue->insert($callback); + } elseif ($callback instanceof SignalCallback) { + if (!isset($this->signalCallbacks[$callback->signal])) { + \set_error_handler(static function (int $errno, string $errstr): bool { + throw new UnsupportedFeatureException( + \sprintf("Failed to register signal handler; Errno: %d; %s", $errno, $errstr) + ); + }); + + // Avoid bug in Psalm handling of first-class callables by assigning to a temp variable. + $handler = $this->handleSignal(...); + + try { + \pcntl_signal($callback->signal, $handler); + } finally { + \restore_error_handler(); + } + } + + $this->signalCallbacks[$callback->signal][$callback->id] = $callback; + } else { + // @codeCoverageIgnoreStart + throw new \Error("Unknown callback type"); + // @codeCoverageIgnoreEnd + } + } + } + + #[\Override] + protected function deactivate(DriverCallback $callback): void + { + if ($callback instanceof StreamReadableCallback) { + $streamId = (int) $callback->stream; + unset($this->readCallbacks[$streamId][$callback->id]); + if (empty($this->readCallbacks[$streamId])) { + unset($this->readCallbacks[$streamId]); + } + + $this->watch($streamId, $callback->stream); + } elseif ($callback instanceof StreamWritableCallback) { + $streamId = (int) $callback->stream; + unset($this->writeCallbacks[$streamId][$callback->id]); + if (empty($this->writeCallbacks[$streamId])) { + unset($this->writeCallbacks[$streamId]); + } + + $this->watch($streamId, $callback->stream); + } elseif ($callback instanceof TimerCallback) { + $this->timerQueue->remove($callback); + } elseif ($callback instanceof SignalCallback) { + if (isset($this->signalCallbacks[$callback->signal])) { + unset($this->signalCallbacks[$callback->signal][$callback->id]); + + if (empty($this->signalCallbacks[$callback->signal])) { + unset($this->signalCallbacks[$callback->signal]); + \set_error_handler(static fn () => true); + try { + \pcntl_signal($callback->signal, \SIG_DFL); + } finally { + \restore_error_handler(); + } + } + } + } else { + // @codeCoverageIgnoreStart + throw new \Error("Unknown callback type"); + // @codeCoverageIgnoreEnd + } + } + + /** + * Brings the watcher of the given stream in line with the callbacks enabled on it. + */ + private function watch(int $streamId, mixed $stream): void + { + $events = []; + if (isset($this->readCallbacks[$streamId])) { + $events[] = Event::Read; + } + if (isset($this->writeCallbacks[$streamId])) { + $events[] = Event::Write; + } + + $watcher = $this->watchers[$streamId] ?? null; + + if (!$events || !\is_resource($stream)) { + if ($watcher !== null) { + unset($this->watchers[$streamId], $this->alwaysReadyStreams[$streamId]); + + // Closing a stream already takes its descriptor out of the context, and removing the + // watcher of a closed stream crashes PHP before php/php-src#23791 + if (\is_resource($stream)) { + $watcher->remove(); + } + } + + return; + } + + try { + if ($watcher !== null) { + $watcher->modifyEvents($events); + + return; + } + + $handle = new \StreamPollHandle($stream); + + try { + $this->watchers[$streamId] = $this->context->add($handle, $events, $streamId); + } catch (FailedHandleAddException $exception) { + $this->alwaysReadyContext ??= new Context(Backend::Poll); + + try { + $this->watchers[$streamId] = $this->alwaysReadyContext->add($handle, $events, $streamId); + } catch (PollException) { + throw $exception; + } + + $this->alwaysReadyStreams[$streamId] = true; + } + } catch (HandleAlreadyWatchedException | InvalidHandleException $exception) { + throw new \Error( + "Polling the stream failed: ensure all callbacks on closed stream resources are cancelled", + previous: $exception, + ); + } + } + + private function poll(float $timeout): void + { + if (!$this->watchers) { + if ($timeout < 0) { // Only signal callbacks are enabled, so sleep indefinitely. + /** @psalm-suppress ArgumentTypeCoercion */ + \usleep(\PHP_INT_MAX); + return; + } + + if ($timeout > 0) { // Sleep until next timer expires. + /** @psalm-suppress ArgumentTypeCoercion $timeout is positive here. */ + \usleep((int) ($timeout * 1_000_000)); + } + + return; + } + + if ($this->alwaysReadyStreams) { + // Those streams are ready by definition, so there is nothing to wait for. + $timeout = 0.0; + } + + if ($timeout < 0) { + $duration = null; + } else { + $seconds = (int) $timeout; + $duration = Duration::fromSeconds($seconds, (int) (($timeout - $seconds) * 1_000_000_000)); + } + + try { + $watchers = $this->context->wait($duration); + } catch (FailedPollWaitException $exception) { + if ($exception->getCode() !== FailedPollOperationException::ERROR_INTERRUPTED) { + throw $exception; + } + + return; // A signal arrived, it is dispatched at the start of the next tick. + } + + if ($this->alwaysReadyContext !== null) { + $watchers = [...$watchers, ...$this->alwaysReadyContext->wait(Duration::fromSeconds(0))]; + } + + foreach ($watchers as $watcher) { + $streamId = $watcher->getData(); + + // Error and HangUp are reported whether they were requested or not. Both sides have to be + // woken up on them, or a level-triggered backend reports them again on every wait(). + $aborted = $watcher->hasTriggered(Event::Error) || $watcher->hasTriggered(Event::HangUp); + + if ($aborted || $watcher->hasTriggered(Event::Read)) { + foreach ($this->readCallbacks[$streamId] ?? [] as $callback) { + $this->enqueueCallback($callback); + } + } + + if ($aborted || $watcher->hasTriggered(Event::Write)) { + foreach ($this->writeCallbacks[$streamId] ?? [] as $callback) { + $this->enqueueCallback($callback); + } + } + } + } + + /** + * @return float Seconds until next timer expires or -1 if there are no pending timers. + */ + private function getTimeout(): float + { + $expiration = $this->timerQueue->peek(); + + if ($expiration === null) { + return -1; + } + + $expiration -= $this->now(); + + return $expiration > 0 ? $expiration : 0.0; + } + + private function handleSignal(int $signal): void + { + // Queue signals, so we don't suspend inside pcntl_signal_dispatch, which disables signals while it runs + $this->signalQueue->enqueue($signal); + } +} diff --git a/src/EventLoop/DriverFactory.php b/src/EventLoop/DriverFactory.php index 9d732fd..efd8bd3 100644 --- a/src/EventLoop/DriverFactory.php +++ b/src/EventLoop/DriverFactory.php @@ -7,6 +7,7 @@ // @codeCoverageIgnoreStart use Revolt\EventLoop\Driver\EvDriver; use Revolt\EventLoop\Driver\EventDriver; +use Revolt\EventLoop\Driver\IoPollDriver; use Revolt\EventLoop\Driver\StreamSelectDriver; use Revolt\EventLoop\Driver\TracingDriver; use Revolt\EventLoop\Driver\UvDriver; @@ -39,6 +40,10 @@ public function create(): Driver return new EventDriver(); } + if (IoPollDriver::isSupported()) { + return new IoPollDriver(); + } + return new StreamSelectDriver(); })(); diff --git a/stubs/io-poll.php b/stubs/io-poll.php new file mode 100644 index 0000000..efcc3bf --- /dev/null +++ b/stubs/io-poll.php @@ -0,0 +1,130 @@ += 8.6, or by symfony/polyfill-io-poll below it. + +namespace Io { + class IoException extends \Exception {} +} + +namespace Io\Poll { + enum Backend + { + case Auto; + case Poll; + case Epoll; + case Kqueue; + case EventPorts; + case WSAPoll; + + /** @return list */ + public static function getAvailableBackends(): array {} + + public function isAvailable(): bool {} + + public function supportsEdgeTriggering(): bool {} + } + + enum Event + { + case Read; + case Write; + case Error; + case HangUp; + case ReadHangUp; + case OneShot; + case EdgeTriggered; + } + + interface Handle {} + + final class Watcher + { + public function getHandle(): Handle {} + + /** @return list */ + public function getWatchedEvents(): array {} + + /** @return list */ + public function getTriggeredEvents(): array {} + + public function getData(): mixed {} + + public function hasTriggered(Event $event): bool {} + + public function isActive(): bool {} + + /** @param list $events */ + public function modify(array $events, mixed $data = null): void {} + + /** @param list $events */ + public function modifyEvents(array $events): void {} + + public function modifyData(mixed $data): void {} + + public function remove(): void {} + } + + final class Context + { + public function __construct(Backend $backend = Backend::Auto) {} + + /** @param list $events */ + public function add(Handle $handle, array $events, mixed $data = null): Watcher {} + + /** @return list */ + public function wait(?\Time\Duration $timeout = null, ?int $maxEvents = null): array {} + + public function getBackend(): Backend {} + } + + class PollException extends \Io\IoException {} + + abstract class FailedPollOperationException extends PollException + { + public const int ERROR_NONE = 0; + public const int ERROR_SYSTEM = 1; + public const int ERROR_NOMEM = 2; + public const int ERROR_INVALID = 3; + public const int ERROR_EXISTS = 4; + public const int ERROR_NOTFOUND = 5; + public const int ERROR_TIMEOUT = 6; + public const int ERROR_INTERRUPTED = 7; + public const int ERROR_PERMISSION = 8; + public const int ERROR_TOOBIG = 9; + public const int ERROR_AGAIN = 10; + public const int ERROR_NOSUPPORT = 11; + } + + class FailedContextInitializationException extends FailedPollOperationException {} + class FailedHandleAddException extends FailedPollOperationException {} + class FailedWatcherModificationException extends FailedPollOperationException {} + class FailedPollWaitException extends FailedPollOperationException {} + class BackendUnavailableException extends PollException {} + class InactiveWatcherException extends PollException {} + class HandleAlreadyWatchedException extends PollException {} + class InvalidHandleException extends PollException {} +} + +namespace Time { + final class Duration + { + public readonly int $seconds; + public readonly int $nanoseconds; + public readonly bool $negative; + + public static function fromSeconds(int $seconds, int $nanoseconds = 0): self {} + } +} + +namespace { + final class StreamPollHandle implements Io\Poll\Handle + { + /** @param resource $stream */ + public function __construct($stream) {} + + /** @return resource */ + public function getStream() {} + + public function isValid(): bool {} + } +} diff --git a/test/Driver/IoPollDriverTest.php b/test/Driver/IoPollDriverTest.php new file mode 100644 index 0000000..233af30 --- /dev/null +++ b/test/Driver/IoPollDriverTest.php @@ -0,0 +1,205 @@ +loop->getHandle()); + } + + public function testAsyncSignals(): void + { + if (\DIRECTORY_SEPARATOR === '\\') { + self::markTestSkipped('Skip on Windows'); + } + + if (!\extension_loaded("pcntl") + || !\function_exists('pcntl_signal_dispatch') + || !\function_exists('pcntl_signal')) { + self::markTestSkipped('Skip, PCNTL functions not available'); + } + + \pcntl_async_signals(true); + + try { + $this->start(function (Driver $loop) use (&$invoked, &$callbackId) { + $callbackId = $loop->onSignal(SIGUSR1, function () use (&$invoked) { + $invoked = true; + }); + + $loop->defer(function () use ($loop, $callbackId) { + \posix_kill(\getmypid(), \SIGUSR1); + + // Two defers, because defer is queued in the first tick and signals only after signals have been + // processed, so the second tick dispatches the signal. At the start of the third tick, we're done! + $loop->defer(function () use ($loop, $callbackId) { + $loop->defer(function () use ($loop, $callbackId) { + $loop->cancel($callbackId); + }); + }); + }); + }); + } finally { + \pcntl_async_signals(false); + } + + self::assertTrue($invoked); + + $this->loop->cancel($callbackId); + } + + /** + * @requires extension pcntl + */ + public function testSignalDuringPollIgnored(): void + { + if (\DIRECTORY_SEPARATOR === '\\') { + self::markTestSkipped('Skip on Windows'); + } + + if (!\extension_loaded("pcntl") + || !\function_exists('pcntl_signal_dispatch') + || !\function_exists('pcntl_signal') + ) { + self::markTestSkipped('Skip, PCNTL functions not available'); + } + + $sockets = self::createSocketPair(); + + $this->start(function (Driver $loop) use ($sockets, &$signalCallbackId) { + $socketCallbackIds = [ + $loop->onReadable($sockets[0], function () { + // nothing + }), + $loop->onReadable($sockets[1], function () { + // nothing + }), + ]; + + $signalCallbackId = $loop->onSignal(\SIGUSR2, function ($callbackId) use ($socketCallbackIds, $loop) { + $loop->cancel($callbackId); + + foreach ($socketCallbackIds as $socketCallbackId) { + $loop->cancel($socketCallbackId); + } + + $this->assertTrue(true); + }); + + $loop->delay(0.1, function () { + \proc_open('sh -c "sleep 1; kill -USR2 ' . \getmypid() . '"', [], $pipes); + }); + }); + + $this->loop->cancel($signalCallbackId); + } + + /** + * stream_select() is capped at FD_SETSIZE, which is 1024 on Linux. Poll backends are not. + */ + public function testMoreFileDescriptorsThanFdSetSize(): void + { + if (\stripos(PHP_OS, 'win') === 0) { + self::markTestSkipped('Skip on Windows'); + } + + if (!(new \ReflectionClass(Context::class))->isInternal()) { + self::markTestSkipped('Skip, the polyfill is backed by stream_select() and capped at FD_SETSIZE too'); + } + + $sockets = []; + + for ($i = 0; $i < 700; $i++) { + $sockets[] = self::createSocketPair(); + } + + $invoked = false; + + try { + $this->start(function (Driver $loop) use ($sockets, &$invoked) { + foreach ($sockets as [$left, $right]) { + $loop->onReadable($left, static function () { + // nothing + }); + + $loop->onReadable($right, static function () { + // nothing + }); + } + + [$left, $right] = \end($sockets); + \fwrite($left, "."); + + $loop->onReadable($right, function (string $callbackId) use ($loop, &$invoked) { + $invoked = true; + $loop->stop(); + }); + + $loop->delay(1, function () use ($loop) { + $loop->stop(); + }); + }); + } finally { + foreach ($sockets as [$left, $right]) { + \fclose($left); + \fclose($right); + } + } + + self::assertTrue($invoked); + } + + public function testCancelAfterStreamIsClosed(): void + { + [$left, $right] = self::createSocketPair(); + + $callbackId = $this->loop->onReadable($left, static function () { + // nothing + }); + + $this->loop->defer(function () use ($callbackId, $left): void { + \fclose($left); + $this->loop->cancel($callbackId); + }); + + $this->loop->delay(0.1, function (): void { + $this->loop->stop(); + }); + + $this->loop->run(); + + \fclose($right); + + self::assertNotContains($callbackId, $this->loop->getIdentifiers()); + } + + public function testSupportedOnlyWhenNative(): void + { + // the polyfill runs the driver, but it is backed by stream_select() and slower than using it directly + self::assertSame(\PHP_VERSION_ID >= 80600, IoPollDriver::isSupported()); + } +} diff --git a/test/event_loop_destruction_order_io_poll.phpt b/test/event_loop_destruction_order_io_poll.phpt new file mode 100644 index 0000000..a0db8f7 --- /dev/null +++ b/test/event_loop_destruction_order_io_poll.phpt @@ -0,0 +1,56 @@ +--TEST-- +Issue #105: Ensure the callback fiber is always alive as long as the event loop lives (io_poll driver) +--SKIPIF-- + +--FILE-- +resume(...)); + $suspension->suspend(); + echo "Finished " . self::class, "\n"; + } +} + +EventLoop::defer(function () { + echo "start\n"; +}); + +a::getInstance(); + +EventLoop::run(); + +?> +--EXPECT-- +start +Destroying a +Finished a From c286c4306d42f98648ae062a4a8ee8db68799e9c Mon Sep 17 00:00:00 2001 From: Nicolas Grekas Date: Mon, 21 Sep 2026 22:17:59 +0200 Subject: [PATCH 2/2] Address review feedback getTimeout() returns null rather than -1 when no timer is pending, the cancel-after-close test moves to the shared suite since every driver passes it, the signal test now blocks the loop in wait() and interrupts it with pcntl_alarm() so it exercises ERROR_INTERRUPTED for real, Windows is detected in one place, and the polyfill is no longer suggested since it is slower than stream_select(). --- .github/workflows/ci.yml | 2 +- composer.json | 4 -- src/EventLoop/Driver/IoPollDriver.php | 27 +++++----- test/Driver/DriverTest.php | 31 ++++++++++- test/Driver/IoPollDriverTest.php | 74 +++++++++------------------ 5 files changed, 70 insertions(+), 68 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2060709..8048501 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -69,7 +69,7 @@ jobs: - name: Get Composer cache directory id: composer-cache - run: echo "::set-output name=dir::$(composer config cache-dir)" + run: echo "dir=$(composer config cache-dir)" >> "$GITHUB_OUTPUT" - name: Cache dependencies uses: actions/cache@v6 diff --git a/composer.json b/composer.json index ec65881..6176de5 100644 --- a/composer.json +++ b/composer.json @@ -38,10 +38,6 @@ "jetbrains/phpstorm-stubs": "^2019.3", "psalm/phar": "6.16.*" }, - "suggest": { - "symfony/polyfill-io-poll": "To use IoPollDriver on PHP < 8.6, together with symfony/polyfill-time", - "symfony/polyfill-time": "To use IoPollDriver on PHP < 8.6, together with symfony/polyfill-io-poll" - }, "autoload": { "psr-4": { "Revolt\\": "src" diff --git a/src/EventLoop/Driver/IoPollDriver.php b/src/EventLoop/Driver/IoPollDriver.php index 62944da..f466fa6 100644 --- a/src/EventLoop/Driver/IoPollDriver.php +++ b/src/EventLoop/Driver/IoPollDriver.php @@ -29,13 +29,14 @@ final class IoPollDriver extends AbstractDriver /** * Whether this driver is worth choosing over StreamSelectDriver. * - * The driver also runs on symfony/polyfill-io-poll below PHP 8.6, but that polyfill is backed by - * stream_select() itself, so it is slower than using stream_select() directly and is not picked - * automatically. Set REVOLT_DRIVER to this class to use it there anyway. + * The Io\Poll API ships with every PHP 8.6 build. The driver also runs on symfony/polyfill-io-poll + * below that, but that polyfill is backed by stream_select() itself, so it is slower than using + * stream_select() directly and is not picked automatically. Set REVOLT_DRIVER to this class to + * use it there anyway. */ public static function isSupported(): bool { - return \PHP_VERSION_ID >= 80600 && \class_exists(Context::class) && \class_exists(Duration::class); + return \PHP_VERSION_ID >= 80600; } private readonly Context $context; @@ -250,8 +251,7 @@ private function watch(int $streamId, mixed $stream): void if ($watcher !== null) { unset($this->watchers[$streamId], $this->alwaysReadyStreams[$streamId]); - // Closing a stream already takes its descriptor out of the context, and removing the - // watcher of a closed stream crashes PHP before php/php-src#23791 + // Removing the watcher of a closed stream crashes 8.6 builds before php/php-src#23791 if (\is_resource($stream)) { $watcher->remove(); } @@ -277,7 +277,7 @@ private function watch(int $streamId, mixed $stream): void try { $this->watchers[$streamId] = $this->alwaysReadyContext->add($handle, $events, $streamId); } catch (PollException) { - throw $exception; + throw $exception; // the refusal of the preferred backend is the one worth reporting } $this->alwaysReadyStreams[$streamId] = true; @@ -290,10 +290,10 @@ private function watch(int $streamId, mixed $stream): void } } - private function poll(float $timeout): void + private function poll(?float $timeout): void { if (!$this->watchers) { - if ($timeout < 0) { // Only signal callbacks are enabled, so sleep indefinitely. + if ($timeout === null) { // Only signal callbacks are enabled, so sleep indefinitely. /** @psalm-suppress ArgumentTypeCoercion */ \usleep(\PHP_INT_MAX); return; @@ -312,7 +312,7 @@ private function poll(float $timeout): void $timeout = 0.0; } - if ($timeout < 0) { + if ($timeout === null) { $duration = null; } else { $seconds = (int) $timeout; @@ -334,6 +334,7 @@ private function poll(float $timeout): void } foreach ($watchers as $watcher) { + /** @var int $streamId */ $streamId = $watcher->getData(); // Error and HangUp are reported whether they were requested or not. Both sides have to be @@ -355,14 +356,14 @@ private function poll(float $timeout): void } /** - * @return float Seconds until next timer expires or -1 if there are no pending timers. + * @return float|null Seconds until next timer expires or null if there are no pending timers. */ - private function getTimeout(): float + private function getTimeout(): ?float { $expiration = $this->timerQueue->peek(); if ($expiration === null) { - return -1; + return null; } $expiration -= $this->now(); diff --git a/test/Driver/DriverTest.php b/test/Driver/DriverTest.php index f8f8616..418d51d 100644 --- a/test/Driver/DriverTest.php +++ b/test/Driver/DriverTest.php @@ -26,10 +26,15 @@ abstract class DriverTest extends TestCase /** * @return array{resource, resource} */ + protected static function isWindows(): bool + { + return \DIRECTORY_SEPARATOR === "\\"; + } + protected static function createSocketPair(): array { $sockets = \stream_socket_pair( - \DIRECTORY_SEPARATOR === "\\" ? STREAM_PF_INET : STREAM_PF_UNIX, + self::isWindows() ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP, ); @@ -61,6 +66,30 @@ public function tearDown(): void unset($this->loop); } + public function testCancelAfterStreamIsClosed(): void + { + [$left, $right] = self::createSocketPair(); + + $callbackId = $this->loop->onReadable($left, static function () { + // nothing + }); + + $this->loop->defer(function () use ($callbackId, $left): void { + \fclose($left); + $this->loop->cancel($callbackId); + }); + + $this->loop->delay(0.1, function (): void { + $this->loop->stop(); + }); + + $this->loop->run(); + + \fclose($right); + + self::assertNotContains($callbackId, $this->loop->getIdentifiers()); + } + public function testCorrectTimeoutIfBlockingBeforeActivate(): void { $start = 0; diff --git a/test/Driver/IoPollDriverTest.php b/test/Driver/IoPollDriverTest.php index 233af30..900280c 100644 --- a/test/Driver/IoPollDriverTest.php +++ b/test/Driver/IoPollDriverTest.php @@ -19,6 +19,8 @@ public function getFactory(): callable public function setUp(): void { + // Not isSupported(): that is deliberately false below 8.6, while the driver still has to be + // exercised there through the polyfill. if (!\class_exists(Context::class) || !\class_exists(Duration::class)) { self::markTestSkipped("Skip, the Io\\Poll API requires PHP 8.6 or symfony/polyfill-io-poll"); } @@ -33,7 +35,7 @@ public function testHandle(): void public function testAsyncSignals(): void { - if (\DIRECTORY_SEPARATOR === '\\') { + if (self::isWindows()) { self::markTestSkipped('Skip on Windows'); } @@ -73,49 +75,47 @@ public function testAsyncSignals(): void } /** + * A signal arriving while the loop is blocked in wait() makes it throw ERROR_INTERRUPTED. The + * driver has to turn that into the signal callback running, not into an exception. + * * @requires extension pcntl */ - public function testSignalDuringPollIgnored(): void + public function testSignalInterruptingWaitIsDispatched(): void { - if (\DIRECTORY_SEPARATOR === '\\') { + if (self::isWindows()) { self::markTestSkipped('Skip on Windows'); } if (!\extension_loaded("pcntl") || !\function_exists('pcntl_signal_dispatch') || !\function_exists('pcntl_signal') + || !\function_exists('pcntl_alarm') ) { self::markTestSkipped('Skip, PCNTL functions not available'); } - $sockets = self::createSocketPair(); + [$left, $right] = self::createSocketPair(); + $invoked = false; - $this->start(function (Driver $loop) use ($sockets, &$signalCallbackId) { - $socketCallbackIds = [ - $loop->onReadable($sockets[0], function () { - // nothing - }), - $loop->onReadable($sockets[1], function () { - // nothing - }), - ]; + $this->start(function (Driver $loop) use ($left, &$invoked): void { + // keeps the loop blocked in wait() until the kernel delivers SIGALRM a second later + $readableId = $loop->onReadable($left, static function (): void { + // nothing + }); - $signalCallbackId = $loop->onSignal(\SIGUSR2, function ($callbackId) use ($socketCallbackIds, $loop) { + $loop->onSignal(\SIGALRM, function (string $callbackId) use ($loop, $readableId, &$invoked): void { + $invoked = true; $loop->cancel($callbackId); - - foreach ($socketCallbackIds as $socketCallbackId) { - $loop->cancel($socketCallbackId); - } - - $this->assertTrue(true); + $loop->cancel($readableId); }); - $loop->delay(0.1, function () { - \proc_open('sh -c "sleep 1; kill -USR2 ' . \getmypid() . '"', [], $pipes); - }); + \pcntl_alarm(1); }); - $this->loop->cancel($signalCallbackId); + \fclose($left); + \fclose($right); + + self::assertTrue($invoked); } /** @@ -123,7 +123,7 @@ public function testSignalDuringPollIgnored(): void */ public function testMoreFileDescriptorsThanFdSetSize(): void { - if (\stripos(PHP_OS, 'win') === 0) { + if (self::isWindows()) { self::markTestSkipped('Skip on Windows'); } @@ -173,30 +173,6 @@ public function testMoreFileDescriptorsThanFdSetSize(): void self::assertTrue($invoked); } - public function testCancelAfterStreamIsClosed(): void - { - [$left, $right] = self::createSocketPair(); - - $callbackId = $this->loop->onReadable($left, static function () { - // nothing - }); - - $this->loop->defer(function () use ($callbackId, $left): void { - \fclose($left); - $this->loop->cancel($callbackId); - }); - - $this->loop->delay(0.1, function (): void { - $this->loop->stop(); - }); - - $this->loop->run(); - - \fclose($right); - - self::assertNotContains($callbackId, $this->loop->getIdentifiers()); - } - public function testSupportedOnlyWhenNative(): void { // the polyfill runs the driver, but it is backed by stream_select() and slower than using it directly