Fibers & Cooperative Multitasking

Creating and Managing Fibers

class FiberBasics
  def self.simple_fiber_demo
    puts "=== Basic Fiber Example ==="

    # Create a fiber
    fiber = Fiber.new do
      puts "Fiber started"
      Fiber.yield "First yield"
      puts "Fiber resumed"
      Fiber.yield "Second yield"
      puts "Fiber finishing"
      "Final result"
    end

    puts "Main: Created fiber"

    # Resume the fiber
    result1 = fiber.resume
    puts "Main: Got #{result1}"

    result2 = fiber.resume
    puts "Main: Got #{result2}"

    result3 = fiber.resume
    puts "Main: Got #{result3}"

    puts "Fiber alive? #{fiber.alive?}"
  end

  def self.fibonacci_generator
    puts "\n=== Fibonacci Generator ==="

    fibonacci = Fiber.new do
      a, b = 0, 1
      loop do
        Fiber.yield a
        a, b = b, a + b
      end
    end

    puts "First 10 Fibonacci numbers:"
    10.times do |i|
      puts "fib(#{i}) = #{fibonacci.resume}"
    end
  end

  def self.parameter_passing_demo
    puts "\n=== Parameter Passing ==="

    calculator = Fiber.new do |operation|
      result = 0
      loop do
        case operation
        when 'add'
          x = Fiber.yield "Enter first number:"
          y = Fiber.yield "Enter second number:"
          result = x + y
          operation = Fiber.yield "Result: #{result}"
        when 'multiply'
          x = Fiber.yield "Enter first number:"
          y = Fiber.yield "Enter second number:"
          result = x * y
          operation = Fiber.yield "Result: #{result}"
        when 'quit'
          break
        else
          operation = Fiber.yield "Unknown operation: #{operation}"
        end
      end
      "Calculator finished"
    end

    puts calculator.resume('add')      # Enter first number:
    puts calculator.resume(5)          # Enter second number:
    puts calculator.resume(3)          # Result: 8

    puts calculator.resume('multiply') # Enter first number:
    puts calculator.resume(4)          # Enter second number:
    puts calculator.resume(7)          # Result: 28

    puts calculator.resume('quit')     # Calculator finished
  end

  def self.fiber_state_demo
    puts "\n=== Fiber States ==="

    fiber = Fiber.new do
      puts "Fiber executing"
      Fiber.yield "Paused"
      puts "Fiber resumed"
      "Completed"
    end

    puts "Before resume: alive? #{fiber.alive?}"

    result = fiber.resume
    puts "After first resume: #{result}, alive? #{fiber.alive?}"

    result = fiber.resume
    puts "After second resume: #{result}, alive? #{fiber.alive?}"

    begin
      fiber.resume  # This will raise FiberError
    rescue FiberError => e
      puts "Error: #{e.message}"
    end
  end
end

FiberBasics.simple_fiber_demo
FiberBasics.fibonacci_generator
FiberBasics.parameter_passing_demo
FiberBasics.fiber_state_demo

💡 Fiber vs Thread Key Differences

• Scheduling: Fibers are cooperative (explicit yield), threads are preemptive
• Memory: Fibers are much lighter (~4KB vs ~8MB per thread stack)
• Synchronization: No locks needed with fibers (single-threaded)
• Control: Complete control over when execution switches

Enumerator Implementation with Fibers

