相互恢复的协程

Question

对于这个问题的长度，我事先表示歉意。我试图让它尽可能简洁，但它只是一个相当复杂的野兽。

---

在 Ierusalimschy 的 Lua 编程（第 4 版）的第 24 章中，作者展示了任何异步 I/O 库的玩具（“丑陋”）实现，例如这个¹ :

-- filename: async.lua

-- Based (with several modifications) on Listing 24.3 (p. 246) of *Programming
-- in Lua*, 4th edition.

local async = {}
local queue = {}

local function enqueue (command) table.insert(queue, command) end

function async.readline (stream, callback)
  enqueue(function () callback(stream:read()) end)
end

function async.writeline (stream, line, callback)
  enqueue(function () callback(stream:write(line)) end)
end

function async.stop () enqueue("stop") end

function async.runloop ()
  while true do
    local next_command = table.remove(queue, 1)
    if next_command == "stop" then break end
    next_command()
  end
end

return async

作者用这个玩具库来说明协程的一些应用，比如下图运行“异步库之上的同步代码”的方案²：

-- Based (with several modifications) on Listing 24.5 (p. 248) of *Programming
-- in Lua*, 4th edition.

local async = require "async"

function run (synchronous_code)
  local co = coroutine.create(function ()
    synchronous_code()
    async.stop()
  end)
  local wrapper = function ()
    local status, result = assert(coroutine.resume(co))
    return result
  end
  wrapper()
  async.runloop()
end

function getline (stream)
  local co = coroutine.running()
  local callback = function (line) assert(coroutine.resume(co, line)) end
  async.readline(stream, callback)
  local line = coroutine.yield()
  return line
end

function putline (stream, line)
  local co = coroutine.running()
  local callback = function () assert(coroutine.resume(co)) end
  async.writeline(stream, line, callback)
  coroutine.yield()
end

作者使用这种技术实现了一个函数，它以相反的顺序从 stdin 读取的行打印到 stdout：

function synchronous_code ()
  local lines = {}
  local input = io.input()
  local output = io.output()

  while true do
    local line = getline(input)
    if not line then break end
    table.insert(lines, line)
  end

  for i = #lines, 1, -1 do putline(output, lines[i] .. "\n") end

end

run(synchronous_code)

一般的想法是run 函数创建一个协程，该协程将自身（通过getline 和putline 创建的回调）“注册”到异步库的主循环中。每当异步库的主循环执行这些回调之一时，它就会恢复协程，协程可以做更多的工作，包括向主循环注册下一个回调。

run 函数通过调用 wrapper 函数来启动，这反过来又会“恢复”（实际上是启动）协程。协程然后运行直到它遇到第一个 yield 语句，在这个例子中，它发生在 getline 内，就在 getline 将回调注册到 async 库的队列之后。然后wrapper 函数重新获得控制权并返回。最后，run 调用async.runloop。当async.runloop 开始处理它的队列时，它会恢复协程，然后我们就出发了。 “同步代码”（在协程中运行）一直持续到下一个 getline 或 putline 产生（在注册回调之后），然后 async 的主循环再次接管。

到目前为止一切顺利。但是，在练习 24.4（第 249 页）中，作者问道：

练习 24.4： 为基于协程的库（清单 24.5）编写一个行迭代器，以便您可以使用 for 循环读取文件。

(“Listing 24.5”指的是上面第二个代码片段中的代码，其中定义了run、getline和putline。)

我完全被这个难住了。在上面的示例中，协程通过将读取的行“传递”到标准输出来“传递”它们，它可以自己完成所有操作。相比之下，练习 24.4 请求的迭代器必须将其行传递给一个不同的协程，也就是执行迭代的那个。

我能想象到这可能发生的唯一方法是两个协程可以相互恢复。这甚至可能吗？我无法构建一个简单的例子，希望看到能做到这一点的代码³。

另外，在我看来，要使其完全起作用，需要使用 write 方法（以便可以将其传递给 putline）实现一个对象，该方法最终负责传递行（不知何故）到迭代器的协程。

---

^{1我修改了一些表面的细节，比如变量名、缩进等，整体结构和功能不变。}

^{2我再次更改了一些无关紧要的细节，以使代码更易于理解。}

^{3 值得注意的是，本章剩下的两个练习（24.5 和 24.6）都是关于实现涉及多个并发协同程序的系统。因此，想象练习 24.4 也是关于让两个协程相互交谈的想法并不牵强。}

Answer 1

我相信你完全想多了这个练习。按照我的理解，您只是为迭代器编写一个同步样式，该迭代器在提供给run 函数的同步代码中运行。以第三个代码块为基础：

function for_file(file)
  return function(file)
    return getline(file)
  end, file, nil
end

function synchronous_code ()
  local lines = {}
  local input = io.input()
  local output = io.output()

  for line in for_line(input) do
    table.insert(lines, line)
  end

  for i = #lines, 1, -1 do putline(output, lines[i] .. "\n") end

end

run(synchronous_code)

如您所见，您根本不需要了解协程即可使其工作，这正是库的重点。