From 872491e9680a3d83c8a686e6a4edd7fea71fd440 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Thu, 17 Dec 2015 16:25:24 +0800 Subject: [PATCH] Add skynet.coroutine module --- lualib/skynet.lua | 24 ++++---- lualib/skynet/coroutine.lua | 108 ++++++++++++++++++++++++++++++++++++ test/testcoroutine.lua | 24 ++++++++ 3 files changed, 145 insertions(+), 11 deletions(-) create mode 100644 lualib/skynet/coroutine.lua create mode 100644 test/testcoroutine.lua diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 97a7f6b9..c0a615ee 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -8,8 +8,8 @@ local pcall = pcall local profile = require "profile" -coroutine.resume = profile.resume -coroutine.yield = profile.yield +local coroutine_resume = profile.resume +local coroutine_yield = profile.yield local proto = {} local skynet = { @@ -69,7 +69,7 @@ local function dispatch_error_queue() if session then local co = session_id_coroutine[session] session_id_coroutine[session] = nil - return suspend(co, coroutine.resume(co, false)) + return suspend(co, coroutine_resume(co, false)) end end @@ -96,7 +96,6 @@ end -- coroutine reuse local coroutine_pool = {} -local coroutine_yield = coroutine.yield local function co_create(f) local co = table.remove(coroutine_pool) @@ -111,7 +110,7 @@ local function co_create(f) end end) else - coroutine.resume(co, f) + coroutine_resume(co, f) end return co end @@ -123,7 +122,7 @@ local function dispatch_wakeup() local session = sleep_session[co] if session then session_id_coroutine[session] = "BREAK" - return suspend(co, coroutine.resume(co, false, "BREAK")) + return suspend(co, coroutine_resume(co, false, "BREAK")) end end end @@ -178,7 +177,7 @@ function suspend(co, result, command, param, size) c.trash(param, size) ret = false end - return suspend(co, coroutine.resume(co, ret)) + return suspend(co, coroutine_resume(co, ret)) elseif command == "RESPONSE" then local co_session = session_coroutine_id[co] local co_address = session_coroutine_address[co] @@ -227,7 +226,7 @@ function suspend(co, result, command, param, size) watching_service[co_address] = watching_service[co_address] + 1 session_response[co] = true unresponse[response] = true - return suspend(co, coroutine.resume(co, response)) + return suspend(co, coroutine_resume(co, response)) elseif command == "EXIT" then -- coroutine exit local address = session_coroutine_address[co] @@ -238,6 +237,9 @@ function suspend(co, result, command, param, size) elseif command == "QUIT" then -- service exit return + elseif command == "USER" then + -- See skynet.coutine for detail + error("Call skynet.coroutine.yield out of skynet.coroutine.resume\n" .. debug.traceback(co)) elseif command == nil then -- debug trace return @@ -468,7 +470,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...) unknown_response(session, source, msg, sz) else session_id_coroutine[session] = nil - suspend(co, coroutine.resume(co, true, msg, sz)) + suspend(co, coroutine_resume(co, true, msg, sz)) end else local p = proto[prototype] @@ -491,7 +493,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...) local co = co_create(f) session_coroutine_id[co] = session session_coroutine_address[co] = source - suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz, ...))) + suspend(co, coroutine_resume(co, session,source, p.unpack(msg,sz, ...))) else unknown_request(session, source, msg, sz, proto[prototype].name) end @@ -506,7 +508,7 @@ function skynet.dispatch_message(...) break end fork_queue[key] = nil - local fork_succ, fork_err = pcall(suspend,co,coroutine.resume(co)) + local fork_succ, fork_err = pcall(suspend,co,coroutine_resume(co)) if not fork_succ then if succ then succ = false diff --git a/lualib/skynet/coroutine.lua b/lualib/skynet/coroutine.lua new file mode 100644 index 00000000..69cd3130 --- /dev/null +++ b/lualib/skynet/coroutine.lua @@ -0,0 +1,108 @@ +-- You should use this module (skynet.coroutine) instead of origin lua coroutine in skynet framework + +local coroutine = coroutine +-- origin lua coroutine module +local coroutine_resume = coroutine.resume +local coroutine_yield = coroutine.yield +local coroutine_status = coroutine.status + +local select = select +local skynetco = {} + +skynetco.create = coroutine.create +skynetco.isyieldable = coroutine.isyieldable +skynetco.running = coroutine.running +skynetco.status = coroutine.status + +local skynet_coroutines = setmetatable({}, { __mode = "kv" }) + +function skynetco.create(f) + local co = coroutine.create(f) + -- mark co as a skynet coroutine + skynet_coroutines[co] = true + return co +end + +function skynetco.isskynetcoroutine(co) + co = co or coroutine.running() + return skynet_coroutines[co] ~= nil +end + +do -- begin skynetco.resume + + local profile = require "profile" + -- skynet use profile.resume/yield instead of coroutine.resume/yield + -- read skynet.lua for detail + local skynet_resume = profile.resume + local skynet_yield = profile.yield + + local function unlock(co, ...) + skynet_coroutines[co] = true + return ... + end + + local function skynet_yielding(co, ...) + skynet_coroutines[co] = false + return unlock(co, skynet_resume(co, skynet_yield(...))) + end + + local function resume(co, ok, ...) + if not ok then + return ok, ... + elseif coroutine_status(co) == "dead" then + -- the main function exit + skynet_coroutines[co] = nil + return true, ... + elseif (...) == "USER" then + return true, select(2, ...) + else + -- blocked in skynet framework, so raise the yielding message + return resume(co, skynet_yielding(co, ...)) + end + end + +function skynetco.resume(co, ...) + local co_status = skynet_coroutines[co] + if not co_status then + if co_status == false then + -- is running + return false, "cannot resume a skynet coroutine suspend by skynet framework" + end + if coroutine_status(co) == "dead" then + -- always return false, "cannot resume dead coroutine" + return coroutine_resume(co, ...) + else + return false, "cannot resume none skynet coroutine" + end + end + return resume(co, coroutine_resume(co, ...)) +end + +end -- end of skynetco.resume + +function skynetco.yield(...) + return coroutine_yield("USER", ...) +end + +do -- begin skynetco.wrap + + local function wrap_co(ok, ...) + if ok then + return ... + else + error(...) + end + end + +function skynetco.wrap(f) + local co = skynetco.create(function(...) + return f(...) + end) + return function(...) + return wrap_co(skynetco.resume(co, ...)) + end +end + +end -- end of skynetco.wrap + +return skynetco diff --git a/test/testcoroutine.lua b/test/testcoroutine.lua new file mode 100644 index 00000000..78a433a6 --- /dev/null +++ b/test/testcoroutine.lua @@ -0,0 +1,24 @@ +local skynet = require "skynet" +-- You should use skynet.coroutine instead of origin coroutine in skynet +local coroutine = require "skynet.coroutine" + +local function test(n) + print ("begin", coroutine.isskynetcoroutine()) + for i=1,n do + skynet.sleep(100) + coroutine.yield(i) + end + print "end" + return false +end + +skynet.start(function() + print("Is the main thead a skynet coroutine ?", coroutine.isskynetcoroutine(coroutine.running())) -- always false + print(coroutine.resume(coroutine.running())) -- always return false + local f = coroutine.wrap(test) + repeat + local n = f(5) + print(n) + until not n + skynet.exit() +end)