Fix IoPollDriver spinning on and missing EOF of files and unpollable streams

.github/workflows/ci.yml

@@ -6,7 +6,7 @@ on:
 
 jobs:
   tests:
-    timeout-minutes: 5
+    timeout-minutes: 10
     strategy:
       fail-fast: false
       matrix:
@@ -44,9 +44,28 @@ jobs:
             io-poll-polyfill: 'yes'
             psalm: 'skip'
 
+          # kqueue and WSAPoll back IoPollDriver there, instead of epoll
+          - operating-system: 'macos-latest'
+            php-version: 8.6
+            composer-flags: '--ignore-platform-req=php+'
+            job-description: 'on macOS'
+            extensions: ':xdebug'
+            psalm: 'skip'
+
+          - operating-system: 'windows-latest'
+            php-version: 8.6
+            composer-flags: '--ignore-platform-req=php+'
+            job-description: 'on Windows'
+            extensions: ':xdebug'
+            psalm: 'skip'
+
     name: PHP ${{ matrix.php-version }} ${{ matrix.job-description }}
     runs-on: ${{ matrix.operating-system }}
 
+    defaults:
+      run:
+        shell: bash
+
     steps:
       - name: Set git to use LF
         run: |
@@ -60,13 +79,14 @@ jobs:
         run: |
           sudo apt-get update
           sudo apt-get install libuv1-dev libevent-dev
+        if: runner.os == 'Linux'
 
       - name: Setup PHP
         uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php-version }}
-          extensions: uv-amphp/ext-uv@0.3.x, event, :xdebug
-          tools: pie
+          extensions: ${{ matrix.extensions || 'uv-amphp/ext-uv@0.3.x, event, :xdebug' }}
+          tools: ${{ runner.os == 'Linux' && 'pie' || '' }}
         env:
           # Workaround for https://github.com/shivammathur/setup-php/issues/1146 (cd "" fails on bash 5.3)
           UV_PATH: .
@@ -78,6 +98,7 @@ jobs:
           # PIE refuses to call sudo itself in a non-interactive terminal, but the runner has passwordless sudo
           sudo --preserve-env=GITHUB_TOKEN pie install osmanov/pecl-ev
           php --ri ev
+        if: runner.os == 'Linux'
 
       - name: Get Composer cache directory
         id: composer-cache
@@ -125,3 +146,4 @@ jobs:
         run: |
           composer install --optimize-autoloader --no-interaction --no-progress --working-dir=tools/php-cs-fixer
           tools/php-cs-fixer/vendor/bin/php-cs-fixer --diff --dry-run -v fix
+        if: runner.os == 'Linux'

src/EventLoop/Driver/IoPollDriver.php

@@ -12,7 +12,6 @@
 use Io\Poll\FailedPollWaitException;
 use Io\Poll\HandleAlreadyWatchedException;
 use Io\Poll\InvalidHandleException;
-use Io\Poll\PollException;
 use Io\Poll\Watcher;
 use Revolt\EventLoop\Internal\AbstractDriver;
 use Revolt\EventLoop\Internal\DriverCallback;
