Fibers and the Fiber Scheduler

# frozen_string_literal: true

# A fiber does not run until it is resumed.
fiber = Fiber.new do |first|
  puts "started with #{first.inspect}"
  second = Fiber.yield(first + 1)
  puts "resumed with #{second.inspect}"
  third = Fiber.yield(second * 2)
  puts "resumed again with #{third.inspect}"
  :finished
end

fiber.alive?     # => true
fiber.resume(10) # => 11
fiber.resume(20) # => 40
fiber.resume(30) # => :finished
fiber.alive?     # => false

begin
  fiber.resume
rescue FiberError => e
  e.message # => "attempt to resume a terminated fiber"
end

started with 10
resumed with 20
resumed again with 30

# frozen_string_literal: true

# Fiber#inspect ends with the state: "#<Fiber:0x... file.rb:5 (created)>".
state = ->(fiber) { fiber.inspect[/\((.+)\)>\z/, 1] }

fiber = Fiber.new { Fiber.yield 1 }
state.call(fiber)         # => "created"
fiber.resume              # => 1
state.call(fiber)         # => "suspended"
fiber.resume              # => nil
state.call(fiber)         # => "terminated"
state.call(Fiber.current) # => "resumed"
fiber.alive?              # => false

# frozen_string_literal: true

def message
  yield
rescue FiberError => e
  e.message
end

done = Fiber.new { 1 }
done.resume
message { done.resume } # => "attempt to resume a terminated fiber"
message { Fiber.yield } # => "attempt to yield on a not resumed fiber"

me = Fiber.new { me.resume }
message { me.resume } # => "attempt to resume the current fiber"

outer = Fiber.new { Fiber.new { outer.resume }.resume }
message { outer.resume } # => "attempt to resume a resumed fiber (double resume)"

suspended = Fiber.new { Fiber.yield }
suspended.resume
other_thread = Thread.new { message { suspended.resume } }
other_thread.value # => "fiber called across threads"

message { suspended.transfer } # => "attempt to transfer to a yielding fiber"
message { done.transfer } # => "dead fiber called"
message { Fiber.new { 1 }.raise("boom") } # => "cannot raise exception on unborn fiber"

# frozen_string_literal: true

def inner = Fiber.yield
def outer = inner

fiber = Fiber.new { outer }
fiber.backtrace # => []

fiber.resume
fiber.backtrace
# => ["fiber_backtrace.rb:3:in 'Fiber.yield'",
#     "fiber_backtrace.rb:3:in 'Object#inner'",
#     "fiber_backtrace.rb:4:in 'Object#outer'",
#     "fiber_backtrace.rb:6:in 'block in <main>'"]
fiber.backtrace(1, 2).first # => "fiber_backtrace.rb:3:in 'Object#inner'"

location = fiber.backtrace_locations.first
[location.label, location.lineno] # => ["Fiber.yield", 3]

fiber.resume
fiber.backtrace # => []

# frozen_string_literal: true

# How deep can a method recurse before SystemStackError?
def depth(level = 0)
  depth(level + 1)
rescue SystemStackError
  level
end

in_main = depth
in_thread = Thread.new { depth }.value
in_fiber = Fiber.new { depth }.resume

puts "main=#{in_main} thread=#{in_thread} fiber=#{in_fiber}"
puts in_fiber < in_main

$ ruby -w fiber_stack.rb
main=10917 thread=10917 fiber=1360
true
$ RUBY_FIBER_VM_STACK_SIZE=524288 ruby -w fiber_stack.rb
main=10917 thread=10917 fiber=5456
true

# frozen_string_literal: true

main = Fiber.current
Thread.current[:branch] = "Central" # fiber-local, despite the Thread receiver

where = Enumerator.new do |yielder|
  yielder << Fiber.current.equal?(main)
  yielder << Thread.current[:branch]
end

# Internal iteration (each, to_a, map ...) runs the block in the caller's fiber.
where.to_a # => [true, "Central"]

# External iteration (next, peek) runs it in a fiber of its own.
where.next # => false
where.next # => nil

# The end of the sequence raises StopIteration, which loop rescues.
begin
  where.next
rescue StopIteration => e
  e.message # => "iteration reached an end"
end

where.rewind.next # => false

# frozen_string_literal: true

