add skynet.response for delay response

HISTORY.md

@@ -1,3 +1,10 @@
+Dev version
+-----------
+* add sharedata
+* bugfix: service exit before init would not report back
+* add skynet.response and check multicall skynet.ret
+* skynet.newservice throw error when lanuch faild
+
 v0.5.0 (2014-7-28)
 -----------
 * skynet.exit will quit service immediately.

lualib-src/lua-skynet.c

@@ -299,6 +299,26 @@ lpackstring(lua_State *L) {
 	return 1;
 }
 
+static int
+ltrash(lua_State *L) {
+	int t = lua_type(L,1);
+	switch (t) {
+	case LUA_TSTRING: {
+		break;
+	}
+	case LUA_TLIGHTUSERDATA: {
+		void * msg = lua_touserdata(L,1);
+		luaL_checkinteger(L,2);
+		skynet_free(msg);
+		break;
+	}
+	default:
+		luaL_error(L, "skynet.trash invalid param %s", lua_typename(L,t));
+	}
+
+	return 0;
+}
+
 int
 luaopen_skynet_c(lua_State *L) {
 	luaL_checkversion(L);
@@ -314,6 +334,7 @@ luaopen_skynet_c(lua_State *L) {
 		{ "pack", _luaseri_pack },
 		{ "unpack", _luaseri_unpack },
 		{ "packstring", lpackstring },
+		{ "trash" , ltrash },
 		{ "callback", _callback },
 		{ NULL, NULL },
 	};

lualib/sharedata.lua

@@ -14,7 +14,7 @@ local function monitor(name, obj, cobj)
 	while true do
 		newobj = skynet.call(service, "lua", "monitor", name, newobj)
 		if newobj == nil then
-			return
+			break
 		end
 		sd.update(obj, newobj)
 	end

lualib/skynet.lua

@@ -43,12 +43,14 @@ end
 local session_id_coroutine = {}
 local session_coroutine_id = {}
 local session_coroutine_address = {}
+local session_response = {}
 
 local wakeup_session = {}
 local sleep_session = {}
 
 local watching_service = {}
 local watching_session = {}
+local dead_service = {}
 local error_queue = {}
 
 -- suspend is function
@@ -73,7 +75,9 @@ local function _error_dispatch(error_session, error_source)
 	if error_session == 0 then
 		-- service is down
 		--  Don't remove from watching_service , because user may call dead service
-		watching_service[error_source] = false
+		if watching_service[error_source] then
+			dead_service[error_source] = true
+		end
 		for session, srv in pairs(watching_session) do
 			if srv == error_source then
 				table.insert(error_queue, session)
@@ -122,6 +126,18 @@ local function dispatch_wakeup()
 	end
 end
 
+local function release_watching(address)
+	local ref = watching_service[address]
+	if ref then
+		ref = ref - 1
+		if ref > 0 then
+			watching_service[address] = ref
+		else
+			watching_service[address] = nil
+		end
+	end
+end
+
 -- suspend is local function
 function suspend(co, result, command, param, size)
 	if not result then
@@ -142,15 +158,50 @@ function suspend(co, result, command, param, size)
 	elseif command == "RETURN" then
 		local co_session = session_coroutine_id[co]
 		local co_address = session_coroutine_address[co]
-		if param == nil then
+		if param == nil or session_response[co] then
 			error(debug.traceback(co))
 		end
-		c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size)
-		return suspend(co, coroutine.resume(co))
+		session_response[co] = true
+		local ret
+		if not dead_service[co_address] then
+			ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) >= 0
+		elseif size == nil then
+			c.trash(param, size)
+			ret = false
+		end
+		return suspend(co, coroutine.resume(co, ret))
+	elseif command == "RESPONSE" then
+		local co_session = session_coroutine_id[co]
+		local co_address = session_coroutine_address[co]
+		if session_response[co] then
+			error(debug.traceback(co))
+		end
+		local f = param
+		local function response(ok, ...)
+			local ret
+			if not dead_service[co_address] then
+				if ok then
+					ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) >=0
+				else
+					ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") >=0
+				end
+			else
+				ret = false
+			end
+			release_watching(co_address)
+			f = nil
+			return ret
+		end
+		watching_service[co_address] = watching_service[co_address] + 1
+		session_response[co] = response
+		return suspend(co, coroutine.resume(co, response))
 	elseif command == "EXIT" then
 		-- coroutine exit
