1
0
mirror of https://github.com/danog/amp.git synced 2024-12-13 18:07:30 +01:00
amp/lib/NativeReactor.php

466 lines
16 KiB
PHP
Raw Normal View History

2013-08-05 22:05:08 +02:00
<?php
2014-09-23 04:38:32 +02:00
namespace Amp;
2013-08-05 22:05:08 +02:00
2015-03-16 20:00:10 +01:00
class NativeReactor implements Reactor {
private $watchers = [];
private $immediates = [];
private $timerOrder = [];
2013-08-05 22:05:08 +02:00
private $readStreams = [];
private $writeStreams = [];
private $readWatchers = [];
private $writeWatchers = [];
private $isTimerSortNeeded;
2015-03-19 16:14:21 +01:00
private $lastWatcherId = "a";
2014-02-23 22:26:28 +01:00
private $isRunning = false;
2014-12-08 15:50:09 +01:00
private $isTicking = false;
private $onError;
private $onCoroutineResolution;
private static $instanceCount = 0;
2014-09-22 22:47:48 +02:00
public function __construct() {
self::$instanceCount++;
$this->onCoroutineResolution = function($e = null, $r = null) {
if (empty($e)) {
return;
2015-04-30 19:41:14 +02:00
} elseif ($this->onError) {
call_user_func($this->onError, $e);
} else {
2014-09-22 22:47:48 +02:00
throw $e;
}
};
}
public function __destruct() {
self::$instanceCount--;
}
public function __debugInfo() {
$timers = $immediates = $readers = $writers = $disabled = 0;
foreach ($this->watchers as $watcher) {
switch ($watcher->type) {
case Watcher::TIMER_ONCE:
// fallthrough
case Watcher::TIMER_REPEAT:
if ($watcher->isEnabled) { $timers++; } else { $disabled++; }
break;
2015-05-10 06:54:02 +02:00
case Watcher::IO_READER:
if ($watcher->isEnabled) { $readers++; } else { $disabled++; }
break;
case Watcher::IO_WRITER:
if ($watcher->isEnabled) { $writers++; } else { $disabled++; }
break;
case Watcher::IMMEDIATE:
if ($watcher->isEnabled) { $immediates++; } else { $disabled++; }
break;
}
}
return [
'timers' => $timers,
'immediates' => $immediates,
'io_readers' => $readers,
'io_writers' => $writers,
'disabled' => $disabled,
'last_watcher_id' => $this->lastWatcherId,
'instances' => self::$instanceCount,
];
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2014-08-07 07:35:05 +02:00
public function run(callable $onStart = null) {
if ($this->isRunning) {
return;
}
2014-02-23 22:26:28 +01:00
$this->isRunning = true;
if ($onStart) {
$this->immediately($onStart);
}
$this->enableTimers();
while ($this->isRunning || $this->immediates) {
$this->tick();
2013-08-05 22:05:08 +02:00
}
}
private function enableTimers() {
2014-02-23 22:26:28 +01:00
$now = microtime(true);
foreach ($this->watchers as $watcherId => $watcher) {
if (!($watcher->type & Watcher::TIMER) || isset($watcher->nextExecutionAt) || !$watcher->isEnabled) {
continue;
}
$watcher->nextExecutionAt = $now + $watcher->msDelay;
$this->timerOrder[$watcherId] = $watcher->nextExecutionAt;
2013-08-05 22:05:08 +02:00
}
$this->isTimerSortNeeded = true;
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2014-02-23 22:26:28 +01:00
public function stop() {
2014-12-08 17:56:26 +01:00
$this->isRunning = $this->isTicking = false;
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function tick($noWait = false) {
try {
2015-04-30 19:41:14 +02:00
$noWait = (bool) $noWait;
2014-12-08 15:50:09 +01:00
$this->isTicking = true;
if (!$this->isRunning) {
$this->enableTimers();
}
2013-08-05 22:05:08 +02:00
if ($immediates = $this->immediates) {
foreach ($immediates as $watcherId => $watcher) {
unset(
$this->immediates[$watcherId],
$this->watchers[$watcherId]
);
2015-04-30 19:41:14 +02:00
$result = call_user_func($watcher->callback, $this, $watcherId, $watcher->callbackData);
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onCoroutineResolution);
}
2014-09-22 22:47:48 +02:00
}
}
// If an immediately watcher called stop() we pull out here
2014-12-08 15:50:09 +01:00
if (!$this->isTicking) {
return;
}
if ($this->immediates || $noWait) {
$timeToNextAlarm = 0;
} elseif (empty($this->timerOrder)) {
$timeToNextAlarm = 1;
} else {
if ($this->isTimerSortNeeded) {
asort($this->timerOrder);
$this->isTimerSortNeeded = false;
}
$nextTimerAt = current($this->timerOrder);
$timeToNextAlarm = round($nextTimerAt - microtime(true), 4);
$timeToNextAlaram = ($timeToNextAlarm > 0) ? $timeToNextAlarm : 0;
}
2013-08-05 22:05:08 +02:00
if ($this->readStreams || $this->writeStreams) {
$this->selectActionableStreams($timeToNextAlarm);
} elseif (!($this->timerOrder || $this->immediates)) {
$this->stop();
} elseif ($timeToNextAlarm > 0) {
usleep($timeToNextAlarm * 1000000);
}
2013-08-05 22:05:08 +02:00
if ($this->timerOrder || $this->immediates) {
$this->executeTimers();
}
2014-12-08 15:50:09 +01:00
$this->isTicking = false;
} catch (\Exception $error) {
$errorHandler = $this->onCoroutineResolution;
$errorHandler($error);
2013-08-05 22:05:08 +02:00
}
}
private function selectActionableStreams($timeout) {
$r = $this->readStreams;
$w = $this->writeStreams;
2014-08-07 07:35:05 +02:00
$e = null;
2013-08-05 22:05:08 +02:00
if ($timeout <= 0) {
$sec = 0;
$usec = 0;
} else {
$sec = floor($timeout);
$usec = ($timeout - $sec) * 1000000;
}
2014-04-14 23:03:38 +02:00
if (@stream_select($r, $w, $e, $sec, $usec)) {
2013-08-05 22:05:08 +02:00
foreach ($r as $readableStream) {
$streamId = (int) $readableStream;
foreach ($this->readWatchers[$streamId] as $watcherId => $watcher) {
2015-04-30 19:41:14 +02:00
$result = call_user_func($watcher->callback, $this, $watcherId, $readableStream, $watcher->callbackData);
2014-09-22 22:47:48 +02:00
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onCoroutineResolution);
2014-09-22 22:47:48 +02:00
}
2013-08-05 22:05:08 +02:00
}
}
foreach ($w as $writableStream) {
$streamId = (int) $writableStream;
foreach ($this->writeWatchers[$streamId] as $watcherId => $watcher) {
2015-04-30 19:41:14 +02:00
$result = call_user_func($watcher->callback, $this, $watcherId, $writableStream, $watcher->callbackData);
2014-09-22 22:47:48 +02:00
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onCoroutineResolution);
2014-09-22 22:47:48 +02:00
}
2013-08-05 22:05:08 +02:00
}
}
}
}
private function executeTimers() {
2014-02-23 22:26:28 +01:00
$now = microtime(true);
if ($this->isTimerSortNeeded) {
asort($this->timerOrder);
$this->isTimerSortNeeded = false;
}
2013-08-05 22:05:08 +02:00
foreach ($this->timerOrder as $watcherId => $executionCutoff) {
2014-09-22 22:47:48 +02:00
if ($executionCutoff > $now) {
2013-08-05 22:05:08 +02:00
break;
}
$watcher = $this->watchers[$watcherId];
2013-08-05 22:05:08 +02:00
2015-04-30 19:41:14 +02:00
$result = call_user_func($watcher->callback, $this, $watcherId, $watcher->callbackData);
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onCoroutineResolution);
}
if ($watcher->type === Watcher::TIMER_REPEAT && $watcher->isEnabled) {
$this->isTimerSortNeeded = true;
$watcher->nextExecutionAt = $now + $watcher->msInterval;
$this->timerOrder[$watcherId] = $watcher->nextExecutionAt;
2014-09-22 22:47:48 +02:00
} else {
unset(
$this->watchers[$watcherId],
$this->timerOrder[$watcherId]
2014-09-22 22:47:48 +02:00
);
}
}
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function immediately(callable $callback, array $options = []) {
$watcher = new Watcher;
$watcher->id = $watcherId = $this->lastWatcherId++;
$watcher->type = Watcher::IMMEDIATE;
$watcher->callback = $callback;
2015-04-30 19:41:14 +02:00
$watcher->callbackData = @$options["callback_data"];
$watcher->isEnabled = isset($options["enable"]) ? (bool) $options["enable"] : true;
if ($watcher->isEnabled) {
$this->watchers[$watcherId] = $this->immediates[$watcherId] = $watcher;
}
return $watcherId;
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function once(callable $callback, $msDelay, array $options = []) {
$msDelay = (int) $msDelay;
assert(($msDelay >= 0), "\$msDelay at Argument 2 expects integer >= 0");
2015-04-30 19:41:14 +02:00
/* In the php7 branch we use an anonymous class with Struct for this.
* Using a stdclass isn't terribly readable and it's prone to error but
* it's the easiest way to minimize the distance between 5.x and 7 code
* and keep maintenance simple.
*/
$watcher = new \StdClass;
$watcher->id = $watcherId = $this->lastWatcherId++;
$watcher->type = Watcher::TIMER_ONCE;
$watcher->callback = $callback;
2015-04-30 19:41:14 +02:00
$watcher->callbackData = @$options["callback_data"];
$watcher->isEnabled = isset($options["enable"]) ? (bool) $options["enable"] : true;
$watcher->msDelay = round(($msDelay / 1000), 3);
$watcher->nextExecutionAt = null;
if ($watcher->isEnabled && $this->isRunning) {
$nextExecutionAt = microtime(true) + $watcher->msDelay;
$watcher->nextExecutionAt = $nextExecutionAt;
$this->timerOrder[$watcherId] = $nextExecutionAt;
$this->isTimerSortNeeded = true;
2013-08-05 22:05:08 +02:00
}
$this->watchers[$watcherId] = $watcher;
2013-08-05 22:05:08 +02:00
return $watcherId;
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function repeat(callable $callback, $msInterval, array $options = []) {
$msInterval = (int) $msInterval;
assert(($msInterval >= 0), "\$msInterval at Argument 2 expects integer >= 0");
2015-04-30 19:41:14 +02:00
$msDelay = isset($options["ms_delay"]) ? $options["ms_delay"] : (int) $msInterval;
assert(($msDelay >= 0), "ms_delay option expects integer >= 0");
2015-04-30 19:41:14 +02:00
/* In the php7 branch we use an anonymous class with Struct for this.
* Using a stdclass isn't terribly readable and it's prone to error but
* it's the easiest way to minimize the distance between 5.x and 7 code
* and keep maintenance simple.
*/
$watcher = new \StdClass;
$watcher->id = $watcherId = $this->lastWatcherId++;
$watcher->type = Watcher::TIMER_REPEAT;
$watcher->callback = $callback;
2015-04-30 19:41:14 +02:00
$watcher->callbackData = @$options["callback_data"];
$watcher->isEnabled = isset($options["enable"]) ? (bool) $options["enable"] : true;
$watcher->msInterval = round(($msInterval / 1000), 3);
$watcher->msDelay = round(($msDelay / 1000), 3);
2015-04-30 19:41:14 +02:00
$watcher->nextExecutionAt = null; // only needed for php5.x
if ($watcher->isEnabled && $this->isRunning) {
$increment = (isset($watcher->msDelay) ? $watcher->msDelay : $watcher->msInterval);
$nextExecutionAt = microtime(true) + $increment;
$this->timerOrder[$watcherId] = $watcher->nextExecutionAt = $nextExecutionAt;
$this->isTimerSortNeeded = true;
}
$this->watchers[$watcherId] = $watcher;
2013-08-05 22:05:08 +02:00
return $watcherId;
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function onReadable($stream, callable $callback, array $options = []) {
return $this->registerIoWatcher($stream, $callback, $options, Watcher::IO_READER);
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function onWritable($stream, callable $callback, array $options = []) {
return $this->registerIoWatcher($stream, $callback, $options, Watcher::IO_WRITER);
2013-08-05 22:05:08 +02:00
}
2015-04-30 19:41:14 +02:00
private function registerIoWatcher($stream, $callback, $options, $type) {
/* In the php7 branch we use an anonymous class with Struct for this.
* Using a stdclass isn't terribly readable and it's prone to error but
* it's the easiest way to minimize the distance between 5.x and 7 code
* and keep maintenance simple.
*/
$watcher = new \StdClass;
$watcher->id = $watcherId = $this->lastWatcherId++;
$watcher->type = $type;
$watcher->callback = $callback;
2015-04-30 19:41:14 +02:00
$watcher->callbackData = @$options["callback_data"];
$watcher->isEnabled = isset($options["enable"]) ? (bool) $options["enable"] : true;
$watcher->stream = $stream;
$watcher->streamId = $streamId = (int) $stream;
if ($watcher->isEnabled) {
if ($type === Watcher::IO_READER) {
$this->readStreams[$streamId] = $stream;
$this->readWatchers[$streamId][$watcherId] = $watcher;
} else {
$this->writeStreams[$streamId] = $stream;
$this->writeWatchers[$streamId][$watcherId] = $watcher;
}
2013-08-05 22:05:08 +02:00
}
$this->watchers[$watcherId] = $watcher;
2013-08-05 22:05:08 +02:00
return $watcherId;
}
2013-08-05 22:05:08 +02:00
/**
* {@inheritDoc}
*/
2015-04-30 19:41:14 +02:00
public function cancel($watcherId) {
$this->disable($watcherId);
unset($this->watchers[$watcherId]);
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function enable($watcherId) {
if (!isset($this->watchers[$watcherId])) {
2013-08-05 22:05:08 +02:00
return;
}
$watcher = $this->watchers[$watcherId];
if ($watcher->isEnabled) {
// If the watcher is already enabled we're finished here
return;
}
2013-08-05 22:05:08 +02:00
$watcher->isEnabled = true;
2013-08-05 22:05:08 +02:00
switch ($watcher->type) {
case Watcher::TIMER_ONCE:
case Watcher::TIMER_REPEAT:
if (!isset($watcher->nextExecutionAt)) {
$watcher->nextExecutionAt = microtime(true) + $watcher->msDelay;
2013-08-05 22:05:08 +02:00
}
$this->isTimerSortNeeded = true;
$this->timerOrder[$watcherId] = $watcher->nextExecutionAt;
2013-08-05 22:05:08 +02:00
break;
case Watcher::IO_READER:
$streamId = (int) $watcher->stream;
$this->readStreams[$streamId] = $watcher->stream;
$this->readWatchers[$streamId][$watcherId] = $watcher;
2013-08-05 22:05:08 +02:00
break;
case Watcher::IO_WRITER:
$streamId = (int) $watcher->stream;
$this->writeStreams[$streamId] = $watcher->stream;
$this->writeWatchers[$streamId][$watcherId] = $watcher;
2013-08-05 22:05:08 +02:00
break;
case Watcher::IMMEDIATE:
$this->immediates[$watcherId] = $watcher;
break;
default:
assert(false, "Unexpected Watcher type constant encountered");
2013-08-05 22:05:08 +02:00
}
}
2014-08-08 18:26:08 +02:00
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
2014-08-08 18:26:08 +02:00
*/
2015-04-30 19:41:14 +02:00
public function disable($watcherId) {
if (!isset($this->watchers[$watcherId])) {
return;
}
2013-08-05 22:05:08 +02:00
$watcher = $this->watchers[$watcherId];
if (!$watcher->isEnabled) {
return;
}
2013-08-05 22:05:08 +02:00
$watcher->isEnabled = false;
switch ($watcher->type) {
case Watcher::TIMER_ONCE:
case Watcher::TIMER_REPEAT:
unset($this->timerOrder[$watcherId]);
break;
case Watcher::IO_READER:
$streamId = $watcher->streamId;
unset($this->readWatchers[$streamId][$watcherId]);
if (empty($this->readWatchers[$streamId])) {
unset($this->readStreams[$streamId]);
}
break;
case Watcher::IO_WRITER:
$streamId = $watcher->streamId;
unset($this->writeWatchers[$streamId][$watcherId]);
if (empty($this->writeWatchers[$streamId])) {
unset($this->writeStreams[$streamId]);
}
break;
case Watcher::IMMEDIATE:
unset($this->immediates[$watcherId]);
break;
default:
assert(false, "Unexpected Watcher type constant encountered");
2013-08-05 22:05:08 +02:00
}
}
/**
2015-03-19 16:14:21 +01:00
* {@inheritDoc}
*/
2015-03-19 16:14:21 +01:00
public function onError(callable $func) {
$this->onError = $func;
}
}