class CustomEnumerators
  def self.tree_traversal_demo
    # Binary tree node
    Node = Struct.new(:value, :left, :right)

    # Build a sample tree
    #       1
    #      / \
    #     2   3
    #    / \
    #   4   5
    root = Node.new(1,
      Node.new(2,
        Node.new(4),
        Node.new(5)
      ),
      Node.new(3)
    )

    # In-order traversal using fiber
    def self.inorder_traversal(node)
      Fiber.new do
        traverse_inorder(node) { |value| Fiber.yield value }
      end
    end

    def self.traverse_inorder(node, &block)
      return unless node

      traverse_inorder(node.left, &block)
      block.call(node.value)
      traverse_inorder(node.right, &block)
    end

    puts "=== Tree Traversal with Fibers ==="
    traversal = inorder_traversal(root)

    values = []
    while traversal.alive?
      values << traversal.resume
    end

    puts "In-order traversal: #{values.compact}"
  end

  def self.lazy_sequence_demo
    puts "\n=== Lazy Infinite Sequences ==="

    # Prime number generator
    def self.prime_generator
      Fiber.new do
        primes = []
        candidate = 2

        loop do
          is_prime = primes.none? { |p| candidate % p == 0 }

          if is_prime
            primes << candidate
            Fiber.yield candidate
          end

          candidate += 1
        end
      end
    end

    primes = prime_generator
    puts "First 15 prime numbers:"
    15.times do |i|
      puts "Prime #{i + 1}: #{primes.resume}"
    end
  end

  def self.data_processing_pipeline
    puts "\n=== Data Processing Pipeline ==="

    # Source: generates data
    def self.data_source(count)
      Fiber.new do
        count.times do |i|
          Fiber.yield "data_#{i}"
        end
      end
    end

    # Filter: processes data
    def self.data_filter(source)
      Fiber.new do
        while source.alive?
          data = source.resume
          next unless data

          # Only process even-numbered data
          if data.end_with?('0', '2', '4', '6', '8')
            Fiber.yield "filtered_#{data}"
          end
        end
      end
    end

    # Transform: modifies data
    def self.data_transformer(source)
      Fiber.new do
        while source.alive?
          data = source.resume
          next unless data

          Fiber.yield data.upcase
        end
      end
    end

    # Build pipeline
    source = data_source(10)
    filtered = data_filter(source)
    transformed = data_transformer(filtered)

    puts "Pipeline results:"
    while transformed.alive?
      result = transformed.resume
      puts result if result
    end
  end
end

CustomEnumerators.tree_traversal_demo
CustomEnumerators.lazy_sequence_demo
CustomEnumerators.data_processing_pipeline

Cooperative Multitasking Scheduler

class CooperativeScheduler
  def initialize
    @tasks = []
    @current_task = nil
  end

  def add_task(name, &block)
    fiber = Fiber.new do |scheduler|
      begin
        puts "Task #{name} starting"
        block.call(scheduler)
        puts "Task #{name} completed"
      rescue => e
        puts "Task #{name} failed: #{e.message}"
      end
    end

    @tasks << { name: name, fiber: fiber, priority: 1 }
  end

  def yield_control(reason = "yielding")
    puts "  Task #{@current_task[:name]} #{reason}"
    Fiber.yield
  end

  def sleep(duration)
    puts "  Task #{@current_task[:name]} sleeping for #{duration}s"
    # In a real implementation, this would integrate with an event loop
    # For demo purposes, we'll just yield
    Fiber.yield
  end

  def run
    puts "=== Cooperative Scheduler Demo ==="

    while @tasks.any? { |task| task[:fiber].alive? }
      @tasks.each do |task|
        next unless task[:fiber].alive?

        @current_task = task
        puts "Running task: #{task[:name]}"

        begin
          task[:fiber].resume(self)
        rescue FiberError
          puts "Task #{task[:name]} finished"
        end
      end

      # Remove completed tasks
      @tasks.reject! { |task| !task[:fiber].alive? }
    end

    puts "All tasks completed"
  end

  def self.demo
    scheduler = new

    # Add cooperative tasks
    scheduler.add_task("FileProcessor") do |sched|
      3.times do |i|
        puts "    Processing file #{i + 1}"
        sched.yield_control("processed file #{i + 1}")
      end
    end

    scheduler.add_task("NetworkClient") do |sched|
      2.times do |i|
        puts "    Making network request #{i + 1}"
        sched.sleep(0.1)
        puts "    Request #{i + 1} completed"
        sched.yield_control("finished request #{i + 1}")
      end
    end

    scheduler.add_task("DataAnalyzer") do |sched|
      4.times do |i|
        puts "    Analyzing data chunk #{i + 1}"
        sched.yield_control("analyzed chunk #{i + 1}")
      end
    end

    scheduler.run
  end
end

CooperativeScheduler.demo

Event Loop Implementation