+		local address = session_coroutine_address[co]
+		release_watching(address)
 		session_coroutine_id[co] = nil
 		session_coroutine_address[co] = nil
+		session_response[co] = nil
 	elseif command == "QUIT" then
 		-- service exit
 		return
@@ -259,13 +310,21 @@ end
 
 function skynet.exit()
 	skynet.send(".launcher","lua","REMOVE",skynet.self())
+	-- report the sources that call me
 	for co, session in pairs(session_coroutine_id) do
 		local address = session_coroutine_address[co]
-		local self = skynet.self()
 		if session~=0 and address then
-			skynet.redirect(address, self, "error", session, "")
+			c.redirect(address, 0, skynet.PTYPE_ERROR, session, "")
 		end
 	end
+	-- report the sources I call but haven't return
+	local tmp = {}
+	for session, address in pairs(watching_session) do
+		tmp[address] = true
+	end
+	for address in pairs(tmp) do
+		c.redirect(address, 0, skynet.PTYPE_ERROR, 0, "")
+	end
 	c.command("EXIT")
 	-- quit service
 	coroutine_yield "QUIT"
@@ -294,9 +353,6 @@ end
 
 function skynet.send(addr, typename, ...)
 	local p = proto[typename]
-	if watching_service[addr] == false then
-		error("Service is dead")
-	end
 	return c.send(addr, p.id, 0 , p.pack(...))
 end
 
@@ -321,9 +377,6 @@ end
 
 function skynet.call(addr, typename, ...)
 	local p = proto[typename]
-	if watching_service[addr] == false then
-		error("Service is dead")
-	end
 	local session = c.send(addr, p.id , nil , p.pack(...))
 	if session == nil then
 		error("call to invalid address " .. skynet.address(addr))
@@ -333,16 +386,18 @@ end
 
 function skynet.rawcall(addr, typename, msg, sz)
 	local p = proto[typename]
-	if watching_service[addr] == false then
-		error("Service is dead")
-	end
 	local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address")
 	return yield_call(addr, session)
 end
 
 function skynet.ret(msg, sz)
 	msg = msg or ""
-	coroutine_yield("RETURN", msg, sz)
+	return coroutine_yield("RETURN", msg, sz)
+end
+
+function skynet.response(pack)
+	pack = pack or skynet.pack
+	return coroutine_yield("RESPONSE", pack)
 end
 
 function skynet.retpack(...)
@@ -412,6 +467,12 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
 		local p = assert(proto[prototype], prototype)
 		local f = p.dispatch
 		if f then
+			local ref = watching_service[source]
+			if ref then
+				watching_service[source] = ref + 1
+			else
+				watching_service[source] = 1
+			end
 			local co = co_create(f)
 			session_coroutine_id[co] = session
 			session_coroutine_address[co] = source

service/bootstrap.lua

