diff --git a/HISTORY.md b/HISTORY.md index 81718ed3..23303874 100644 --- a/HISTORY.md +++ b/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. diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index 0189b61b..479fcea0 100644 --- a/lualib-src/lua-skynet.c +++ b/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 }, }; diff --git a/lualib/sharedata.lua b/lualib/sharedata.lua index 5b83d703..58ecf28d 100644 --- a/lualib/sharedata.lua +++ b/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 diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 5e8e95b0..a5d2caf5 100644 --- a/lualib/skynet.lua +++ b/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 diff --git a/service/bootstrap.lua b/service/bootstrap.lua index 11303577..06e67d7e 100644 --- a/service/bootstrap.lua +++ b/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) diff --git a/service/console.lua b/service/console.lua index 86bf6cc1..c0eba879 100644 --- a/service/console.lua +++ b/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) diff --git a/service/cslave.lua b/service/cslave.lua index 6f005b2f..8a21aa43 100644 --- a/service/cslave.lua +++ b/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)) diff --git a/service/datacenterd.lua b/service/datacenterd.lua index 1a0fee65..5929b0e3 100644 --- a/service/datacenterd.lua +++ b/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]) diff --git a/service/debug_console.lua b/service/debug_console.lua index a587f69d..aa552ceb 100644 --- a/service/debug_console.lua +++ b/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 diff --git a/service/launcher.lua b/service/launcher.lua index 5c1c1a8b..9cb194b4 100644 --- a/service/launcher.lua +++ b/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 diff --git a/service/sharedatad.lua b/service/sharedatad.lua index af7e1169..eb3346c4 100644 --- a/service/sharedatad.lua +++ b/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 diff --git a/test/testblockcall.lua b/test/testblockcall.lua deleted file mode 100644 index b0d7e76f..00000000 --- a/test/testblockcall.lua +++ /dev/null @@ -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) diff --git a/test/testdeadcall.lua b/test/testdeadcall.lua index 9017d380..b34dccb8 100644 --- a/test/testdeadcall.lua +++ b/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 \ No newline at end of file +end