require_relative "mini_scheduler"

# Without a scheduler, blocking and non-blocking fibers behave alike.
Fiber.scheduler                       # => nil
Fiber.current.blocking?               # => true
Fiber.blocking?                       # => 1
Fiber.new(blocking: false) { Fiber.current_scheduler }.resume # => nil

states = {}
Shelf::MiniScheduler.run do
  states[:thread_root] = [Fiber.blocking?, Fiber.current_scheduler.class]

  Fiber.schedule do
    states[:scheduled] = [Fiber.blocking?, Fiber.current_scheduler.class]
  end

  Fiber.new { states[:plain_new] = Fiber.blocking? }.resume
  Fiber.new(blocking: true) { states[:blocking_new] = [Fiber.blocking?, Fiber.current_scheduler] }.resume
  Fiber.blocking { states[:blocking_block] = [Fiber.blocking?, Fiber.current_scheduler] }
end

pp states

{thread_root: [1, NilClass],
 scheduled: [false, Shelf::MiniScheduler],
 plain_new: false,
 blocking_new: [1, nil],
 blocking_block: [1, nil]}

# frozen_string_literal: true

REQUIRED = %i[block unblock kernel_sleep io_wait].freeze

# Which hooks does Fiber.set_scheduler insist on?
REQUIRED.each do |missing|
  scheduler = Class.new do
    (REQUIRED - [missing]).each { |hook| define_method(hook) { |*| nil } }
    def fiber_interrupt(*); end
  end.new
  begin
    Fiber.set_scheduler(scheduler)
  rescue ArgumentError => e
    puts e.message
  end
end

# fiber, close, scheduler_close, io_read and the others are optional.
minimal = Class.new do
  REQUIRED.each { |hook| define_method(hook) { |*| nil } }
  def fiber_interrupt(*); end
end
Fiber.set_scheduler(minimal.new).respond_to?(:close) # => false
Fiber.set_scheduler(nil) # => nil

# Without a scheduler there is nothing for Fiber.schedule to call.
begin
  Fiber.schedule { 1 }
rescue RuntimeError => e
  e.message # => "No scheduler is available!"
end

Scheduler must implement #block
Scheduler must implement #unblock
Scheduler must implement #kernel_sleep
Scheduler must implement #io_wait

# frozen_string_literal: true

