Enforce timer interval as minimum time to execution (#319)

Co-authored-by: Aaron Piotrowski <aaron@trowski.com>

.travis.yml

@@ -31,7 +31,7 @@ install:
   - travis/install-ev.sh
   - travis/install-event.sh
 
-  - if [[ ${TRAVIS_PHP_VERSION:0:3} == "7.0" ]]; then composer remove --dev vimeo/psalm; fi
+  - if [[ ${TRAVIS_PHP_VERSION:0:3} == "7.0" ]]; then composer remove --dev psalm/phar; fi
   - composer update -n --prefer-dist
 
   - mkdir -p coverage/cov coverage/bin
@@ -45,7 +45,7 @@ script:
   - php vendor/bin/phpunit --verbose --group memoryleak
   - php vendor/bin/phpunit --verbose --exclude-group memoryleak --coverage-php coverage/cov/main.cov
   - PHP_CS_FIXER_IGNORE_ENV=1 php vendor/bin/php-cs-fixer --diff --dry-run -v fix
-  - if [[ ${TRAVIS_PHP_VERSION:0:3} == "7.0" ]]; then echo "Skipped psalm static analysis"; else vendor/bin/psalm; fi
+  - if [[ ${TRAVIS_PHP_VERSION:0:3} == "7.0" ]]; then echo "Skipped psalm static analysis"; else vendor/bin/psalm.phar; fi
 
 after_script:
   - curl -OL https://github.com/php-coveralls/php-coveralls/releases/download/v1.0.0/coveralls.phar

composer.json

@@ -41,7 +41,7 @@
         "amphp/php-cs-fixer-config": "dev-master",
         "react/promise": "^2",
         "phpunit/phpunit": "^6.0.9 | ^7",
-        "vimeo/psalm": "^3.11@dev",
+        "psalm/phar": "^3.11@dev",
         "jetbrains/phpstorm-stubs": "^2019.3"
     },
     "autoload": {

lib/Loop/Driver.php

@@ -227,6 +227,7 @@ abstract class Driver
         $watcher->id = $this->nextId++;
         $watcher->callback = $callback;
         $watcher->value = $delay;
+        $watcher->expiration = $this->now() + $delay;
         $watcher->data = $data;
 
         $this->watchers[$watcher->id] = $watcher;
@@ -263,6 +264,7 @@ abstract class Driver
         $watcher->id = $this->nextId++;
         $watcher->callback = $callback;
         $watcher->value = $interval;
+        $watcher->expiration = $this->now() + $interval;
         $watcher->data = $data;
 
         $this->watchers[$watcher->id] = $watcher;
@@ -408,6 +410,14 @@ abstract class Driver
                 $this->nextTickQueue[$watcher->id] = $watcher;
                 break;
 
+            case Watcher::REPEAT:
+            case Watcher::DELAY:
+                \assert(\is_int($watcher->value));
+
+                $watcher->expiration = $this->now() + $watcher->value;
+                $this->enableQueue[$watcher->id] = $watcher;
+                break;
+
             default:
                 $this->enableQueue[$watcher->id] = $watcher;
                 break;

lib/Loop/EvDriver.php

@@ -32,8 +32,6 @@ class EvDriver extends Driver
     private $signals = [];
     /** @var int Internal timestamp for now. */
     private $now;
-    /** @var bool */
-    private $nowUpdateNeeded = false;
     /** @var int Loop time offset */
     private $nowOffset;
 
@@ -213,10 +211,7 @@ class EvDriver extends Driver
      */
     public function now(): int
     {
-        if ($this->nowUpdateNeeded) {
+        $this->now = getCurrentTime() - $this->nowOffset;
-            $this->now = getCurrentTime() - $this->nowOffset;
-            $this->nowUpdateNeeded = false;
-        }
 
         return $this->now;
     }
@@ -236,7 +231,6 @@ class EvDriver extends Driver
      */
     protected function dispatch(bool $blocking)
     {
-        $this->nowUpdateNeeded = true;
         $this->handle->run($blocking ? \Ev::RUN_ONCE : \Ev::RUN_ONCE | \Ev::RUN_NOWAIT);
     }
 
@@ -247,6 +241,9 @@ class EvDriver extends Driver
      */
     protected function activate(array $watchers)
     {
+        $this->handle->nowUpdate();
+        $now = $this->now();
+
         foreach ($watchers as $watcher) {
             if (!isset($this->events[$id = $watcher->id])) {
                 switch ($watcher->type) {
@@ -273,7 +270,7 @@ class EvDriver extends Driver
 
                         $interval = $watcher->value / self::MILLISEC_PER_SEC;
                         $this->events[$id] = $this->handle->timer(
-                            $interval,
+                            \max(0, ($watcher->expiration - $now) / self::MILLISEC_PER_SEC),
                             ($watcher->type & Watcher::REPEAT) ? $interval : 0,
                             $this->timerCallback,
                             $watcher

lib/Loop/EventDriver.php

@@ -31,9 +31,6 @@ class EventDriver extends Driver
     /** @var \Event[] */
     private $signals = [];
 
-    /** @var bool */
-    private $nowUpdateNeeded = false;
-
     /** @var int Internal timestamp for now. */
     private $now;
 
@@ -235,10 +232,7 @@ class EventDriver extends Driver
      */
     public function now(): int
     {
-        if ($this->nowUpdateNeeded) {
+        $this->now = getCurrentTime() - $this->nowOffset;
-            $this->now = getCurrentTime() - $this->nowOffset;
-            $this->nowUpdateNeeded = false;
-        }
 
         return $this->now;
     }
@@ -258,7 +252,6 @@ class EventDriver extends Driver
      */
     protected function dispatch(bool $blocking)
     {
-        $this->nowUpdateNeeded = true;
         $this->handle->loop($blocking ? \EventBase::LOOP_ONCE : \EventBase::LOOP_ONCE | \EventBase::LOOP_NONBLOCK);
     }
 
@@ -269,7 +262,7 @@ class EventDriver extends Driver
      */
     protected function activate(array $watchers)
     {
-        $now = getCurrentTime() - $this->nowOffset;
+        $now = $this->now();
 
         foreach ($watchers as $watcher) {
             if (!isset($this->events[$id = $watcher->id])) {
@@ -335,7 +328,7 @@ class EventDriver extends Driver
                 case Watcher::REPEAT:
                     \assert(\is_int($watcher->value));
 
-                    $interval = $watcher->value - ($now - $this->now());
+                    $interval = \max(0, $watcher->expiration - $now);
                     $this->events[$id]->add($interval > 0 ? $interval / self::MILLISEC_PER_SEC : 0);
                     break;
 

lib/Loop/Internal/TimerQueue.php

@@ -19,15 +19,17 @@ final class TimerQueue
      * Inserts the watcher into the queue. Time complexity: O(log(n)).
      *
      * @param Watcher $watcher
-     * @param int     $expiration
      *
      * @psalm-param Watcher<int> $watcher
      *
      * @return void
      */
-    public function insert(Watcher $watcher, int $expiration)
+    public function insert(Watcher $watcher)
     {
-        $entry = new TimerQueueEntry($watcher, $expiration);
+        \assert($watcher->expiration !== null);
+        \assert(!isset($this->pointers[$watcher->id]));
+
+        $entry = new TimerQueueEntry($watcher, $watcher->expiration);
 
         $node = \count($this->data);
         $this->data[$node] = $entry;

lib/Loop/NativeDriver.php

@@ -31,9 +31,6 @@ class NativeDriver extends Driver
     /** @var Watcher[][] */
     private $signalWatchers = [];
 
-    /** @var bool */
-    private $nowUpdateNeeded = false;
-
     /** @var int Internal timestamp for now. */
     private $now;
 
@@ -71,10 +68,7 @@ class NativeDriver extends Driver
      */
     public function now(): int
     {
-        if ($this->nowUpdateNeeded) {
+        $this->now = getCurrentTime() - $this->nowOffset;
-            $this->now = getCurrentTime() - $this->nowOffset;
-            $this->nowUpdateNeeded = false;
-        }
 
         return $this->now;
     }
@@ -96,8 +90,6 @@ class NativeDriver extends Driver
      */
     protected function dispatch(bool $blocking)
     {
-        $this->nowUpdateNeeded = true;
-
         $this->selectStreams(
             $this->readStreams,
             $this->writeStreams,
@@ -292,9 +284,7 @@ class NativeDriver extends Driver
                 case Watcher::DELAY:
                 case Watcher::REPEAT:
                     \assert(\is_int($watcher->value));
-
+                    $this->timerQueue->insert($watcher);
-                    $expiration = $this->now() + $watcher->value;
-                    $this->timerQueue->insert($watcher, $expiration);
                     break;
 
                 case Watcher::SIGNAL:

lib/Loop/UvDriver.php

@@ -190,6 +190,8 @@ class UvDriver extends Driver
      */
     public function now(): int
     {
+        \uv_update_time($this->handle);
+
         /** @psalm-suppress TooManyArguments */
         return \uv_now($this->handle);
     }
@@ -220,6 +222,8 @@ class UvDriver extends Driver
      */
     protected function activate(array $watchers)
     {
+        $now = $this->now();
+
         foreach ($watchers as $watcher) {
             $id = $watcher->id;
 
@@ -264,7 +268,7 @@ class UvDriver extends Driver
 
                     \uv_timer_start(
                         $event,
-                        $watcher->value,
+                        \max(0, $watcher->expiration - $now),
                         ($watcher->type & Watcher::REPEAT) ? $watcher->value : 0,
                         $this->timerCallback
                     );

lib/Loop/Watcher.php

@@ -51,4 +51,7 @@ class Watcher
      * @psalm-var TValue
      */
     public $value;
+
+    /** @var int|null */
+    public $expiration;
 }

test/Loop/DriverTest.php

@@ -11,6 +11,7 @@ use Amp\Loop\InvalidWatcherError;
 use Amp\Loop\UnsupportedFeatureException;
 use PHPUnit\Framework\TestCase;
 use React\Promise\RejectedPromise as RejectedReactPromise;
+use function Amp\getCurrentTime;
 
 if (!\defined("SIGUSR1")) {
     \define("SIGUSR1", 30);
@@ -106,6 +107,52 @@ abstract class DriverTest extends TestCase
         });
     }
 
+    public function testCorrectTimeoutIfBlockingBeforeActivate()
+    {
+        $start = 0;
+        $invoked = 0;
+
+        $this->start(function (Driver $loop) use (&$start, &$invoked) {
+            $loop->defer(function () use ($loop, &$start, &$invoked) {
+                $start = getCurrentTime();
+
+                $loop->delay(1000, function () use (&$invoked) {
+                    $invoked = getCurrentTime();
+                });
+
+                \usleep(500000);
+            });
+        });
+
+        $this->assertNotSame(0, $start);
+        $this->assertNotSame(0, $invoked);
+
+        $this->assertGreaterThanOrEqual(999, $invoked - $start);
+        $this->assertLessThan(1100, $invoked - $start);
+    }
+
+    public function testCorrectTimeoutIfBlockingBeforeDelay()
+    {
+        $start = 0;
+        $invoked = 0;
+
+        $this->start(function (Driver $loop) use (&$start, &$invoked) {
+            $start = getCurrentTime();
+
+            \usleep(500000);
+
+            $loop->delay(1000, function () use (&$invoked) {
+                $invoked = getCurrentTime();
+            });
+        });
+
+        $this->assertNotSame(0, $start);
+        $this->assertNotSame(0, $invoked);
+
+        $this->assertGreaterThanOrEqual(1500, $invoked - $start);
+        $this->assertLessThan(1600, $invoked - $start);
+    }
+
     public function testLoopTerminatesWithOnlyUnreferencedWatchers()
     {
         $this->start(function (Driver $loop) use (&$end) {
@@ -272,18 +319,53 @@ abstract class DriverTest extends TestCase
     public function provideRegistrationArgs()
     {
         $args = [
-            ["defer", [function () {
+            [
-            }]],
+                "defer",
-            ["delay", [5, function () {
+                [
-            }]],
+                    function () {
-            ["repeat", [5, function () {
+                    },
-            }]],
+                ],
-            ["onWritable", [\STDOUT, function () {
+            ],
-            }]],
+            [
-            ["onReadable", [\STDIN, function () {
+                "delay",
-            }]],
+                [
-            ["onSignal", [\SIGUSR1, function () {
+                    5,
-            }]],
+                    function () {
+                    },
+                ],
+            ],
+            [
+                "repeat",
+                [
+                    5,
+                    function () {
+                    },
+                ],
+            ],
+            [
+                "onWritable",
+                [
+                    \STDOUT,
+                    function () {
+                    },
+                ],
+            ],
+            [
+                "onReadable",
+                [
+                    \STDIN,
+                    function () {
+                    },
+                ],
+            ],
+            [
+                "onSignal",
+                [
+                    \SIGUSR1,
+                    function () {
+                    },
+                ],
+            ],
         ];
 
         return $args;
@@ -301,7 +383,11 @@ abstract class DriverTest extends TestCase
 
         $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);
+                $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 {
@@ -574,7 +660,14 @@ abstract class DriverTest extends TestCase
                             }
                             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);
+                                $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);
@@ -621,7 +714,10 @@ abstract class DriverTest extends TestCase
      */
     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->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.
@@ -1415,7 +1511,11 @@ abstract class DriverTest extends TestCase
                             break;
 
                         case "onReadable":
-                            $ends = \stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+                            $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;
@@ -1458,7 +1558,11 @@ abstract class DriverTest extends TestCase
 
     public function testMultipleWatchersOnSameDescriptor()
     {
-        $sockets = \stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        $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;
@@ -1505,11 +1609,12 @@ abstract class DriverTest extends TestCase
 
     public function testTimerIntervalCountedWhenNotRunning()
     {
-        \usleep(600000); // 600ms instead of 500ms to allow for variations in timing.
+        $this->loop->delay(1000, function () use (&$start) {
-        $start = \microtime(true);
-        $this->loop->delay(1000, function () use ($start) {
             $this->assertLessThan(0.5, \microtime(true) - $start);
         });
+
+        \usleep(600000); // 600ms instead of 500ms to allow for variations in timing.
+        $start = \microtime(true);
         $this->loop->run();
     }
 
@@ -1565,9 +1670,6 @@ abstract class DriverTest extends TestCase
             // 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();
     }

test/LoopTest.php

@@ -62,9 +62,6 @@ class LoopTest extends BaseTest
                 // Allow a few milliseconds of inaccuracy.
                 $this->assertGreaterThanOrEqual($now - 1, $new);
                 $this->assertLessThanOrEqual($now + 100, $new);
-
-                // Same time should be returned from later call.
-                $this->assertSame($new, Loop::now());
             });
         });
     }

test/PsalmTest.php

@@ -12,7 +12,7 @@ class PsalmTest extends TestCase
     public function test()
     {
         $issues = \json_decode(
-            \shell_exec('./vendor/bin/psalm --output-format=json --no-progress --config=psalm.examples.xml'),
+            \shell_exec('./vendor/bin/psalm.phar --output-format=json --no-progress --config=psalm.examples.xml'),
             true
         );