class Async::Scheduler

Handles scheduling of fibers. Implements the fiber scheduler interface.

def self.supported?

@public Since `stable-v1`.
Whether the fiber scheduler is supported.
def self.supported?
	true
end

def address_resolve(hostname)

@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

def async(*arguments, **options, &block)

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

def block(blocker, timeout)

@asynchronous May only be called on same thread as fiber scheduler.
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.
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

def close

@public Since `stable-v1`.
Terminate all child tasks and close the scheduler.
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

def closed?

@public Since `stable-v1`.
@returns [Boolean] Whether the scheduler has been closed.
def closed?
	@selector.nil?
end

def fiber(...)

def fiber(...)
	return async(...).fiber
end

def get_timeout(io)

def get_timeout(io)
out

def get_timeout(io)

def get_timeout(io)

def initialize(parent = nil, selector: nil)

@parameter selector [IO::Event::Selector] The selector to use for event handling.
@parameter parent [Node | Nil] The parent node to use for task hierarchy.
@public Since `stable-v1`.

Create a new scheduler.
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

def interrupt

@asynchronous May be called from any thread.
Interrupt the event loop and cause it to exit.
def interrupt
	@interrupted = true
	@selector&.wakeup
end

def interrupted?

@returns [Boolean] Whether the reactor has been interrupted.
Checks and clears the interrupted state of the scheduler.
def interrupted?
errupted
rupted = false
 true
ad.pending_interrupt?
 true
false

def io_read(io, buffer, length, offset = 0)

def io_read(io, buffer, length, offset = 0)
	fiber = Fiber.current
	
	if timeout = get_timeout(io)
		timer = @timers.after(timeout) do
			fiber.raise(::IO::TimeoutError, "Timeout while waiting for IO to become readable!")
		end
	end
	
	@selector.io_read(fiber, io, buffer, length, offset)
ensure
	timer&.cancel!
end

def io_wait(io, events, timeout = nil)

@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

def io_write(io, buffer, length, offset = 0)

def io_write(io, buffer, length, offset = 0)
	fiber = Fiber.current
	
	if timeout = get_timeout(io)
		timer = @timers.after(timeout) do
			fiber.raise(::IO::TimeoutError, "Timeout while waiting for IO to become writable!")
		end
	end
	
	@selector.io_write(fiber, io, buffer, length, offset)
ensure
	timer&.cancel!
end

def kernel_sleep(duration = nil)

@asynchronous May be non-blocking..
def kernel_sleep(duration = nil)
	if duration
		self.block(nil, duration)
	else
		self.transfer
	end
end

def load

@returns [Float] The load of the scheduler. 0.0 means no load, 1.0 means fully loaded or over-loaded.
Compute the scheduler load according to the busy and idle times that are updated by the run loop.
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 process_wait(pid, flags)

@asynchronous May be non-blocking..
@returns [Process::Status] A process status instance.
@parameter flags [Integer] A bit-mask of flags suitable for `Process::Status.wait`.
@parameter pid [Integer] The process ID to wait for.
Wait for the specified process ID to exit.
def process_wait(pid, flags)
	return @selector.process_wait(Fiber.current, pid, flags)
end

def push(fiber)

@parameter fiber [Fiber | Object] The object to be resumed on the next iteration of the run-loop.
Schedule a fiber (or equivalent object) to be resumed on the next loop through the reactor.
def push(fiber)
	@selector.push(fiber)
end

def raise(...)

@parameter *arguments [Array] The arguments to pass to the fiber.
@parameter fiber [Fiber] The fiber to raise the exception on.

This internally schedules the current fiber to be ready, before raising the exception, so that it will later resume execution.

Raise an exception on a specified fiber with the given arguments.
def raise(...)
	@selector.raise(...)
end

def resume(fiber, *arguments)

@parameter arguments [Array] The arguments to pass to the fiber.
@parameter fiber [Fiber] The fiber to resume.

Resume execution of the specified fiber.
def resume(fiber, *arguments)
	@selector.resume(fiber, *arguments)
end

def run(...)

Run the reactor until all tasks are finished. Proxies arguments to {#async} immediately before entering the loop, if a block is provided.
def run(...)
	Kernel::raise ClosedError if @selector.nil?
	
	initial_task = self.async(...) if block_given?
	interrupt = nil
	
	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 => interrupt
		Thread.handle_interrupt(::SignalException => :never) do
			self.stop
		end
		
		retry
	end
	
	# If the event loop was interrupted, and we finished exiting normally (due to the interrupt), we need to re-raise the interrupt so that the caller can handle it too.
	Kernel.raise interrupt if interrupt
	
	return initial_task
ensure
	Console.debug(self) {"Exiting run-loop because #{$! ? $! : 'finished'}."}
end

def run_once(timeout = nil)

@returns [Boolean] Whether there is more work to do.
@parameter timeout [Float | Nil] The maximum timeout, or if nil, indefinite.
Does not handle interrupts.
Run one iteration of the event loop.
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_once!(timeout = 0)

@returns [Boolean] Whether there is more work to do.
@parameter timeout [Float | Nil] The maximum timeout, or if nil, indefinite.

When terminating the event loop, we already know we are finished. So we don't need to check the task tree. This is a logical requirement because `run_once` ignores transient tasks. For example, a single top level transient task is not enough to keep the reactor running, but during termination we must still process it in order to terminate child tasks.

Run one iteration of the event loop.
def run_once!(timeout = 0)
ime = Async::Clock.now
l = @timers.wait_interval
ere is no interval to wait (thus no timers), and no tasks, we could be done:
rval.nil?
w the user to specify a maximum interval if we would otherwise be sleeping indefinitely:
al = timeout
nterval < 0
ave timers ready to fire, don't sleep in the selctor:
al = 0
imeout and interval > timeout
al = timeout
tor.select(interval)
Errno::EINTR
re.
.fire
te load:
e = Async::Clock.now
uration = end_time - start_time
ration = @selector.idle_duration
ration = total_duration - idle_duration
ime += busy_duration
ime += idle_duration
eactor still has work to do:
true

def scheduler_close

Executes the run loop until all tasks are finished, then closes the scheduler.

Invoked when the fiber scheduler is being closed.
def scheduler_close
	# If the execution context (thread) was handling an exception, we want to exit as quickly as possible:
	unless $!
		self.run
	end
ensure
	self.close
end

def terminate

Terminate the scheduler. We deliberately ignore interrupts here, as this code can be called from an interrupt, and we don't want to be interrupted while cleaning up.
def terminate
	Thread.handle_interrupt(::Interrupt => :never) do
		super
	end
end

def timeout_after(duration, exception, message, &block)

def timeout_after(duration, exception, message, &block)
	with_timeout(duration, exception, message) do |timer|
		yield duration
	end
end

def to_s

@returns [String] A description of the scheduler.
def to_s
	"\#<#{self.description} #{@children&.size || 0} children (#{stopped? ? 'stopped' : 'running'})>"
end

def transfer

Transfer from the calling fiber to the event loop.
def transfer
	@selector.transfer
end

def unblock(blocker, fiber)

@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

def with_timeout(duration, exception = TimeoutError, message = "execution expired", &block)

@parameter duration [Numeric] The time in seconds, in which the task should complete.
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.
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

def yield

Yield the current fiber and resume it on the next iteration of the event loop.
def yield
	@selector.yield
end