module Shelf
  # A toy Fiber::Scheduler: it understands sleeping, IO readiness, blocking
  # on a lock or a queue, and Timeout. It is for learning, not for production.
  class MiniScheduler
    def initialize
      @ready = []   # fibers woken by #unblock, possibly from another thread
      @timers = {}  # fiber => deadline of a blocked fiber (Infinity: none)
      @io = {}      # io => [fiber, events] for fibers waiting on a descriptor
      @limits = {}  # fiber => [deadline, klass, message] set by Timeout
      @interrupts = {} # fiber => exception to raise when it wakes
      @lock = Thread::Mutex.new
      @wake_reader, @wake_writer = IO.pipe
    end

    # Runs the block in a fresh thread that owns a scheduler. When the block
    # returns, the thread exits and Ruby calls #close, which runs the loop.
    def self.run
      Thread.new do
        Fiber.set_scheduler(new)
        yield
      end.join
    end

    # Called by Fiber.schedule: create the fiber and start it right away.
    def fiber(&)
      Fiber.new(blocking: false, &).tap(&:resume)
    end

    def kernel_sleep(duration = nil) = block(:sleep, duration)

    def block(_blocker, timeout = nil)
      @timers[Fiber.current] = timeout ? now + timeout : Float::INFINITY
      Fiber.yield
    end

    # Thread safe: it may be called from any thread, so it uses the lock
    # and wakes the event loop through a pipe.
    def unblock(_blocker, fiber)
      @lock.synchronize { @ready << fiber }
      @wake_writer.write_nonblock(".")
    end

    # Called, for example, by IO#close for fibers blocked on that IO.
    def fiber_interrupt(fiber, exception)
      @lock.synchronize do
        @interrupts[fiber] = exception
        @ready << fiber
      end
      @wake_writer.write_nonblock(".")
    end

    def io_wait(io, events, timeout)
      @io[io] = [Fiber.current, events]
      @timers[Fiber.current] = timeout ? now + timeout : Float::INFINITY
      Fiber.yield
    end

    # Waiting for a child process runs in a helper thread; Thread#value
    # blocks this fiber through #block and #unblock.
    def process_wait(pid, flags)
      Thread.new { Process::Status.wait(pid, flags) }.value
    end

    def timeout_after(duration, klass, message, &)
      @limits[Fiber.current] = [now + duration, klass, message]
      yield duration
    ensure
      @limits.delete(Fiber.current)
    end

    # Called when the thread exits: keep running until nothing is left.
    def close
      run
    ensure
      @wake_reader.close
      @wake_writer.close
    end

    def run
      step until idle?
    end

    private

    def now = Process.clock_gettime(Process::CLOCK_MONOTONIC)

    def idle? = @timers.empty? && @lock.synchronize { @ready.empty? }

    def step
      wait_for_events
      @lock.synchronize { @ready.slice!(0..) }.each { |fiber| wake(fiber) }
      @timers.select { |_, at| at <= now }.each_key { |fiber| wake(fiber) }
      @limits.select { |_, (at, *)| at <= now }.each { |fiber, (_, klass, message)| expire(fiber, klass, message) }
    end

    def next_deadline = [*@timers.values, *@limits.values.map(&:first)].min || Float::INFINITY

    def wait_for_events
      @io.delete_if { |io, _| io.closed? } # closed by IO#close: see fiber_interrupt
      timeout = [next_deadline - now, 0].max
      readable, = IO.select(@io.keys + [@wake_reader], nil, nil, timeout)
      readable&.each { |io| io == @wake_reader ? @wake_reader.read_nonblock(64) : wake_io(io) }
    end

    def wake_io(io)
      fiber, events = @io[io]
      wake(fiber, events)
    end

    def wake(fiber, value = nil)
      forget(fiber)
      return unless fiber.alive?

      exception = @lock.synchronize { @interrupts.delete(fiber) }
      exception ? fiber.raise(exception) : fiber.resume(value)
    end

    def expire(fiber, klass, message)
      forget(fiber)
      @limits.delete(fiber)
      fiber.raise(klass, message)
    end

    def forget(fiber)
      @timers.delete(fiber)
      @io.delete_if { |_, (waiting, _)| waiting == fiber }
    end
  end
end

# frozen_string_literal: true

require_relative "mini_scheduler"

def timed
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  yield
  (Process.clock_gettime(Process::CLOCK_MONOTONIC) - started).round(1)
end

lookup = lambda do |isbn|
  sleep 0.2 # stands in for a slow HTTP request
  "record for #{isbn}"
end

isbns = %w[9780131103627 9780596516178 9781492078005]

sequential = timed { isbns.map(&lookup) }

records = []
concurrent = timed do
  Shelf::MiniScheduler.run do
    isbns.each { |isbn| Fiber.schedule { records << lookup.call(isbn) } }
  end
end

[sequential, concurrent] # => [0.6, 0.2]
records.size             # => 3

# frozen_string_literal: true

require_relative "mini_scheduler"

def timed
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  yield
  (Process.clock_gettime(Process::CLOCK_MONOTONIC) - started).round(1)
end

# Fibers resume in the order their sleeps end, not the order they started.
events = []
Shelf::MiniScheduler.run do
  { "slow" => 0.3, "fast" => 0.1, "medium" => 0.2 }.each do |name, seconds|
    Fiber.schedule do
      sleep seconds
      events << name
    end
  end
  events << "all scheduled"
end
events.join(", ") # => "all scheduled, fast, medium, slow"

# A blocking fiber does not give control to the scheduler: its sleep is a
# plain, thread-blocking sleep and the other fibers wait for it.
polite = timed { Shelf::MiniScheduler.run { 2.times { Fiber.schedule { sleep 0.2 } } } }
rude = timed { Shelf::MiniScheduler.run { 2.times { Fiber.schedule { Fiber.blocking { sleep 0.2 } } } } }
[polite, rude] # => [0.2, 0.4]

# A child process is awaited through the process_wait hook, so these overlap.
processes = timed { Shelf::MiniScheduler.run { 3.times { Fiber.schedule { system("sleep 0.2") } } } }
processes.between?(0.2, 0.3) # => true

# frozen_string_literal: true

