Ruby 3: Non-blocking Fiber

class Scheduler
  # trigger by events: IO#wait, IO#wait_readable, IO#wait_writeable
  def io_wait(io, events, timeout)
  end

  # trigger by events: Kernel#sleep, Mutex#sleep
  def kernel_sleep(duration = nil)
  end

  # trigger by events: Process::Status.wait
  def process_wait(pid, flags)
  end

  # trigger by events: Thread#join, Mutex
  def block(blocker, timeout = nil)
  end

  # trigger when a previous block called is unblock
  def unblock(blocker, fiber)
  end
  
  # Called when current thread exits
  def close
  end
end

require 'fiber'

class SimpleScheduler
  def initialize
    # our internal Hash to track which Fiber is sleeping and for how long
    @waiting = {}
  end

  def run
    # we will loop until there is no more event
    # Our event loop for now will only check for sleeping fibers
    while @waiting.any?
      @waiting.keys.each do |fiber|
        # fiber needs to wake up
        if current_time > @waiting[fiber]
          @waiting.delete(fiber)
          fiber.resume
        end
      end
    end
  end

  def kernel_sleep(duration = nil)
    # this function(and other hooks) will run in context of the fiber calling sleep
    # hence Fiber.current will be our target fiber that need to be halted
    @waiting[Fiber.current] = current_time + duration
    # halt this fiber and transfer control to its parent
    Fiber.yield
    return true
  end

  def close
    run
  end

  private
  def current_time
    Process.clock_gettime(Process::CLOCK_MONOTONIC)
  end
end

scheduler = SimpleScheduler.new
Fiber.set_scheduler(scheduler)

# now we're ready to call sleep
# let's create two fiber
Fiber.new do
  puts "Fiber 1: sleep for 2s"
  sleep(2)
  puts "Fiber 1: wake up"
end.resume

Fiber.new do
  puts "Fiber 2: sleep for 3s"
  sleep(3)
  puts "Fiber 2: wake up"
end.resume

Fiber 1: Sleep 2s
Fiber 2: Sleep 3s
Fiber 1: wakeup
Fiber 2: wakeup

require 'net/http'
Fiber.new do
  Net::HTTP.get(URI('https://www.google.com'))
end.resume

class SimpleScheduler
  def initialize
    # our internal Hash to track which Fiber is waiting for a read event
    @readable = {}
    # our internal Hash to track which Fiber is waiting for a write event
    @writable = {}
    @waiting = {}
  end

  def io_wait(io, events, timeout)
    # events is a bit mask of IO::READABLE, IO::WRITABLE, and IO::PRIORITY.
    unless (events & IO::READABLE).zero?
      @readable[io] = Fiber.current
    end
  
    unless (events & IO::WRITABLE).zero?
      @writable[io] = Fiber.current
    end

    # small exercise: add IO::PRIORITY
  
    # halt fiber and transfer control to its parent
    Fiber.yield
    # when it's ready to read/write, resume fiber
    return events
  end
end

class SimpleScheduler
  def run
    while @readable.any? or @writable.any? or @waiting.any?
      # select's signature:
      # select(read_array [, write_array [, error_array [, timeout]]]) → array or nil
      readable, writable = IO.select(@readable.keys, @writable.keys, [], 0)

      # when there's an io that's ready
      # we remove it from our @readable/@writable and resume its corresponding fiber
      readable&.each do |io|
        fiber = @readable.delete(io)
        fiber.resume
      end

      writable&.each do |io|
        fiber = @writable.delete(io)
        fiber.resume
      end

      @waiting.keys.each do |fiber|
        if current_time > @waiting[fiber]
          @waiting.delete(fiber)
          fiber.resume
        end
      end
    end
  end
end

# our block's implementation is pretty simple
# we maintain @blocking variable, tracking how many fibers are blocked
# increase @blocking when there's block event
# decrease it when it's resumed
def block(blocker, timeout = nil)
  @blocking += 1
  begin
    Fiber.yield
  ensure
    @blocking -= 1
  end
end

# this is when previously block fiber is unblock
# we'll use an IO.pipe to notify event-loop that there's fiber unblocked
#  ie: @urgent = IO.pipe
# @ready is an array containing fiber that are unblocked
def unblock(blocker, fiber)
  @ready << fiber
  # we'll write to the pipe to notify event-loop
  io = @urgent.last
  io.write_nonblock('.')
end

# we also need to update our event-loop
def run
  # let's review what we're tracking:
  # @readable: for fibers that are waiting to read
  # @writable: for fibers that are waiting to write
  # @waiting: for fibers that are sleeping
  # @blocking: for fibers that are blocked
  # @ready: for fibers that are unblocked
  while @readable.any? or @writable.any? or @waiting.any? or @blocking.positive? or @ready.any?
    readable, writable = IO.select(@readable.keys + [@urgent.first], @writable.keys, [], 0)

    readable&.each do |io|
      if fiber = @readable.delete(io)
        fiber.resume
      end
    end

    writable&.each do |io|
      if fiber = @writable.delete(io)
        fiber.resume
      end
    end

    @waiting.keys.each do |fiber|
      if current_time > @waiting[fiber]
        @waiting.delete(fiber)
        fiber.resume
      end
    end

    # we simply resume all fibers in @ready
    ready, @ready = @ready, []
    ready.each do |fiber|
      fiber.resume
    end
  end
end