diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 9b18467a..1585c391 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -69,6 +69,101 @@ local watching_session = {} local error_queue = {} local fork_queue = { h = 1, t = 0 } +local auxsend, auxtimeout +do ---- avoid session rewind conflict + local csend = c.send + local cintcommand = c.intcommand + local dangerzone + local dangerzone_size = 0x1000 + local dangerzone_low = 0x70000000 + local dangerzone_up = dangerzone_low + dangerzone_size + + local set_checkrewind -- set auxsend and auxtimeout for safezone + local set_checkconflict -- set auxsend and auxtimeout for dangerzone + + local function reset_dangerzone(session) + dangerzone_up = session + dangerzone_low = session + dangerzone = { [session] = true } + for s in pairs(session_id_coroutine) do + if s < dangerzone_low then + dangerzone_low = s + elseif s > dangerzone_up then + dangerzone_up = s + end + dangerzone[s] = true + end + dangerzone_low = dangerzone_low - dangerzone_size + end + + -- in dangerzone, we should check if the next session already exist. + local function checkconflict(session) + local next_session = session + 1 + if next_session > dangerzone_up then + -- leave dangerzone + reset_dangerzone(session) + assert(next_session > dangerzone_up) + set_checkrewind() + return + end + while true do + if not dangerzone[next_session] then + return + end + if not session_id_coroutine[next_session] then + reset_dangerzone(session) + return + end + -- skip the session already exist. + next_session = c.genid() + 1 + end + end + + local function auxsend_checkconflict(addr, proto, msg, sz) + local session = csend(addr, proto, nil, msg, sz) + checkconflict(session) + return session + end + + local function auxtimeout_checkconflict(timeout) + local session = cintcommand("TIMEOUT", timeout) + checkconflict(session) + return session + end + + local function auxsend_checkrewind(addr, proto, msg, sz) + local session = csend(addr, proto, nil, msg, sz) + if session and session > dangerzone_low and session < dangerzone_up then + -- enter dangerzone + set_checkconflict(session) + end + return session + end + + local function auxtimeout_checkrewind(timeout) + local session = cintcommand("TIMEOUT", timeout) + if session and session > dangerzone_low and session < dangerzone_up then + -- enter dangerzone + set_checkconflict(session) + end + return session + end + + set_checkrewind = function() + auxsend = auxsend_checkrewind + auxtimeout = auxtimeout_checkrewind + end + + set_checkconflict = function(session) + reset_dangerzone(session) + auxsend = auxsend_checkconflict + auxtimeout = auxtimeout_checkconflict + end + + -- in safezone at the beginning + set_checkrewind() +end + do ---- request/select local function send_requests(self) local sessions = {} @@ -85,7 +180,7 @@ do ---- request/select 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))) + local session = auxsend(addr, p.id , p.pack(tunpack(req, 3, req.n))) if session == nil then err = err or {} err[#err+1] = req @@ -188,7 +283,7 @@ do ---- request/select self._error = send_requests(self) self._resp = {} if timeout then - self._timeout = c.intcommand("TIMEOUT",timeout) + self._timeout = auxtimeout(timeout) session_id_coroutine[self._timeout] = self._thread end @@ -373,7 +468,7 @@ end skynet.trace_timeout(false) -- turn off by default function skynet.timeout(ti, func) - local session = c.intcommand("TIMEOUT",ti) + local session = auxtimeout(ti) assert(session) local co = co_create_for_timeout(func, ti) assert(session_id_coroutine[session] == nil) @@ -392,7 +487,7 @@ local function suspend_sleep(session, token) end function skynet.sleep(ti, token) - local session = c.intcommand("TIMEOUT",ti) + local session = auxtimeout(ti) assert(session) token = token or coroutine.running() local succ, ret = suspend_sleep(session, token) @@ -605,7 +700,7 @@ function skynet.call(addr, typename, ...) end local p = proto[typename] - local session = c.send(addr, p.id , nil , p.pack(...)) + local session = auxsend(addr, p.id , p.pack(...)) if session == nil then error("call to invalid address " .. skynet.address(addr)) end @@ -619,7 +714,7 @@ function skynet.rawcall(addr, typename, msg, sz) c.send(addr, skynet.PTYPE_TRACE, 0, tag) end local p = proto[typename] - local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address") + local session = assert(auxsend(addr, p.id , msg, sz), "call to invalid address") return yield_call(addr, session) end @@ -627,7 +722,7 @@ function skynet.tracecall(tag, addr, typename, msg, sz) c.trace(tag, "tracecall begin") c.send(addr, skynet.PTYPE_TRACE, 0, tag) local p = proto[typename] - local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address") + local session = assert(auxsend(addr, p.id , msg, sz), "call to invalid address") local msg, sz = yield_call(addr, session) c.trace(tag, "tracecall end") return msg, sz