@@ -26,17 +25,30 @@
 
 final class IoPollDriver extends AbstractDriver
 {
+    /**
+     * Codes of a FailedHandleAddException for handles the backend cannot poll: epoll reports EPERM
+     * as ERROR_NOSUPPORT, kqueue reports EINVAL as ERROR_INVALID.
+     */
+    private const UNPOLLABLE_HANDLE_ERRORS = [
+        FailedPollOperationException::ERROR_NOSUPPORT,
+        FailedPollOperationException::ERROR_INVALID,
+    ];
+
     /**
      * Whether this driver is worth choosing over StreamSelectDriver.
      *
      * 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.
+     *
+     * On Windows, the only backend is WSAPoll, which accepts pipes such as STDOUT, but never reports
+     * them, so callbacks on them never run. stream_select() handles them, so the driver is not
+     * supported there for now.
      */
     public static function isSupported(): bool
     {
-        return \PHP_VERSION_ID >= 80600;
+        return \PHP_VERSION_ID >= 80600 && \PHP_OS_FAMILY !== 'Windows';
     }
 
     private readonly Context $context;
@@ -45,13 +57,13 @@ public static function isSupported(): bool
     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.
+     * Regular files and the streams the backend refuses to poll, keyed by stream id. epoll and kqueue
+     * only refuse handles that never block, such as /dev/null, so they are always ready, the way
+     * stream_select() reports them. Watching them makes sense for STDIN, which may be redirected
+     * from a file.
+     *
+     * @var array<int, true>
      */
-    private ?Context $alwaysReadyContext = null;
-
-    /** @var array<int, true> Streams watched in the context above, keyed by stream id. */
     private array $alwaysReadyStreams = [];
 
     /** @var array<int, array<string, StreamReadableCallback>> */
@@ -248,8 +260,10 @@ private function watch(int $streamId, mixed $stream): void
         $watcher = $this->watchers[$streamId] ?? null;
 
         if (!$events || !\is_resource($stream)) {
+            unset($this->alwaysReadyStreams[$streamId]);
+
             if ($watcher !== null) {
-                unset($this->watchers[$streamId], $this->alwaysReadyStreams[$streamId]);
+                unset($this->watchers[$streamId]);
 
                 // Removing the watcher of a closed stream crashes 8.6 builds before php/php-src#23791
                 if (\is_resource($stream)) {
@@ -260,24 +274,36 @@ private function watch(int $streamId, mixed $stream): void
             return;
         }
 
+        if (isset($this->alwaysReadyStreams[$streamId])) {
+            return; // ready for whichever events are enabled, there is nothing to update
+        }
+
         try {
             if ($watcher !== null) {
                 $watcher->modifyEvents($events);
 
                 return;
             }
 
+            // kqueue accepts regular files, but stops reporting them readable at EOF, so a reader would
+            // never see the end. Other backends refuse them or always report them ready. On macOS and BSD,
+            // PHP never flags pipes or devices as seekable, so there the flag means regular files and
+            // directories. PHP determines it when opening the stream.
+            if ($this->context->getBackend() === Backend::Kqueue && \stream_get_meta_data($stream)['seekable']) {
+                $this->alwaysReadyStreams[$streamId] = true;
+
+                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; // the refusal of the preferred backend is the one worth reporting
+                // Any other failure, such as running out of epoll watches, says nothing about the
+                // stream being ready, so treating it as always ready would turn into a busy loop.
+                if (!\in_array($exception->getCode(), self::UNPOLLABLE_HANDLE_ERRORS, true)) {
+                    throw $exception;
                 }
 
                 $this->alwaysReadyStreams[$streamId] = true;
@@ -292,7 +318,7 @@ private function watch(int $streamId, mixed $stream): void
 
     private function poll(?float $timeout): void
     {
-        if (!$this->watchers) {
+        if (!$this->watchers && !$this->alwaysReadyStreams) {
             if ($timeout === null) { // Only signal callbacks are enabled, so sleep indefinitely.
                 /** @psalm-suppress ArgumentTypeCoercion */
                 \usleep(\PHP_INT_MAX);
@@ -308,8 +334,7 @@ private function poll(?float $timeout): void
         }
 
         if ($this->alwaysReadyStreams) {
-            // Those streams are ready by definition, so there is nothing to wait for.
-            $timeout = 0.0;
+            $timeout = 0.0; // their callbacks run in this tick, so there is nothing to wait for
         }
 
         if ($timeout === null) {
@@ -329,8 +354,14 @@ private function poll(?float $timeout): void
             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 ($this->alwaysReadyStreams as $streamId => $_) {
+            foreach ($this->readCallbacks[$streamId] ?? [] as $callback) {
+                $this->enqueueCallback($callback);
+            }
+
+            foreach ($this->writeCallbacks[$streamId] ?? [] as $callback) {
+                $this->enqueueCallback($callback);
+            }
         }
 
         foreach ($watchers as $watcher) {

test/Driver/DriverTest.php

@@ -90,6 +90,76 @@ public function testCancelAfterStreamIsClosed(): void
         self::assertNotContains($callbackId, $this->loop->getIdentifiers());
     }
 
+    /**
+     * epoll and kqueue refuse devices such as /dev/null and /dev/urandom, so a driver needs a fallback
+     * to watch them at all. That fallback must not turn into a busy loop while there is nothing to do.
+     *
+     * The test measures the CPU time of the process, because counting callback invocations cannot
+     * tell a busy loop apart: a loop can spin without invoking the callback at all, while a callback
+     * left enabled on a stream that is always ready is rightly invoked on every tick. The loop itself
+     * has no hook to count its ticks.
+     */
+    public function testRefusedStreamDoesNotSpin(): void
+    {
+        if (self::isWindows()) {
+            self::markTestSkipped('Skip on Windows');
+        }
+
+        // PHP flags /dev/null as seekable on Linux, while /dev/urandom isn't, and epoll refuses both
+        $stream = \fopen(\PHP_OS_FAMILY === 'Linux' ? '/dev/urandom' : '/dev/null', 'r');
+        $before = \getrusage();
+
+        try {
+            $this->start(function (Driver $loop) use ($stream): void {
+                // runs once where the stream is reported readable, never where it is not
+                $readableId = $loop->onReadable($stream, static function (string $callbackId) use ($loop): void {
+                    $loop->cancel($callbackId);
+                });
+
+                $loop->delay(0.5, function () use ($loop, $readableId): void {
+                    $loop->cancel($readableId);
+                });
+            });
+        } finally {
+            \fclose($stream);
+        }
+
+        $after = \getrusage();
+
+        $cpu = static fn (array $usage): float => $usage['ru_utime.tv_sec'] + $usage['ru_utime.tv_usec'] / 1e6
+            + $usage['ru_stime.tv_sec'] + $usage['ru_stime.tv_usec'] / 1e6;
+
+        // Idle waiting for half a second costs about a millisecond of CPU, spinning costs a lot more.
+        self::assertLessThan(0.05, $cpu($after) - $cpu($before));
+    }
+
+    /**
+     * stream_select() reports regular files as readable, at EOF too, so a reader sees the end.
+     */
+    public function testEmptyRegularFileIsReadable(): void
+    {
+        $file = \tmpfile();
+        $invoked = false;
+
+        $this->loop->onReadable($file, function (string $callbackId) use (&$invoked): void {
+            $invoked = true;
+            $this->loop->cancel($callbackId);
+        });
+
+        $guardId = $this->loop->delay(0.5, function (): void {
+            $this->loop->stop();
+        });
+
+        try {
+            $this->loop->run();
+        } finally {
+            $this->loop->cancel($guardId);
+            \fclose($file);
+        }
+
+        self::assertTrue($invoked);
+    }
+
     public function testCorrectTimeoutIfBlockingBeforeActivate(): void
     {
         $start = 0;
@@ -110,8 +180,11 @@ public function testCorrectTimeoutIfBlockingBeforeActivate(): void
         self::assertNotSame(0, $start);
         self::assertNotSame(0, $invoked);
 
-        self::assertGreaterThanOrEqual(0.9995, $invoked - $start);
-        self::assertLessThan(1.1, $invoked - $start);
+        // The tolerance is for timers firing slightly early, see the test below.
+        self::assertGreaterThanOrEqual(0.995, $invoked - $start);
+        // Counting from the activation after blocking would take 1.5 seconds. The margin to that
+        // leaves room for runners that oversleep, which CI runners on macOS do.
+        self::assertLessThan(1.25, $invoked - $start);
     }
 
     public function testCorrectTimeoutIfBlockingBeforeDelay(): void
@@ -132,8 +205,12 @@ public function testCorrectTimeoutIfBlockingBeforeDelay(): void
         self::assertNotSame(0, $start);
         self::assertNotSame(0, $invoked);
 
-        self::assertGreaterThanOrEqual(1.5, $invoked - $start);
-        self::assertLessThan(1.6, $invoked - $start);
+        // Counting from the tick start before blocking would take only 1 second. The tolerance is for
+        // timers firing slightly early: microtime() here and the clocks of the drivers drift apart, as
+        // they do on Windows, and libevent rounds timeouts to milliseconds.
+        self::assertGreaterThanOrEqual(1.495, $invoked - $start);
+        // Generous, as CI runners on macOS oversleep by more than 0.1 seconds.
+        self::assertLessThan(1.75, $invoked - $start);
     }
 
     public function testLoopTerminatesWithOnlyUnreferencedCallbacks(): void

test/Driver/EventDriverTest.php

@@ -25,4 +25,14 @@ public function testSupported(): void
     {
         self::assertTrue(EventDriver::isSupported());
     }
+
+    public function testRefusedStreamDoesNotSpin(): void
+    {
+        self::markTestSkipped('libevent refuses to watch devices such as /dev/null, epoll rejects them with EPERM');
+    }
+
+    public function testEmptyRegularFileIsReadable(): void
+    {
+        self::markTestSkipped('libevent refuses to watch regular files, epoll rejects them with EPERM');
+    }
 }

test/Driver/IoPollDriverTest.php

@@ -4,7 +4,9 @@
 
 namespace Revolt\EventLoop\Driver;
 
+use Io\Poll\Backend;
 use Io\Poll\Context;
+use PHPUnit\Framework\ExpectationFailedException;
 use Revolt\EventLoop\Driver;
 use Time\Duration;
 
@@ -25,9 +27,33 @@ public function setUp(): void
             self::markTestSkipped("Skip, the Io\\Poll API requires PHP 8.6 or symfony/polyfill-io-poll");
         }
 
+        if (self::isWindows()) {
+            self::markTestSkipped("Skip on Windows, WSAPoll never reports pipes such as STDOUT, so tests hang");
+        }
+
         parent::setUp();
     }
 
+    /**
+     * @dataProvider provideRegistrationArgs
+     * @group memoryleak
+     */
+    public function testNoMemoryLeak(string $type, array $args): void
+    {
+        try {
+            parent::testNoMemoryLeak($type, $args);
+        } catch (ExpectationFailedException $exception) {
+            // the kqueue backend leaks on every Watcher::modifyEvents(), see https://github.com/php/php-src/issues/25086
+            if (\in_array($type, ['onReadable', 'onWritable'], true)
+                && $this->loop->getHandle()->getBackend() === Backend::Kqueue
+            ) {
+                self::markTestSkipped('Io\\Poll leaks memory with kqueue, see php/php-src#25086: ' . $exception->getMessage());
+            }
+
+            throw $exception;
+        }
+    }
+
     public function testHandle(): void
     {
         self::assertInstanceOf(Context::class, $this->loop->getHandle());
@@ -173,9 +199,9 @@ public function testMoreFileDescriptorsThanFdSetSize(): void
         self::assertTrue($invoked);
     }
 
-    public function testSupportedOnlyWhenNative(): void
+    public function testSupportedOnlyWhenNativeAndNotOnWindows(): 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());
+        self::assertSame(\PHP_VERSION_ID >= 80600 && !self::isWindows(), IoPollDriver::isSupported());
     }
 }