diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f0c17e0..8048501 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 }} @@ -57,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 @@ -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/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..f466fa6 --- /dev/null +++ b/src/EventLoop/Driver/IoPollDriver.php @@ -0,0 +1,379 @@ += 80600; + } + + 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]); + + // Removing the watcher of a closed stream crashes 8.6 builds 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; // the refusal of the preferred backend is the one worth reporting + } + + $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 === null) { // 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 === null) { + $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) { + /** @var int $streamId */ + $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|null Seconds until next timer expires or null if there are no pending timers. + */ + private function getTimeout(): ?float + { + $expiration = $this->timerQueue->peek(); + + if ($expiration === null) { + return null; + } + + $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/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 new file mode 100644 index 0000000..900280c --- /dev/null +++ b/test/Driver/IoPollDriverTest.php @@ -0,0 +1,181 @@ +loop->getHandle()); + } + + public function testAsyncSignals(): void + { + if (self::isWindows()) { + 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); + } + + /** + * 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 testSignalInterruptingWaitIsDispatched(): void + { + 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'); + } + + [$left, $right] = self::createSocketPair(); + $invoked = false; + + $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 + }); + + $loop->onSignal(\SIGALRM, function (string $callbackId) use ($loop, $readableId, &$invoked): void { + $invoked = true; + $loop->cancel($callbackId); + $loop->cancel($readableId); + }); + + \pcntl_alarm(1); + }); + + \fclose($left); + \fclose($right); + + self::assertTrue($invoked); + } + + /** + * stream_select() is capped at FD_SETSIZE, which is 1024 on Linux. Poll backends are not. + */ + public function testMoreFileDescriptorsThanFdSetSize(): void + { + if (self::isWindows()) { + 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 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