class SimpleEventLoop
  def initialize
    @ready_queue = []
    @waiting_tasks = {}
    @running = false
    @task_id = 0
  end

  def schedule_task(&block)
    task_id = (@task_id += 1)
    fiber = Fiber.new do
      begin
        result = block.call(self)
        puts "Task #{task_id} completed with result: #{result}"
      rescue => e
        puts "Task #{task_id} failed: #{e.message}"
      end
    end

    @ready_queue << { id: task_id, fiber: fiber }
    task_id
  end

  def sleep(duration)
    task_id = @current_task[:id]
    wake_time = Time.now + duration

    puts "Task #{task_id} sleeping for #{duration}s"
    @waiting_tasks[wake_time] = @current_task

    Fiber.yield  # Give up control
  end

  def async_operation(name, duration)
    puts "Starting async operation: #{name}"
    sleep(duration)
    puts "Completed async operation: #{name}"
    "#{name}_result"
  end

  def run
    @running = true
    puts "=== Event Loop Starting ==="

    while @running && (@ready_queue.any? || @waiting_tasks.any?)
      # Process sleeping tasks
      current_time = Time.now
      @waiting_tasks.keys.select { |wake_time| wake_time <= current_time }.each do |wake_time|
        task = @waiting_tasks.delete(wake_time)
        @ready_queue << task
        puts "Task #{task[:id]} woke up"
      end

      # Process ready tasks
      if @ready_queue.any?
        @current_task = @ready_queue.shift
        fiber = @current_task[:fiber]

        if fiber.alive?
          puts "Resuming task #{@current_task[:id]}"
          fiber.resume

          # If task yielded but didn't sleep, put it back in queue
          if fiber.alive? && !@waiting_tasks.values.include?(@current_task)
            @ready_queue << @current_task
          end
        end
      else
        # No ready tasks, sleep briefly
        sleep(0.01)
      end

      # Stop if no more tasks
      @running = false if @ready_queue.empty? && @waiting_tasks.empty?
    end

    puts "Event loop finished"
  end

  def stop
    @running = false
  end

  def self.demo
    loop = new

    # Schedule async tasks
    loop.schedule_task do |event_loop|
      puts "Task A: Starting"
      result1 = event_loop.async_operation("download", 0.1)
      result2 = event_loop.async_operation("process", 0.05)
      puts "Task A: Got results: #{result1}, #{result2}"
      "Task A complete"
    end

    loop.schedule_task do |event_loop|
      puts "Task B: Starting"
      3.times do |i|
        event_loop.async_operation("step_#{i + 1}", 0.03)
      end
      puts "Task B: All steps complete"
      "Task B complete"
    end

    loop.schedule_task do |event_loop|
      puts "Task C: Quick task"
      "Task C complete"
    end

    loop.run
  end
end

SimpleEventLoop.demo

Async/Await Pattern with Fibers

class AsyncAwait
  # Promise-like object
  class Promise
    def initialize(&block)
      @fiber = Fiber.new(&block)
      @resolved = false
      @value = nil
      @error = nil
    end

    def then(&block)
      if @resolved
        if @error
          raise @error
        else
          Promise.new { block.call(@value) }
        end
      else
        Promise.new do
          begin
            @value = @fiber.resume
            @resolved = true
            block.call(@value)
          rescue => e
            @error = e
            @resolved = true
            raise e
          end
        end
      end
    end

    def await
      return @value if @resolved && !@error
      raise @error if @resolved && @error

      begin
        @value = @fiber.resume
        @resolved = true
        @value
      rescue => e
        @error = e
        @resolved = true
        raise e
      end
    end

    def self.resolve(value)
      Promise.new { value }
    end

    def self.reject(error)
      Promise.new { raise error }
    end
  end

  def self.async_fetch(url, delay = 0.1)
    Promise.new do
      puts "Fetching #{url}..."
      sleep(delay)  # Simulate network delay
      if url.include?('error')
        raise "Failed to fetch #{url}"
      else
        "Data from #{url}"
      end
    end
  end

  def self.async_process(data, delay = 0.05)
    Promise.new do
      puts "Processing #{data}..."
      sleep(delay)
      "Processed: #{data}"
    end
  end

  def self.demo
    puts "=== Async/Await Pattern Demo ==="

    # Example 1: Basic async/await
    begin
      promise = async_fetch("https://api.example.com/users")
      data = promise.await
      puts "Received: #{data}"

      processed = async_process(data).await
      puts "Final result: #{processed}"
    rescue => e
      puts "Error: #{e.message}"
    end

    puts "\n--- Chaining Promises ---"

    # Example 2: Promise chaining
    begin
      result = async_fetch("https://api.example.com/posts")
                .then { |data| async_process(data).await }
                .await

      puts "Chained result: #{result}"
    rescue => e
      puts "Chained error: #{e.message}"
    end

    puts "\n--- Error Handling ---"

    # Example 3: Error handling
    begin
      error_result = async_fetch("https://api.error.com/data").await
    rescue => e
      puts "Caught error: #{e.message}"
    end

    puts "\n--- Multiple Async Operations ---"

    # Example 4: Multiple async operations
    start_time = Time.now

    promises = [
      async_fetch("https://api.example.com/users", 0.1),
      async_fetch("https://api.example.com/posts", 0.08),
      async_fetch("https://api.example.com/comments", 0.12)
    ]

    results = promises.map do |promise|
      begin
        promise.await
      rescue => e
        "Error: #{e.message}"
      end
    end

    end_time = Time.now
    puts "All operations completed in #{(end_time - start_time).round(3)}s"
    results.each_with_index do |result, i|
      puts "Result #{i + 1}: #{result}"
    end
  end
