Ruby 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 ):
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= 了。
No comments.