Event loop updates.

Author: ioquatix

async.gemspec

@@ -13,11 +13,11 @@ Gem::Specification.new do |spec|
 
 	spec.files = Dir.glob('{lib}/**/*', File::FNM_DOTMATCH, base: __dir__)
 
-	spec.required_ruby_version = ">= 3.1.0"
+	spec.required_ruby_version = ">= 3.0.2"
 
 	spec.add_dependency "console", "~> 1.10"
 
-	spec.add_dependency "event", "~> 0.5.0"
+	spec.add_dependency "event", "~> 0.9.3"
 	spec.add_dependency "timers", "~> 4.1"
 
 	spec.add_development_dependency "async-rspec", "~> 1.1"

lib/async/debug/selector.rb

@@ -27,7 +27,7 @@ module Async
 	module Debug
 		class Selector < Event::Debug::Selector
 			def initialize(selector = nil)
-				super(selector || Event::Backend.new(Fiber.current))
+				super(selector || Event::Selector.new(Fiber.current))
 			end
 		end
 	end

lib/async/scheduler.rb

@@ -36,19 +36,14 @@ def self.supported?
 		def initialize(parent = nil, selector: nil)
 			super(parent)
 
-			@selector = selector || Event::Backend.new(Fiber.current)
-			@timers = Timers::Group.new
-			
-			@ready = []
-			@running = []
+			@selector = selector || ::Event::Selector.new(Fiber.current)
+			@timers = ::Timers::Group.new
 
 			@guard = Mutex.new
 			@interrupted = false
 			@blocked = 0
 			@unblocked = []
 
-			@loop = nil
-			
 			@interrupt = Interrupt.new(@selector) do |event|
 				case event
 				when '!'
@@ -59,44 +54,47 @@ def initialize(parent = nil, selector: nil)
 
 		def set!
 			Fiber.set_scheduler(self)
-			@loop = Fiber.current
 		end
 
 		def clear!
 			Fiber.set_scheduler(nil)
-			@loop = nil
 		end
 
+		# Interrupt the event loop.
 		def interrupt
 			@interrupt.signal('!')
 		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
-			@ready << fiber
+		# Transfer from the calling fiber to the event loop.
+		def transfer
+			@selector.transfer
 		end
 
 		# Yield the current fiber and resume it on the next iteration of the event loop.
 		def yield
-			@ready << Fiber.current
-			@loop.transfer
+			@selector.yield
+		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 push(fiber)
+			@selector.push(fiber)
+		end
+		
+		alias << push
+		
+		def raise(*arguments)
+			@selector.raise(*arguments)
 		end
 
 		def resume(fiber, *arguments)
-			if @loop
-				@ready << Fiber.current
-				fiber.transfer(*arguments)
+			if Fiber.scheduler
+				@selector.resume(fiber, *arguments)
 			else
-				@ready << fiber
+				@selector.push(fiber)
 			end
 		end
 
-		# Transfer from te calling fiber to the event loop.
-		def transfer
-			@loop.transfer
-		end
-		
 		# Invoked when a fiber tries to perform a blocking operation which cannot continue. A corresponding call {unblock} must be performed to allow this fiber to continue.
 		# @reentrant Not thread safe.
 		def block(blocker, timeout)
@@ -113,7 +111,7 @@ def block(blocker, timeout)
 
 			begin
 				@blocked += 1
-				@loop.transfer
+				@selector.transfer
 			ensure
 				@blocked -= 1
 			end
@@ -159,33 +157,27 @@ def io_wait(io, events, timeout = nil)
 			timer&.cancel
 		end
 
+		# def io_read(io, buffer, length)
+		# 	@selector.io_read(Fiber.current, io, buffer, length)
+		# end
+		# 
+		# def io_write(io, buffer, length)
+		# 	@selector.io_write(Fiber.current, io, buffer, length)
+		# end
+		
 		# Wait for the specified process ID to exit.
 		# @parameter pid [Integer] The process ID to wait for.
 		# @parameter flags [Integer] A bit-mask of flags suitable for `Process::Status.wait`.
 		# @returns [Process::Status] A process status instance.
 		def process_wait(pid, flags)
