diff --git a/spec/std/crystal/evented/timers_spec.cr b/spec/std/crystal/evented/timers_spec.cr index d40917910d1d..9dccbf4f56f2 100644 --- a/spec/std/crystal/evented/timers_spec.cr +++ b/spec/std/crystal/evented/timers_spec.cr @@ -10,6 +10,9 @@ describe Crystal::Evented::Timers do event = Crystal::Evented::Event.new(:sleep, Fiber.current, timeout: 7.seconds) timers.add(pointerof(event)) timers.empty?.should be_false + + timers.delete(pointerof(event)) + timers.empty?.should be_true end it "#next_ready?" do @@ -18,9 +21,18 @@ describe Crystal::Evented::Timers do timers.next_ready?.should be_nil # with events - event = Crystal::Evented::Event.new(:sleep, Fiber.current, timeout: 5.seconds) - timers.add(pointerof(event)) - timers.next_ready?.should eq(event.wake_at?) + event1s = Crystal::Evented::Event.new(:sleep, Fiber.current, timeout: 1.second) + event3m = Crystal::Evented::Event.new(:sleep, Fiber.current, timeout: 3.minutes) + event5m = Crystal::Evented::Event.new(:sleep, Fiber.current, timeout: 5.minutes) + + timers.add(pointerof(event5m)) + timers.next_ready?.should eq(event5m.wake_at?) + + timers.add(pointerof(event1s)) + timers.next_ready?.should eq(event1s.wake_at?) + + timers.add(pointerof(event3m)) + timers.next_ready?.should eq(event1s.wake_at?) end it "#dequeue_ready" do @@ -66,16 +78,6 @@ 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), - ]) - timers.empty?.should be_false end it "#delete" do diff --git a/spec/std/crystal/pointer_pairing_heap_spec.cr b/spec/std/crystal/pointer_pairing_heap_spec.cr new file mode 100644 index 000000000000..7aca79b37f07 --- /dev/null +++ b/spec/std/crystal/pointer_pairing_heap_spec.cr @@ -0,0 +1,150 @@ +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 + + def inspect(io : IO, indent = 0) : Nil + prv = @heap_previous + nxt = @heap_next + chd = @heap_child + + indent.times { io << ' ' } + io << "Node value=" << key + io << " prv=" << prv.try(&.value.key) + io << " nxt=" << nxt.try(&.value.key) + io << " chd=" << chd.try(&.value.key) + io.puts + + node = heap_child? + while node + node.value.inspect(io, indent + 2) + node = node.value.heap_next? + end + 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| + node = heap.shift? + node.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) } + + # insert in random order + (0..9).to_a.shuffle.each do |i| + heap.add nodes.to_unsafe + i + end + + # remove some values + heap.delete(nodes.to_unsafe + 3) + heap.delete(nodes.to_unsafe + 7) + heap.delete(nodes.to_unsafe + 1) + + # remove tail + heap.delete(nodes.to_unsafe + 9) + + # remove head + heap.delete(nodes.to_unsafe + 0) + + # 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 shift while we add 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 diff --git a/src/crystal/pointer_pairing_heap.cr b/src/crystal/pointer_pairing_heap.cr new file mode 100644 index 000000000000..1b0d73d06bcf --- /dev/null +++ b/src/crystal/pointer_pairing_heap.cr @@ -0,0 +1,158 @@ +# :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 : Pointer(T)? + + private def head=(head) + @head = head + head.value.heap_previous = nil if head + head + end + + def empty? + @head.nil? + end + + def first? : Pointer(T)? + @head + end + + def shift? : Pointer(T)? + if node = @head + self.head = merge_pairs(node.value.heap_child?) + node.value.heap_child = nil + node + end + end + + def add(node : Pointer(T)) : Nil + 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) + end + + def delete(node : Pointer(T)) : Nil + 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 + + subtree = merge_pairs(node.value.heap_child?) + clear(node) + self.head = meld(@head, subtree) + else + # removing head + self.head = merge_pairs(node.value.heap_child?) + node.value.heap_child = nil + 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 + clear(node) + end + + private def meld(a : Pointer(T), b : Pointer(T)) : Pointer(T) + if a.value.heap_compare(b) + add_child(a, b) + else + add_child(b, a) + end + end + + private def meld(a : Pointer(T), b : Nil) : Pointer(T) + a + end + + private def meld(a : Nil, b : Pointer(T)) : Pointer(T) + b + end + + private def meld(a : Nil, b : Nil) : Nil + end + + private def add_child(parent : Pointer(T), node : Pointer(T)) : Pointer(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 + + private def merge_pairs(node : Pointer(T)?) : Pointer(T)? + return unless node + + # 1st pass: meld children into pairs (left to right) + tail = nil + + while a = node + if b = a.value.heap_next? + node = b.value.heap_next? + root = meld(a, b) + root.value.heap_previous = tail + tail = root + else + a.value.heap_previous = tail + tail = a + break + end + end + + # 2nd pass: meld the pairs back into a single tree (right to left) + root = nil + + while tail + node = tail.value.heap_previous? + root = meld(root, tail) + tail = node + end + + root.value.heap_next = nil if root + root + end + + private def clear(node) : Nil + node.value.heap_previous = nil + node.value.heap_next = nil + node.value.heap_child = nil + end +end diff --git a/src/crystal/system/unix/evented/event.cr b/src/crystal/system/unix/evented/event.cr index b33130df53c2..e6937cf4d044 100644 --- a/src/crystal/system/unix/evented/event.cr +++ b/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 diff --git a/src/crystal/system/unix/evented/timers.cr b/src/crystal/system/unix/evented/timers.cr index ace4fefcf09b..7b6deac4f543 100644 --- a/src/crystal/system/unix/evented/timers.cr +++ b/src/crystal/system/unix/evented/timers.cr @@ -1,86 +1,58 @@ +require "crystal/pointer_pairing_heap" + # List of `Event` ordered by `Event#wake_at` ascending. Optimized for fast # dequeue and determining when is the next timer event. # -# Thread unsafe: parallel accesses much be protected. +# Thread unsafe: parallel accesses much be protected! # -# NOTE: this is a struct because it only wraps a const pointer to a deque +# NOTE: this is a struct because it only wraps a const pointer to an object # allocated in the heap. -# -# OPTIMIZE: consider a skiplist for faster lookups (add/delete). -# -# OPTIMIZE: we could avoid memmove on add/delete by allocating a buffer, putting -# entries at whatever available index in the buffer, and linking entries in -# order (using indices so we can realloc the buffer); we'd have to keep a list -# of free indexes, too. It could be a good combo of unbounded linked list while -# retaining some memory locality. It should even be compatible with a skiplist -# (e.g. make entries a fixed height tower instead of prev/next node). struct Crystal::Evented::Timers def initialize - @list = Deque(Evented::Event*).new + @heap = PointerPairingHeap(Evented::Event).new end def empty? : Bool - @list.empty? + @heap.empty? end - # Returns the time at which the next timer is supposed to run. + # Returns the time of the next ready timer (if any). def next_ready? : Time::Span? - @list.first?.try(&.value.wake_at) + @heap.first?.try(&.value.wake_at) end # Dequeues and yields each ready timer (their `#wake_at` is lower than # `System::Time.monotonic`) from the oldest to the most recent (i.e. time # ascending). def dequeue_ready(& : Evented::Event* -> Nil) : Nil - return if @list.empty? - seconds, nanoseconds = System::Time.monotonic now = Time::Span.new(seconds: seconds, nanoseconds: nanoseconds) - n = 0 - @list.each do |event| + while event = @heap.first? break if event.value.wake_at > now + @heap.shift? yield event - n += 1 end - - # OPTIMIZE: consume the n entries at once - n.times { @list.shift } end # Add a new timer into the list. Returns true if it is the next ready timer. def add(event : Evented::Event*) : Bool - if @list.empty? - @list << event - true - elsif index = lookup(event.value.wake_at) - @list.insert(index, event) - index == 0 - else - @list.push(event) - false - end - end - - private def lookup(wake_at) - @list.each_with_index do |event, index| - return index if event.value.wake_at >= wake_at - end + @heap.add(event) + @heap.first? == event end # Remove a timer from the list. Returns a tuple(dequeued, was_next_ready) of # booleans. The first bool tells whether the event was dequeued, in which case # the second one tells if it was the next ready event. def delete(event : Evented::Event*) : {Bool, Bool} - if index = @list.index(event) - @list.delete_at(index) - {true, index.zero?} + if @heap.first? == event + @heap.shift? + {true, true} + elsif event.value.heap_previous? + @heap.delete(event) + {true, false} else {false, false} end end - - def each(&) : Nil - @list.each { |event| yield event } - end end