class Scheduler

public

Since stable-v1.

def self.supported?
	true
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 terminate
	Thread.handle_interrupt(::Interrupt => :never) do
		super
	end
end

public

Since stable-v1.

def close
	# It's critical to stop all tasks. Otherwise they might be holding on to resources which are never closed/released correctly.
	until self.terminate
		self.run_once!
	end
	
	Kernel.raise "Closing scheduler with blocked operations!" if @blocked > 0
	
	# We depend on GVL for consistency:
	# @guard.synchronize do
	
	# 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
	
	# end
	
	consume
end

returns　Boolean

Whether the scheduler has been closed.

public

Since stable-v1.

def closed?
	@selector.nil?
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

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 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 we are finished, we stop the task tree and exit:
	if self.finished?
		return false
	end
	
	return run_once!(timeout)
end

def run(...)
	Kernel::raise ClosedError if @selector.nil?
	
	initial_task = self.async(...) if block_given?
	
	begin
		# In theory, we could use Exception here to be a little bit safer, but we've only shown the case for SignalException to be a problem, so let's not over-engineer this.
		Thread.handle_interrupt(::SignalException => :never) do
			while true
				# If we are interrupted, we need to exit:
				break if self.interrupted?
				
				# If we are finished, we need to exit:
				break unless self.run_once
			end
		end
	rescue Interrupt
		Thread.handle_interrupt(::SignalException => :never) do
			self.stop
		end
		
		retry
	end
		
	return initial_task
ensure
	Console.logger.debug(self) {"Exiting run-loop because #{$! ? $! : 'finished'}."}
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.logger.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 if timer
end