Add priority heap and timers implementation. (#100)

.contributors.yaml

@@ -0,0 +1,59 @@
+- path: lib/io/event/priority_heap.rb
+  time: 2021-02-12T12:19:44+01:00
+  author:
+    name: Wander Hillen
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2021-02-12T13:19:56+01:00
+  author:
+    name: Wander Hillen
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2021-02-12T13:28:58+01:00
+  author:
+    name: Wander Hillen
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2021-02-13T18:44:46+13:00
+  author:
+    name: Samuel Williams
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2021-02-13T12:40:15+01:00
+  author:
+    name: Wander Hillen
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2021-02-13T12:47:51+01:00
+  author:
+    name: Wander Hillen
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2022-09-02T13:45:20+12:00
+  author:
+    name: Samuel Williams
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2022-09-02T13:45:20+12:00
+  author:
+    name: Samuel Williams
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2022-10-13T11:06:34+13:00
+  author:
+    name: Samuel Williams
+    email: [email protected]
+
+- path: lib/io/event/priority_heap.rb
+  time: 2023-04-12T17:25:59+12:00
+  author:
+    name: Samuel Williams
+    email: [email protected]

io-event.gemspec

@@ -7,7 +7,7 @@ Gem::Specification.new do |spec|
 	spec.version = IO::Event::VERSION
 
 	spec.summary = "An event loop."
-	spec.authors = ["Samuel Williams", "Math Ieu", "Benoit Daloze", "Bruno Sutic", "Alex Matchneer", "Delton Ding", "Pavel Rosický"]
+	spec.authors = ["Samuel Williams", "Math Ieu", "Wander Hillen", "Benoit Daloze", "Bruno Sutic", "Alex Matchneer", "Delton Ding", "Pavel Rosický"]
 	spec.license = "MIT"
 
 	spec.cert_chain  = ['release.cert']

lib/io/event.rb

@@ -1,10 +1,11 @@
 # frozen_string_literal: true
 
 # Released under the MIT License.
-# Copyright, 2021-2023, by Samuel Williams.
+# Copyright, 2021-2024, by Samuel Williams.
 
 require_relative 'event/version'
 require_relative 'event/selector'
+require_relative 'event/timers'
 
 begin
 	require 'IO_Event'

lib/io/event/priority_heap.rb

@@ -0,0 +1,148 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2021, by Wander Hillen.
+# Copyright, 2021-2024, by Samuel Williams.
+
+class IO
+	module Event
+		# A priority queue implementation using a standard binary minheap. It uses straight comparison
+		# of its contents to determine priority.
+		# See <https://en.wikipedia.org/wiki/Binary_heap> for explanations of the main methods.
+		class PriorityHeap
+			def initialize
+				# The heap is represented with an array containing a binary tree. See
+				# https://en.wikipedia.org/wiki/Binary_heap#Heap_implementation for how this array
+				# is built up.
+				@contents = []
+			end
+
+			# Returns the earliest timer or nil if the heap is empty.
+			def peek
+				@contents[0]
+			end
+
+			# Returns the number of elements in the heap
+			def size
+				@contents.size
+			end
+
+			# Returns the earliest timer if the heap is non-empty and removes it from the heap.
+			# Returns nil if the heap is empty. (and doesn't change the heap in that case)
+			def pop
+				# If the heap is empty:
+				if @contents.empty?
+					return nil
+				end
+
+				# If we have only one item, no swapping is required:
+				if @contents.size == 1
+					return @contents.pop
+				end
+
+				# Take the root of the tree:
+				value = @contents[0]
+
+				# Remove the last item in the tree:
+				last = @contents.pop
+
+				# Overwrite the root of the tree with the item:
+				@contents[0] = last
+
+				# Bubble it down into place:
+				bubble_down(0)
+
+				# validate!
+
+				return value
+			end
+
+			# Inserts a new timer into the heap, then rearranges elements until the heap invariant is true again.
+			def push(element)
+				# Insert the item at the end of the heap:
+				@contents.push(element)
+
+				# Bubble it up into position:
+				bubble_up(@contents.size - 1)
+
+				# validate!
+
+				return self
+			end
+
+			# Empties out the heap, discarding all elements
+			def clear!
+				@contents = []
+			end
+
+			# Validate the heap invariant. Every element except the root must not be smaller than
+			# its parent element. Note that it MAY be equal.
+			def valid?
+				# notice we skip index 0 on purpose, because it has no parent
+				(1..(@contents.size - 1)).all? { |e| @contents[e] >= @contents[(e - 1) / 2] }
+			end
+
+			private
+
+			# Left here for reference, but unused.
+			# def swap(i, j)
+			# 	@contents[i], @contents[j] = @contents[j], @contents[i]
+			# end
+
+			def bubble_up(index)
+				parent_index = (index - 1) / 2 # watch out, integer division!
+
+				while index > 0 && @contents[index] < @contents[parent_index]
+					# if the node has a smaller value than its parent, swap these nodes
+					# to uphold the minheap invariant and update the index of the 'current'
+					# node. If the node is already at index 0, we can also stop because that
+					# is the root of the heap.
+					# swap(index, parent_index)
+					@contents[index], @contents[parent_index] = @contents[parent_index], @contents[index]
+
+					index = parent_index
+					parent_index = (index - 1) / 2 # watch out, integer division!
+				end
+			end
+
+			def bubble_down(index)
+				swap_value = 0
+				swap_index = nil
+
+				while true
+					left_index = (2 * index) + 1
+					left_value = @contents[left_index]
+
+					if left_value.nil?
+						# This node has no children so it can't bubble down any further.
+						# We're done here!
+						return
+					end
+
+					# Determine which of the child nodes has the smallest value:
+					right_index = left_index + 1
+					right_value = @contents[right_index]
+
+					if right_value.nil? or right_value > left_value
+						swap_value = left_value
+						swap_index = left_index
+					else
+						swap_value = right_value
+						swap_index = right_index
+					end
+
+					if @contents[index] < swap_value
+						# No need to swap, the minheap invariant is already satisfied:
+						return
+					else
+						# At least one of the child node has a smaller value than the current node, swap current node with that child and update current node for if it might need to bubble down even further:
+						# swap(index, swap_index)
+						@contents[index], @contents[swap_index] = @contents[swap_index], @contents[index]
+
+						index = swap_index
+					end
+				end
+			end
+		end
+	end
+end

lib/io/event/timers.rb

@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2024, by Samuel Williams.
+
+require_relative 'priority_heap'
+
+class IO
+	module Event
+		class Timers
+			class Handle
+				def initialize(offset, block)
+					@offset = offset
+					@block = block
+				end
+
+				def < other
+					@offset < other.offset
+				end
+
+				def > other
+					@offset > other.offset
+				end
+
+				attr :offset
+				attr :block
+
+				def call(...)
+					@block.call(...)
+				end
+
+				def cancel!
+					@block = nil
+				end
+
+				def cancelled?
+					@block.nil?
+				end
+			end
+
+			def initialize
+				@heap = PriorityHeap.new
+				@scheduled = []
+			end
+
+			def size
+				flush!
+
+				return @heap.size
+			end
+
+			def schedule(offset, block)
+				handle = Handle.new(offset, block)
+				@scheduled << handle
+
+				return handle
+			end
+
+			def after(timeout, &block)
+				schedule(now + timeout, block)
+			end
+
+			def wait_interval(now = self.now)
+				flush!
+
+				while handle = @heap.peek
+					if handle.cancelled?
+						@heap.pop
+					else
+						return handle.offset - now
+					end
+				end
+			end
+
+			def now
+				::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
+			end
+
+			def fire(now = self.now)
+				# Flush scheduled timers into the heap:
+				flush!
+
+				# Get the earliest timer:
+				while handle = @heap.peek
+					if handle.cancelled?
+						@heap.pop
+					elsif handle.offset <= now
+						# Remove the earliest timer from the heap:
+						@heap.pop
+
+						# Call the block:
+						handle.call(now)
+					else
+						break
+					end
+				end
+			end
+
+			protected def flush!
+				while handle = @scheduled.pop
+					@heap.push(handle) unless handle.cancelled?
+				end
+			end
+		end
+	end
+end

license.md

@@ -1,5 +1,6 @@
 # MIT License
 
+Copyright, 2021, by Wander Hillen.  
 Copyright, 2021-2024, by Samuel Williams.  
 Copyright, 2021, by Delton Ding.  
 Copyright, 2021-2024, by Benoit Daloze.