Rework reactor interface. Use Scheduler as base class.

Author: ioquatix

.github/workflows/development.yml

@@ -14,7 +14,7 @@ jobs:
           - macos
 
         ruby:
-          - 3.0
+          - head
 
         experimental: [false]
         env: [""]

async.gemspec

@@ -13,7 +13,7 @@ Gem::Specification.new do |spec|
 
 	spec.files = Dir.glob('{lib}/**/*', File::FNM_DOTMATCH, base: __dir__)
 
-	spec.required_ruby_version = ">= 3.0.1"
+	spec.required_ruby_version = ">= 3.1.0"
 
 	spec.add_dependency "console", "~> 1.10"
 

gems.rb

@@ -4,6 +4,9 @@
 
 gemspec
 
+# gem "event", path: "../event"
+# gem "async-rspec", path: "../async-rspec"
+
 group :maintenance, optional: true do
 	gem "bake-bundler"
 	gem "bake-modernize"

lib/async.rb

@@ -21,15 +21,14 @@
 # THE SOFTWARE.
 
 require_relative "async/version"
-require_relative "async/logger"
 require_relative "async/reactor"
 
 require_relative "kernel/async"
 require_relative "kernel/sync"
 
 module Async
 	# Invoke `Reactor.run` with all arguments/block.
-	def self.run(*arguments, &block)
-		Reactor.run(*arguments, &block)
+	def self.run(...)
+		Reactor.run(...)
 	end
 end

lib/async/logger.rb renamed to lib/async/debug/selector.rb

@@ -1,6 +1,6 @@
 # frozen_string_literal: true
 
-# Copyright, 2017, by Samuel G. D. Williams. <http://www.codeotaku.com>
+# Copyright, 2021, by Samuel G. D. Williams. <http://www.codeotaku.com>
 # 
 # Permission is hereby granted, free of charge, to any person obtaining a copy
 # of this software and associated documentation files (the "Software"), to deal
@@ -20,9 +20,15 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 # THE SOFTWARE.
 
-require 'console'
-require_relative 'task'
+require 'fiber'
+require 'event/debug/selector'
 
 module Async
-	extend Console
+	module Debug
+		class Selector < Event::Debug::Selector
+			def initialize(selector = nil)
+				super(selector || Event::Backend.new(Fiber.current))
+			end
+		end
+	end
 end

lib/async/scheduler/interrupt.rb renamed to lib/async/interrupt.rb

@@ -19,32 +19,32 @@
 # THE SOFTWARE.
 
 module Async
-	class Scheduler
-		# A thread safe synchronisation primative.
-		class Interrupt
-			def initialize(scheduler)
-				@scheduler = scheduler
-				@input, @output = IO.pipe
-				
-				@fiber = Fiber.new do
-					while true
-						@scheduler.io_wait(@fiber, @input, ::Event::READABLE)
-						@input.read_nonblock(1)
-					end
-				end
-				
-				@fiber.transfer
-			end
+	# A thread safe synchronisation primative.
+	class Interrupt
+		def initialize(scheduler, &block)
+			@scheduler = scheduler
+			@input, @output = IO.pipe
 
-			def signal
-				@output.write('.')
-				@output.flush
+			@fiber = Fiber.new do
+				while true
+					@scheduler.io_wait(@fiber, @input, ::Event::READABLE)
+					block.call(@input.read_nonblock(1))
+				end
 			end
 
-			def close
-				@input.close
-				@output.close
-			end
+			@fiber.transfer
+		end
+		
+		def signal(event = '.')
+			@output.write('.')
+			@output.flush
+		end
+		
+		def close
+			@input.close
+			@output.close
 		end
 	end
+	
+	private_constant :Interrupt
 end

lib/async/node.rb

@@ -175,6 +175,10 @@ def initialize(parent = nil, annotation: nil, transient: false)
 			end
 		end
 
+		def root
+			@parent&.root || self
+		end
+		
 		# You should not directly rely on these pointers but instead use `#children`.
 		# List pointers:
 		attr_accessor :head
@@ -225,7 +229,7 @@ def backtrace(*arguments)
 		end
 
 		def to_s
-			"\#<#{description}>"
+			"\#<#{self.description}>"
 		end
 
 		# Change the parent of this node.
@@ -306,6 +310,10 @@ def stop
 			@children&.each(&:stop)
 		end
 
+		def stopped?
+			@children.nil?
+		end
+		
 		def print_hierarchy(out = $stdout, backtrace: true)
 			self.traverse do |node, level|
 				indent = "\t" * level

lib/async/notification.rb

@@ -30,7 +30,7 @@ class Notification < Condition
 		def signal(value = nil, task: Task.current)
 			return if @waiting.empty?
 
-			task.reactor << Signal.new(@waiting, value)
+			Fiber.scheduler << Signal.new(@waiting, value)
 
 			@waiting = []
 

lib/async/reactor.rb

