在LUA中使用异步IO的思考

Posted 2022-11-24 bywayboy

LUA协程的介绍

lua 有一套非常高效的协程机制, 这一套实现非常轻量级, 虽然简单意味着高效, 然而它并不是真正意义上的对称式协程. lua中使用 coroutine.create 来创建一个协程, 使用 coroutine.resume 来执行协程。使用 coroutine.yield 来让出当前正在执行的协程. 通过这两个函数,你可以在两个协程之间相互传递参数, 直到协程执行完成.

local function f(a1)
    -- 让出协程,传递 a1+1 值给resume函数
    local ret = coroutine.yield(a1+1)
    -- 让出协程,传递 ret+1 值给resume函数
    return coroutine.yield(ret+1)
end

-- 这里创建一个协程
local co = coroutine.create(f)
-- 执行f 协程,  当前协程会被挂起
print(coroutine.resume(co, 1))
print(coroutine.resume(co, 12))

通过上面的例子可以看出, lua的协程只能yield 给 resume方. 也就是说 你在协程内任意地方yield 那么resume 会立刻返回.

异步IO

假定我们希望使用一套异步io库来结合LUA的协程来做开发， 会遇到哪些问题呢？首先异步IO会有一个事件循环, 比如 libuv 的 uv_run. 它们所有的操作都是通过回调来完成的. 如果采用完全相同的编程方式,这肯定没太多意义, 对简化编程帮助也不大。编程同步化更加符合人的思维习惯, 开发起来也会更加简单.

想的方案是, 当我们调用一个io操作的时候, 挂起当前协程, 在收到网络事件的时候再恢复它的执行. 编程方法将会变成下面这样:

-- 连接通信函数
local function connhandler(conn)
    while true do
        --- 循环从客户端读取数据
        local data, reason = conn.read(10000)
        if 110 == data then
            -- 超时
        elseif 'string' == type(data) then
            --读取到了数据
            conn.write(data)
        else
            -- 出错了, 或者连接断开
            break
        end
    end
    print('conn has gone')
end


-- 服务器循环等待客户端连接
local function serverloop(server)
    while true do
        -- 等待一个连接
        local retv, message = server.wait(10000)
        if false == conn then
            break
        elseif 110 == conn then
            -- wait timeout
        else
            -- 得到了一个连接 retv is conn
            coroutine.resume(connhandler, retv)
        end
    end
end


local server = tcp.server('0.0.0.0', 80)
coroutine.resume(serverloop)

--这里是一个事件循环
uv.run()

这里完美将回调函数给隐藏了, 使用一个 while 循环来等待客户端的连接. 在得到一个连接的时候, 我们创建了一个协程来处理这个连接的所有业务.
在 和连接通信的时候 conn.read conn.write 我们会挂起当前协程， 将控制权转移给 serverloop 以让 server 能继续去接受新的连接. 这样做的并发能力还是不错的. 通常情况下, 执行不会有问题的。例如如下的流程:

1. connhandler 进行IO操作 挂起 -> serverloop 开始执行.
2. serverloop 执行等待连接操作, 挂起, 这个时候进入了 lua的 mainthread.
3. 第一个连接有数据到达.
4. mainthread 恢复 connhandler 协程.

lua 的一个协程创建后, 你可以在任何地方 resume 它, 然而它的 yield 却只能 将执行控制权交给 resume 方.

然而 这样做也还是存在一些问题. 继续分析一下:
原因在于 connhandler 挂起并不总是为了等待 conn的io操作返回. 假定我们在 connhandler 中执行了另外一个异步io, 比如 发起一个http请求, 这个时候 这个conn 刚好有数据回来. 整套执行流程就乱了. ^_

解决办法:

我们只有在 显式uv_read_start, 在lua中 conn.read 的时候执行 uv_read_start, on_read 回调中取到数据后立即 uv_read_stop. 即可.

接下来我们可以放心地在 connhandler 去调用其它的异步IO.操作. 我们可以确保一个io操作导致的等待， 只会被该io操作的回调结果 resume 即可.

如何做?

下面给出示意代码:

    // tcp.read 函数
    uv_read_start(conn, uv_onread);
    lua_pushthread(L);
    conn->read_ref = luaL_ref(L, LUA_REGISTERYINDEX, -1);
    return lua_yield(L, 0);

    // 读取到数据的回调结果.
    uv_onread(...)
    uv_read_stop(conn)
    // resume read_ref 协程 并传递读取结果.

至此, 我们已经实现了将异步io同步化的改造, 在框架中 我们仅需要实现 tcpserver tcpclient udp pipe 进程池 等的封装, 包括ssl支持的具体的上层应用可以全部使用lua来完成.

Channel 实现

本来打算利用纯lua来实现channel的， 毕竟 协程之间的相互 yield 和resume 不就是现成的channel么? 然而，看起来很香, 事实上可能没那么简单. 原因是 脱离了 libuv 的事件循环后,纯语言层 进行 yield resume 优先级太高了. 假定你在业务场景中死循环使用channel 通信, 只在两个协程中间相互 yield resume的话， 这个时候 main thread 将永远没机会执行。其它io事件都没机会执行.

虽然 lua 的协程之间的 yield resume 效率很高, 但还是不得不做一些取舍, 借助 uv_async_send 来实现 channel.

1. push 的时候, 我们将数据加入到一个队列中. 并执行 uv_async_send.
2. channel->pop 执行的时候, 先检查队列,如果有数据直接返回这条数据. 如果没有. 挂起当前协程. 等到 uv_async_cb 的 resume 唤醒.
3. 可以使用 uv_timer 来实现超时. 这里不多作说明.

后注

这套实现已经在项目中得到应用, 目前工作良好.