UvDriver.php

<?php

declare(strict_types=1);

namespace Revolt\EventLoop\Driver;

use Revolt\EventLoop\FiberFactory;
use Revolt\EventLoop\Internal\AbstractDriver;
use Revolt\EventLoop\Internal\DriverCallback;
use Revolt\EventLoop\Internal\SignalCallback;
use Revolt\EventLoop\Internal\StreamCallback;
use Revolt\EventLoop\Internal\StreamReadableCallback;
use Revolt\EventLoop\Internal\StreamWritableCallback;
use Revolt\EventLoop\Internal\TimerCallback;

final class UvDriver extends AbstractDriver
{
    public static function isSupported(): bool
    {
        return \extension_loaded("uv");
    }

    /** @var resource|\UVLoop A uv_loop resource created with uv_loop_new() */
    private $handle;
    /** @var array<string, resource> */
    private array $events = [];
    /** @var array<int, array<array-key, DriverCallback>> */
    private array $uvCallbacks = [];
    /** @var array<int, resource> */
    private array $streams = [];
    private readonly \Closure $ioCallback;
    private readonly \Closure $timerCallback;
    private readonly \Closure $signalCallback;

    public function __construct(?FiberFactory $fiberFactory = null)
    {
        parent::__construct($fiberFactory);

        $this->handle = \uv_loop_new();

        $this->ioCallback = function ($event, $status, $events, $resource): void {
            $callbacks = $this->uvCallbacks[(int) $event];

            // Invoke the callback on errors, as this matches behavior with other loop back-ends.
            // Re-enable callback as libuv disables the callback on non-zero status.
            if ($status !== 0) {
                $flags = 0;
                foreach ($callbacks as $callback) {
                    \assert($callback instanceof StreamCallback);

                    $flags |= $callback->invokable ? $this->getStreamCallbackFlags($callback) : 0;
                }
                \uv_poll_start($event, $flags, $this->ioCallback);
            }

            foreach ($callbacks as $callback) {
                \assert($callback instanceof StreamCallback);

                // $events is ORed with 4 to trigger callback if no events are indicated (0) or on UV_DISCONNECT (4).
                // http://docs.libuv.org/en/v1.x/poll.html
                if (!($this->getStreamCallbackFlags($callback) & $events || ($events | 4) === 4)) {
                    continue;
                }

                $this->enqueueCallback($callback);
            }
        };

        $this->timerCallback = function ($event): void {
            $callback = $this->uvCallbacks[(int) $event][0];

            \assert($callback instanceof TimerCallback);

            $this->enqueueCallback($callback);
        };

        $this->signalCallback = function ($event): void {
            $callback = $this->uvCallbacks[(int) $event][0];

            $this->enqueueCallback($callback);
        };
    }

    /**
     * {@inheritdoc}
     */
    public function cancel(string $callbackId): void
    {
        parent::cancel($callbackId);

        if (!isset($this->events[$callbackId])) {
            return;
        }

        $event = $this->events[$callbackId];
        $eventId = (int) $event;

        if (isset($this->uvCallbacks[$eventId][0])) { // All except IO callbacks.
            unset($this->uvCallbacks[$eventId]);
        } elseif (isset($this->uvCallbacks[$eventId][$callbackId])) {
            $callback = $this->uvCallbacks[$eventId][$callbackId];
            unset($this->uvCallbacks[$eventId][$callbackId]);

            \assert($callback instanceof StreamCallback);

            if (empty($this->uvCallbacks[$eventId])) {
                unset($this->uvCallbacks[$eventId], $this->streams[(int) $callback->stream]);
            }
        }

        unset($this->events[$callbackId]);
    }

    /**
     * @return \UVLoop|resource
     */
    public function getHandle(): mixed
    {
        return $this->handle;
    }

    protected function now(): float
    {
        \uv_update_time($this->handle);

        /** @psalm-suppress TooManyArguments */
        return \uv_now($this->handle) / 1000;
    }

