1
0
mirror of https://github.com/danog/amp.git synced 2024-12-13 01:47:33 +01:00
amp/lib/Loop/UvDriver.php

321 lines
9.4 KiB
PHP
Raw Normal View History

2016-06-09 19:57:46 +02:00
<?php
namespace Amp\Loop;
use Amp\Promise;
2020-11-01 16:43:21 +01:00
final class UvDriver extends DriverFoundation
2018-06-18 20:00:01 +02:00
{
/** @var resource|\UVLoop A uv_loop resource created with uv_loop_new() */
2016-06-09 19:57:46 +02:00
private $handle;
2017-02-17 05:36:32 +01:00
/** @var resource[] */
private array $events = [];
2016-06-09 19:57:46 +02:00
/** @var Watcher[][] */
private array $watchers = [];
2016-06-09 19:57:46 +02:00
/** @var resource[] */
private array $streams = [];
private \Closure $ioCallback;
2016-06-09 19:57:46 +02:00
private \Closure $timerCallback;
2017-01-16 17:39:24 +01:00
private \Closure $signalCallback;
2016-06-09 19:57:46 +02:00
2018-06-18 20:00:01 +02:00
public function __construct()
{
2016-06-09 19:57:46 +02:00
$this->handle = \uv_loop_new();
2020-03-28 22:20:44 +01:00
/**
* @param $event
* @param $status
* @param $events
* @param $resource
*
* @return void
*/
$this->ioCallback = function ($event, $status, $events, $resource): void {
$watchers = $this->watchers[(int) $event];
2016-06-09 19:57:46 +02:00
switch ($status) {
case 0: // OK
break;
default: // Invoke the callback on errors, as this matches behavior with other loop back-ends.
// Re-enable watcher as libuv disables the watcher on non-zero status.
2017-05-13 17:11:00 +02:00
$flags = 0;
foreach ($watchers as $watcher) {
2017-05-13 17:11:00 +02:00
$flags |= $watcher->enabled ? $watcher->type : 0;
}
\uv_poll_start($event, $flags, $this->ioCallback);
break;
2016-06-09 19:57:46 +02:00
}
foreach ($watchers as $watcher) {
2018-04-08 20:09:38 +02:00
// $events is OR'ed with 4 to trigger watcher if no events are indicated (0) or on UV_DISCONNECT (4).
// http://docs.libuv.org/en/v1.x/poll.html
if (!($watcher->enabled && ($watcher->type & $events || ($events | 4) === 4))) {
continue;
}
try {
$result = ($watcher->callback)($watcher->id, $resource, $watcher->data);
if ($result instanceof Promise) {
Promise\rethrow($result);
}
} catch (\Throwable $exception) {
$this->error($exception);
}
2016-06-09 19:57:46 +02:00
}
};
2020-03-28 22:20:44 +01:00
/**
* @param $event
*
* @return void
*/
$this->timerCallback = function ($event): void {
$watcher = $this->watchers[(int) $event][0];
2016-06-09 19:57:46 +02:00
if ($watcher->type & Watcher::DELAY) {
unset($this->events[$watcher->id], $this->watchers[(int) $event]); // Avoid call to uv_is_active().
$this->cancel($watcher->id); // Remove reference to watcher in parent.
2017-05-23 20:17:34 +02:00
} elseif ($watcher->value === 0) {
// Disable and re-enable so it's not executed repeatedly in the same tick
// See https://github.com/amphp/amp/issues/131
$this->disable($watcher->id);
$this->enable($watcher->id);
}
try {
$result = ($watcher->callback)($watcher->id, $watcher->data);
if ($result instanceof Promise) {
Promise\rethrow($result);
}
} catch (\Throwable $exception) {
$this->error($exception);
}
2016-06-09 19:57:46 +02:00
};
2020-03-28 22:20:44 +01:00
/**
* @param $event
* @param $signo
*
* @return void
*/
$this->signalCallback = function ($event, $signo): void {
$watcher = $this->watchers[(int) $event][0];
2016-06-09 19:57:46 +02:00
try {
$result = ($watcher->callback)($watcher->id, $signo, $watcher->data);
if ($result instanceof Promise) {
Promise\rethrow($result);
}
} catch (\Throwable $exception) {
$this->error($exception);
}
2016-06-09 19:57:46 +02:00
};
}
/**
* {@inheritdoc}
*/
2020-09-24 18:52:22 +02:00
public function cancel(string $watcherId): void
2018-06-18 20:00:01 +02:00
{
parent::cancel($watcherId);
if (!isset($this->events[$watcherId])) {
return;
}
$event = $this->events[$watcherId];
$eventId = (int) $event;
if (isset($this->watchers[$eventId][0])) { // All except IO watchers.
unset($this->watchers[$eventId]);
} elseif (isset($this->watchers[$eventId][$watcherId])) {
$watcher = $this->watchers[$eventId][$watcherId];
unset($this->watchers[$eventId][$watcherId]);
if (empty($this->watchers[$eventId])) {
unset($this->watchers[$eventId], $this->streams[(int) $watcher->value]);
}
}
unset($this->events[$watcherId]);
}
2018-06-18 20:00:01 +02:00
public static function isSupported(): bool
{
return \extension_loaded("uv");
}
2018-01-06 03:32:57 +01:00
/**
* {@inheritdoc}
*/
public function now(): int
{
\uv_update_time($this->handle);
2020-03-28 20:27:42 +01:00
/** @psalm-suppress TooManyArguments */
2018-01-06 03:32:57 +01:00
return \uv_now($this->handle);
}
/**
* {@inheritdoc}
*/
2018-06-18 20:00:01 +02:00
public function getHandle()
{
return $this->handle;
}
2016-06-09 19:57:46 +02:00
/**
* {@inheritdoc}
2020-03-28 22:20:44 +01:00
*
* @return void
2016-06-09 19:57:46 +02:00
*/
2020-09-24 18:52:22 +02:00
protected function dispatch(bool $blocking): void
2018-06-18 20:00:01 +02:00
{
2020-03-28 20:27:42 +01:00
/** @psalm-suppress TooManyArguments */
2016-12-28 23:16:09 +01:00
\uv_run($this->handle, $blocking ? \UV::RUN_ONCE : \UV::RUN_NOWAIT);
2016-06-09 19:57:46 +02:00
}
/**
* {@inheritdoc}
2020-03-28 22:20:44 +01:00
*
* @return void
2016-06-09 19:57:46 +02:00
*/
2020-09-24 18:52:22 +02:00
protected function activate(array $watchers): void
2018-06-18 20:00:01 +02:00
{
$now = $this->now();
2016-06-09 19:57:46 +02:00
foreach ($watchers as $watcher) {
$id = $watcher->id;
switch ($watcher->type) {
case Watcher::READABLE:
case Watcher::WRITABLE:
2020-04-19 15:38:22 +02:00
\assert(\is_resource($watcher->value));
2016-06-09 19:57:46 +02:00
$streamId = (int) $watcher->value;
if (isset($this->streams[$streamId])) {
$event = $this->streams[$streamId];
2016-06-09 19:57:46 +02:00
} elseif (isset($this->events[$id])) {
$event = $this->streams[$streamId] = $this->events[$id];
2016-06-09 19:57:46 +02:00
} else {
2020-04-19 15:38:22 +02:00
/** @psalm-suppress UndefinedFunction */
$event = $this->streams[$streamId] = \uv_poll_init_socket($this->handle, $watcher->value);
2016-06-09 19:57:46 +02:00
}
$eventId = (int) $event;
2016-06-09 19:57:46 +02:00
$this->events[$id] = $event;
$this->watchers[$eventId][$id] = $watcher;
2016-06-09 19:57:46 +02:00
$flags = 0;
foreach ($this->watchers[$eventId] as $w) {
$flags |= $w->enabled ? $w->type : 0;
2016-06-09 19:57:46 +02:00
}
\uv_poll_start($event, $flags, $this->ioCallback);
2016-06-09 19:57:46 +02:00
break;
case Watcher::DELAY:
case Watcher::REPEAT:
2020-04-19 15:38:22 +02:00
\assert(\is_int($watcher->value));
2016-06-09 19:57:46 +02:00
if (isset($this->events[$id])) {
$event = $this->events[$id];
} else {
$event = $this->events[$id] = \uv_timer_init($this->handle);
}
$this->watchers[(int) $event] = [$watcher];
2016-06-09 19:57:46 +02:00
\uv_timer_start(
$event,
\max(0, $watcher->expiration - $now),
($watcher->type & Watcher::REPEAT) ? $watcher->value : 0,
2016-06-09 19:57:46 +02:00
$this->timerCallback
);
break;
case Watcher::SIGNAL:
2020-04-19 15:38:22 +02:00
\assert(\is_int($watcher->value));
2016-06-09 19:57:46 +02:00
if (isset($this->events[$id])) {
$event = $this->events[$id];
} else {
2020-04-19 15:38:22 +02:00
/** @psalm-suppress UndefinedFunction */
2016-06-09 19:57:46 +02:00
$event = $this->events[$id] = \uv_signal_init($this->handle);
}
2017-01-16 17:39:24 +01:00
$this->watchers[(int) $event] = [$watcher];
2016-06-09 19:57:46 +02:00
2020-04-19 15:38:22 +02:00
/** @psalm-suppress UndefinedFunction */
2016-06-09 19:57:46 +02:00
\uv_signal_start($event, $this->signalCallback, $watcher->value);
break;
default:
// @codeCoverageIgnoreStart
throw new \Error("Unknown watcher type");
2018-06-18 20:00:01 +02:00
// @codeCoverageIgnoreEnd
2016-06-09 19:57:46 +02:00
}
}
}
/**
* {@inheritdoc}
2020-03-28 22:20:44 +01:00
*
* @return void
2016-06-09 19:57:46 +02:00
*/
2020-09-24 18:52:22 +02:00
protected function deactivate(Watcher $watcher): void
2018-06-18 20:00:01 +02:00
{
2016-06-09 19:57:46 +02:00
$id = $watcher->id;
if (!isset($this->events[$id])) {
return;
}
$event = $this->events[$id];
if (!\uv_is_active($event)) {
return;
}
2016-06-09 19:57:46 +02:00
switch ($watcher->type) {
case Watcher::READABLE:
case Watcher::WRITABLE:
$flags = 0;
foreach ($this->watchers[(int) $event] as $w) {
$flags |= $w->enabled ? $w->type : 0;
}
2017-04-23 14:39:19 +02:00
if ($flags) {
\uv_poll_start($event, $flags, $this->ioCallback);
} else {
\uv_poll_stop($event);
2016-06-09 19:57:46 +02:00
}
break;
case Watcher::DELAY:
case Watcher::REPEAT:
\uv_timer_stop($event);
2016-06-09 19:57:46 +02:00
break;
case Watcher::SIGNAL:
\uv_signal_stop($event);
2016-06-09 19:57:46 +02:00
break;
default:
// @codeCoverageIgnoreStart
throw new \Error("Unknown watcher type");
2018-06-18 20:00:01 +02:00
// @codeCoverageIgnoreEnd
2016-06-09 19:57:46 +02:00
}
}
}