lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-24 20:23:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

skynet.wait use token instead of coroutine.running()

Browse Source
This commit is contained in:
Cloud Wu
2018-05-25 10:42:58 +08:00
parent 2fde0a0ae4
commit 348e53245b
2 changed files with 30 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
lualib/skynet.lua
View File

@@ -126,10 +126,11 @@ local function co_create(f)
end end
local function dispatch_wakeup() local function dispatch_wakeup()
local co = table.remove(wakeup_queue,1) local token = table.remove(wakeup_queue,1)
if co then if token then
local session = sleep_session[co] local session = sleep_session[token]
if session then if session then
local co = session_id_coroutine[session]
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
@@ -149,7 +150,7 @@ local function release_watching(address)
end end
-- suspend is local function -- suspend is local function
function suspend(co, result, command, param, size) function suspend(co, result, command, param, param2)
if not result then if not result then
local session = session_coroutine_id[co] local session = session_coroutine_id[co]
if session then -- coroutine may fork by others (session is nil) if session then -- coroutine may fork by others (session is nil)
@@ -167,13 +168,16 @@ function suspend(co, result, command, param, size)
session_id_coroutine[param] = co session_id_coroutine[param] = co
elseif command == "SLEEP" then elseif command == "SLEEP" then
session_id_coroutine[param] = co session_id_coroutine[param] = co
sleep_session[co] = param if sleep_session[param2] then
error(debug.traceback(co, "token duplicative"))
end
sleep_session[param2] = param
elseif command == "RETURN" then elseif command == "RETURN" then
local co_session = session_coroutine_id[co] local co_session = session_coroutine_id[co]
session_coroutine_id[co] = nil session_coroutine_id[co] = nil
if co_session == 0 then if co_session == 0 then
if size ~= nil then if param2 ~= nil then
c.trash(param, size) c.trash(param, param2)
end end
return suspend(co, coroutine_resume(co, false)) -- send don't need ret return suspend(co, coroutine_resume(co, false)) -- send don't need ret
end end
@@ -183,13 +187,13 @@ function suspend(co, result, command, param, size)
end end
local ret local ret
if not dead_service[co_address] then if not dead_service[co_address] then
ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) ~= nil ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, param2) ~= nil
if not ret then if not ret then
-- If the package is too large, returns nil. so we should report error back -- If the package is too large, returns nil. so we should report error back
c.send(co_address, skynet.PTYPE_ERROR, co_session, "") c.send(co_address, skynet.PTYPE_ERROR, co_session, "")
end end
elseif size ~= nil then elseif param2 ~= nil then
c.trash(param, size) c.trash(param, param2)
ret = false ret = false
end end
return suspend(co, coroutine_resume(co, ret)) return suspend(co, coroutine_resume(co, ret))
@@ -279,11 +283,12 @@ function skynet.timeout(ti, func)
session_id_coroutine[session] = co session_id_coroutine[session] = co
end end
function skynet.sleep(ti) function skynet.sleep(ti, token)
local session = c.intcommand("TIMEOUT",ti) local session = c.intcommand("TIMEOUT",ti)
assert(session) assert(session)
local succ, ret = coroutine_yield("SLEEP", session) token = token or coroutine.running()
sleep_session[coroutine.running()] = nil local succ, ret = coroutine_yield("SLEEP", session, token)
sleep_session[token] = nil
if succ then if succ then
return return
end end
@@ -298,11 +303,11 @@ function skynet.yield()
return skynet.sleep(0) return skynet.sleep(0)
end end
function skynet.wait(co) function skynet.wait(token)
local session = c.genid() local session = c.genid()
local ret, msg = coroutine_yield("SLEEP", session) token = token or coroutine.running()
co = co or coroutine.running() local ret, msg = coroutine_yield("SLEEP", session, token)
sleep_session[co] = nil sleep_session[token] = nil
session_id_coroutine[session] = nil session_id_coroutine[session] = nil
end end
@@ -429,9 +434,9 @@ function skynet.retpack(...)
return skynet.ret(skynet.pack(...)) return skynet.ret(skynet.pack(...))
end end
function skynet.wakeup(co) function skynet.wakeup(token)
if sleep_session[co] then if sleep_session[token] then
table.insert(wakeup_queue, co) table.insert(wakeup_queue, token)
return true return true
end end
end end

11
test/testtimeout.lua
View File

@@ -14,18 +14,15 @@ local function test_service()
end end
local function timeout_call(ti, ...) local function timeout_call(ti, ...)
local co = coroutine.running() local token = {}
local ret local ret
skynet.fork(function(...) skynet.fork(function(...)
ret = table.pack(pcall(skynet.call, ...)) ret = table.pack(pcall(skynet.call, ...))
if co then skynet.wakeup(token)
skynet.wakeup(co)
end
end, ...) end, ...)
skynet.sleep(ti) skynet.sleep(ti, token)
co = nil -- prevent wakeup after call
if ret then if ret then
if ret[1] then if ret[1] then
return table.unpack(ret, 1, ret.n) return table.unpack(ret, 1, ret.n)
@@ -45,6 +42,8 @@ skynet.start(function()
skynet.error("2", skynet.now()) skynet.error("2", skynet.now())
skynet.error(timeout_call(50, test, "lua")) skynet.error(timeout_call(50, test, "lua"))
skynet.error("3", skynet.now()) skynet.error("3", skynet.now())
skynet.error(timeout_call(150, test, "lua"))
skynet.error("4", skynet.now())
skynet.exit() skynet.exit()
end) end)