mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-24 12:20:41 +00:00
Add skynet.coroutine module
This commit is contained in:
@@ -8,8 +8,8 @@ local pcall = pcall
|
|||||||
|
|
||||||
local profile = require "profile"
|
local profile = require "profile"
|
||||||
|
|
||||||
coroutine.resume = profile.resume
|
local coroutine_resume = profile.resume
|
||||||
coroutine.yield = profile.yield
|
local coroutine_yield = profile.yield
|
||||||
|
|
||||||
local proto = {}
|
local proto = {}
|
||||||
local skynet = {
|
local skynet = {
|
||||||
@@ -69,7 +69,7 @@ local function dispatch_error_queue()
|
|||||||
if session then
|
if session then
|
||||||
local co = session_id_coroutine[session]
|
local co = session_id_coroutine[session]
|
||||||
session_id_coroutine[session] = nil
|
session_id_coroutine[session] = nil
|
||||||
return suspend(co, coroutine.resume(co, false))
|
return suspend(co, coroutine_resume(co, false))
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -96,7 +96,6 @@ end
|
|||||||
-- coroutine reuse
|
-- coroutine reuse
|
||||||
|
|
||||||
local coroutine_pool = {}
|
local coroutine_pool = {}
|
||||||
local coroutine_yield = coroutine.yield
|
|
||||||
|
|
||||||
local function co_create(f)
|
local function co_create(f)
|
||||||
local co = table.remove(coroutine_pool)
|
local co = table.remove(coroutine_pool)
|
||||||
@@ -111,7 +110,7 @@ local function co_create(f)
|
|||||||
end
|
end
|
||||||
end)
|
end)
|
||||||
else
|
else
|
||||||
coroutine.resume(co, f)
|
coroutine_resume(co, f)
|
||||||
end
|
end
|
||||||
return co
|
return co
|
||||||
end
|
end
|
||||||
@@ -123,7 +122,7 @@ local function dispatch_wakeup()
|
|||||||
local session = sleep_session[co]
|
local session = sleep_session[co]
|
||||||
if session then
|
if session then
|
||||||
session_id_coroutine[session] = "BREAK"
|
session_id_coroutine[session] = "BREAK"
|
||||||
return suspend(co, coroutine.resume(co, false, "BREAK"))
|
return suspend(co, coroutine_resume(co, false, "BREAK"))
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
@@ -178,7 +177,7 @@ function suspend(co, result, command, param, size)
|
|||||||
c.trash(param, size)
|
c.trash(param, size)
|
||||||
ret = false
|
ret = false
|
||||||
end
|
end
|
||||||
return suspend(co, coroutine.resume(co, ret))
|
return suspend(co, coroutine_resume(co, ret))
|
||||||
elseif command == "RESPONSE" then
|
elseif command == "RESPONSE" then
|
||||||
local co_session = session_coroutine_id[co]
|
local co_session = session_coroutine_id[co]
|
||||||
local co_address = session_coroutine_address[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
|
watching_service[co_address] = watching_service[co_address] + 1
|
||||||
session_response[co] = true
|
session_response[co] = true
|
||||||
unresponse[response] = true
|
unresponse[response] = true
|
||||||
return suspend(co, coroutine.resume(co, response))
|
return suspend(co, coroutine_resume(co, response))
|
||||||
elseif command == "EXIT" then
|
elseif command == "EXIT" then
|
||||||
-- coroutine exit
|
-- coroutine exit
|
||||||
local address = session_coroutine_address[co]
|
local address = session_coroutine_address[co]
|
||||||
@@ -238,6 +237,9 @@ function suspend(co, result, command, param, size)
|
|||||||
elseif command == "QUIT" then
|
elseif command == "QUIT" then
|
||||||
-- service exit
|
-- service exit
|
||||||
return
|
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
|
elseif command == nil then
|
||||||
-- debug trace
|
-- debug trace
|
||||||
return
|
return
|
||||||
@@ -468,7 +470,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
|
|||||||
unknown_response(session, source, msg, sz)
|
unknown_response(session, source, msg, sz)
|
||||||
else
|
else
|
||||||
session_id_coroutine[session] = nil
|
session_id_coroutine[session] = nil
|
||||||
suspend(co, coroutine.resume(co, true, msg, sz))
|
suspend(co, coroutine_resume(co, true, msg, sz))
|
||||||
end
|
end
|
||||||
else
|
else
|
||||||
local p = proto[prototype]
|
local p = proto[prototype]
|
||||||
@@ -491,7 +493,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
|
|||||||
local co = co_create(f)
|
local co = co_create(f)
|
||||||
session_coroutine_id[co] = session
|
session_coroutine_id[co] = session
|
||||||
session_coroutine_address[co] = source
|
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
|
else
|
||||||
unknown_request(session, source, msg, sz, proto[prototype].name)
|
unknown_request(session, source, msg, sz, proto[prototype].name)
|
||||||
end
|
end
|
||||||
@@ -506,7 +508,7 @@ function skynet.dispatch_message(...)
|
|||||||
break
|
break
|
||||||
end
|
end
|
||||||
fork_queue[key] = nil
|
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 not fork_succ then
|
||||||
if succ then
|
if succ then
|
||||||
succ = false
|
succ = false
|
||||||
|
|||||||
108
lualib/skynet/coroutine.lua
Normal file
108
lualib/skynet/coroutine.lua
Normal file
@@ -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
|
||||||
24
test/testcoroutine.lua
Normal file
24
test/testcoroutine.lua
Normal file
@@ -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)
|
||||||
Reference in New Issue
Block a user