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

530 lines
17 KiB
PHP
Raw Normal View History

2015-04-30 19:41:14 +02:00
<?php
namespace Amp;
2015-07-29 07:15:43 +02:00
class LibeventReactor implements Reactor {
2015-07-30 05:23:53 +02:00
use Struct;
2015-04-30 19:41:14 +02:00
private $base;
private $keepAliveBase;
2015-04-30 19:41:14 +02:00
private $watchers = [];
private $immediates = [];
private $keepAliveCount = 0;
private $state = self::STOPPED;
2015-04-30 19:41:14 +02:00
private $stopException;
private $onError;
private $onCoroutineResolution;
/* Pre-PHP7 closure GC hack vars */
private $gcEvent;
private $garbage = [];
private $isGcScheduled = false;
public function __construct() {
2015-07-22 17:38:17 +02:00
// @codeCoverageIgnoreStart
if (!\extension_loaded("libevent")) {
2015-07-21 04:40:06 +02:00
throw new \RuntimeException(
2015-07-22 17:38:17 +02:00
"The pecl libevent extension is required to use " . __CLASS__
2015-07-21 04:40:06 +02:00
);
2015-04-30 19:41:14 +02:00
}
2015-07-22 17:38:17 +02:00
// @codeCoverageIgnoreEnd
2015-04-30 19:41:14 +02:00
2015-07-29 07:15:43 +02:00
$this->base = \event_base_new();
$this->keepAliveBase = \event_base_new();
2015-04-30 19:41:14 +02:00
/**
* Prior to PHP7 we can't cancel closure watchers inside their own callbacks
* because PHP will fatal. In legacy versions we schedule manual GC workarounds.
*
* @link https://bugs.php.net/bug.php?id=62452
*/
if (PHP_MAJOR_VERSION < 7) {
$this->gcEvent = event_new();
2015-07-29 07:15:43 +02:00
\event_timer_set($this->gcEvent, function() {
2015-04-30 19:41:14 +02:00
$this->garbage = [];
$this->isGcScheduled = false;
2015-07-29 07:15:43 +02:00
\event_del($this->gcEvent);
2015-04-30 19:41:14 +02:00
});
\event_base_set($this->gcEvent, $this->keepAliveBase);
2015-04-30 19:41:14 +02:00
}
$this->onCoroutineResolution = function($e = null, $r = null) {
if ($e) {
$this->onCallbackError($e);
}
};
}
/**
* {@inheritDoc}
*/
public function run(callable $onStart = null) {
if ($this->state !== self::STOPPED) {
throw new \LogicException(
"Cannot run() recursively; event reactor already active"
);
2015-04-30 19:41:14 +02:00
}
if ($onStart) {
$this->state = self::STARTING;
$onStartWatcherId = $this->immediately($onStart);
$this->tryImmediate($this->watchers[$onStartWatcherId]);
if (empty($this->keepAliveCount) && empty($this->stopException)) {
$this->state = self::STOPPED;
}
} else {
$this->state = self::RUNNING;
2015-04-30 19:41:14 +02:00
}
while ($this->state > self::STOPPED) {
$immediates = $this->immediates;
foreach ($immediates as $watcher) {
if (!$this->tryImmediate($watcher)) {
break;
}
2015-04-30 19:41:14 +02:00
}
if (empty($this->keepAliveCount) || $this->state <= self::STOPPED) {
2015-04-30 19:41:14 +02:00
break;
}
$flags = \EVLOOP_ONCE | \EVLOOP_NONBLOCK;
\event_base_loop($this->base, $flags);
$flags = empty($this->immediates) ? \EVLOOP_ONCE : (\EVLOOP_ONCE | \EVLOOP_NONBLOCK);
\event_base_loop($this->keepAliveBase, $flags);
2015-04-30 19:41:14 +02:00
}
if ($this->watchers) {
foreach (array_keys($this->watchers) as $watcherId) {
$this->cancel($watcherId);
}
}
$this->state = self::STOPPED;
2015-04-30 19:41:14 +02:00
if ($this->stopException) {
$e = $this->stopException;
$this->stopException = null;
throw $e;
}
}
private function tryImmediate($watcher) {
try {
$this->keepAliveCount -= $watcher->keepAlive;
unset(
$this->watchers[$watcher->id],
$this->immediates[$watcher->id]
);
$out = \call_user_func($watcher->callback, $watcher->id, $watcher->cbData);
if ($out instanceof \Generator) {
resolve($out)->when($this->onCoroutineResolution);
2015-04-30 19:41:14 +02:00
}
} catch (\Throwable $e) {
// @TODO Remove coverage ignore block once PHP5 support is no longer required
// @codeCoverageIgnoreStart
$this->onCallbackError($e);
// @codeCoverageIgnoreEnd
} catch (\Exception $e) {
// @TODO Remove this catch block once PHP5 support is no longer required
$this->onCallbackError($e);
2015-04-30 19:41:14 +02:00
}
return $this->state;
2015-04-30 19:41:14 +02:00
}
/**
* {@inheritDoc}
*/
public function tick($noWait = false) {
if ($this->state) {
throw new \LogicException(
"Cannot tick() recursively; event reactor already active"
);
}
$this->state = self::TICKING;
$immediates = $this->immediates;
foreach ($immediates as $watcher) {
if (!$this->tryImmediate($watcher)) {
break;
}
}
// Check the conditional again because a manual stop() could've changed the state
if ($this->state) {
$flags = \EVLOOP_ONCE | \EVLOOP_NONBLOCK;
\event_base_loop($this->base, \EVLOOP_ONCE | \EVLOOP_NONBLOCK);
2015-04-30 19:41:14 +02:00
$flags = $noWait || !empty($this->immediates) ? (EVLOOP_ONCE | EVLOOP_NONBLOCK) : EVLOOP_ONCE;
\event_base_loop($this->keepAliveBase, $flags);
2015-04-30 19:41:14 +02:00
}
$this->state = self::STOPPED;
2015-04-30 19:41:14 +02:00
if ($this->stopException) {
$e = $this->stopException;
$this->stopException = null;
throw $e;
}
}
/**
* {@inheritDoc}
*/
public function stop() {
if ($this->state !== self::STOPPED) {
\event_base_loopexit($this->base);
\event_base_loopexit($this->keepAliveBase);
$this->state = self::STOPPING;
} else {
throw new \LogicException(
"Cannot stop(); event reactor not currently active"
);
}
2015-04-30 19:41:14 +02:00
}
/**
* {@inheritDoc}
*/
public function immediately(callable $callback, array $options = []) {
$watcher = $this->initWatcher(Watcher::IMMEDIATE, $callback, $options);
if ($watcher->isEnabled) {
$this->immediates[$watcher->id] = $watcher;
}
return $watcher->id;
}
private function initWatcher($type, $callback, $options) {
2015-07-30 04:10:24 +02:00
$watcher = new \StdClass;
$watcher->id = \spl_object_hash($watcher);
2015-04-30 19:41:14 +02:00
$watcher->type = $type;
$watcher->callback = $callback;
$watcher->cbData = isset($options["cb_data"]) ? $options["cb_data"] : null;
2015-04-30 19:41:14 +02:00
$watcher->isEnabled = isset($options["enable"]) ? (bool) $options["enable"] : true;
$watcher->keepAlive = isset($options["keep_alive"]) ? (bool) $options["keep_alive"] : true;
$this->keepAliveCount += ($watcher->isEnabled && $watcher->keepAlive);
2015-04-30 19:41:14 +02:00
if ($type !== Watcher::IMMEDIATE) {
$watcher->eventResource = \event_new();
$watcher->eventBase = $watcher->keepAlive ? $this->keepAliveBase : $this->base;
2015-04-30 19:41:14 +02:00
}
$this->watchers[$watcher->id] = $watcher;
2015-07-30 04:10:24 +02:00
2015-04-30 19:41:14 +02:00
return $watcher;
}
/**
* {@inheritDoc}
*/
public function once(callable $callback, $msDelay, array $options = []) {
assert(($msDelay >= 0), "\$msDelay at Argument 2 expects integer >= 0");
$watcher = $this->initWatcher(Watcher::TIMER_ONCE, $callback, $options);
$watcher->msDelay = ($msDelay * 1000);
\event_timer_set($watcher->eventResource, $this->wrapCallback($watcher));
\event_base_set($watcher->eventResource, $watcher->eventBase);
2015-04-30 19:41:14 +02:00
if ($watcher->isEnabled) {
\event_add($watcher->eventResource, $watcher->msDelay);
2015-04-30 19:41:14 +02:00
}
return $watcher->id;
}
private function wrapCallback($watcher) {
return function() use ($watcher) {
try {
switch ($watcher->type) {
case Watcher::IO_READER:
// fallthrough
case Watcher::IO_WRITER:
$result = \call_user_func($watcher->callback, $watcher->id, $watcher->stream, $watcher->cbData);
2015-04-30 19:41:14 +02:00
break;
case Watcher::TIMER_ONCE:
$result = \call_user_func($watcher->callback, $watcher->id, $watcher->cbData);
2015-04-30 19:41:14 +02:00
$this->cancel($watcher->id);
break;
case Watcher::TIMER_REPEAT:
$result = \call_user_func($watcher->callback, $watcher->id, $watcher->cbData);
2015-04-30 19:41:14 +02:00
// If the watcher cancelled itself this will no longer exist
if (isset($this->watchers[$watcher->id])) {
event_add($watcher->eventResource, $watcher->msInterval);
}
break;
case Watcher::SIGNAL:
$result = \call_user_func($watcher->callback, $watcher->id, $watcher->signo, $watcher->cbData);
2015-04-30 19:41:14 +02:00
break;
}
if ($result instanceof \Generator) {
2015-07-29 07:15:43 +02:00
resolve($result)->when($this->onCoroutineResolution);
2015-04-30 19:41:14 +02:00
}
2015-07-22 17:38:17 +02:00
} catch (\Throwable $e) {
// @TODO Remove coverage ignore block once PHP5 support is no longer required
// @codeCoverageIgnoreStart
$this->onCallbackError($e);
// @codeCoverageIgnoreEnd
2015-04-30 19:41:14 +02:00
} catch (\Exception $e) {
2015-07-22 17:38:17 +02:00
// @TODO Remove this catch block once PHP5 support is no longer required
2015-04-30 19:41:14 +02:00
$this->onCallbackError($e);
}
};
}
2015-07-22 17:38:17 +02:00
/**
*@TODO Add a \Throwable typehint once PHP5 is no longer required
*/
private function onCallbackError($e) {
2015-04-30 19:41:14 +02:00
if (empty($this->onError)) {
$this->stopException = $e;
$this->stop();
} else {
$this->tryUserErrorCallback($e);
}
}
2015-07-22 17:38:17 +02:00
/**
*@TODO Add a \Throwable typehint once PHP5 is no longer required
*/
private function tryUserErrorCallback($e) {
2015-04-30 19:41:14 +02:00
try {
2015-07-22 17:38:17 +02:00
\call_user_func($this->onError, $e);
} catch (\Throwable $e) {
// @TODO Remove coverage ignore block once PHP5 support is no longer required
// @codeCoverageIgnoreStart
$this->stopException = $e;
$this->stop();
// @codeCoverageIgnoreEnd
2015-04-30 19:41:14 +02:00
} catch (\Exception $e) {
2015-07-22 17:38:17 +02:00
// @TODO Remove this catch block once PHP5 support is no longer required
2015-04-30 19:41:14 +02:00
$this->stopException = $e;
$this->stop();
}
}
/**
* {@inheritDoc}
*/
public function repeat(callable $callback, $msInterval, array $options = []) {
assert(($msInterval >= 0), "\$msInterval at Argument 2 expects integer >= 0");
if (isset($options["ms_delay"])) {
$msDelay = (int) $options["ms_delay"];
assert(($msDelay >= 0), "ms_delay option expects integer >= 0");
$msDelay = ($msDelay * 1000);
2015-04-30 19:41:14 +02:00
} else {
$msDelay = $msInterval;
}
$watcher = $this->initWatcher(Watcher::TIMER_REPEAT, $callback, $options);
$watcher->msInterval = (int) $msInterval;
\event_timer_set($watcher->eventResource, $this->wrapCallback($watcher));
\event_base_set($watcher->eventResource, $watcher->eventBase);
2015-04-30 19:41:14 +02:00
if ($watcher->isEnabled) {
\event_add($watcher->eventResource, $msDelay);
2015-04-30 19:41:14 +02:00
}
return $watcher->id;
}
/**
* {@inheritDoc}
*/
public function onReadable($stream, callable $callback, array $options = []) {
$watcher = $this->initWatcher(Watcher::IO_READER, $callback, $options);
2015-07-19 19:53:26 +02:00
$watcher->stream = $stream;
$evFlags = \EV_PERSIST | \EV_READ;
\event_set($watcher->eventResource, $stream, $evFlags, $this->wrapCallback($watcher));
\event_base_set($watcher->eventResource, $watcher->eventBase);
2015-04-30 19:41:14 +02:00
if ($watcher->isEnabled) {
\event_add($watcher->eventResource);
2015-04-30 19:41:14 +02:00
}
return $watcher->id;
}
/**
* {@inheritDoc}
*/
public function onWritable($stream, callable $callback, array $options = []) {
$watcher = $this->initWatcher(Watcher::IO_WRITER, $callback, $options);
2015-07-19 19:53:26 +02:00
$watcher->stream = $stream;
$evFlags = \EV_PERSIST | \EV_WRITE;
\event_set($watcher->eventResource, $stream, $evFlags, $this->wrapCallback($watcher));
\event_base_set($watcher->eventResource, $watcher->eventBase);
2015-04-30 19:41:14 +02:00
if ($watcher->isEnabled) {
\event_add($watcher->eventResource);
2015-04-30 19:41:14 +02:00
}
return $watcher->id;
}
/**
* {@inheritDoc}
*/
public function onSignal($signo, callable $callback, array $options = []) {
$watcher = $this->initWatcher(Watcher::SIGNAL, $callback, $options);
$watcher->signo = $signo = (int) $signo;
$evFlags = \EV_SIGNAL | \EV_PERSIST;
\event_set($watcher->eventResource, $signo, $evFlags, $this->wrapCallback($watcher));
\event_base_set($watcher->eventResource, $watcher->eventBase);
2015-04-30 19:41:14 +02:00
if ($watcher->isEnabled) {
\event_add($watcher->eventResource);
2015-04-30 19:41:14 +02:00
}
return $watcher->id;
}
/**
* {@inheritDoc}
*/
public function cancel($watcherId) {
if (empty($this->watchers[$watcherId])) {
return;
}
$watcher = $this->watchers[$watcherId];
if ($watcher->isEnabled) {
$this->keepAliveCount -= $watcher->keepAlive;
2015-04-30 19:41:14 +02:00
}
if ($watcher->type === Watcher::IMMEDIATE) {
unset(
$this->immediates[$watcherId],
$this->watchers[$watcherId]
);
} else {
\event_del($watcher->eventResource);
2015-04-30 19:41:14 +02:00
unset($this->watchers[$watcherId]);
}
if (PHP_MAJOR_VERSION < 7) {
$this->garbage[] = $watcher;
$this->scheduleGarbageCollection();
}
}
/**
* {@inheritDoc}
*/
public function disable($watcherId) {
if (empty($this->watchers[$watcherId])) {
return;
}
$watcher = $this->watchers[$watcherId];
if (!$watcher->isEnabled) {
return;
}
$watcher->isEnabled = false;
$this->keepAliveCount -= $watcher->keepAlive;
2015-04-30 19:41:14 +02:00
switch ($watcher->type) {
case Watcher::IMMEDIATE:
unset($this->immediates[$watcherId]);
break;
case Watcher::IO_READER: // fallthrough
case Watcher::IO_WRITER: // fallthrough
case Watcher::SIGNAL: // fallthrough
case Watcher::TIMER_ONCE: // fallthrough
case Watcher::TIMER_REPEAT:
\event_del($watcher->eventResource);
2015-04-30 19:41:14 +02:00
break;
}
}
/**
* {@inheritDoc}
*/
public function enable($watcherId) {
if (empty($this->watchers[$watcherId])) {
return;
}
$watcher = $this->watchers[$watcherId];
if ($watcher->isEnabled) {
return;
}
$watcher->isEnabled = true;
$this->keepAliveCount += $watcher->keepAlive;
2015-04-30 19:41:14 +02:00
switch ($watcher->type) {
case Watcher::IMMEDIATE:
$this->immediates[$watcherId] = $watcher;
break;
case Watcher::TIMER_ONCE:
\event_add($watcher->eventResource, $watcher->msDelay);
2015-04-30 19:41:14 +02:00
break;
case Watcher::TIMER_REPEAT:
\event_add($watcher->eventResource, $watcher->msInterval);
2015-04-30 19:41:14 +02:00
break;
case Watcher::IO_READER: // fallthrough
case Watcher::IO_WRITER: // fallthrough
case Watcher::SIGNAL:
\event_add($watcher->eventResource);
2015-04-30 19:41:14 +02:00
break;
}
}
private function scheduleGarbageCollection() {
if (!$this->isGcScheduled) {
\event_add($this->gcEvent, 0);
2015-04-30 19:41:14 +02:00
$this->isGcScheduled = true;
}
}
/**
* {@inheritDoc}
*/
2015-07-20 21:42:56 +02:00
public function onError(callable $callback) {
$this->onError = $callback;
2015-04-30 19:41:14 +02:00
}
2015-07-29 07:15:43 +02:00
/**
* {@inheritDoc}
*/
public function info() {
$once = $repeat = $immediately = $onReadable = $onWritable = $onSignal = [
"enabled" => 0,
"disabled" => 0,
];
2015-04-30 19:41:14 +02:00
foreach ($this->watchers as $watcher) {
2015-07-29 07:15:43 +02:00
switch ($watcher->type) {
case Watcher::IMMEDIATE: $arr =& $immediately; break;
case Watcher::TIMER_ONCE: $arr =& $once; break;
case Watcher::TIMER_REPEAT: $arr =& $repeat; break;
case Watcher::IO_READER: $arr =& $onReadable; break;
case Watcher::IO_WRITER: $arr =& $onWritable; break;
case Watcher::SIGNAL: $arr =& $onSignal; break;
2015-07-22 05:54:52 +02:00
}
2015-07-29 07:15:43 +02:00
if ($watcher->isEnabled) {
$arr["enabled"] += 1;
} else {
$arr["disabled"] += 1;
2015-04-30 19:41:14 +02:00
}
}
return [
2015-07-29 07:15:43 +02:00
"immediately" => $immediately,
"once" => $once,
"repeat" => $repeat,
"on_readable" => $onReadable,
"on_writable" => $onWritable,
"on_signal" => $onSignal,
"keep_alive" => $this->keepAliveCount,
2015-04-30 19:41:14 +02:00
];
}
2015-07-29 07:15:43 +02:00
/**
* Access the underlying libevent extension event base
*
* This method provides access to the underlying libevent event loop resource for
* code that wishes to interact with lower-level libevent extension functionality.
*
* @return resource
*/
public function getLoop() {
return $this->keepAliveBase;
2015-07-29 07:15:43 +02:00
}
public function __debugInfo() {
return $this->info();
}
2015-04-30 19:41:14 +02:00
}