Emulate libev/libuv behavior in Native and Event Driver

amphp · Apr 19, 2017 · ab59ef3 · ab59ef3
1 parent 89570e8
commit ab59ef3
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 8 deletions.
diff --git a/lib/Loop/EventDriver.php b/lib/Loop/EventDriver.php
@@ -22,9 +22,12 @@ class EventDriver extends Driver {
     private $signalCallback;
     /** @var \Event[] */
     private $signals = [];
+    /** @var int Internal timestamp for now. */
+    private $now;
 
     public function __construct() {
         $this->handle = new \EventBase;
+        $this->now = (int) (\microtime(true) * self::MILLISEC_PER_SEC);
 
         if (self::$activeSignals === null) {
             self::$activeSignals = &$this->signals;
@@ -53,6 +56,8 @@ public function __construct() {
         $this->timerCallback = function ($resource, $what, Watcher $watcher) {
             if ($watcher->type & Watcher::DELAY) {
                 $this->cancel($watcher->id);
+            } else {
+                $this->events[$watcher->id]->add($watcher->value / self::MILLISEC_PER_SEC);
             }
 
             try {
@@ -202,7 +207,7 @@ protected function activate(array $watchers) {
                         $this->events[$id] = new \Event(
                             $this->handle,
                             -1,
-                            \Event::TIMEOUT | \Event::PERSIST,
+                            \Event::TIMEOUT,
                             $this->timerCallback,
                             $watcher
                         );
@@ -228,18 +233,21 @@ protected function activate(array $watchers) {
             switch ($watcher->type) {
                 case Watcher::DELAY:
                 case Watcher::REPEAT:
-                    $this->events[$id]->add($watcher->value / self::MILLISEC_PER_SEC);
+                    $interval = $watcher->value - ($diff ?? ($diff = (int) (\microtime(true) * self::MILLISEC_PER_SEC) - $this->now));
+                    $this->events[$id]->add($interval > 0 ? $interval / self::MILLISEC_PER_SEC : 0);
                     break;
 
                 case Watcher::SIGNAL:
                     $this->signals[$id] = $this->events[$id];
-                // No break
+                    // No break
 
                 default:
                     $this->events[$id]->add();
                     break;
             }
         }
+
+        $this->now = (int) (\microtime(true) * self::MILLISEC_PER_SEC);
     }
 
     /**

diff --git a/lib/Loop/NativeDriver.php b/lib/Loop/NativeDriver.php
@@ -29,12 +29,16 @@ class NativeDriver extends Driver {
     /** @var \Amp\Loop\Watcher[][] */
     private $signalWatchers = [];
 
+    /** @var int Internal timestamp for now. */
+    private $now;
+
     /** @var bool */
     private $signalHandling;
 
     public function __construct() {
         $this->timerQueue = new \SplPriorityQueue();
         $this->signalHandling = \extension_loaded("pcntl");
+        $this->now = (int) (\microtime(true) * self::MILLISEC_PER_SEC);
     }
 
     /**
@@ -65,8 +69,6 @@ protected function dispatch(bool $blocking) {
         );
 
         if (!empty($this->timerExpires)) {
-            $time = (int) (\microtime(true) * self::MILLISEC_PER_SEC);
-
             while (!$this->timerQueue->isEmpty()) {
                 list($watcher, $expiration) = $this->timerQueue->top();
 
@@ -77,14 +79,16 @@ protected function dispatch(bool $blocking) {
                     continue;
                 }
 
-                if ($this->timerExpires[$id] > $time) { // Timer at top of queue has not expired.
+                if ($this->timerExpires[$id] > $this->now) { // Timer at top of queue has not expired.
                     break;
                 }
 
                 $this->timerQueue->extract();
 
                 if ($watcher->type & Watcher::REPEAT) {
-                    $this->activate([$watcher]);
+                    $expiration = $this->now + $watcher->value;
+                    $this->timerExpires[$watcher->id] = $expiration;
+                    $this->timerQueue->insert([$watcher, $expiration], -$expiration);
                 } else {
                     $this->cancel($id);
                 }
@@ -251,7 +255,7 @@ protected function activate(array $watchers) {
 
                 case Watcher::DELAY:
                 case Watcher::REPEAT:
-                    $expiration = (int) (\microtime(true) * self::MILLISEC_PER_SEC) + $watcher->value;
+                    $expiration = $this->now + $watcher->value;
                     $this->timerExpires[$watcher->id] = $expiration;
                     $this->timerQueue->insert([$watcher, $expiration], -$expiration);
                     break;
@@ -276,6 +280,8 @@ protected function activate(array $watchers) {
                     // @codeCoverageIgnoreEnd
             }
         }
+
+        $this->now = (int) (\microtime(true) * self::MILLISEC_PER_SEC);
     }
 
     /**