1
0
mirror of https://github.com/danog/amp.git synced 2024-12-14 10:27:58 +01:00
amp/lib/NativeReactor.php

459 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
class NativeReactor implements Reactor {
private $alarms = [];
private $immediates = [];
2013-08-05 22:05:08 +02:00
private $alarmOrder = [];
private $readStreams = [];
private $writeStreams = [];
private $readCallbacks = [];
private $writeCallbacks = [];
private $watcherIdReadStreamIdMap = [];
private $watcherIdWriteStreamIdMap = [];
private $disabledWatchers = [];
private $resolution = 1000;
2014-08-21 19:54:02 +02:00
private $lastWatcherId = 1;
2014-02-23 22:26:28 +01:00
private $isRunning = false;
2014-09-22 22:47:48 +02:00
private $onGeneratorError;
2013-08-05 22:05:08 +02:00
private static $DISABLED_ALARM = 0;
private static $DISABLED_READ = 1;
private static $DISABLED_WRITE = 2;
private static $DISABLED_IMMEDIATE = 3;
private static $MICROSECOND = 1000000;
2013-08-05 22:05:08 +02:00
2014-09-22 22:47:48 +02:00
public function __construct() {
$this->onGeneratorError = function($e, $r) {
if ($e) {
throw $e;
}
};
}
2014-08-08 18:26:08 +02:00
/**
* Start the event reactor and assume program flow control
*
* @param callable $onStart Optional callback to invoke immediately upon reactor start
* @throws \Exception Will throw if code executed during the event loop throws
* @return void
*/
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->enableAlarms();
while ($this->isRunning || $this->immediates) {
$this->tick();
2013-08-05 22:05:08 +02:00
}
}
private function enableAlarms() {
2014-02-23 22:26:28 +01:00
$now = microtime(true);
2013-08-05 22:05:08 +02:00
$enabled = 0;
foreach ($this->alarms as $watcherId => $alarmStruct) {
$nextExecution = $alarmStruct[1];
if (!$nextExecution) {
$enabled++;
$delay = $alarmStruct[2];
$nextExecution = $now + $delay;
$alarmStruct[1] = $nextExecution;
$this->alarms[$watcherId] = $alarmStruct;
$this->alarmOrder[$watcherId] = $nextExecution;
}
}
}
2014-08-08 18:26:08 +02:00
/**
* Stop the event reactor
*
* @return void
*/
2014-02-23 22:26:28 +01:00
public function stop() {
$this->isRunning = false;
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
* Execute a single event loop iteration
*
* @throws \Exception will throw any uncaught exception encountered during the loop iteration
* @return void
*/
2014-02-23 22:26:28 +01:00
public function tick() {
2013-08-05 22:05:08 +02:00
if (!$this->isRunning) {
$this->enableAlarms();
}
if ($immediates = $this->immediates) {
$this->immediates = [];
foreach ($immediates as $watcherId => $callback) {
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId);
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onGeneratorError);
2014-09-22 22:47:48 +02:00
}
}
}
if ($this->immediates) {
$timeToNextAlarm = 0;
} elseif ($this->alarmOrder) {
$timeToNextAlarm = round(min($this->alarmOrder) - microtime(true), 4);
} else {
$timeToNextAlarm = 1;
}
2013-08-05 22:05:08 +02:00
if ($this->readStreams || $this->writeStreams) {
$this->selectActionableStreams($timeToNextAlarm);
} elseif ((!($this->alarmOrder) || !$this->immediates)) {
$this->stop();
2013-08-05 22:05:08 +02:00
} elseif ($timeToNextAlarm > 0) {
usleep($timeToNextAlarm * self::$MICROSECOND);
2013-08-05 22:05:08 +02:00
}
if ($this->alarmOrder) {
$this->executeAlarms();
}
}
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) * self::$MICROSECOND;
}
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->readCallbacks[$streamId] as $watcherId => $callback) {
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId, $readableStream);
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onGeneratorError);
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->writeCallbacks[$streamId] as $watcherId => $callback) {
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId, $writableStream);
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onGeneratorError);
2014-09-22 22:47:48 +02:00
}
2013-08-05 22:05:08 +02:00
}
}
}
}
private function executeAlarms() {
2014-02-23 22:26:28 +01:00
$now = microtime(true);
2013-08-05 22:05:08 +02:00
asort($this->alarmOrder);
foreach ($this->alarmOrder as $watcherId => $executionCutoff) {
2014-09-22 22:47:48 +02:00
if ($executionCutoff > $now) {
2013-08-05 22:05:08 +02:00
break;
}
2014-09-22 22:47:48 +02:00
list($callback, $nextExecution, $interval, $isRepeating) = $this->alarms[$watcherId];
2013-08-05 22:05:08 +02:00
2014-09-22 22:47:48 +02:00
if ($isRepeating) {
$nextExecution += $interval;
$this->alarms[$watcherId] = [$callback, $nextExecution, $interval, $isRepeating];
$this->alarmOrder[$watcherId] = $nextExecution;
} else {
unset(
$this->alarms[$watcherId],
$this->alarmOrder[$watcherId]
);
}
2013-08-12 17:35:18 +02:00
2014-09-22 22:47:48 +02:00
$result = $callback($this, $watcherId);
if ($result instanceof \Generator) {
resolve($result, $this)->when($this->onGeneratorError);
2014-09-22 22:47:48 +02:00
}
}
2013-08-05 22:05:08 +02:00
}
/**
* Schedule an event to trigger once at the specified time
*
* @param callable $callback Any valid PHP callable
* @param mixed[int|string] $unixTimeOrStr A future unix timestamp or string parsable by strtotime()
* @throws \InvalidArgumentException On invalid future time
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
*/
public function at(callable $callback, $unixTimeOrStr) {
2013-08-05 22:05:08 +02:00
$now = time();
if (is_int($unixTimeOrStr) && $unixTimeOrStr > $now) {
$secondsUntil = ($unixTimeOrStr - $now);
} elseif (($executeAt = @strtotime($unixTimeOrStr)) && $executeAt > $now) {
$secondsUntil = ($executeAt - $now);
} else {
2013-08-05 22:05:08 +02:00
throw new \InvalidArgumentException(
'Unix timestamp or future time string (parsable by strtotime()) required'
2013-08-05 22:05:08 +02:00
);
}
$msDelay = $secondsUntil * $this->resolution;
2013-08-05 22:05:08 +02:00
return $this->once($callback, $msDelay);
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
* Schedule a callback for immediate invocation in the next event loop iteration
*
* @param callable $callback Any valid PHP callable
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-08 18:26:08 +02:00
*/
2014-02-23 22:26:28 +01:00
public function immediately(callable $callback) {
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
$this->immediates[$watcherId] = $callback;
return $watcherId;
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
* Schedule a callback to execute once
*
* @param callable $callback Any valid PHP callable
* @param int $msDelay The delay in milliseconds before the callback will trigger (may be zero)
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-08 18:26:08 +02:00
*/
public function once(callable $callback, $msDelay) {
return $this->scheduleAlarm($callback, $msDelay, $isRepeating = false);
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
/**
* Schedule a recurring callback to execute every $interval seconds until cancelled
*
* @param callable $callback Any valid PHP callable
* @param int $msDelay The interval in milliseconds between callback invocations
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-08 18:26:08 +02:00
*/
public function repeat(callable $callback, $msDelay) {
return $this->scheduleAlarm($callback, $msDelay, $isRepeating = true);
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
private function scheduleAlarm($callback, $msDelay, $isRepeating) {
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
2014-08-08 18:26:08 +02:00
$msDelay = round(($msDelay / $this->resolution), 3);
2013-08-05 22:05:08 +02:00
if ($this->isRunning) {
2014-08-08 18:26:08 +02:00
$nextExecution = (microtime(true) + $msDelay);
2013-08-05 22:05:08 +02:00
$this->alarmOrder[$watcherId] = $nextExecution;
} else {
2014-08-07 07:35:05 +02:00
$nextExecution = null;
2013-08-05 22:05:08 +02:00
}
2014-08-08 18:26:08 +02:00
$alarmStruct = [$callback, $nextExecution, $msDelay, $isRepeating];
2013-08-05 22:05:08 +02:00
$this->alarms[$watcherId] = $alarmStruct;
return $watcherId;
}
2014-08-08 18:26:08 +02:00
/**
* Watch a stream resource for IO readable data and trigger the callback when actionable
*
* @param resource $stream A stream resource to watch for readable data
* @param callable $callback Any valid PHP callable
* @param bool $enableNow Should the watcher be enabled now or held for later use?
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-08 18:26:08 +02:00
*/
2014-02-23 22:26:28 +01:00
public function onReadable($stream, callable $callback, $enableNow = true) {
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
2013-08-05 22:05:08 +02:00
if ($enableNow) {
$streamId = (int) $stream;
$this->readStreams[$streamId] = $stream;
$this->readCallbacks[$streamId][$watcherId] = $callback;
$this->watcherIdReadStreamIdMap[$watcherId] = $streamId;
} else {
$this->disabledWatchers[$watcherId] = [self::$DISABLED_READ, [$stream, $callback]];
}
return $watcherId;
}
2014-08-08 18:26:08 +02:00
/**
* Watch a stream resource to become writable and trigger the callback when actionable
*
* @param resource $stream A stream resource to watch for writability
* @param callable $callback Any valid PHP callable
* @param bool $enableNow Should the watcher be enabled now or held for later use?
2014-08-21 19:54:02 +02:00
* @return string Returns a unique watcher ID
2014-08-08 18:26:08 +02:00
*/
2014-02-23 22:26:28 +01:00
public function onWritable($stream, callable $callback, $enableNow = true) {
2014-08-21 19:54:02 +02:00
$watcherId = (string) $this->lastWatcherId++;
2013-08-05 22:05:08 +02:00
if ($enableNow) {
$streamId = (int) $stream;
$this->writeStreams[$streamId] = $stream;
$this->writeCallbacks[$streamId][$watcherId] = $callback;
$this->watcherIdWriteStreamIdMap[$watcherId] = $streamId;
} else {
$this->disabledWatchers[$watcherId] = [self::$DISABLED_WRITE, [$stream, $callback]];
}
return $watcherId;
}
2014-08-08 18:26:08 +02:00
/**
* Cancel an existing watcher
*
2014-08-21 19:54:02 +02:00
* @param string $watcherId
2014-08-08 18:26:08 +02:00
* @return void
*/
2014-02-23 22:26:28 +01:00
public function cancel($watcherId) {
2013-08-05 22:05:08 +02:00
if (isset($this->alarms[$watcherId])) {
unset(
$this->alarms[$watcherId],
$this->alarmOrder[$watcherId]
);
} elseif (isset($this->watcherIdReadStreamIdMap[$watcherId])) {
$this->cancelReadWatcher($watcherId);
} elseif (isset($this->watcherIdWriteStreamIdMap[$watcherId])) {
$this->cancelWriteWatcher($watcherId);
} elseif (isset($this->disabledWatchers[$watcherId])) {
unset($this->disabledWatchers[$watcherId]);
} elseif (isset($this->immediates[$watcherId])) {
unset($this->immediates[$watcherId]);
2013-08-05 22:05:08 +02:00
}
}
private function cancelReadWatcher($watcherId) {
$streamId = $this->watcherIdReadStreamIdMap[$watcherId];
unset(
$this->readCallbacks[$streamId][$watcherId],
$this->watcherIdReadStreamIdMap[$watcherId],
$this->disabledWatchers[$watcherId]
);
if (empty($this->readCallbacks[$streamId])) {
unset($this->readStreams[$streamId]);
}
}
private function cancelWriteWatcher($watcherId) {
$streamId = $this->watcherIdWriteStreamIdMap[$watcherId];
unset(
$this->writeCallbacks[$streamId][$watcherId],
$this->watcherIdWriteStreamIdMap[$watcherId],
$this->disabledWatchers[$watcherId]
);
if (empty($this->writeCallbacks[$streamId])) {
unset($this->writeStreams[$streamId]);
}
}
2014-08-08 18:26:08 +02:00
/**
* Enable a disabled timer/stream watcher
*
2014-08-21 19:54:02 +02:00
* @param string $watcherId
2014-08-08 18:26:08 +02:00
* @return void
*/
2014-02-23 22:26:28 +01:00
public function enable($watcherId) {
2013-08-05 22:05:08 +02:00
if (!isset($this->disabledWatchers[$watcherId])) {
return;
}
list($type, $watcherStruct) = $this->disabledWatchers[$watcherId];
unset($this->disabledWatchers[$watcherId]);
switch ($type) {
case self::$DISABLED_ALARM:
if (!$nextExecution = $watcherStruct[1]) {
2014-02-23 22:26:28 +01:00
$nextExecution = microtime(true) + $watcherStruct[2];
2013-08-05 22:05:08 +02:00
$watcherStruct[1] = $nextExecution;
}
$this->alarms[$watcherId] = $watcherStruct;
$this->alarmOrder[$watcherId] = $nextExecution;
break;
case self::$DISABLED_READ:
list($stream, $callback) = $watcherStruct;
$streamId = (int) $stream;
$this->readCallbacks[$streamId][$watcherId] = $callback;
$this->watcherIdReadStreamIdMap[$watcherId] = $streamId;
$this->readStreams[$streamId] = $stream;
break;
case self::$DISABLED_WRITE:
list($stream, $callback) = $watcherStruct;
$streamId = (int) $stream;
$this->writeCallbacks[$streamId][$watcherId] = $callback;
$this->watcherIdWriteStreamIdMap[$watcherId] = $streamId;
$this->writeStreams[$streamId] = $stream;
break;
case self::$DISABLED_IMMEDIATE:
$this->immediates[$watcherId] = $watcherStruct;
break;
2013-08-05 22:05:08 +02:00
}
}
2014-08-08 18:26:08 +02:00
/**
* Temporarily disable (but don't cancel) an existing timer/stream watcher
*
2014-08-21 19:54:02 +02:00
* @param string $watcherId
2014-08-08 18:26:08 +02:00
* @return void
*/
2014-02-23 22:26:28 +01:00
public function disable($watcherId) {
2013-08-05 22:05:08 +02:00
if (isset($this->alarms[$watcherId])) {
$alarmStruct = $this->alarms[$watcherId];
$this->disabledWatchers[$watcherId] = [self::$DISABLED_ALARM, $alarmStruct];
unset(
$this->alarms[$watcherId],
$this->alarmOrder[$watcherId]
);
} elseif (isset($this->watcherIdReadStreamIdMap[$watcherId])) {
$streamId = $this->watcherIdReadStreamIdMap[$watcherId];
$stream = $this->readStreams[$streamId];
$callback = $this->readCallbacks[$streamId][$watcherId];
unset(
$this->readCallbacks[$streamId][$watcherId],
$this->watcherIdReadStreamIdMap[$watcherId]
);
if (empty($this->readCallbacks[$streamId])) {
unset($this->readStreams[$streamId]);
}
$this->disabledWatchers[$watcherId] = [self::$DISABLED_READ, [$stream, $callback]];
} elseif (isset($this->watcherIdWriteStreamIdMap[$watcherId])) {
$streamId = $this->watcherIdWriteStreamIdMap[$watcherId];
$stream = $this->writeStreams[$streamId];
$callback = $this->writeCallbacks[$streamId][$watcherId];
unset(
$this->writeCallbacks[$streamId][$watcherId],
$this->watcherIdWriteStreamIdMap[$watcherId]
);
if (empty($this->writeCallbacks[$streamId])) {
unset($this->writeStreams[$streamId]);
}
$this->disabledWatchers[$watcherId] = [self::$DISABLED_WRITE, [$stream, $callback]];
} elseif (isset($this->immediates[$watcherId])) {
$this->disabledWatchers[$watcherId] = [self::$DISABLED_IMMEDIATE, $this->immediates[$watcherId]];
unset($this->immediates[$watcherId]);
2013-08-05 22:05:08 +02:00
}
}
}