check conflict (#1798)

* rewind check conflict

* Use auxsend instead of c.send

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