    /**
     * {@inheritdoc}
     */
    protected function dispatch(bool $blocking): void
    {
        /** @psalm-suppress TooManyArguments */
        \uv_run($this->handle, $blocking ? \UV::RUN_ONCE : \UV::RUN_NOWAIT);
    }

    /**
     * {@inheritdoc}
     */
    protected function activate(array $callbacks): void
    {
        $now = $this->now();

        foreach ($callbacks as $callback) {
            $id = $callback->id;

            if ($callback instanceof StreamCallback) {
                \assert(\is_resource($callback->stream));

                $streamId = (int) $callback->stream;

                if (isset($this->streams[$streamId])) {
                    $event = $this->streams[$streamId];
                } elseif (isset($this->events[$id])) {
                    $event = $this->streams[$streamId] = $this->events[$id];
                } else {
                    /** @psalm-suppress TooManyArguments */
                    $event = $this->streams[$streamId] = \uv_poll_init_socket($this->handle, $callback->stream);
                }

                $eventId = (int) $event;
                $this->events[$id] = $event;
                $this->uvCallbacks[$eventId][$id] = $callback;

                $flags = 0;
                foreach ($this->uvCallbacks[$eventId] as $w) {
                    \assert($w instanceof StreamCallback);

                    $flags |= $w->enabled ? ($this->getStreamCallbackFlags($w)) : 0;
                }
                \uv_poll_start($event, $flags, $this->ioCallback);
            } elseif ($callback instanceof TimerCallback) {
                if (isset($this->events[$id])) {
                    $event = $this->events[$id];
                } else {
                    $event = $this->events[$id] = \uv_timer_init($this->handle);
                }

                $this->uvCallbacks[(int) $event] = [$callback];

                \uv_timer_start(
                    $event,
                    (int) \min(\max(0, \ceil(($callback->expiration - $now) * 1000)), \PHP_INT_MAX),
                    $callback->repeat ? (int) \min(\max(0, \ceil($callback->interval * 1000)), \PHP_INT_MAX) : 0,
                    $this->timerCallback
                );
            } elseif ($callback instanceof SignalCallback) {
                if (isset($this->events[$id])) {
                    $event = $this->events[$id];
                } else {
                    /** @psalm-suppress TooManyArguments */
                    $event = $this->events[$id] = \uv_signal_init($this->handle);
                }

                $this->uvCallbacks[(int) $event] = [$callback];

                /** @psalm-suppress TooManyArguments */
                \uv_signal_start($event, $this->signalCallback, $callback->signal);
            } else {
                // @codeCoverageIgnoreStart
                throw new \Error("Unknown callback type");
                // @codeCoverageIgnoreEnd
            }
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function deactivate(DriverCallback $callback): void
    {
        $id = $callback->id;

        if (!isset($this->events[$id])) {
            return;
        }

        $event = $this->events[$id];

        if (!\uv_is_active($event)) {
            return;
        }

        if ($callback instanceof StreamCallback) {
            $flags = 0;
            foreach ($this->uvCallbacks[(int) $event] as $w) {
                \assert($w instanceof StreamCallback);

                $flags |= $w->invokable ? ($this->getStreamCallbackFlags($w)) : 0;
            }

            if ($flags) {
                \uv_poll_start($event, $flags, $this->ioCallback);
            } else {
                \uv_poll_stop($event);
            }
        } elseif ($callback instanceof TimerCallback) {
            \uv_timer_stop($event);
        } elseif ($callback instanceof SignalCallback) {
            \uv_signal_stop($event);
        } else {
            // @codeCoverageIgnoreStart
            throw new \Error("Unknown callback type");
            // @codeCoverageIgnoreEnd
        }
    }

    private function getStreamCallbackFlags(StreamCallback $callback): int
    {
        if ($callback instanceof StreamWritableCallback) {
            return \UV::WRITABLE;
        }

        if ($callback instanceof StreamReadableCallback) {
            return \UV::READABLE;
        }

        throw new \Error('Invalid callback type');
    }
}