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