Store use a min (pairing) heap

spec/std/crystal/evented/timers_spec.cr

@@ -67,14 +67,16 @@ describe Crystal::Evented::Timers do
     event0.wake_at = -1.minute
     timers.add(pointerof(event0)).should be_true # added new head (next ready)
 
-    events = [] of Crystal::Evented::Event*
-    timers.each { |event| events << event }
-    events.should eq([
-      pointerof(event0),
-      pointerof(event1),
-      pointerof(event3),
-      pointerof(event2),
-    ])
+    # TODO: forcefully dequeue the next ready event, then check the order
+
+    # events = [] of Crystal::Evented::Event*
+    # timers.each { |event| events << event }
+    # events.should eq([
+    #   pointerof(event0),
+    #   pointerof(event1),
+    #   pointerof(event3),
+    #   pointerof(event2),
+    # ])
     timers.empty?.should be_false
   end
 

spec/std/crystal/pairing_heap_spec.cr

@@ -0,0 +1,137 @@
+require "spec"
+require "../../../src/crystal/pointer_pairing_heap"
+
+private struct Node
+  getter key : Int32
+
+  include Crystal::PointerPairingHeap::Node
+
+  def initialize(@key : Int32)
+  end
+
+  def heap_compare(other : Pointer(self)) : Bool
+    key < other.value.key
+  end
+end
+
+describe Crystal::PointerPairingHeap do
+  it "#add" do
+    heap = Crystal::PointerPairingHeap(Node).new
+    node1 = Node.new(1)
+    node2 = Node.new(2)
+    node2b = Node.new(2)
+    node3 = Node.new(3)
+
+    # can add distinct nodes
+    heap.add(pointerof(node3))
+    heap.add(pointerof(node1))
+    heap.add(pointerof(node2))
+
+    # can add duplicate key (different nodes)
+    heap.add(pointerof(node2b))
+
+    # can't add same node twice
+    expect_raises(ArgumentError) { heap.add(pointerof(node1)) }
+
+    # can re-add removed nodes
+    heap.delete(pointerof(node3))
+    heap.add(pointerof(node3))
+
+    heap.shift?.should eq(pointerof(node1))
+    heap.add(pointerof(node1))
+  end
+
+  it "#shift?" do
+    heap = Crystal::PointerPairingHeap(Node).new
+    nodes = StaticArray(Node, 10).new { |i| Node.new(i) }
+
+    # insert in random order
+    (0..9).to_a.shuffle.each do |i|
+      heap.add nodes.to_unsafe + i
+    end
+
+    # removes in ascending order
+    10.times do |i|
+      heap.shift?.should eq(nodes.to_unsafe + i)
+    end
+  end
+
+  it "#delete" do
+    heap = Crystal::PointerPairingHeap(Node).new
+    nodes = StaticArray(Node, 10).new { |i| Node.new(i) }
+
+    # noop: empty
+    heap.delete(nodes.to_unsafe + 0).should eq({false, false})
+
+    # insert in random order
+    (0..9).to_a.shuffle.each do |i|
+      heap.add nodes.to_unsafe + i
+    end
+
+    # noop: unknown node
+    node11 = Node.new(11)
+    heap.delete(pointerof(node11)).should eq({false, false})
+
+    # remove some values
+    heap.delete(nodes.to_unsafe + 3).should eq({true, false})
+    heap.delete(nodes.to_unsafe + 7).should eq({true, false})
+    heap.delete(nodes.to_unsafe + 1).should eq({true, false})
+
+    # remove tail
+    heap.delete(nodes.to_unsafe + 9).should eq({true, false})
+
+    # remove head
+    heap.delete(nodes.to_unsafe + 0).should eq({true, true})
+
+    # repeatedly delete min
+    [2, 4, 5, 6, 8].each do |i|
+      heap.shift?.should eq(nodes.to_unsafe + i)
+    end
+    heap.shift?.should be_nil
+  end
+
+  it "adds 1000 nodes then shifts them in order" do
+    heap = Crystal::PointerPairingHeap(Node).new
+
+    nodes = StaticArray(Node, 1000).new { |i| Node.new(i) }
+    (0..999).to_a.shuffle.each { |i| heap.add(nodes.to_unsafe + i) }
+
+    i = 0
+    while node = heap.shift?
+      node.value.key.should eq(i)
+      i += 1
+    end
+    i.should eq(1000)
+
+    heap.shift?.should be_nil
+  end
+
+  it "randomly adds while we shift nodes" do
+    heap = Crystal::PointerPairingHeap(Node).new
+
+    nodes = uninitialized StaticArray(Node, 1000)
+    (0..999).to_a.shuffle.each_with_index { |i, j| nodes[j] = Node.new(i) }
+
+    i = 0
+    removed = 0
+
+    # regularly calls delete-min while we insert
+    loop do
+      if rand(0..5) == 0
+        removed += 1 if heap.shift?
+      else
+        heap.add(nodes.to_unsafe + i)
+        break if (i += 1) == 1000
+      end
+    end
+
+    # exhaust the heap
+    while heap.shift?
+      removed += 1
+    end
+
+    # we must have added and removed all nodes _once_
+    i.should eq(1000)
+    removed.should eq(1000)
+  end
+end