end

AsyncAwait.demo

Advanced Generator Patterns

class AdvancedGenerators
  def self.file_line_generator(filename)
    Fiber.new do
      begin
        File.open(filename, 'r') do |file|
          line_number = 0
          file.each_line do |line|
            line_number += 1
            Fiber.yield({ number: line_number, content: line.chomp })
          end
        end
      rescue Errno::ENOENT
        Fiber.yield({ error: "File not found: #{filename}" })
      end
    end
  end

  def self.csv_parser_generator(csv_content)
    Fiber.new do
      lines = csv_content.split("\n")
      headers = lines.first&.split(',')

      return unless headers

      lines[1..-1].each_with_index do |line, index|
        values = line.split(',')
        row = headers.zip(values).to_h
        Fiber.yield({ row_number: index + 2, data: row })
      end
    end
  end

  def self.recursive_directory_generator(path)
    Fiber.new do
      def self.traverse(current_path, depth = 0)
        return unless Dir.exist?(current_path)

        Dir.entries(current_path).each do |entry|
          next if entry == '.' || entry == '..'

          full_path = File.join(current_path, entry)

          if File.directory?(full_path)
            Fiber.yield({
              type: :directory,
              path: full_path,
              depth: depth,
              name: entry
            })
            traverse(full_path, depth + 1)
          else
            Fiber.yield({
              type: :file,
              path: full_path,
              depth: depth,
              name: entry,
              size: File.size(full_path)
            })
          end
        end
      end

      traverse(path)
    end
  end

  def self.streaming_json_parser(json_stream)
    Fiber.new do
      buffer = ""
      brace_count = 0
      in_string = false
      escape_next = false

      json_stream.each_char do |char|
        buffer += char

        unless escape_next
          case char
          when '"'
            in_string = !in_string unless escape_next
          when '\\'
            escape_next = true
            next
          when '{'
            brace_count += 1 unless in_string
          when '}'
            brace_count -= 1 unless in_string

            if brace_count == 0 && !in_string
              begin
                parsed = JSON.parse(buffer.strip)
                Fiber.yield(parsed)
                buffer = ""
              rescue JSON::ParserError => e
                Fiber.yield({ error: e.message, buffer: buffer })
                buffer = ""
              end
            end
          end
        end

        escape_next = false
      end
    end
  end

  def self.demo
    puts "=== Advanced Generator Patterns ==="

    # CSV parsing generator
    csv_data = <<~CSV
      name,age,city
      Alice,30,New York
      Bob,25,San Francisco
      Carol,35,Chicago
    CSV

    puts "CSV Parser Generator:"
    csv_gen = csv_parser_generator(csv_data)
    while csv_gen.alive?
      row = csv_gen.resume
      puts "  Row #{row[:row_number]}: #{row[:data]}" if row
    end

    puts "\nStreaming JSON Parser:"
    json_stream = '{"id":1,"name":"Alice"}{"id":2,"name":"Bob"}{"id":3,"name":"Carol"}'
    json_gen = streaming_json_parser(json_stream)
    while json_gen.alive?
      result = json_gen.resume
      if result
        if result[:error]
          puts "  Error: #{result[:error]}"
        else
          puts "  Parsed: #{result}"
        end
      end
    end

    puts "\nMathematical Sequence Generators:"

    # Collatz sequence generator
    def self.collatz_generator(n)
      Fiber.new do
        current = n
        Fiber.yield current

        while current != 1
          if current.even?
            current = current / 2
          else
            current = 3 * current + 1
          end
          Fiber.yield current
        end
      end
    end

    collatz = collatz_generator(13)
    sequence = []
    while collatz.alive?
      sequence << collatz.resume
    end
    puts "Collatz sequence for 13: #{sequence.compact.join(' → ')}"
  end
end

AdvancedGenerators.demo