mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-25 04:33:05 +00:00
Update to lua 5.4 (#1174)
* update to lua 5.4.0-rc1 * remove LUA_ERRGCMM * use age bits for shared * remove unused lbitlib.c * use luaV_fastget * fix #1174 * suspend函数tailcall优化 * 防止没有session导致未调用suspend * luaH_setint check shared (#1184) * set LUA_GCGEN by default * fix #1174 (#1186) checkshared in sweeplist * update to lua 5.4 rc2 * update to lua 5.4 rc4 * move skynet.profile into snlua * Use lua_sethook to interrupt tight loops * remove lua_checksig * critical condition for signals * update to lua 5.4 released * lua 5.4 bugfix * update lua bugfix * fix lua_sharestring (#1224) * update lua * update lua 5.4 * update README * add skynet.select * add test * test error & discard * yield session * request error has no error message * add timeout to skynet.select * bugfix * for lua 5.4 * new version * bugfix * bugfix * use if instead of while * make local * yield in select for * update lua 5.4.1 * change lua version (#1245) Co-authored-by: xiaojin <xiaojin@onemt.com.cn> * update lua version number Co-authored-by: hong <hongling0@gmail.com> Co-authored-by: hong <hongling-0@qq.com> Co-authored-by: 风---自由 <996442717qqcom@gmail.com> Co-authored-by: xiaojin <xiaojin@onemt.com.cn>
This commit is contained in:
云风
@@ -6,13 +6,14 @@ local assert = assert
|
||||
local pairs = pairs
|
||||
local pcall = pcall
|
||||
local table = table
|
||||
local next = next
|
||||
local tremove = table.remove
|
||||
local tinsert = table.insert
|
||||
local tpack = table.pack
|
||||
local tunpack = table.unpack
|
||||
local traceback = debug.traceback
|
||||
|
||||
local profile = require "skynet.profile"
|
||||
|
||||
local cresume = profile.resume
|
||||
local cresume = coroutine.resume
|
||||
local running_thread = nil
|
||||
local init_thread = nil
|
||||
|
||||
@@ -20,7 +21,7 @@ local function coroutine_resume(co, ...)
|
||||
running_thread = co
|
||||
return cresume(co, ...)
|
||||
end
|
||||
local coroutine_yield = profile.yield
|
||||
local coroutine_yield = coroutine.yield
|
||||
local coroutine_create = coroutine.create
|
||||
|
||||
local proto = {}
|
||||
@@ -66,10 +67,147 @@ local watching_session = {}
|
||||
local error_queue = {}
|
||||
local fork_queue = {}
|
||||
|
||||
do ---- request/select
|
||||
local function send_requests(self)
|
||||
local sessions = {}
|
||||
self._sessions = sessions
|
||||
local request_n = 0
|
||||
local err
|
||||
for i = 1, #self do
|
||||
local req = self[i]
|
||||
local addr = req[1]
|
||||
local p = proto[req[2]]
|
||||
assert(p.unpack)
|
||||
local tag = session_coroutine_tracetag[running_thread]
|
||||
if tag then
|
||||
c.trace(tag, "call", 4)
|
||||
c.send(addr, skynet.PTYPE_TRACE, 0, tag)
|
||||
end
|
||||
local session = c.send(addr, p.id , nil , p.pack(tunpack(req, 3, req.n)))
|
||||
if session == nil then
|
||||
err = err or {}
|
||||
err[#err+1] = req
|
||||
else
|
||||
sessions[session] = req
|
||||
watching_session[session] = addr
|
||||
session_id_coroutine[session] = self._thread
|
||||
request_n = request_n + 1
|
||||
end
|
||||
end
|
||||
self._request = request_n
|
||||
return err
|
||||
end
|
||||
|
||||
local function request_thread(self)
|
||||
while true do
|
||||
local succ, msg, sz, session = coroutine_yield "SUSPEND"
|
||||
if session == self._timeout then
|
||||
self._timeout = nil
|
||||
self.timeout = true
|
||||
else
|
||||
watching_session[session] = nil
|
||||
local req = self._sessions[session]
|
||||
local p = proto[req[2]]
|
||||
if succ then
|
||||
self._resp[session] = tpack( p.unpack(msg, sz) )
|
||||
else
|
||||
self._resp[session] = false
|
||||
end
|
||||
end
|
||||
skynet.wakeup(self)
|
||||
end
|
||||
end
|
||||
|
||||
local function request_iter(self)
|
||||
return function()
|
||||
if self._error then
|
||||
-- invalid address
|
||||
local e = tremove(self._error)
|
||||
if e then
|
||||
return e
|
||||
end
|
||||
self._error = nil
|
||||
end
|
||||
local session, resp = next(self._resp)
|
||||
if session == nil then
|
||||
if self._request == 0 then
|
||||
return
|
||||
end
|
||||
if self.timeout then
|
||||
return
|
||||
end
|
||||
skynet.wait(self)
|
||||
if self.timeout then
|
||||
return
|
||||
end
|
||||
session, resp = next(self._resp)
|
||||
end
|
||||
|
||||
self._request = self._request - 1
|
||||
local req = self._sessions[session]
|
||||
self._resp[session] = nil
|
||||
self._sessions[session] = nil
|
||||
return req, resp
|
||||
end
|
||||
end
|
||||
|
||||
local request_meta = {} ; request_meta.__index = request_meta
|
||||
|
||||
function request_meta:add(obj)
|
||||
assert(type(obj) == "table" and not self._thread)
|
||||
self[#self+1] = obj
|
||||
return self
|
||||
end
|
||||
|
||||
request_meta.__call = request_meta.add
|
||||
|
||||
function request_meta:close()
|
||||
if self._request > 0 then
|
||||
local resp = self._resp
|
||||
for session, req in pairs(self._sessions) do
|
||||
if not resp[session] then
|
||||
session_id_coroutine[session] = "BREAK"
|
||||
watching_session[session] = nil
|
||||
end
|
||||
end
|
||||
self._request = 0
|
||||
end
|
||||
if self._timeout then
|
||||
session_id_coroutine[self._timeout] = "BREAK"
|
||||
self._timeout = nil
|
||||
end
|
||||
end
|
||||
|
||||
request_meta.__close = request_meta.close
|
||||
|
||||
function request_meta:select(timeout)
|
||||
assert(self._thread == nil)
|
||||
self._thread = coroutine_create(request_thread)
|
||||
self._error = send_requests(self)
|
||||
self._resp = {}
|
||||
if timeout then
|
||||
self._timeout = c.intcommand("TIMEOUT",timeout)
|
||||
session_id_coroutine[self._timeout] = self._thread
|
||||
end
|
||||
|
||||
local running = running_thread
|
||||
coroutine_resume(self._thread, self)
|
||||
running_thread = running
|
||||
return request_iter(self), nil, nil, self
|
||||
end
|
||||
|
||||
function skynet.request(obj)
|
||||
local ret = setmetatable({}, request_meta)
|
||||
if obj then
|
||||
return ret(obj)
|
||||
end
|
||||
return ret
|
||||
end
|
||||
end
|
||||
|
||||
-- suspend is function
|
||||
local suspend
|
||||
|
||||
|
||||
----- monitor exit
|
||||
|
||||
local function dispatch_error_queue()
|
||||
@@ -77,7 +215,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, nil, nil, session))
|
||||
end
|
||||
end
|
||||
|
||||
@@ -151,17 +289,22 @@ local function co_create(f)
|
||||
end
|
||||
|
||||
local function dispatch_wakeup()
|
||||
local token = tremove(wakeup_queue,1)
|
||||
if token then
|
||||
local session = sleep_session[token]
|
||||
if session then
|
||||
local co = session_id_coroutine[session]
|
||||
local tag = session_coroutine_tracetag[co]
|
||||
if tag then c.trace(tag, "resume") end
|
||||
session_id_coroutine[session] = "BREAK"
|
||||
return suspend(co, coroutine_resume(co, false, "BREAK"))
|
||||
while true do
|
||||
local token = tremove(wakeup_queue,1)
|
||||
if token then
|
||||
local session = sleep_session[token]
|
||||
if session then
|
||||
local co = session_id_coroutine[session]
|
||||
local tag = session_coroutine_tracetag[co]
|
||||
if tag then c.trace(tag, "resume") end
|
||||
session_id_coroutine[session] = "BREAK"
|
||||
return suspend(co, coroutine_resume(co, false, "BREAK", nil, session))
|
||||
end
|
||||
else
|
||||
break
|
||||
end
|
||||
end
|
||||
return dispatch_error_queue()
|
||||
end
|
||||
|
||||
-- suspend is local function
|
||||
@@ -184,8 +327,7 @@ function suspend(co, result, command)
|
||||
error(traceback(co,tostring(command)))
|
||||
end
|
||||
if command == "SUSPEND" then
|
||||
dispatch_wakeup()
|
||||
dispatch_error_queue()
|
||||
return dispatch_wakeup()
|
||||
elseif command == "QUIT" then
|
||||
-- service exit
|
||||
return
|
||||
@@ -572,7 +714,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source)
|
||||
local tag = session_coroutine_tracetag[co]
|
||||
if tag then c.trace(tag, "resume") end
|
||||
session_id_coroutine[session] = nil
|
||||
suspend(co, coroutine_resume(co, true, msg, sz))
|
||||
suspend(co, coroutine_resume(co, true, msg, sz, session))
|
||||
end
|
||||
else
|
||||
local p = proto[prototype]
|
||||
|
||||
@@ -24,24 +24,17 @@ function skynetco.create(f)
|
||||
end
|
||||
|
||||
do -- begin skynetco.resume
|
||||
|
||||
local profile = require "skynet.profile"
|
||||
-- skynet use profile.resume_co/yield_co instead of coroutine.resume/yield
|
||||
|
||||
local skynet_resume = profile.resume_co
|
||||
local skynet_yield = profile.yield_co
|
||||
|
||||
local function unlock(co, ...)
|
||||
skynet_coroutines[co] = true
|
||||
return ...
|
||||
end
|
||||
|
||||
local function skynet_yielding(co, from, ...)
|
||||
local function skynet_yielding(co, ...)
|
||||
skynet_coroutines[co] = false
|
||||
return unlock(co, skynet_resume(co, from, skynet_yield(from, ...)))
|
||||
return unlock(co, coroutine_resume(co, coroutine_yield(...)))
|
||||
end
|
||||
|
||||
local function resume(co, from, ok, ...)
|
||||
local function resume(co, ok, ...)
|
||||
if not ok then
|
||||
return ok, ...
|
||||
elseif coroutine_status(co) == "dead" then
|
||||
@@ -52,41 +45,41 @@ do -- begin skynetco.resume
|
||||
return true, select(2, ...)
|
||||
else
|
||||
-- blocked in skynet framework, so raise the yielding message
|
||||
return resume(co, from, skynet_yielding(co, from, ...))
|
||||
return resume(co, skynet_yielding(co, ...))
|
||||
end
|
||||
end
|
||||
|
||||
-- record the root of coroutine caller (It should be a skynet thread)
|
||||
local coroutine_caller = setmetatable({} , { __mode = "kv" })
|
||||
|
||||
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"
|
||||
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
|
||||
local from = coroutine_running()
|
||||
local caller = coroutine_caller[from] or from
|
||||
coroutine_caller[co] = caller
|
||||
return resume(co, coroutine_resume(co, ...))
|
||||
end
|
||||
local from = coroutine_running()
|
||||
local caller = coroutine_caller[from] or from
|
||||
coroutine_caller[co] = caller
|
||||
return resume(co, caller, coroutine_resume(co, ...))
|
||||
end
|
||||
|
||||
function skynetco.thread(co)
|
||||
co = co or coroutine_running()
|
||||
if skynet_coroutines[co] ~= nil then
|
||||
return coroutine_caller[co] , false
|
||||
else
|
||||
return co, true
|
||||
function skynetco.thread(co)
|
||||
co = co or coroutine_running()
|
||||
if skynet_coroutines[co] ~= nil then
|
||||
return coroutine_caller[co] , false
|
||||
else
|
||||
return co, true
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
end -- end of skynetco.resume
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ local table = require "table"
|
||||
local packbytes
|
||||
local packvalue
|
||||
|
||||
if _VERSION == "Lua 5.3" then
|
||||
if _VERSION == "Lua 5.4" then
|
||||
function packbytes(str)
|
||||
return string.pack("<s4",str)
|
||||
end
|
||||
|
||||
Reference in New Issue
Block a user