@@ -13,31 +13,31 @@ skynet.start(function()
 		standalone = true
 		skynet.setenv("standalone", "true")
 
-		local slave = skynet.newservice "cdummy"
-		if slave == nil then
+		local ok, slave = pcall(skynet.newservice, "cdummy")
+		if not ok then
 			skynet.abort()
 		end
 		skynet.name(".slave", slave)
 
 	else
 		if standalone then
-			if not skynet.newservice "cmaster" then
+			if not pcall(skynet.newservice,"cmaster") then
 				skynet.abort()
 			end
 		end
 
-		local slave = skynet.newservice "cslave"
-		if slave == nil then
+		local ok, slave = pcall(skynet.newservice, "cslave")
+		if not ok then
 			skynet.abort()
 		end
 		skynet.name(".slave", slave)
 	end
 
 	if standalone then
-		local datacenter = assert(skynet.newservice "datacenterd")
+		local datacenter = skynet.newservice "datacenterd"
 		skynet.name("DATACENTER", datacenter)
 	end
-	assert(skynet.newservice "service_mgr")
-	assert(skynet.newservice(skynet.getenv "start" or "main"))
+	skynet.newservice "service_mgr"
+	pcall(skynet.newservice,skynet.getenv "start" or "main")
 	skynet.exit()
 end)

service/console.lua

@@ -7,10 +7,7 @@ local function console_main_loop()
 	while true do
 		local cmdline = socket.readline(stdin, "\n")
 		if cmdline ~= "" then
-			local handle = skynet.newservice(cmdline)
-			if handle == nil then
-				print("Launch error:",cmdline)
-			end
+			pcall(skynet.newservice,cmdline)
 		end
 	end
 	socket.unlock(stdin)

service/cslave.lua

@@ -28,7 +28,7 @@ local function monitor_clear(id)
 	if v then
 		monitor[id] = nil
 		for _, v in ipairs(v) do
-			skynet.redirect(v.address, 0, "response", v.session, "")
+			v()
 		end
 	end
 end
@@ -149,30 +149,30 @@ local function monitor_harbor(master_fd)
 	end
 end
 
-function harbor.REGISTER(_,_, fd, name, handle)
+function harbor.REGISTER(fd, name, handle)
 	assert(globalname[name] == nil)
 	globalname[name] = handle
 	socket.write(fd, pack_package("R", name, handle))
 	skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name)
 end
 
-function harbor.LINK(session, source, fd, id)
+function harbor.LINK(fd, id)
 	if slaves[id] then
 		if monitor[id] == nil then
 			monitor[id] = {}
 		end
-		table.insert(monitor[id], { address = source, session = session })
+		table.insert(monitor[id], skynet.response(true))
 	else
 		skynet.ret()
 	end
 end
 
-function harbor.CONNECT(session, source, fd, id)
+function harbor.CONNECT(fd, id)
 	if not slaves[id] then
 		if monitor[id] == nil then
 			monitor[id] = {}
 		end
-		table.insert(monitor[id], { address = source, session = session })
+		table.insert(monitor[id], skynet.response(true))
 	else
 		skynet.ret()
 	end
