1
0
mirror of https://github.com/danog/amp.git synced 2024-11-30 04:29:08 +01:00
amp/test/Loop/DriverTest.php
Bob Weinand 7075ef7d74 Fix #251 - 100% CPU usage in NativeDriver
Also use better bounds for timer - it must not execute before, but may execute a little later (on my system it were up to 7 ms actually)
2018-12-11 11:31:37 +01:00

1592 lines
54 KiB
PHP

<?php
namespace Amp\Test\Loop;
use Amp\Coroutine;
use Amp\Delayed;
use Amp\Failure;
use Amp\Loop;
use Amp\Loop\Driver;
use Amp\Loop\InvalidWatcherError;
use Amp\Loop\UnsupportedFeatureException;
use PHPUnit\Framework\TestCase;
use React\Promise\RejectedPromise as RejectedReactPromise;
if (!\defined("SIGUSR1")) {
\define("SIGUSR1", 30);
}
if (!\defined("SIGUSR2")) {
\define("SIGUSR2", 31);
}
if (!\defined("PHP_INT_MIN")) {
\define("PHP_INT_MIN", ~PHP_INT_MAX);
}
abstract class DriverTest extends TestCase
{
/**
* The DriverFactory to run this test on.
*
* @return callable
*/
abstract public function getFactory(): callable;
/** @var Driver */
public $loop;
public function setUp()
{
$this->loop = ($this->getFactory())();
if (!$this->loop instanceof Driver) {
$this->fail("Factory did not return a loop Driver");
}
// Required for error handler to work
Loop::set($this->loop);
\gc_collect_cycles();
}
public function tearDown()
{
unset($this->loop);
}
public function start($cb)
{
$cb($this->loop);
$this->loop->run();
}
public function testEmptyLoop()
{
$this->assertNull($this->loop->run());
}
public function testStopWorksEvenIfNotCurrentlyRunning()
{
$this->assertNull($this->loop->stop());
}
// Note: The running nesting is important for being able to continue actually still running loops (i.e. running flag set, if the driver has one) inside register_shutdown_function() for example
public function testLoopRunsCanBeConsecutiveAndNested()
{
$this->expectOutputString("123456");
$this->start(function (Driver $loop) {
$loop->stop();
$loop->defer(function () use (&$run) {
echo $run = 1;
});
$loop->run();
if (!$run) {
$this->fail("A loop stop before a run must not impact that run");
}
$loop->defer(function () use ($loop) {
$loop->run();
echo 5;
$loop->defer(function () use ($loop) {
echo 6;
$loop->stop();
$loop->defer(function () {
$this->fail("A loop stopped at all levels must not execute further defers");
});
});
$loop->run();
});
$loop->defer(function () use ($loop) {
echo 2;
$loop->defer(function () {
echo 4;
});
});
$loop->defer(function () {
echo 3;
});
});
}
public function testLoopTerminatesWithOnlyUnreferencedWatchers()
{
$this->start(function (Driver $loop) use (&$end) {
$loop->unreference($loop->onReadable(STDIN, function () {
}));
$w = $loop->delay(10000000, function () {
});
$loop->defer(function () use ($loop, $w) {
$loop->cancel($w);
});
$end = true;
});
$this->assertTrue($end);
}
/** This MUST NOT have a "test" prefix, otherwise it's executed as test and marked as risky. */
public function checkForSignalCapability()
{
try {
$watcher = $this->loop->onSignal(SIGUSR1, function () {
});
$this->loop->cancel($watcher);
} catch (UnsupportedFeatureException $e) {
$this->markTestSkipped("The loop is not capable of handling signals properly. Skipping.");
}
}
public function testWatcherUnrefRerefRunResult()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->defer(function () use (&$invoked) {
$invoked = true;
});
$loop->unreference($watcher);
$loop->unreference($watcher);
$loop->reference($watcher);
});
$this->assertTrue($invoked);
}
public function testDeferWatcherUnrefRunResult()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->defer(function () use (&$invoked) {
$invoked = true;
});
$loop->unreference($watcher);
});
$this->assertFalse($invoked);
}
public function testOnceWatcherUnrefRunResult()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->delay(2000, function () use (&$invoked) {
$invoked = true;
});
$loop->unreference($watcher);
$loop->unreference($watcher);
});
$this->assertFalse($invoked);
}
public function testRepeatWatcherUnrefRunResult()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->repeat(2000, function () use (&$invoked) {
$invoked = true;
});
$loop->unreference($watcher);
});
$this->assertFalse($invoked);
}
public function testOnReadableWatcherUnrefRunResult()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->onReadable(STDIN, function () use (&$invoked) {
$invoked = true;
});
$loop->unreference($watcher);
});
$this->assertFalse($invoked);
}
public function testOnWritableWatcherKeepAliveRunResult()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->onWritable(STDOUT, function () use (&$invoked) {
$invoked = true;
});
$loop->unreference($watcher);
});
$this->assertFalse($invoked);
}
public function testOnSignalWatcherKeepAliveRunResult()
{
$this->checkForSignalCapability();
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->onSignal(SIGUSR1, function () {
// empty
});
$watcher = $loop->delay(100, function () use (&$invoked, $loop, $watcher) {
$invoked = true;
$loop->unreference($watcher);
});
$loop->unreference($watcher);
});
$this->assertTrue($invoked);
}
public function testUnreferencedDeferWatcherStillExecutes()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcher = $loop->defer(function () use (&$invoked) {
$invoked = true;
});
$loop->unreference($watcher);
$loop->defer(function () {
// just to keep loop running
});
});
$this->assertTrue($invoked);
}
public function testLoopDoesNotBlockOnNegativeTimerExpiration()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$loop->delay(1, function () use (&$invoked) {
$invoked = true;
});
\usleep(1000 * 10);
});
$this->assertTrue($invoked);
}
public function testDisabledDeferReenableInSubsequentTick()
{
$this->expectOutputString("123");
$this->start(function (Driver $loop) {
$watcherId = $loop->defer(function ($watcherId) {
echo 3;
});
$loop->disable($watcherId);
$loop->defer(function () use ($loop, $watcherId) {
$loop->enable($watcherId);
echo 2;
});
echo 1;
});
}
public function provideRegistrationArgs()
{
$args = [
["defer", [function () {
}]],
["delay", [5, function () {
}]],
["repeat", [5, function () {
}]],
["onWritable", [\STDOUT, function () {
}]],
["onReadable", [\STDIN, function () {
}]],
["onSignal", [\SIGUSR1, function () {
}]],
];
return $args;
}
/**
* @requires PHP 7
* @dataProvider provideRegistrationArgs
*/
public function testWeakTypes($type, $args)
{
if ($type == "onSignal") {
$this->checkForSignalCapability();
}
$this->start(function (Driver $loop) use ($type, $args, &$invoked) {
if ($type == "onReadable") {
$ends = \stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
\fwrite($ends[0], "trigger readability watcher");
$args = [$ends[1]];
} else {
\array_pop($args);
}
$expectedData = 20.75;
if (\substr($type, 0, 2) == "on") {
$args[] = function ($watcherId, $arg, int $data) use ($loop, &$invoked, $expectedData) {
$invoked = true;
$this->assertSame((int) $expectedData, $data);
$loop->unreference($watcherId);
};
} else {
$args[] = function ($watcherId, int $data) use ($loop, &$invoked, $expectedData, $type) {
$invoked = true;
$this->assertSame((int) $expectedData, $data);
if ($type == "repeat") {
$loop->unreference($watcherId);
}
};
}
$args[] = $expectedData;
\call_user_func_array([$loop, $type], $args);
if ($type == "onSignal") {
$loop->defer(function () {
\posix_kill(\getmypid(), \SIGUSR1);
});
}
});
$this->assertTrue($invoked);
}
/** @dataProvider provideRegistrationArgs */
public function testDisableWithConsecutiveCancel($type, $args)
{
if ($type === "onSignal") {
$this->checkForSignalCapability();
}
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked, $type, $args) {
$func = [$loop, $type];
$watcherId = \call_user_func_array($func, $args);
$loop->disable($watcherId);
$loop->defer(function () use (&$invoked, $loop, $watcherId) {
$loop->cancel($watcherId);
$invoked = true;
});
$this->assertFalse($invoked);
});
$this->assertTrue($invoked);
}
/** @dataProvider provideRegistrationArgs */
public function testWatcherReferenceInfo($type, $args)
{
if ($type === "onSignal") {
$this->checkForSignalCapability();
}
$loop = $this->loop;
$func = [$loop, $type];
if (\substr($type, 0, 2) === "on") {
$type = "on_" . \lcfirst(\substr($type, 2));
}
// being referenced is the default
$watcherId1 = \call_user_func_array($func, $args);
$info = $loop->getInfo();
$expected = ["enabled" => 1, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
$expected = ["referenced" => 1, "unreferenced" => 0];
$this->assertSame($expected, $info["enabled_watchers"]);
// explicitly reference() even though it's the default setting
$argsCopy = $args;
$watcherId2 = \call_user_func_array($func, $argsCopy);
$loop->reference($watcherId2);
$loop->reference($watcherId2);
$info = $loop->getInfo();
$expected = ["enabled" => 2, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
$expected = ["referenced" => 2, "unreferenced" => 0];
$this->assertSame($expected, $info["enabled_watchers"]);
// disabling a referenced watcher should decrement the referenced count
$loop->disable($watcherId2);
$loop->disable($watcherId2);
$loop->disable($watcherId2);
$info = $loop->getInfo();
$expected = ["referenced" => 1, "unreferenced" => 0];
$this->assertSame($expected, $info["enabled_watchers"]);
// enabling a referenced watcher should increment the referenced count
$loop->enable($watcherId2);
$loop->enable($watcherId2);
$info = $loop->getInfo();
$expected = ["referenced" => 2, "unreferenced" => 0];
$this->assertSame($expected, $info["enabled_watchers"]);
// cancelling an referenced watcher should decrement the referenced count
$loop->cancel($watcherId2);
$info = $loop->getInfo();
$expected = ["referenced" => 1, "unreferenced" => 0];
$this->assertSame($expected, $info["enabled_watchers"]);
// unreference() should just increment unreferenced count
$watcherId2 = \call_user_func_array($func, $args);
$loop->unreference($watcherId2);
$info = $loop->getInfo();
$expected = ["enabled" => 2, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
$expected = ["referenced" => 1, "unreferenced" => 1];
$this->assertSame($expected, $info["enabled_watchers"]);
$loop->cancel($watcherId1);
$loop->cancel($watcherId2);
}
/** @dataProvider provideRegistrationArgs */
public function testWatcherRegistrationAndCancellationInfo($type, $args)
{
if ($type === "onSignal") {
$this->checkForSignalCapability();
}
$loop = $this->loop;
$func = [$loop, $type];
if (\substr($type, 0, 2) === "on") {
$type = "on_" . \lcfirst(\substr($type, 2));
}
$watcherId = \call_user_func_array($func, $args);
$this->assertInternalType("string", $watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 1, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
// invoke enable() on active watcher to ensure it has no side-effects
$loop->enable($watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 1, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
// invoke disable() twice to ensure it has no side-effects
$loop->disable($watcherId);
$loop->disable($watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 0, "disabled" => 1];
$this->assertSame($expected, $info[$type]);
$loop->cancel($watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 0, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
$watcherId = \call_user_func_array($func, $args);
$info = $loop->getInfo();
$expected = ["enabled" => 1, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
$loop->disable($watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 0, "disabled" => 1];
$this->assertSame($expected, $info[$type]);
$loop->enable($watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 1, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
$loop->cancel($watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 0, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
$loop->disable($watcherId);
$info = $loop->getInfo();
$expected = ["enabled" => 0, "disabled" => 0];
$this->assertSame($expected, $info[$type]);
}
/**
* @dataProvider provideRegistrationArgs
* @group memoryleak
*/
public function testNoMemoryLeak($type, $args)
{
if ($this->getTestResultObject()->getCollectCodeCoverageInformation()) {
$this->markTestSkipped("Cannot run this test with code coverage active [code coverage consumes memory which makes it impossible to rely on memory_get_usage()]");
}
$runs = 2000;
if ($type === "onSignal") {
$this->checkForSignalCapability();
}
$this->start(function (Driver $loop) use ($type, $args, $runs) {
$initialMem = \memory_get_usage();
$cb = function ($runs) use ($loop, $type, $args) {
$func = [$loop, $type];
for ($watchers = [], $i = 0; $i < $runs; $i++) {
$watchers[] = \call_user_func_array($func, $args);
}
foreach ($watchers as $watcher) {
$loop->cancel($watcher);
}
for ($watchers = [], $i = 0; $i < $runs; $i++) {
$watchers[] = \call_user_func_array($func, $args);
}
foreach ($watchers as $watcher) {
$loop->disable($watcher);
$loop->cancel($watcher);
}
for ($watchers = [], $i = 0; $i < $runs; $i++) {
$watchers[] = \call_user_func_array($func, $args);
}
if ($type === "repeat") {
$loop->delay($msInterval = 7, function () use ($loop, $watchers) {
foreach ($watchers as $watcher) {
$loop->cancel($watcher);
}
});
} elseif ($type !== "defer" && $type !== "delay") {
$loop->defer(function () use ($loop, $watchers) {
foreach ($watchers as $watcher) {
$loop->cancel($watcher);
}
});
}
$loop->run();
if ($type === "defer") {
$loop->defer($fn = function ($watcherId, $i) use (&$fn, $loop) {
if ($i) {
$loop->defer($fn, --$i);
}
}, $runs);
$loop->run();
}
if ($type === "delay") {
$loop->delay($msDelay = 0, $fn = function ($watcherId, $i) use (&$fn, $loop) {
if ($i) {
$loop->delay($msDelay = 0, $fn, --$i);
}
}, $runs);
$loop->run();
}
if ($type === "repeat") {
$loop->repeat($msDelay = 0, $fn = function ($watcherId, $i) use (&$fn, $loop) {
$loop->cancel($watcherId);
if ($i) {
$loop->repeat($msDelay = 0, $fn, --$i);
}
}, $runs);
$loop->run();
}
if ($type === "onWritable") {
$loop->defer(function ($watcherId, $runs) use ($loop) {
$fn = function ($watcherId, $socket, $i) use (&$fn, $loop) {
$loop->cancel($watcherId);
if ($socket) {
\fwrite($socket, ".");
}
if ($i) {
// explicitly use *different* streams with *different* resource ids
$ends = \stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
$loop->onWritable($ends[0], $fn, --$i);
$loop->onReadable($ends[1], function ($watcherId) use ($loop) {
$loop->cancel($watcherId);
});
}
};
$fn($watcherId, null, $runs);
}, $runs + 1);
$loop->run();
}
if ($type === "onSignal") {
$sendSignal = function () {
\posix_kill(\getmypid(), \SIGUSR1);
};
$loop->onSignal(\SIGUSR1, $fn = function ($watcherId, $signo, $i) use (&$fn, $loop, $sendSignal) {
if ($i) {
$loop->onSignal(\SIGUSR1, $fn, --$i);
$loop->defer($sendSignal);
}
$loop->cancel($watcherId);
}, $runs);
$loop->defer($sendSignal);
$loop->run();
}
};
$closureMem = \memory_get_usage() - $initialMem;
$cb($runs); /* just to set up eventual structures inside loop without counting towards memory comparison */
\gc_collect_cycles();
$initialMem = \memory_get_usage() - $closureMem;
$cb($runs);
unset($cb);
\gc_collect_cycles();
$endMem = \memory_get_usage();
/* this is allowing some memory usage due to runtime caches etc., but nothing actually leaking */
$this->assertLessThan($runs * 4, $endMem - $initialMem); // * 4, as 4 is minimal sizeof(void *)
});
}
/**
* The first number of each tuple indicates the tick in which the watcher is supposed to execute, the second digit
* indicates the order within the tick.
*/
public function testExecutionOrderGuarantees()
{
$this->expectOutputString("01 02 03 04 " . \str_repeat("05 ", 8) . "10 11 12 " . \str_repeat("13 ", 4) . "20 " . \str_repeat("21 ", 4) . "30 40 41 ");
$this->start(function (Driver $loop) {
// Wrap in extra defer, so driver creation time doesn't count for timers, as timers are driver creation
// relative instead of last tick relative before first tick.
$loop->defer(function () use ($loop) {
$f = function () use ($loop) {
$args = \func_get_args();
return function ($watcherId) use ($loop, &$args) {
if (!$args) {
$this->fail("Watcher callback called too often");
}
$loop->cancel($watcherId);
echo \array_shift($args) . \array_shift($args), " ";
};
};
$loop->onWritable(STDOUT, $f(0, 5));
$writ1 = $loop->onWritable(STDOUT, $f(0, 5));
$writ2 = $loop->onWritable(STDOUT, $f(0, 5));
$loop->delay($msDelay = 0, $f(0, 5));
$del1 = $loop->delay($msDelay = 0, $f(0, 5));
$del2 = $loop->delay($msDelay = 0, $f(0, 5));
$del3 = $loop->delay($msDelay = 0, $f());
$del4 = $loop->delay($msDelay = 0, $f(1, 3));
$del5 = $loop->delay($msDelay = 0, $f(2, 0));
$loop->defer(function () use ($loop, $del5) {
$loop->disable($del5);
});
$loop->cancel($del3);
$loop->disable($del1);
$loop->disable($del2);
$writ3 = $loop->onWritable(STDOUT, $f());
$loop->cancel($writ3);
$loop->disable($writ1);
$loop->disable($writ2);
$loop->enable($writ1);
$writ4 = $loop->onWritable(STDOUT, $f(1, 3));
$loop->onWritable(STDOUT, $f(0, 5));
$loop->enable($writ2);
$loop->disable($writ4);
$loop->defer(function () use ($loop, $writ4, $f) {
$loop->enable($writ4);
$loop->onWritable(STDOUT, $f(1, 3));
});
$loop->enable($del1);
$loop->delay($msDelay = 0, $f(0, 5));
$loop->enable($del2);
$loop->disable($del4);
$loop->defer(function () use ($loop, $del4, $f) {
$loop->enable($del4);
$loop->onWritable(STDOUT, $f(1, 3));
});
$loop->delay($msDelay = 1000, $f(4, 1));
$loop->delay($msDelay = 600, $f(3, 0));
$loop->delay($msDelay = 500, $f(2, 1));
$loop->repeat($msDelay = 500, $f(2, 1));
$rep1 = $loop->repeat($msDelay = 250, $f(2, 1));
$loop->disable($rep1);
$loop->delay($msDelay = 500, $f(2, 1));
$loop->enable($rep1);
$loop->defer($f(0, 1));
$def1 = $loop->defer($f(0, 3));
$def2 = $loop->defer($f(1, 1));
$def3 = $loop->defer($f());
$loop->defer($f(0, 2));
$loop->disable($def1);
$loop->cancel($def3);
$loop->enable($def1);
$loop->defer(function () use ($loop, $def2, $del5, $f) {
$tick = $f(0, 4);
$tick("invalid");
$loop->defer($f(1, 0));
$loop->enable($def2);
$loop->defer($f(1, 2));
$loop->defer(function () use ($loop, $del5, $f) {
$loop->enable($del5);
$loop->defer(function () use ($loop, $f) {
\usleep(700000); // to have $msDelay == 500 and $msDelay == 600 run at the same tick (but not $msDelay == 150)
$loop->defer(function () use ($loop, $f) {
$loop->defer($f(4, 0));
});
});
});
});
$loop->disable($def2);
});
});
}
public function testSignalExecutionOrder()
{
$this->checkForSignalCapability();
$this->expectOutputString("122222");
$this->start(function (Driver $loop) {
$f = function ($i) use ($loop) {
return function ($watcherId) use ($loop, $i) {
$loop->cancel($watcherId);
echo $i;
};
};
$loop->defer($f(1));
$loop->onSignal(SIGUSR1, $f(2));
$sig1 = $loop->onSignal(SIGUSR1, $f(2));
$sig2 = $loop->onSignal(SIGUSR1, $f(2));
$sig3 = $loop->onSignal(SIGUSR1, $f(" FAIL - MUST NOT BE CALLED "));
$loop->disable($sig1);
$loop->onSignal(SIGUSR1, $f(2));
$loop->disable($sig2);
$loop->enable($sig1);
$loop->cancel($sig3);
$loop->onSignal(SIGUSR1, $f(2));
$loop->defer(function () use ($loop, $sig2) {
$loop->enable($sig2);
$loop->defer(function () use ($loop) {
\posix_kill(\getmypid(), \SIGUSR1);
$loop->delay($msDelay = 10, function () use ($loop) {
$loop->stop();
});
});
});
});
}
/** @expectedException \Amp\Loop\InvalidWatcherError */
public function testExceptionOnEnableNonexistentWatcher()
{
try {
$this->loop->enable("nonexistentWatcher");
} catch (InvalidWatcherError $e) {
$this->assertSame("nonexistentWatcher", $e->getWatcherId());
throw $e;
}
}
public function testSuccessOnDisableNonexistentWatcher()
{
$this->loop->disable("nonexistentWatcher");
// Otherwise risky, throwing fails the test
$this->assertTrue(true);
}
public function testSuccessOnCancelNonexistentWatcher()
{
$this->loop->cancel("nonexistentWatcher");
// Otherwise risky, throwing fails the test
$this->assertTrue(true);
}
/** @expectedException \Amp\Loop\InvalidWatcherError */
public function testExceptionOnReferenceNonexistentWatcher()
{
try {
$this->loop->reference("nonexistentWatcher");
} catch (InvalidWatcherError $e) {
$this->assertSame("nonexistentWatcher", $e->getWatcherId());
throw $e;
}
}
public function testSuccessOnUnreferenceNonexistentWatcher()
{
$this->loop->unreference("nonexistentWatcher");
// Otherwise risky, throwing fails the test
$this->assertTrue(true);
}
/** @expectedException \Amp\Loop\InvalidWatcherError */
public function testWatcherInvalidityOnDefer()
{
$this->start(function (Driver $loop) {
$loop->defer(function ($watcher) use ($loop) {
$loop->enable($watcher);
});
});
}
/** @expectedException \Amp\Loop\InvalidWatcherError */
public function testWatcherInvalidityOnDelay()
{
$this->start(function (Driver $loop) {
$loop->delay($msDelay = 0, function ($watcher) use ($loop) {
$loop->enable($watcher);
});
});
}
public function testEventsNotExecutedInSameTickAsEnabled()
{
$this->start(function (Driver $loop) {
$loop->defer(function () use ($loop) {
$loop->defer(function () use ($loop, &$diswatchers, &$watchers) {
$loop->defer(function () use ($loop, $diswatchers) {
foreach ($diswatchers as $watcher) {
$loop->disable($watcher);
}
$loop->defer(function () use ($loop, $diswatchers) {
$loop->defer(function () use ($loop, $diswatchers) {
foreach ($diswatchers as $watcher) {
$loop->cancel($watcher);
}
});
foreach ($diswatchers as $watcher) {
$loop->enable($watcher);
}
});
});
foreach ($watchers as $watcher) {
$loop->cancel($watcher);
}
foreach ($diswatchers as $watcher) {
$loop->disable($watcher);
$loop->enable($watcher);
}
});
$f = function () use ($loop) {
$watchers[] = $loop->defer([$this, "fail"]);
$watchers[] = $loop->delay($msDelay = 0, [$this, "fail"]);
$watchers[] = $loop->repeat($msDelay = 0, [$this, "fail"]);
$watchers[] = $loop->onWritable(STDIN, [$this, "fail"]);
return $watchers;
};
$watchers = $f();
$diswatchers = $f();
});
});
// Otherwise risky, as we only rely on $this->fail()
$this->assertTrue(true);
}
public function testEnablingWatcherAllowsSubsequentInvocation()
{
$loop = $this->loop;
$increment = 0;
$watcherId = $loop->defer(function () use (&$increment) {
$increment++;
});
$loop->disable($watcherId);
$loop->delay($msDelay = 5, [$loop, "stop"]);
$loop->run();
$this->assertSame(0, $increment);
$loop->enable($watcherId);
$loop->delay($msDelay = 5, [$loop, "stop"]);
$loop->run();
$this->assertSame(1, $increment);
}
public function testUnresolvedEventsAreReenabledOnRunFollowingPreviousStop()
{
$increment = 0;
$this->start(function (Driver $loop) use (&$increment) {
$loop->defer([$loop, "stop"]);
$loop->run();
$loop->defer(function () use (&$increment, $loop) {
$loop->delay($msDelay = 100, function () use ($loop, &$increment) {
$increment++;
$loop->stop();
});
});
$this->assertSame(0, $increment);
\usleep(5000);
});
$this->assertSame(1, $increment);
}
public function testTimerWatcherParameterOrder()
{
$this->start(function (Driver $loop) {
$counter = 0;
$loop->defer(function ($watcherId) use ($loop, &$counter) {
$this->assertInternalType("string", $watcherId);
if (++$counter === 3) {
$loop->stop();
}
});
$loop->delay($msDelay = 5, function ($watcherId) use ($loop, &$counter) {
$this->assertInternalType("string", $watcherId);
if (++$counter === 3) {
$loop->stop();
}
});
$loop->repeat($msDelay = 5, function ($watcherId) use ($loop, &$counter) {
$this->assertInternalType("string", $watcherId);
$loop->cancel($watcherId);
if (++$counter === 3) {
$loop->stop();
}
});
});
}
public function testStreamWatcherParameterOrder()
{
$this->start(function (Driver $loop) use (&$invoked) {
$invoked = 0;
$loop->onWritable(STDOUT, function ($watcherId, $stream) use ($loop, &$invoked) {
$this->assertInternalType("string", $watcherId);
$this->assertSame(STDOUT, $stream);
$invoked++;
$loop->cancel($watcherId);
});
});
$this->assertSame(1, $invoked);
}
public function testDisablingWatcherPreventsSubsequentInvocation()
{
$this->start(function (Driver $loop) {
$increment = 0;
$watcherId = $loop->defer(function () use (&$increment) {
$increment++;
});
$loop->disable($watcherId);
$loop->delay($msDelay = 5, [$loop, "stop"]);
$this->assertSame(0, $increment);
});
}
public function testImmediateExecution()
{
$loop = $this->loop;
$increment = 0;
$this->start(function (Driver $loop) use (&$increment) {
$loop->defer(function () use (&$increment) {
$increment++;
});
$loop->defer([$loop, "stop"]);
});
$this->assertSame(1, $increment);
}
public function testImmediatelyCallbacksDoNotRecurseInSameTick()
{
$increment = 0;
$this->start(function (Driver $loop) use (&$increment) {
$loop->defer(function () use ($loop, &$increment) {
$increment++;
$loop->defer(function () use (&$increment) {
$increment++;
});
});
$loop->defer([$loop, "stop"]);
});
$this->assertSame(1, $increment);
}
public function testRunExecutesEventsUntilExplicitlyStopped()
{
$increment = 0;
$this->start(function (Driver $loop) use (&$increment) {
$loop->repeat($msInterval = 5, function ($watcherId) use ($loop, &$increment) {
$increment++;
if ($increment === 10) {
$loop->cancel($watcherId);
}
});
});
$this->assertSame(10, $increment);
}
/**
* @expectedException \Exception
* @expectedExceptionMessage loop error
*/
public function testLoopAllowsExceptionToBubbleUpDuringStart()
{
$this->start(function (Driver $loop) {
$loop->defer(function () {
throw new \Exception("loop error");
});
});
}
/**
* @expectedException \RuntimeException
* @expectedExceptionMessage test
*/
public function testLoopAllowsExceptionToBubbleUpFromRepeatingAlarmDuringStart()
{
$this->start(function (Driver $loop) {
$loop->repeat($msInterval = 1, function () {
throw new \RuntimeException("test");
});
});
}
public function testErrorHandlerCapturesUncaughtException()
{
$msg = "";
$this->loop->setErrorHandler($f = function () {
});
$oldErrorHandler = $this->loop->setErrorHandler(function (\Exception $error) use (&$msg) {
$msg = $error->getMessage();
});
$this->assertSame($f, $oldErrorHandler);
$this->start(function (Driver $loop) {
$loop->defer(function () {
throw new \Exception("loop error");
});
});
$this->assertSame("loop error", $msg);
}
/**
* @expectedException \Exception
* @expectedExceptionMessage errorception
*/
public function testOnErrorFailure()
{
$this->loop->setErrorHandler(function () {
throw new \Exception("errorception");
});
$this->start(function (Driver $loop) {
$loop->delay($msDelay = 5, function () {
throw new \Exception("error");
});
});
}
/**
* @expectedException \RuntimeException
* @expectedExceptionMessage test
*/
public function testLoopException()
{
$this->start(function (Driver $loop) {
$loop->defer(function () use ($loop) {
// force next tick, outside of primary startup tick
$loop->defer(function () {
throw new \RuntimeException("test");
});
});
});
}
public function testOnSignalWatcher()
{
$this->checkForSignalCapability();
$this->expectOutputString("caught SIGUSR1");
$this->start(function (Driver $loop) {
$loop->delay($msDelay = 1, function () use ($loop) {
\posix_kill(\getmypid(), \SIGUSR1);
$loop->delay($msDelay = 10, function () use ($loop) {
$loop->stop();
});
});
$loop->onSignal(SIGUSR1, function ($watcherId) use ($loop) {
$loop->cancel($watcherId);
echo "caught SIGUSR1";
});
});
}
public function testInitiallyDisabledOnSignalWatcher()
{
$this->checkForSignalCapability();
$this->expectOutputString("caught SIGUSR1");
$this->start(function (Driver $loop) {
$stop = $loop->delay($msDelay = 100, function () use ($loop) {
echo "ERROR: manual stop";
$loop->stop();
});
$watcherId = $loop->onSignal(SIGUSR1, function ($watcherId) use ($loop, $stop) {
echo "caught SIGUSR1";
$loop->disable($stop);
$loop->disable($watcherId);
});
$loop->disable($watcherId);
$loop->delay($msDelay = 1, function () use ($loop, $watcherId) {
$loop->enable($watcherId);
$loop->delay($msDelay = 1, function () {
\posix_kill(\getmypid(), SIGUSR1);
});
});
});
}
public function testNestedLoopSignalDispatch()
{
$this->checkForSignalCapability();
$this->expectOutputString("inner SIGUSR2\nouter SIGUSR1\n");
$this->start(function (Driver $loop) {
$loop->delay($msDelay = 300, function () use ($loop) {
$loop->stop();
});
$loop->onSignal(SIGUSR1, function () use ($loop) {
echo "outer SIGUSR1\n";
$loop->stop();
});
$loop->delay($msDelay = 1, function () {
/** @var Driver $loop */
$loop = ($this->getFactory())();
$stop = $loop->delay($msDelay = 100, function () use ($loop) {
echo "ERROR: manual stop";
$loop->stop();
});
$loop->onSignal(SIGUSR2, function ($watcherId) use ($loop, $stop) {
echo "inner SIGUSR2\n";
$loop->cancel($stop);
$loop->cancel($watcherId);
});
$loop->delay($msDelay = 1, function () {
\posix_kill(\getmypid(), SIGUSR2);
});
$loop->run();
});
$loop->delay($msDelay = 20, function () {
\posix_kill(\getmypid(), \SIGUSR1);
});
});
}
public function testCancelRemovesWatcher()
{
$invoked = false;
$this->start(function (Driver $loop) use (&$invoked) {
$watcherId = $loop->delay($msDelay = 10, function () {
$this->fail('Watcher was not cancelled as expected');
});
$loop->defer(function () use ($loop, $watcherId, &$invoked) {
$loop->cancel($watcherId);
$invoked = true;
});
$loop->delay($msDelay = 5, [$loop, "stop"]);
});
$this->assertTrue($invoked);
}
public function testOnWritableWatcher()
{
$flag = false;
$this->start(function (Driver $loop) use (&$flag) {
$loop->onWritable(STDOUT, function () use ($loop, &$flag) {
$flag = true;
$loop->stop();
});
$loop->delay($msDelay = 5, [$loop, "stop"]);
});
$this->assertTrue($flag);
}
public function testInitiallyDisabledWriteWatcher()
{
$increment = 0;
$this->start(function (Driver $loop) {
$watcherId = $loop->onWritable(STDOUT, function () use (&$increment) {
$increment++;
});
$loop->disable($watcherId);
$loop->delay($msDelay = 5, [$loop, "stop"]);
});
$this->assertSame(0, $increment);
}
public function testInitiallyDisabledWriteWatcherIsTriggeredOnceEnabled()
{
$this->expectOutputString("12");
$this->start(function (Driver $loop) {
$watcherId = $loop->onWritable(STDOUT, function () use ($loop) {
echo 2;
$loop->stop();
});
$loop->disable($watcherId);
$loop->defer(function () use ($loop, $watcherId) {
$loop->enable($watcherId);
echo 1;
});
});
}
/** @expectedException \RuntimeException */
public function testStreamWatcherDoesntSwallowExceptions()
{
$this->start(function (Driver $loop) {
$loop->onWritable(STDOUT, function () {
throw new \RuntimeException;
});
$loop->delay($msDelay = 5, [$loop, "stop"]);
});
}
public function testReactorRunsUntilNoWatchersRemain()
{
$var1 = $var2 = 0;
$this->start(function (Driver $loop) use (&$var1, &$var2) {
$loop->repeat($msDelay = 1, function ($watcherId) use ($loop, &$var1) {
if (++$var1 === 3) {
$loop->cancel($watcherId);
}
});
$loop->onWritable(STDOUT, function ($watcherId) use ($loop, &$var2) {
if (++$var2 === 4) {
$loop->cancel($watcherId);
}
});
});
$this->assertSame(3, $var1);
$this->assertSame(4, $var2);
}
public function testReactorRunsUntilNoWatchersRemainWhenStartedDeferred()
{
$var1 = $var2 = 0;
$this->start(function (Driver $loop) use (&$var1, &$var2) {
$loop->defer(function () use ($loop, &$var1, &$var2) {
$loop->repeat($msDelay = 1, function ($watcherId) use ($loop, &$var1) {
if (++$var1 === 3) {
$loop->cancel($watcherId);
}
});
$loop->onWritable(STDOUT, function ($watcherId) use ($loop, &$var2) {
if (++$var2 === 4) {
$loop->cancel($watcherId);
}
});
});
});
$this->assertSame(3, $var1);
$this->assertSame(4, $var2);
}
public function testOptionalCallbackDataPassedOnInvocation()
{
$callbackData = new \StdClass;
$this->start(function (Driver $loop) use ($callbackData) {
$loop->defer(function ($watcherId, $callbackData) {
$callbackData->defer = true;
}, $callbackData);
$loop->delay($msDelay = 1, function ($watcherId, $callbackData) {
$callbackData->delay = true;
}, $callbackData);
$loop->repeat($msDelay = 1, function ($watcherId, $callbackData) use ($loop) {
$callbackData->repeat = true;
$loop->cancel($watcherId);
}, $callbackData);
$loop->onWritable(STDERR, function ($watcherId, $stream, $callbackData) use ($loop) {
$callbackData->onWritable = true;
$loop->cancel($watcherId);
}, $callbackData);
});
$this->assertTrue($callbackData->defer);
$this->assertTrue($callbackData->delay);
$this->assertTrue($callbackData->repeat);
$this->assertTrue($callbackData->onWritable);
}
public function testLoopStopPreventsTimerExecution()
{
$t = \microtime(1);
$this->start(function (Driver $loop) {
$loop->delay($msDelay = 10000, function () {
$this->fail("Timer was executed despite stopped loop");
});
$loop->defer(function () use ($loop) {
$loop->stop();
});
});
$this->assertGreaterThan(\microtime(1), $t + 0.1);
}
public function testDeferEnabledInNextTick()
{
$tick = function () {
$this->loop->defer(function () {
$this->loop->stop();
});
$this->loop->run();
};
$invoked = 0;
$watcher = $this->loop->onWritable(STDOUT, function () use (&$invoked) {
$invoked++;
});
$tick();
$tick();
$tick();
$this->loop->disable($watcher);
$this->loop->enable($watcher);
$tick(); // disable + immediate enable after a tick should have no effect either
$this->assertSame(4, $invoked);
}
// getState and setState are final, but test it here again to be sure
public function testRegistry()
{
$this->assertNull($this->loop->getState("foo"));
$this->loop->setState("foo", NAN);
$this->assertNan($this->loop->getState("foo"));
$this->loop->setState("foo", "1");
$this->assertNull($this->loop->getState("bar"));
$this->loop->setState("baz", -INF);
// running must not affect state
$this->loop->defer([$this->loop, "stop"]);
$this->loop->run();
$this->assertSame(-INF, $this->loop->getState("baz"));
$this->assertSame("1", $this->loop->getState("foo"));
}
/** @dataProvider provideRegistryValues */
public function testRegistryValues($val)
{
$this->loop->setState("foo", $val);
$this->assertSame($val, $this->loop->getState("foo"));
}
public function provideRegistryValues()
{
return [
["string"],
[0],
[PHP_INT_MIN],
[-1.0],
[true],
[false],
[[]],
[null],
[new \StdClass],
];
}
public function testRethrowsFromCallbacks()
{
foreach (["onReadable", "onWritable", "defer", "delay", "repeat", "onSignal"] as $watcher) {
$promises = [
new Failure(new \Exception("rethrow test")),
new RejectedReactPromise(new \Exception("rethrow test")),
new Coroutine((function () {
if (false) {
yield;
}
throw new \Exception("rethrow test");
})()),
(function () {
if (false) {
yield;
}
throw new \Exception("rethrow test");
})(),
null,
];
foreach ($promises as $promise) {
if ($watcher === "onSignal") {
$this->checkForSignalCapability();
}
try {
$args = [];
switch ($watcher) {
case "onSignal":
$args[] = SIGUSR1;
break;
case "onWritable":
$args[] = STDOUT;
break;
case "onReadable":
$ends = \stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
\fwrite($ends[0], "trigger readability watcher");
$args[] = $ends[1];
break;
case "delay":
case "repeat":
$args[] = 5;
break;
}
if ($promise === null) {
$args[] = function ($watcherId) {
$this->loop->cancel($watcherId);
throw new \Exception("rethrow test");
};
} else {
$args[] = function ($watcherId) use ($promise) {
$this->loop->cancel($watcherId);
return $promise;
};
}
\call_user_func_array([$this->loop, $watcher], $args);
if ($watcher == "onSignal") {
$this->loop->delay(100, function () {
\posix_kill(\getmypid(), \SIGUSR1);
});
}
$this->loop->run();
$this->fail("Didn't throw expected exception.");
} catch (\Exception $e) {
$this->assertSame("rethrow test", $e->getMessage());
}
}
}
}
public function testMultipleWatchersOnSameDescriptor()
{
$sockets = \stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
\fwrite($sockets[1], "testing");
$invoked = 0;
$watcher1 = $this->loop->onReadable($sockets[0], function ($watcher) use (&$invoked) {
$invoked += 1;
$this->loop->disable($watcher);
});
$watcher2 = $this->loop->onReadable($sockets[0], function ($watcher) use (&$invoked) {
$invoked += 10;
$this->loop->disable($watcher);
});
$watcher3 = $this->loop->onWritable($sockets[0], function ($watcher) use (&$invoked) {
$invoked += 100;
$this->loop->disable($watcher);
});
$watcher4 = $this->loop->onWritable($sockets[0], function ($watcher) use (&$invoked) {
$invoked += 1000;
$this->loop->disable($watcher);
});
$this->loop->defer(function () use ($watcher1, $watcher3) {
$this->loop->delay(200, function () use ($watcher1, $watcher3) {
$this->loop->enable($watcher1);
$this->loop->enable($watcher3);
});
});
$this->loop->run();
$this->assertSame(1212, $invoked);
$this->loop->enable($watcher1);
$this->loop->enable($watcher3);
$this->loop->delay(100, function () use ($watcher2, $watcher4) {
$this->loop->enable($watcher2);
$this->loop->enable($watcher4);
});
$this->loop->run();
$this->assertSame(2323, $invoked);
}
public function testTimerIntervalCountedWhenNotRunning()
{
\usleep(600000); // 600ms instead of 500ms to allow for variations in timing.
$start = \microtime(true);
$this->loop->delay(1000, function () use ($start) {
$this->assertLessThan(0.5, \microtime(true) - $start);
});
$this->loop->run();
}
public function testShortTimerDoesNotBlockOtherTimers()
{
$this->loop->repeat(0, function () {
static $i = 0;
if (++$i === 5) {
$this->fail("Loop continues with repeat watcher");
}
\usleep(2000);
});
$this->loop->delay(2, function () {
$this->assertTrue(true);
$this->loop->stop();
});
$this->loop->run();
}
public function testTwoShortRepeatTimersWorkAsExpected()
{
$this->loop->repeat(0, function () use (&$j) {
static $i = 0;
if (++$i === 5) {
$this->loop->stop();
}
$j = $i;
});
$this->loop->repeat(0, function () use (&$k) {
static $i = 0;
if (++$i === 5) {
$this->loop->stop();
}
$k = $i;
});
$this->loop->run();
$this->assertLessThan(2, \abs($j - $k));
$this->assertNotSame(0, $j);
}
public function testNow()
{
$now = $this->loop->now();
$this->loop->delay(100, function () use ($now) {
$now += 100;
$new = $this->loop->now();
// Allow a few milliseconds of inaccuracy.
$this->assertGreaterThanOrEqual($now - 1, $new);
$this->assertLessThanOrEqual($now + 10, $new);
// Same time should be returned from later call.
$this->assertSame($new, $this->loop->now());
});
$this->loop->run();
}
public function testBug163ConsecutiveDelayed()
{
$emits = 3;
$this->loop->defer(function () use (&$time, $emits) {
$time = \microtime(true);
for ($i = 0; $i < $emits; ++$i) {
yield new Delayed(100);
}
$time = \microtime(true) - $time;
});
$this->loop->run();
$this->assertGreaterThan(100 * $emits - 1 /* 1ms grace period */, $time * 1000);
}
}