如何让Ruby的Fiber与Enumerator兼容使用？

问题描述 #

我们实现了一个Filter类，用Fiber处理分块到达的数据，通过暂停执行等待更多输入，支持命令式编写处理逻辑：

class Filter
  def initialize
    @fiber = Fiber.new do
      run
    end
    @fiber.resume
  end

  def <<(chunk)
    @fiber.resume(chunk)
  end

  def each_two
    loop do
      a = Fiber.yield
      b = Fiber.yield
      yield a + b
    end
  end

  def run
    each_two do |chunk|
      puts chunk.inspect
    end
  end
end

filter = Filter.new
filter << "Hello"
filter << ", "
filter << "world"
filter << "!\n"

执行后正常输出合并后的分块：

$ ruby filter.rb
"Hello, "
"world!\n"

将each_two包装为Enumerator后，内部迭代（pairs.each）可以正常运行：

def run
  pairs = enum_for(:each_two)
  pairs.each do |chunk|
    puts chunk.inspect
  end
end

但外部迭代（调用pairs.next）会直接崩溃：

def run
  pairs = enum_for(:each_two)
  loop do
    puts pairs.next.inspect
  end
end

报错信息：

$ ruby filter.rb
nil
nil
filter.rb:20:in 'block in Filter#each_two': undefined method '+' for an instance of Fiber (NoMethodError)

      yield a + b
              ^
    from <internal:kernel>:168:in 'Kernel#loop'
    from filter.rb:17:in 'Filter#each_two'
    from filter.rb:in 'Enumerator#each'

根因分析 #

问题核心是Enumerator的外部迭代（next方法）本身依赖Fiber实现，而我们的each_two方法直接调用了Fiber.yield——这导致两个Fiber上下文发生冲突：

• 自定义Fiber用于等待外部输入的chunk
• Enumerator内部的Fiber用于实现外部迭代的暂停/恢复

当调用pairs.next时，Fiber.yield返回的不再是我们传入的chunk，而是Enumerator内部的Fiber实例，所以a和b变成了Fiber对象，自然无法执行+操作。

解决方案 #

要解决这个冲突，需要把Fiber的暂停逻辑与迭代器的yield逻辑解耦，通过一个中间缓冲区来管理输入的chunk，让迭代器从缓冲区取数据，而不是直接依赖Fiber.yield获取输入：

修改后的Filter类：

class Filter
  def initialize
    @buffer = []
    @fiber = Fiber.new do
      run
    end
    @fiber.resume
  end

  def <<(chunk)
    @buffer << chunk
    # 缓冲区有数据时恢复Fiber执行
    @fiber.resume if @fiber.alive?
  end

  def each_two
    loop do
      # 从缓冲区获取下一个chunk，空则暂停Fiber
      a = next_chunk
      b = next_chunk
      yield a + b
    end
  end

  private

  def next_chunk
    # 缓冲区为空时暂停，等待外部输入
    while @buffer.empty?
      Fiber.yield
    end
    @buffer.shift
  end

  def run
    pairs = enum_for(:each_two)
    loop do
      puts pairs.next.inspect
    end
  end
end

filter = Filter.new
filter << "Hello"
filter << ", "
filter << "world"
filter << "!\n"

逻辑说明 #

1. 新增@buffer存储输入的chunk，避免直接通过Fiber.resume传递数据
2. next_chunk方法封装了Fiber暂停逻辑：缓冲区为空时才暂停，有数据则返回第一个chunk
3. each_two不再直接调用Fiber.yield，而是通过next_chunk获取数据，与Enumerator的Fiber上下文彻底隔离

验证结果 #

运行修改后的代码，外部迭代可以正常输出：

$ ruby filter.rb
"Hello, "
"world!\n"

同时内部迭代（pairs.each）的逻辑也不受影响，两种迭代方式都能正常工作。

内容的提问来源于stack exchange，提问作者Tavian Barnes