@@ -186,9 +186,9 @@ skynet.start(function()
 	skynet.error("slave connect to master " .. tostring(master_addr))
 	local master_fd = socket.open(master_addr)
 
-	skynet.dispatch("lua", function (session,source,command,...)
+	skynet.dispatch("lua", function (_,_,command,...)
 		local f = assert(harbor[command])
-		f(session, source, master_fd, ...)
+		f(master_fd, ...)
 	end)
 	skynet.dispatch("text", monitor_harbor(master_fd))
 

service/datacenterd.lua

@@ -45,10 +45,8 @@ local function wakeup(db, key1, key2, value, ...)
 		db[key1] = nil
 		if value then
 			-- throw error because can't wake up a branch
-			for _,v in ipairs(q) do
-				local session = v[1]
-				local source = v[2]
-				skynet.redirect(source, 0, "error", session, "")
+			for _,response in ipairs(q) do
+				response(false)
 			end
 		else
 			return q
@@ -66,15 +64,13 @@ function command.UPDATE(...)
 	end
 	local q = wakeup(wait_queue, ...)
 	if q then
-		for _, v in ipairs(q) do
-			local session = v[1]
-			local source = v[2]
-			skynet.redirect(source, 0, "response", session, skynet.pack(value))
+		for _, response in ipairs(q) do
+			response(true,value)
 		end
 	end
 end
 
-local function waitfor(session, source, db, key1, key2, ...)
+local function waitfor(db, key1, key2, ...)
 	if key2 == nil then
 		-- push queue
 		local q = db[key1]
@@ -84,7 +80,7 @@ local function waitfor(session, source, db, key1, key2, ...)
 		else
 			assert(q[mode] == "queue")
 		end
-		table.insert(q, { session, source })
+		table.insert(q, skynet.response())
 	else
 		local q = db[key1]
 		if q == nil then
@@ -93,18 +89,18 @@ local function waitfor(session, source, db, key1, key2, ...)
 		else
 			assert(q[mode] == "branch")
 		end
-		return waitfor(session, source, q, key2, ...)
+		return waitfor(q, key2, ...)
 	end
 end
 
 skynet.start(function()
-	skynet.dispatch("lua", function (session, source, cmd, ...)
+	skynet.dispatch("lua", function (_, _, cmd, ...)
 		if cmd == "WAIT" then
 			local ret = command.QUERY(...)
 			if ret then
 				skynet.ret(skynet.pack(ret))
 			else
-				waitfor(session, source, wait_queue, ...)
+				waitfor(wait_queue, ...)
 			end
 		else
 			local f = assert(command[cmd])

service/debug_console.lua

@@ -122,8 +122,8 @@ function COMMAND.clearcache()
 end
 
 function COMMAND.start(...)
-	local addr = skynet.newservice(...)
-	if addr then
+	local ok, addr = pcall(skynet.newservice, ...)
+	if ok then
 		return { [skynet.address(addr)] = ... }
 	else
 		return "Failed"
@@ -131,8 +131,8 @@ function COMMAND.start(...)
 end
 
 function COMMAND.snax(...)
-	local s = snax.newservice(...)
-	if s then
+	local ok, s = pcall(snax.newservice, ...)
+	if ok then
 		local addr = s.handle
 		return { [skynet.address(addr)] = ... }
 	else

service/launcher.lua

@@ -28,7 +28,7 @@ function command.STAT()
 	return list
 end
 
-function command.INFO(_, _, handle)
+function command.INFO(_, handle)
 	handle = handle_to_address(handle)
 	if services[handle] == nil then
 		return
@@ -38,7 +38,7 @@ function command.INFO(_, _, handle)
 	end
 end
 
-function command.TASK(_, _, handle)
+function command.TASK(_, handle)
 	handle = handle_to_address(handle)
 	if services[handle] == nil then
 		return
@@ -48,7 +48,7 @@ function command.TASK(_, _, handle)
 	end
 end
 
-function command.KILL(_, _, handle)
+function command.KILL(_, handle)
 	handle = handle_to_address(handle)
 	skynet.kill(handle)
 	local ret = { [skynet.address(handle)] = tostring(services[handle]) }
@@ -72,18 +72,29 @@ function command.GC()
 	return command.MEM()
 end
 
-function command.REMOVE(_,_, handle)
+function command.REMOVE(_, handle)
 	services[handle] = nil
+	local response = instance[handle]
+	if response then
+		-- instance is dead
+		response(false)
+		instance[handle] = nil
+	end
+
 	-- don't return (skynet.ret) because the handle may exit
 	return NORET
 end
 
-function command.LAUNCH(address, session, service, ...)
+local function return_string(str)
+	return str
+end
+
+function command.LAUNCH(_, service, ...)
 	local param = table.concat({...}, " ")
 	local inst = skynet.launch(service, param)
 	if inst then
 		services[inst] = service .. " " .. param
-		instance[inst] = { session = session, address = address }
+		instance[inst] = skynet.response(return_string)
 	else
 		skynet.ret("")	-- launch failed
 	end
@@ -93,9 +104,9 @@ end
 function command.ERROR(address)
 	-- see serivce-src/service_lua.c
 	-- init failed
-	local reply = instance[address]
-	if reply then
-		skynet.redirect(reply.address , 0, "response", reply.session, "")
+	local response = instance[address]
+	if response then
+		response(false)
 		instance[address] = nil
 	end
 	services[address] = nil
@@ -104,9 +115,9 @@ end
 
 function command.LAUNCHOK(address)
 	-- init notice
-	local reply = instance[address]
-	if reply then
-		skynet.redirect(reply.address , 0, "response", reply.session, skynet.address(address))
+	local response = instance[address]
+	if response then
+		response(true, skynet.address(address))
 		instance[address] = nil
 	end
 
@@ -127,7 +138,7 @@ skynet.register_protocol {
 		else
 			-- launch request
 			local service, param = string.match(cmd,"([^ ]+) (.*)")
-			command.LAUNCH(address, session, service, param)
+			command.LAUNCH(_, service, param)
 		end
 	end,
 }
@@ -136,7 +147,7 @@ skynet.dispatch("lua", function(session, address, cmd , ...)
 	cmd = string.upper(cmd)
 	local f = command[cmd]
 	if f then
-		local ret = f(address, session, ...)
+		local ret = f(address, ...)
 		if ret ~= NORET then
 			skynet.ret(skynet.pack(ret))
 		end

service/sharedatad.lua

@@ -55,10 +55,8 @@ function CMD.delete(name)
 	assert(objmap[v.obj])
 	objmap[v.obj] = true
 	sharedata.host.decref(v.obj)
-	for _,v in ipairs(v.watch) do
-		local session = v[1]
-		local address = v[2]
-		skynet.redirect(address, 0, "response", session, skynet.pack(nil))
+	for _,response in ipairs(v.watch) do
+		response(true)
 	end
 end
 
@@ -90,21 +88,19 @@ function CMD.update(name, t)
 	local newobj = pool[name].obj
 	if watch then
 		sharedata.host.markdirty(oldcobj)
-		for _,v in ipairs(watch) do
-			local session = v[1]
-			local address = v[2]
-			skynet.redirect(address, 0, "response", session, skynet.pack(newobj))
+		for _,response in ipairs(watch) do
+			response(true, newobj)
 		end
 	end
 end
 
-function CMD.monitor(session, address, name, obj)
+function CMD.monitor(name, obj)
 	local v = assert(pool[name])
 	if obj ~= v.obj then
 		return v.obj
 	end
 
-	table.insert(v.watch, { session, address })
+	table.insert(v.watch, skynet.response())
 
 	return NORET
 end
@@ -113,12 +109,7 @@ skynet.start(function()
 	skynet.fork(collectobj)
 	skynet.dispatch("lua", function (session, source ,cmd, ...)
 		local f = assert(CMD[cmd])
-		local r
-		if cmd == "monitor" then
-			r = f(session, source, ...)
-		else
-			r = f(...)
-		end
+		local r = f(...)
 		if r ~= NORET then
 			skynet.ret(skynet.pack(r))
 		end

test/testblockcall.lua

@@ -1,10 +0,0 @@
-local skynet = require "skynet"
-
-skynet.start(function()
-	local ping = skynet.newservice("pingserver")
-	skynet.timeout(0,function()
-		print(skynet.call(ping,"lua","PING","ping"))
-	end)
-
-	print(skynet.blockcall(ping,"lua","HELLO"))
-end)

test/testdeadcall.lua

@@ -11,6 +11,15 @@ skynet.start(function()
 	end)
 end)
 
+elseif mode == "dead" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function (...)
+		skynet.sleep(100)
+		print("return", skynet.ret "")
+	end)
+end)
+
 else
 
 	skynet.start(function()
@@ -20,6 +29,9 @@ else
 			skynet.call(test,"lua", "dead call")
 		end))
 
-		skynet.exit()
+		local dead = skynet.newservice(SERVICE_NAME, "dead")	-- launch self in dead mode
+
+		skynet.timeout(0, skynet.exit)	-- exit after a while, so the call never return
+		skynet.call(dead, "lua", "whould not return")
 	end)
-end
+end