协同程序与常规的多线程不同之处:协同程序是非抢占式的。
当一个协同程序运行时,是无法从外部停止它的。只有当协同程序显式地调用yield时,它才会停止。
当不存在抢先时,编程会变得简单很多,无须为同步的bug抓狂。
在程序中所有的同步都是显式的,只需要确保一个协同程序在它的临界区域之外调用yield即可。
对于这样非抢占式的多线程来说,只要有一个线程调用了一个阻塞操作,整个程序在该操作完成前,都会停止下来。
下面用一个有趣的方法来解决这个问题:通过HTTP下载几个远程文件。
下面的例子测试下载lua源代码,其中会用到LuaSocket模块:
local socket = require "socket" local host = "www.lua.org" local file1 = "/ftp/lua-5.3.3.tar.gz " local HTTP = "HTTP/1.0\r\nUser-Agent: Wget/1.12 (linux-gnu)\r\nAccept: */*\r\nHost: www.lua.org\r\nConnection: Keep-Alive\r\n\r\n" local sock = assert(socket.connect(host1,80)) sock:send("GET " .. file1 .. HTTP) repeat local chunk,status,partial = sock:receive(4096) print("chuck:size: ",string.len(chunk or partial),status or "ok") until status == "closed" sock:close()
在正常情况下receive函数会返回一个字符串。若发生错误,则返回nil,并且附加错误码及出错前读取到的内容(partial).
接下来下载几个文件,最笨的办法就是逐个下载,但是太慢。程序大部分时间花费在等待数据的接收上。
更明确地说,是将时间花在了receive阻塞调用上。
采用的解决办法是,当一个链接没有可用数据时,程序便可以从其他链接出读取数据。
很明显协同程序提供了一种简便的方式来构建这种并发下载。
为每个下载任务创建一个新的线程,只要一个线程无可用数据,它就把控制权转让给一个简单的调度程序。
而这个调度程序则会调用其他下载线程。
在以协同程序来重写程序,先将前面的下载代码重新写:
function receive(connection) local s,status,partial = connection:receive(2^10) return s or partial,status end function download(host,file) local sock = assert(socket.connect(host,80)) local count = 0 --记录接收到的字节数 sock:send("GET " .. file .. HTTP) repeat local chunk,status = receive(sock) count = count + #chunk until status == "closed" sock:close() print(file,count) end download(host,file1)
这是下载一个文件的函数封装,只需调用download就可以。单独下载一个文件需要18秒左右。
但是在并发的情况中,receive代码不能阻塞,因此在它没有可用数据时应该挂起:
function receive(connection) connection:settimeout(0) --设置为非阻塞 local s,status,partial = connection:receive(2^10) if status == "timeout" thencoroutine.yield(connection) end return s or partial,status end
settimeout的调用,使得对此链接的操作不会阻塞。
即使在超时的情况下,连接也是会返回已经读取到的内容,即记录在partial变量中。
以下代码用table threads为调度程序保存所有正在运行中的线程。
get函数保证每个下载任务都在一个独立的线程中执行。
调度程序本身主要就是一个循环,遍历所有的线程,逐个唤醒它们的执行。
当有线程完成时,就将该线程从列表中删除。
threads = {} --保存活跃线程的表 function get(host,file) local co = coroutine.create(function() --创建协同程序 download(host,file) end) table.insert(threads,co) --插入列表 end function dispatch() local i = 1 while true do if threads[i] == nil then --没有线程了 if threads[1] == nil then break end --表是空表吗 i = 1 --重新开始循环 end local status,res = coroutine.resume(threads[i]) --唤醒改线程继续下载文件 if not res then --线程是否已经完成了任务 table.remove(threads,i) --移除list中第i个线程 else i = i + 1 --检查下一个线程 end end end
最后,主程序需要创建所有的线程,并调用调度程序。
local file1 = "/ftp/lua-5.3.3.tar.gz " local file2 = "/ftp/lua-5.3.2.tar.gz " local file3 = "/ftp/lua-5.3.1.tar.gz " local file4 = "/ftp/lua-5.3.0.tar.gz " local file5 = "/ftp/lua-5.2.4.tar.gz " local file6 = "/ftp/lua-5.2.3.tar.gz " local file7 = "/ftp/lua-5.2.2.tar.gz " local file8 = "/ftp/lua-5.2.1.tar.gz " local file9 = "/ftp/lua-5.2.0.tar.gz " get(host,file1) get(host,file2) get(host,file3) get(host,file4) get(host,file5) get(host,file6) get(host,file7) get(host,file8) get(host,file9) dispatch() --main loop
同时下载9个文件总共耗时36秒,比串行下载9个文件速度快很多。
但是发现CPU占用率跑到98%。
为了避免这样的情况,可以使用LuaSocket中的select函数(socket.select(recvt, sendt [, timeout]))。
在等待时陷入阻塞状态,若要在当前实现中应用这个函数,只需要修该调度即可:
function dispatch_new() local i = 1 local timedout = {} --Recvt 集合 while true do if threads[i] == nil then --没有线程了 if threads[1] == nil then break end --表是空表 i = 1 --重新开始循环 timedout = {} --遍历完所有线程,开始新一轮的遍历 end local status,res = coroutine.resume(threads[i]) --唤醒该线程继续下载文件 if not res then --若完成了res就为nil,只有status一个返回值true。否则res为yield传入的参数connection。 table.remove(threads,i) --移除list中第i个线程 else i = i + 1 --检查下一个线程 timedout[#timedout +1] = res if #timedout == #threads then --所有线程都阻塞了吗? socket.select(timedout) --如果线程有数据,就返回 end end end end ... ... dispatch_new() --main loop
receive会将超时的连接通过yield传给resume的res。如果所有的连接都超时了,调度程序就用select来等待这些链接的状态发生变化。
最后运行改良版的程序后,9个文件下载总耗时24秒,cpu占用率不到5%。