@@ -1,6 +1,4 @@
-# frozen_string_literal: true
-
-# Copyright, 2017, by Samuel G. D. Williams. <http://www.codeotaku.com>
+# Copyright, 2020, by Samuel G. D. Williams. <http://www.codeotaku.com>
 # 
 # Permission is hereby granted, free of charge, to any person obtaining a copy
 # of this software and associated documentation files (the "Software"), to deal
@@ -20,25 +18,11 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 # THE SOFTWARE.
 
-require_relative 'logger'
-require_relative 'task'
 require_relative 'scheduler'
-
-require 'timers'
-require 'forwardable'
+require_relative 'task'
 
 module Async
-	# Raised if a timeout occurs on a specific Fiber. Handled gracefully by `Task`.
-	class TimeoutError < StandardError
-		def initialize(message = "execution expired")
-			super
-		end
-	end
-	
-	# An asynchronous, cooperatively scheduled event reactor.
-	class Reactor < Node
-		extend Forwardable
-		
+	class Reactor < Scheduler
 		# The preferred method to invoke asynchronous behavior at the top level.
 		#
 		# - When invoked within an existing reactor task, it will run the given block
@@ -60,24 +44,17 @@ def self.run(*arguments, **options, &block)
 			end
 		end
 
-		def initialize(parent = nil, selector: nil, logger: nil)
-			super(parent)
+		def initialize(...)
+			super
 
-			@scheduler = Scheduler.new(self, selector)
-			@logger = logger
+			@closing = false
+			self.set!
 		end
 
-		attr :scheduler
-		attr :logger
-		
 		def to_s
 			"\#<#{self.description} #{@children&.size || 0} children (#{stopped? ? 'stopped' : 'running'})>"
 		end
 
-		def stopped?
-			@children.nil?
-		end
-		
 		# Start an asynchronous task within the specified reactor. The task will be
 		# executed until the first blocking call, at which point it will yield and
 		# and this method will return.
@@ -102,95 +79,17 @@ def async(*arguments, **options, &block)
 			return task
 		end
 
-		# Interrupt the reactor at the earliest convenience. Can be called from a different thread safely.
-		def interrupt
-			@scheduler.interrupt
-		end
-		
-		# Schedule a fiber (or equivalent object) to be resumed on the next loop through the reactor.
-		# @param fiber [#resume] The object to be resumed on the next iteration of the run-loop.
-		def << fiber
-			@scheduler << fiber
-		end
-		
-		# Yield the current fiber and resume it on the next iteration of the event loop.
-		def yield
-			@scheduler.yield
-		end
-		
-		def transfer
-			@scheduler.transfer
-		end
-		
-		def resume(fiber)
-			@scheduler.resume(fiber)
-		end
-		
-		def finished?
-			super && @scheduler.finished?
-		end
-		
-		# Run the reactor until all tasks are finished. Proxies arguments to {#async} immediately before entering the loop, if a block is provided.
-		def run(*arguments, **options, &block)
-			raise RuntimeError, 'Reactor has been closed' if @scheduler.nil?
-			
-			@scheduler&.set!
-			
-			initial_task = self.async(*arguments, **options, &block) if block_given?
-			
-			@scheduler.run
-			
-			return initial_task
-		ensure
-			@scheduler&.clear!
-			Console.logger.debug(self) {"Exiting run-loop because #{$! ? $! : 'finished'}."}
-		end
-		
-		def stop(later = true)
-			@children&.each do |child|
-				# We don't want this process to propagate `Async::Stop` exceptions, so we schedule tasks to stop later.
-				child.stop(later)
-			end
-		end
-		
-		# Stop each of the children tasks and close the selector.
-		# 
-		# @return [void]
 		def close
-			# This is a critical step. Because tasks could be stored as instance variables, and since the reactor is (probably) going out of scope, we need to ensure they are stopped. Otherwise, the tasks will belong to a reactor that will never run again and are not stopped.
-			self.stop(false)
-			
-			@scheduler.close
-			@scheduler = nil
-		end
-		
-		# Check if the selector has been closed.
-		# @return [Boolean]
-		def closed?
-			@scheduler.nil?
-		end
-		
-		# Put the calling fiber to sleep for a given ammount of time.
-		# @param duration [Numeric] The time in seconds, to sleep for.
-		def sleep(duration)
-			@scheduler.kernel_sleep(duration)
+			if @closing
+				super
+			else
+				@closing = true
+				self.clear!
+			end
 		end
 
-		# Invoke the block, but after the specified timeout, raise {TimeoutError} in any currenly blocking operation. If the block runs to completion before the timeout occurs or there are no non-blocking operations after the timeout expires, the code will complete without any exception.
-		# @param duration [Numeric] The time in seconds, in which the task should 
-		#   complete.
-		def with_timeout(timeout, exception = TimeoutError)
-			fiber = Fiber.current
-			
-			timer = @scheduler.after(timeout) do
-				if fiber.alive?
-					fiber.raise(exception, "execution expired")
-				end
-			end
-			
-			yield timer
-		ensure
-			timer.cancel if timer
+		def with_timeout(timeout, exception = TimeoutError, message = "execution expired", &block)
+			timeout_after(timeout, exception, message, &block)
 		end
 	end
 end