src/crystal/pointer_pairing_heap.cr

@@ -0,0 +1,161 @@
+# :nodoc:
+#
+# Tree of `T` structs referenced as pointers.
+# `T` must include `Crystal::PointerPairingHeap::Node`.
+class Crystal::PointerPairingHeap(T)
+  module Node
+    macro included
+      property? heap_previous : Pointer(self)?
+      property? heap_next : Pointer(self)?
+      property? heap_child : Pointer(self)?
+    end
+
+    # Compare self with other. For example:
+    #
+    # Use `<` to create a min heap.
+    # Use `>` to create a max heap.
+    abstract def heap_compare(other : Pointer(self)) : Bool
+  end
+
+  @head : T* | Nil
+
+  private def head=(head)
+    @head = head
+    head.value.heap_previous = nil if head
+    head
+  end
+
+  def empty?
+    @head.nil?
+  end
+
+  def first? : T* | Nil
+    @head
+  end
+
+  def shift? : T* | Nil
+    if node = @head
+      self.head = merge_pairs(node.value.heap_child?)
+      node.value.heap_child = nil
+      node
+    end
+  end
+
+  def add(node : T*) : Bool
+    if node.value.heap_previous? || node.value.heap_next? || node.value.heap_child?
+      raise ArgumentError.new("The node is already in a Pairing Heap tree")
+    end
+    self.head = meld(@head, node)
+    node == @head
+  end
+
+  def delete(node : T*) : {Bool, Bool}
+    if node == @head
+      self.head = merge_pairs(node.value.heap_child?)
+      node.value.heap_child = nil
+      return {true, true}
+    end
+
+    if remove?(node)
+      subtree = merge_pairs(node.value.heap_child?)
+      self.head = meld(@head, subtree)
+      unlink(node)
+      return {true, false}
+    end
+
+    {false, false}
+  end
+
+  private def remove?(node)
+    if previous_node = node.value.heap_previous?
+      next_sibling = node.value.heap_next?
+
+      if previous_node.value.heap_next? == node
+        previous_node.value.heap_next = next_sibling
+      else
+        previous_node.value.heap_child = next_sibling
+      end
+
+      if next_sibling
+        next_sibling.value.heap_previous = previous_node
+      end
+
+      true
+    else
+      false
+    end
+  end
+
+  def clear : Nil
+    if node = @head
+      clear_recursive(node)
+      @head = nil
+    end
+  end
+
+  private def clear_recursive(node)
+    child = node.value.heap_child?
+    while child
+      clear_recursive(child)
+      child = child.value.heap_next?
+    end
+    unlink(node)
+  end
+
+  private def meld(a : T*, b : T*) : T*
+    if a.value.heap_compare(b)
+      add_child(a, b)
+    else
+      add_child(b, a)
+    end
+  end
+
+  private def meld(a : T*, b : Nil) : T*
+    a
+  end
+
+  private def meld(a : Nil, b : T*) : T*
+    b
+  end
+
+  private def meld(a : Nil, b : Nil) : Nil
+  end
+
+  private def add_child(parent : T*, node : T*) : T*
+    first_child = parent.value.heap_child?
+    parent.value.heap_child = node
+
+    first_child.value.heap_previous = node if first_child
+    node.value.heap_previous = parent
+    node.value.heap_next = first_child
+
+    parent
+  end
+
+  # Twopass merge of the children of *node* into pairs of two.
+  private def merge_pairs(a : T*) : T* | Nil
+    a.value.heap_previous = nil
+
+    if b = a.value.heap_next?
+      a.value.heap_next = nil
+      b.value.heap_previous = nil
+    else
+      return a
+    end
+
+    rest = merge_pairs(b.value.heap_next?)
+    b.value.heap_next = nil
+
+    pair = meld(a, b)
+    meld(pair, rest)
+  end
+
+  private def merge_pairs(node : Nil) : Nil
+  end
+
+  private def unlink(node) : Nil
+    node.value.heap_previous = nil
+    node.value.heap_next = nil
+    node.value.heap_child = nil
+  end
+end

src/crystal/system/unix/evented/event.cr

@@ -1,4 +1,5 @@
 require "crystal/pointer_linked_list"
+require "crystal/pointer_pairing_heap"
 
 # Information about the event that a `Fiber` is waiting on.
 #
@@ -35,6 +36,9 @@ struct Crystal::Evented::Event
   # The event can be added to `Waiters` lists.
   include PointerLinkedList::Node
 
+  # The event can be added to the `Timers` list.
+  include PointerPairingHeap::Node
+
   def initialize(@type : Type, @fiber, @index = nil, timeout : Time::Span? = nil)
     if timeout
       seconds, nanoseconds = System::Time.monotonic
@@ -55,4 +59,8 @@ struct Crystal::Evented::Event
   # NOTE: musn't be changed after registering the event into `Timers`!
   def wake_at=(@wake_at)
   end
+
+  def heap_compare(other : Pointer(self)) : Bool
+    wake_at < other.value.wake_at
+  end
 end