Lua協程

Lua協程

 

協同程序（coroutine）與多線程狀況下的線程比較相似：有本身的堆棧、局部變量、指令指針，但與其它協程共享全局變量等不少信息。

協程相似一種多線程，但與多線程還有不少區別：

1. 協程並不是os線程，因此建立、切換開銷比線程相對要小。
2. 協程與線程同樣有本身的棧、局部變量等，可是協程的棧是在用戶進程空間模擬的，因此建立、切換開銷很小。
3. 多線程程序是多個線程併發執行，也就是說在一瞬間有多個控制流在執行。而協程強調的是一種多個協程間協做的關係，只有當一個協程主動放棄執行權，另外一個協程才能得到執行權，因此在某一瞬間，多個協程間只有一個在運行。
4. 因爲多個協程時只有一個在運行，因此對於臨界區的訪問不須要加鎖，而多線程的狀況則必須加鎖。
5. 多線程程序因爲有多個控制流，因此程序的行爲不可控，而多個協程的執行是由開發者定義的因此是可控的。

 

Lua的協程是不對稱的（asymmetric coroutines），是指「掛起一個正在執行的協同函數」 和 「使一個被掛起的協程再次執行的函數」是不一樣的。

有些語言使用對稱協同（symmetric coroutines），即便用同一個函數負責「執行與掛起間的狀態切換」。

 

一、狀態切換

協程有3個狀態：

一、suspended（掛起）；

二、running（運行）；

三、dead（中止）；

 

協程API： 

co = coroutine.create(function)                         -->建立協程，返回thread類型

coroutine.status(co)                                    -->檢查協程狀態

coroutine.resume(co)                                    -->恢復運行

coroutine.yield()                                       -->掛起

協程建立成功後，處於suspended狀態，此時並未運行。

在協程中使用yield函數，可讓正在運行的代碼掛起。

 一個例子：

co = coroutine.create(function()
            for i=1,5 do
                print("co", i)
                coroutine.yield()
            end
        end)
    
coroutine.resume(co)    -- 1
coroutine.resume(co)    -- 2
coroutine.resume(co)    -- 3
coroutine.resume(co)    -- 4
coroutine.resume(co)    -- 5
print(coroutine.resume(co) )   -- true
print(coroutine.resume(co) )   -- false cannot resume dead coroutine                                                                  

注意：resume運行在保護模式下，若是協程內部存在錯誤，Lua並不會拋出錯誤，而是把錯誤返回給resume函數。

 

協程的另外一個做用是經過resume-yield來交換數據：

co = coroutine.create(function(a, b)
        coroutine.yield(a+b, a-b)
        return 6,7
        end)

print(coroutine.resume(co, 20, 10))     -- true 30 10
print(coroutine.resume(co))     　　　　 -- true 6 7

可見，resume的參數會傳遞給協同函數，yield的參數會做爲resume的返回值。而且，協程結束時的返回值也會傳遞給 resume。

 

二、管道

協同最具備表明性的例子是用來解決生產者-消費者問題，假定有一個函數不斷地生產數據（好比從文件讀取），另外一個函數不斷的處理這些數據（好比寫到一個文件中），函數以下：

function producer()
    while true do
        local x=io.read()
        send(x)
    end
end 

function consumer()
    while true do
        local x=receive()
        io.write(x, '\n')
    end
end

上面的代碼中，生產者和消費者都在不停的循環，而對對方的狀態一無所知，咱們須要改變一下結構，使得二者可以協調工做。

以消費者驅動模型爲例，一開始咱們調用消費者，當消費者須要值時喚起生產者，生產者生產處數據後中止，直到消費者再次請求。

完整的示例代碼以下：

function receive(prod)
    local status, value = coroutine.resume(prod)
    return value
end 

function send(x)
    local coroutine.yield(x)
end 


function producer()
    return coroutine.create(function ()
        while true do
            local x=io.read()
            send(x)
        end
    end)
end

function consumer(prod)
    while true do
        local x=receive(prod)
        io.write(x, '\n')
    end
end 

function filter(prod)
    return coroutine.create(function()
            local line = 1
            while true do
                local x=receive(prod)
                x=string.format("%5d %s", line, x)
                send(x)
                line = line+1
            end
        end)
end 

consumer(filter(producer()))

上面這個例子的工做方式很是相似UNIX的管道（pipe），協程是一種非搶佔式的多線程。

在pipe的方式下，每一個任務在獨立的進程中運行，進程間的切換代價比較高；

而在協同中，每一個任務運行在獨立的協同代碼中，任務間的切換代價較小，與函數調用至關。

 

三、非搶佔式多線程

協程是一種非搶佔式的多線程，這句話的含義是：

當一個協程正在運行時，不能在外部終止它，只能經過顯式調用yield掛起它的執行。

顯然，非搶佔式的多線程比較容易寫，由於不須要考慮線程同步帶來的bug。

非搶佔式的多線程的弊端在於，無論何時，只要有一個線程調用一個阻塞操做（blocking operation），整個程序在阻塞操做完成以前都將中止。

協同的這種弊端有點讓人難以忍受！

 

看一個多線程的例子，經過HTTP協議從遠程服務器下載一些文件。

在下載過程當中，若是遇到阻塞，掛起線程，並使用一個調度器去resume另外一個線程。

require "luasocket"

function receive(connection)
    connection:timeout(0)   -- do not block
    local s,status=connection:receive(2^10)
    if status=="timeout" then
        coroutine.yield(connection)
    end 
    return s, status
end

function download(host, file)
    local c=assert(socket.connection(host, 80))
    local count = 0 
    c:send("GET" .. file .. " HTTP/1.0\r\n\r\n")
    while ture do
        local s, status=receive(c)
        count=count+string.len(s)
        if status=="closed" then break end    
    end 
    c:close()
    print(file, count)
end

threads = {}
function get(host, file)
    local co=coroutine.create(function()
                download(host, file)
            end)

    table.insert(threads, co) 
end

function dispatcher()
    while ture do
        local n=#threads
        if n==0 then break end 
        local connections={}
        for i=1,n do
            local status,res=coroutine.resume(threads[i])
            if not res then　　　　　　-- finish
                table.remove(threads, i)
                break
            else　　-- timeout
                table.insert(connections, res) 
            end 
        end 

        if #connections == n then
            socket.select(connections)
        end
    end
end

host="http://news.163.com/"
get(host, "/14/0330/17/9OJO5ML800014JB6.html")
get(host, "/14/0330/08/9OIQKNS90001124J.html")
get(host, "/14/0330/10/9OJ2M5PN000915BF.html")
dispatch()