class Scheduler

public

Since stable-v1.

def self.supported?
	true
end

public

Since stable-v1.

parameter　parent　Node | Nil

The parent node to use for task hierarchy.

parameter　selector　IO::Event::Selector

The selector to use for event handling.

def initialize(parent = nil, selector: nil)
	super(parent)
	
	@selector = selector || ::IO::Event::Selector.new(Fiber.current)
	@interrupted = false
	
	@blocked = 0
	
	@busy_time = 0.0
	@idle_time = 0.0
	
	@timers = ::IO::Event::Timers.new
end

returns　Float

The load of the scheduler. 0.0 means no load, 1.0 means fully loaded or over-loaded.

def load
	total_time = @busy_time + @idle_time
	
	# If the total time is zero, then the load is zero:
	return 0.0 if total_time.zero?
	
	# We normalize to a 1 second window:
	if total_time > 1.0
		ratio = 1.0 / total_time
		@busy_time *= ratio
		@idle_time *= ratio
		
		# We don't need to divide here as we've already normalised it to a 1s window:
		return @busy_time
	else
		return @busy_time / total_time
	end
end

def scheduler_close(error = $!)
	# If the execution context (thread) was handling an exception, we want to exit as quickly as possible:
	unless error
		self.run
	end
ensure
	self.close
end

def terminate
	# If that doesn't work, take more serious action:
	@children&.each do |child|
		child.terminate
	end
	
	return @children.nil?
end

public

Since stable-v1.

def close
	self.run_loop do
		until self.terminate
			self.run_once!
		end
	end
	
	Kernel.raise "Closing scheduler with blocked operations!" if @blocked > 0
ensure
	# We want `@selector = nil` to be a visible side effect from this point forward, specifically in `#interrupt` and `#unblock`. If the selector is closed, then we don't want to push any fibers to it.
	selector = @selector
	@selector = nil
	
	selector&.close
	
	consume
end

returns　Boolean

Whether the scheduler has been closed.

public

Since stable-v1.

def closed?
	@selector.nil?
end

returns　String

A description of the scheduler.

def to_s
	"\#<#{self.description} #{@children&.size || 0} children (#{stopped? ? 'stopped' : 'running'})>"
end

asynchronous

May be called from any thread.

def interrupt
	@interrupted = true
	@selector&.wakeup
end

def transfer
	@selector.transfer
end

def yield
	@selector.yield
end

parameter　fiber　Fiber | Object

The object to be resumed on the next iteration of the run-loop.

def push(fiber)
	@selector.push(fiber)
end

parameter　fiber　Fiber

The fiber to raise the exception on.

parameter　*arguments　Array

The arguments to pass to the fiber.

def raise(...)
	@selector.raise(...)
end

parameter　fiber　Fiber

The fiber to resume.

parameter　arguments　Array

The arguments to pass to the fiber.

def resume(fiber, *arguments)
	@selector.resume(fiber, *arguments)
end

asynchronous

May only be called on same thread as fiber scheduler.

def block(blocker, timeout)
	# $stderr.puts "block(#{blocker}, #{Fiber.current}, #{timeout})"
	fiber = Fiber.current
	
	if timeout
		timer = @timers.after(timeout) do
			if fiber.alive?
				fiber.transfer(false)
			end
		end
	end
	
	begin
		@blocked += 1
		@selector.transfer
	ensure
		@blocked -= 1
	end
ensure
	timer&.cancel!
end

asynchronous

May be called from any thread.

def unblock(blocker, fiber)
	# $stderr.puts "unblock(#{blocker}, #{fiber})"
	
	# This operation is protected by the GVL:
	if selector = @selector
		selector.push(fiber)
		selector.wakeup
	end
end

asynchronous

May be non-blocking..

def kernel_sleep(duration = nil)
	if duration
		self.block(nil, duration)
	else
		self.transfer
	end
end

asynchronous

May be non-blocking..

def address_resolve(hostname)
	# On some platforms, hostnames may contain a device-specific suffix (e.g. %en0). We need to strip this before resolving.
	# See <https://github.com/socketry/async/issues/180> for more details.
	hostname = hostname.split("%", 2).first
	::Resolv.getaddresses(hostname)
end

asynchronous

May be non-blocking..

def io_wait(io, events, timeout = nil)
	fiber = Fiber.current
	
	if timeout
		# If an explicit timeout is specified, we expect that the user will handle it themselves:
		timer = @timers.after(timeout) do
			fiber.transfer
		end
	elsif timeout = get_timeout(io)
		# Otherwise, if we default to the io's timeout, we raise an exception:
		timer = @timers.after(timeout) do
			fiber.raise(::IO::TimeoutError, "Timeout (#{timeout}s) while waiting for IO to become ready!")
		end
	end
	
	return @selector.io_wait(fiber, io, events)
ensure
	timer&.cancel!
end

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.

asynchronous

May be non-blocking..

def process_wait(pid, flags)
	return @selector.process_wait(Fiber.current, pid, flags)
end

parameter　timeout　Float | Nil

The maximum timeout, or if nil, indefinite.

returns　Boolean

Whether there is more work to do.

def run_once(timeout = nil)
	Kernel.raise "Running scheduler on non-blocking fiber!" unless Fiber.blocking?
	
	if self.finished?
		self.stop
	end
	
	# If we are finished, we stop the task tree and exit:
	if @children.nil?
		return false
	end
	
	return run_once!(timeout)
end

def stop
	@children&.each do |child|
		child.stop
	end
end

def run(...)
	Kernel.raise ClosedError if @selector.nil?
	
	initial_task = self.async(...) if block_given?
	
	self.run_loop do
		run_once
	end
	
	return initial_task
end

yields　{|task| ...}

Executed within the task.

returns　Task

The task that was scheduled into the reactor.

deprecated

With no replacement.

def async(*arguments, **options, &block)
	Kernel.raise ClosedError if @selector.nil?
	
	task = Task.new(Task.current? || self, **options, &block)
	
	# I want to take a moment to explain the logic of this.
	# When calling an async block, we deterministically execute it until the
	# first blocking operation. We don't *have* to do this - we could schedule
	# it for later execution, but it's useful to:
	# - Fail at the point of the method call where possible.
	# - Execute determinstically where possible.
	# - Avoid scheduler overhead if no blocking operation is performed.
	task.run(*arguments)
	
	# Console.debug "Initial execution of task #{fiber} complete (#{result} -> #{fiber.alive?})..."
	return task
end

parameter　duration　Numeric

The time in seconds, in which the task should complete.

def with_timeout(duration, exception = TimeoutError, message = "execution expired", &block)
	fiber = Fiber.current
	
	timer = @timers.after(duration) do
		if fiber.alive?
			fiber.raise(exception, message)
		end
	end
	
	yield timer
ensure
	timer&.cancel!
end