-			fiber = Fiber.current
-			
-			return @selector.process_wait(fiber, pid, flags)
+			return @selector.process_wait(Fiber.current, pid, flags)
 		end
 
 		# Run one iteration of the event loop.
 		# @param timeout [Float | nil] the maximum timeout, or if nil, indefinite.
 		# @return [Boolean] whether there is more work to do.
 		def run_once(timeout = nil)
-			raise "Running scheduler on non-blocking fiber!" unless Fiber.blocking?
-			# Console.logger.info(self) {"@ready = #{@ready} @running = #{@running}"}
-			
-			if @ready.any?
-				# running used to correctly answer on `finished?`, and to reuse Array object.
-				@running, @ready = @ready, @running
-				
-				@running.each do |fiber|
-					fiber.transfer if fiber.alive?
-				end
-				
-				@running.clear
-			end
+			Kernel::raise "Running scheduler on non-blocking fiber!" unless Fiber.blocking?
 
 			if @unblocked.any?
 				unblocked = Array.new
@@ -199,7 +191,7 @@ def run_once(timeout = nil)
 				end
 			end
 
-			if @ready.empty? and @unblocked.empty?
+			if !@selector.ready? and @unblocked.empty?
 				interval = @timers.wait_interval
 			else
 				# if there are tasks ready to execute, don't sleep:
@@ -244,7 +236,7 @@ def run_once(timeout = nil)
 
 		# 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 @selector.nil?
+			Kernel::raise RuntimeError, 'Reactor has been closed' if @selector.nil?
 
 			initial_task = self.async(*arguments, **options, &block) if block_given?
 
@@ -261,7 +253,7 @@ 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.terminate
 
-			raise "Closing scheduler with blocked operations!" if @blocked > 0
+			Kernel::raise "Closing scheduler with blocked operations!" if @blocked > 0
 
 			@guard.synchronize do
 				@interrupt.close

lib/async/task.rb

@@ -190,8 +190,7 @@ def stop(later = false)
 					end
 				elsif @fiber&.alive?
 					begin
-						Fiber.scheduler << Fiber.current
-						@fiber.raise(Stop)
+						Fiber.scheduler.raise(@fiber, Stop)
 					rescue FiberError
 						Fiber.scheduler << Stop::Later.new(self)
 					end

spec/async/barrier_spec.rb

@@ -23,7 +23,6 @@
 require 'async/barrier'
 require 'async/clock'
 require 'async/rspec'
-
 require 'async/semaphore'
 
 require_relative 'chainable_async_examples'

spec/async/reactor_spec.rb

@@ -149,13 +149,12 @@
 
 			subject.async do |task|
 				events << true
+				thread.join
 			end
 
 			subject.run
 
-			puts "join"
-			thread.join
-			
+			expect(thread).to_not be_alive
 			expect(subject).to be_stopped
 		end
 	end

spec/async/scheduler/thread_spec.rb

@@ -24,6 +24,7 @@
 	include_context Async::RSpec::Reactor
 
 	describe ::Thread do
+		# I saw this hang.
 		it "can wait for value" do
 			value = Thread.new do
 				sleep(0)

spec/async/scheduler_spec.rb

@@ -116,8 +116,6 @@
 
 			expect(waiting).to be == 3
 			queue.close
-			
-			puts "Done."
 		end
 	end
 

spec/async/task_spec.rb

@@ -22,6 +22,7 @@
 
 require 'async'
 require 'async/clock'
+require 'async/queue'
 
 RSpec.describe Async::Task do
 	let(:reactor) {Async::Reactor.new}
@@ -229,10 +230,13 @@
 			reactor.run do
 				reactor.async do |task|
 					parent_task = task
+					
 					task.async do |task|
 						child_task = task
+						
 						task.sleep(10)
 					end
+					
 					task.sleep(10)
 				end
 
@@ -244,6 +248,9 @@
 
 				parent_task.stop
 
+				# We need to yield here to allow the tasks to be terminated. The parent task raises an exception in the child task and adds itself to the selector ready queue. It takes at least one iteration for the parent task to exit as well:
+				reactor.yield
+				
 				expect(parent_task).to_not be_alive
 				expect(child_task).to_not be_alive
 			end