Coroutines and Semi-Coroutines
February 19, 2008 – 7:57 pmRuby 1.9 中引入了 Fiber 用于支持 Coroutine 。事实上 Fiber 并不是一个 Coroutine ，而是 Semi-Coroutine ，或者叫做 Asymmetric Coroutine 。因为它只能将执行权返回给调用者，而完整的 Coroutine 可以自由转移运行权。 Ruby 1.9 也提供了 Fiber::Core 用于支持“正宗”的 Coroutine 。关于 Semi-Coroutine 和 Coroutine 的等价性似乎有些争论。下面是一个 Fiber 的例子：
fib = Fiber.new do x, y = 0, 1 loop do Fiber.yield y x,y = y,x+y end end 20.times { puts fib.resume } |
这个 Semi-Coroutine 被用于生成 Fibonacci 数。Semi-Coroutine 看起来就像一个“轻量级的线程（Thread）”，而线程的概念已经为许多人所熟悉了，所以 Semi-Coroutine 看起来也就更“亲切”一些。从例子中可以看到， Fiber.yield
被用于将执行权返回到调用者，而 resume 可以用于启动或恢复 Fiber 的运行。相比之下， Coroutine 则更灵活一些，通过 Fiber#transfer 方法可以将执行权限转移到任意一个 Coroutine ，而不局限于“返回”。但是这种灵活的执行线路的转移通常会让人晕头转向。因此一些语言（例如 Lua ）只提供 Semi-Coroutine 的支持，而 Python 则只提供更局限的（被称作 generator function）支持，在 Python 里只能在 Semi-Coroutine 当前的堆栈层次里 yield ，不能递归或者调用其他函数在 yield ，例如，在 Python 里，用下面的方法来做一个二叉树的 iterator 是不行的：
class Tree: def __init__(self, value, left=None, right=None): self.value = value self.left = left self.right = right def inorder(self): if self.left is not None: self.left.inorder yield self.value if self.right is not None: self.right.inorder |
相比之下，Ruby 1.9 同时提供了 Coroutine （ Fiber::Core ）和 Semi-Coroutine （ Fiber ）的支持。有些人认为同时提供两种支持会让人感到混乱，如果二者实际上是等价的话，像 Lua 那样只支持更容易理解的 Semi-Coroutine 是更好的选择。那么二者究竟等价不等价呢？用 Coroutine 来实现一个Semi-Coroutine 是很容易的事情，相比之下用 Semi-Coroutine 来实现 Coroutine 会难一些。下面让我们来尝试一下：
首先考虑只有两个 Coroutine 互相转移的情况，这是最简单的情况，在执行 transfer 的时候考虑如果目标已经 resume 过了，为了避免二次 resume （这会触发异常），直接调用 Fiber.yield “返回”即可，因为只有两个 Coroutine ，返回一次一定是到另一个 Coroutine 那里；如果目标没有 resume 过，那么直接调用 resume 转移执行权即可：
require 'fiber' class Coroutine @@prev = nil @@current = nil def initialize(&brk) @fib = Fiber.new(&brk) end def transfer(*args) if @@prev == self @@current = self return Fiber.yield(*args) else @@prev = @@current @@current = self return @fib.resume(*args) end end alias :start :transfer end |
下面是一个例子：
producer = consumer = nil producer = Coroutine.new { [1, 2, 3].each { |x| consumer.transfer(x) } consumer.transfer(:stop) } consumer = Coroutine.new { |x| loop do break if x == :stop puts "Consumer get: #{x}" x = producer.transfer end } producer.start |
输出的结果是：
Consumer get: 1 Consumer get: 2 Consumer get: 3 |
但是当有多个 Coroutine 协同工作的时候就不能这么简单了。例如下面的状态：
f ----------> g ----------> h g.resume h.resume |
现在是 h 在执行，如果 h 执行了 f.transfer ，则不能直接调用 f.resume ，这会导致 Double Resume 的异常；也不能直接调用 Fiber.yield ，这会回到 g
中而不是 f 中；唯一的办法就是连续两次 Fiber.yield 。因此，我们需要记录所有被 resume 过的 Coroutine ：
@@prev << @@current @@current = self ret = @fib.resume(*args) |
通过每次 resume 之前把当前 Coroutine 添加到 @@prev 数组中，在实际 transfer 的时候就可以查看目标 Coroutine 是否被 resume 过了，如果有，则设置 @@target 变量为要 transfer 的目标 Coroutine ，并一层一层地“弹出”（通过 Fiber.yield ）：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | def transfer(*args) ret = nil if @@prev.include?(self) @@target = self @@current = @@prev.pop return Fiber.yield(*args) else @@prev << @@current @@current = self ret = @fib.resume(*args) end if @@target == @@current return ret else @@current = @@prev.pop return Fiber.yield(*ret) end end |
注意每次“弹出”后继续执行的位置应该是在第 10 行，所以要在第 13 行检察是否继续“弹”，直到当前 Coroutine 即是目标为止。下面是一个例子：
f = g = h = nil f = Coroutine.new { |x, y| puts "F1: #{x}" x, y = g.transfer(y, x+y) puts "F2: #{x}" x, y = h.transfer(y, x+y) puts "F3: #{x}" } g = Coroutine.new { |x, y| puts "G1: #{x}" x, y = h.transfer(y, x+y) puts "G2: #{x}" x, y = f.transfer(y, x+y) puts "G3: #{x}" } h = Coroutine.new { |x, y| puts "H1: #{x}" x, y = f.transfer(y, x+y) puts "H2: #{x}" x, y = g.transfer(y, x+y) puts "H3: #{x}" } f.start(0, 1) |
执行结果为：
F1: 0 G1: 1 H1: 1 F2: 2 H2: 3 G2: 5 F3: 8 |
当然这样的实现还是有问题，使用了类变量来保存 Coroutine 的历史记录，如果同时有多“组” Coroutines 在运行，就会有问题，因此需要把信息保存在每个线程所独有的变量中，而且限制 Coroutine 不能在不同的线程之间进行跳转。
Updates: 目前 Ruby 1.9 发布的版本中已经没有 Fiber::Core 了， Fiber::Core 和 Fiber 合并为一个东西了，现在 Fiber 既可以做不对称的 yield 也可以做对称的=transfer= 了。
