Correct heap implementation in TimerQueue (#332)

* Correct heap implementation in TimerQueue, which would sometimes extract timers in the wrong order after a node other than the root was removed

* Factor out old heap rebuilding code into its own method to satisfy codeclimate requirements

lib/Loop/Internal/TimerQueue.php

@@ -15,6 +15,60 @@ final class TimerQueue
     /** @var int[] */
     private $pointers = [];
 
+    /**
+     * @param int $node Rebuild the data array from the given node upward.
+     *
+     * @return void
+     */
+    private function heapifyUp(int $node)
+    {
+        $entry = $this->data[$node];
+        while ($node !== 0 && $entry->expiration < $this->data[$parent = ($node - 1) >> 1]->expiration) {
+            $temp = $this->data[$parent];
+            $this->data[$node] = $temp;
+            $this->pointers[$temp->watcher->id] = $node;
+
+            $this->data[$parent] = $entry;
+            $this->pointers[$entry->watcher->id] = $parent;
+
+            $node = $parent;
+        }
+    }
+
+    /**
+     * @param int $node Rebuild the data array from the given node downward.
+     *
+     * @return void
+     */
+    private function heapifyDown(int $node)
+    {
+        $length = \count($this->data);
+        while (($child = ($node << 1) + 1) < $length) {
+            if ($this->data[$child]->expiration < $this->data[$node]->expiration
+                && ($child + 1 >= $length || $this->data[$child]->expiration < $this->data[$child + 1]->expiration)
+            ) {
+                // Left child is less than parent and right child.
+                $swap = $child;
+            } elseif ($child + 1 < $length && $this->data[$child + 1]->expiration < $this->data[$node]->expiration) {
+                // Right child is less than parent and left child.
+                $swap = $child + 1;
+            } else { // Left and right child are greater than parent.
+                break;
+            }
+
+            $left = $this->data[$node];
+            $right = $this->data[$swap];
+
+            $this->data[$node] = $right;
+            $this->pointers[$right->watcher->id] = $node;
+
+            $this->data[$swap] = $left;
+            $this->pointers[$left->watcher->id] = $swap;
+
+            $node = $swap;
+        }
+    }
+
     /**
      * Inserts the watcher into the queue. Time complexity: O(log(n)).
      *
@@ -35,16 +89,7 @@ public function insert(Watcher $watcher)
         $this->data[$node] = $entry;
         $this->pointers[$watcher->id] = $node;
 
-        while ($node !== 0 && $entry->expiration < $this->data[$parent = ($node - 1) >> 1]->expiration) {
-            $temp = $this->data[$parent];
-            $this->data[$node] = $temp;
-            $this->pointers[$temp->watcher->id] = $node;
-
-            $this->data[$parent] = $entry;
-            $this->pointers[$watcher->id] = $parent;
-
-            $node = $parent;
-        }
+        $this->heapifyUp($node);
     }
 
     /**
@@ -105,7 +150,7 @@ public function peek()
     }
 
     /**
-     * @param int $node Remove the given node and then rebuild the data array from that node downward.
+     * @param int $node Remove the given node and then rebuild the data array.
      *
      * @return void
      */
@@ -117,29 +162,13 @@ private function removeAndRebuild(int $node)
         $this->pointers[$left->watcher->id] = $node;
         unset($this->data[$length], $this->pointers[$id]);
 
-        while (($child = ($node << 1) + 1) < $length) {
-            if ($this->data[$child]->expiration < $this->data[$node]->expiration
-                && ($child + 1 >= $length || $this->data[$child]->expiration < $this->data[$child + 1]->expiration)
-            ) {
-                // Left child is less than parent and right child.
-                $swap = $child;
-            } elseif ($child + 1 < $length && $this->data[$child + 1]->expiration < $this->data[$node]->expiration) {
-                // Right child is less than parent and left child.
-                $swap = $child + 1;
-            } else { // Left and right child are greater than parent.
-                break;
+        if ($node < $length) { // don't need to do anything if we removed the last element
+            $parent = ($node - 1) >> 1;
+            if ($parent >= 0 && $this->data[$node]->expiration < $this->data[$parent]->expiration) {
+                $this->heapifyUp($node);
+            } else {
+                $this->heapifyDown($node);
             }
-
-            $left = $this->data[$node];
-            $right = $this->data[$swap];
-
-            $this->data[$node] = $right;
-            $this->pointers[$right->watcher->id] = $node;
-
-            $this->data[$swap] = $left;
-            $this->pointers[$left->watcher->id] = $swap;
-
-            $node = $swap;
         }
     }
 }

test/Loop/TimerQueueTest.php

@@ -0,0 +1,46 @@
+<?php
+
+namespace Amp\Test\Loop;
+
+use Amp\Loop\Internal\TimerQueue;
+use Amp\Loop\Watcher;
+use PHPUnit\Framework\TestCase;
+
+class TimerQueueTest extends TestCase
+{
+    public function testHeapOrder()
+    {
+        $values = [
+            29022197, 29026651, 29026649, 29032037, 29031955, 29032037, 29031870, 29032136, 29032075, 29032144,
+            29032160, 29032101, 29032130, 29032091, 29032107, 29032181, 29032137, 29032142, 29032142, 29032146,
+            29032158, 29032166, 29032177, 29032181, 29032180, 29032184, 29032193, 29032122
+        ];
+        $indexToRemove = 16;
+        $queue = new TimerQueue;
+        $id = 'a';
+        $watchers = [];
+        foreach ($values as $value) {
+            $watcher = new Watcher;
+            $watcher->type = Watcher::DELAY;
+            $watcher->id = $id++;
+            $watcher->callback = static function () {};
+            $watcher->expiration = $watcher->value = $value;
+            $watchers[] = $watcher;
+        }
+
+        $toRemove = $watchers[$indexToRemove];
+        foreach ($watchers as $watcher) {
+            $queue->insert($watcher);
+        }
+        $queue->remove($toRemove);
+
+        \array_splice($values, $indexToRemove, 1);
+        \sort($values);
+        $output = [];
+        while (($extracted = $queue->extract(\PHP_INT_MAX)) !== null) {
+            $output[] = $extracted->expiration;
+        }
+
+        $this->assertSame($values, $output);
+    }
+}