require "timeout"
require_relative "mini_scheduler"

# Records which hooks Ruby calls, in order.
class TracingScheduler < Shelf::MiniScheduler
  attr_reader :calls

  def initialize
    super
    @calls = []
  end

  %i[kernel_sleep block unblock io_wait timeout_after].each do |hook|
    define_method(hook) do |*args, &block|
      @calls << hook
      super(*args, &block)
    end
  end
end

# Runs the block in a scheduled fiber and returns the hooks it triggered.
def trace(&)
  scheduler = TracingScheduler.new
  Thread.new do
    Fiber.set_scheduler(scheduler)
    Fiber.schedule(&)
  end.join
  scheduler.calls.tally
end

reader, writer = IO.pipe

trace { sleep 0.01 } # => {kernel_sleep: 1, block: 1}
trace { Thread::Queue.new.pop(timeout: 0.01) } # => {block: 1}
trace { Thread.new { writer.write("x") }.then { reader.wait_readable } } # => {io_wait: 1}
trace { Thread.new { sleep 0.01 }.join } # => {block: 1, unblock: 1}

expiring = lambda do
  Timeout.timeout(0.01) { sleep }
rescue Timeout::Error
  :expired
end
trace(&expiring) # => {timeout_after: 1, kernel_sleep: 1, block: 1}

# frozen_string_literal: true

require "timeout"
require_relative "mini_scheduler"

outcome = nil
started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
Shelf::MiniScheduler.run do
  Fiber.schedule do
    outcome = Timeout.timeout(0.05) { sleep 5 }
  rescue Timeout::Error => e
    outcome = e.message
  end
end
elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started

[outcome, elapsed < 1] # => ["execution expired", true]

# frozen_string_literal: true

require_relative "mini_scheduler"

outcome = []
Shelf::MiniScheduler.run do
  reader, = IO.pipe

  Fiber.schedule do
    reader.wait_readable # nothing is ever written: the fiber waits in io_wait
  rescue IOError => e
    outcome << "reader: #{e.message}"
  end

  Fiber.schedule do
    sleep 0.05
    reader.close   # Ruby 4.0 calls scheduler.fiber_interrupt for the waiting fiber
    outcome << "closer: done"
  end
end
outcome.join("; ") # => "reader: stream closed in another thread; closer: done"

# frozen_string_literal: true

require_relative "mini_scheduler"

def message
  yield
rescue ThreadError => e
  e.message
end

# Since Ruby 3.0 a Mutex belongs to a fiber, not to a thread.
lock = Thread::Mutex.new
holder = Fiber.new do
  lock.lock
  Fiber.yield
  lock.unlock
end
holder.resume
lock.owned?                  # => false
lock.locked?                 # => true
message { lock.lock }        # => "deadlock; lock already owned by another fiber belonging to the same thread"
message { lock.unlock }      # => "Attempt to unlock a mutex which is locked by another thread/fiber"
holder.resume
lock.locked?                 # => false

# Under a scheduler, a fiber that waits for a lock yields through #block.
order = []
Shelf::MiniScheduler.run do
  Fiber.schedule do
    lock.synchronize do
      sleep 0.1
      order << "first"
    end
  end
  Fiber.schedule { lock.synchronize { order << "second" } }
end
order.join(", ") # => "first, second"

# frozen_string_literal: true

require_relative "mini_scheduler"

# Ruby 4.0 asks every scheduler to implement fiber_interrupt.
bare = Class.new do
  %i[block unblock kernel_sleep io_wait].each { |hook| define_method(hook) { |*| nil } }
end
Fiber.set_scheduler(bare.new)
Fiber.set_scheduler(nil)

# Unbuffered file or socket I/O from a non-blocking fiber builds an IO::Buffer,
# which is still marked experimental in Ruby 4.0.
Shelf::MiniScheduler.run do
  Fiber.schedule { File.read(__FILE__) }
end

$ ruby -w sched_warnings.rb
sched_warnings.rb:9: warning: Scheduler should implement #fiber_interrupt
sched_warnings.rb:15: warning: IO::Buffer is experimental and both the Ruby and C interface may change in the future!
$ ruby -W:no-experimental sched_warnings.rb
sched_warnings.rb:9: warning: Scheduler should implement #fiber_interrupt