local c = require "skynet.c" local tostring = tostring local tonumber = tonumber local coroutine = coroutine local assert = assert local pairs = pairs local pcall = pcall local profile = require "profile" coroutine.resume = profile.resume coroutine.yield = profile.yield local proto = {} local skynet = { -- read skynet.h PTYPE_TEXT = 0, PTYPE_RESPONSE = 1, -- PTYPE_MULTICAST = 2, -- DEPRECATED PTYPE_CLIENT = 3, PTYPE_SYSTEM = 4, PTYPE_HARBOR = 5, PTYPE_SOCKET = 6, PTYPE_ERROR = 7, PTYPE_QUEUE = 8, PTYPE_DEBUG = 9, PTYPE_LUA = 10 } -- code cache skynet.cache = require "skynet.codecache" function skynet.register_protocol(class) local name = class.name local id = class.id assert(proto[name] == nil) assert(type(name) == "string" and type(id) == "number" and id >=0 and id <=255) proto[name] = class proto[id] = class end local session_id_coroutine = {} local session_coroutine_id = {} local session_coroutine_address = {} local wakeup_session = {} local sleep_session = {} local watching_service = {} local watching_session = {} local error_queue = {} -- suspend is function local suspend local function string_to_handle(str) return tonumber("0x" .. string.sub(str , 2)) end ----- monitor exit local function dispatch_error_queue() local session = table.remove(error_queue,1) if session then local co = session_id_coroutine[session] session_id_coroutine[session] = nil return suspend(co, coroutine.resume(co, false)) end end local function _error_dispatch(error_session, monitor, service) if service then -- service is down -- Don't remove from watching_service , because user may call dead service watching_service[service] = false for session, srv in pairs(watching_session) do if srv == service then table.insert(error_queue, session) end end else -- capture an error for error_session if watching_session[error_session] then table.insert(error_queue, error_session) end end end local watch_monitor function skynet.watch(service) assert(type(service) == "number") if watch_monitor == nil then watch_monitor = string_to_handle(c.command("MONITOR")) assert(watch_monitor, "Need a monitor") end if watching_service[service] == nil then watching_service[service] = true -- read lualib/simplemonitor.lua assert(skynet.call(watch_monitor, "lua", "WATCH", service), "watch a dead service") end end -- coroutine reuse local coroutine_pool = {} local coroutine_yield = coroutine.yield local function co_create(f) local co = table.remove(coroutine_pool) if co == nil then co = coroutine.create(function(...) f(...) while true do f = nil coroutine_pool[#coroutine_pool+1] = co f = coroutine_yield "EXIT" f(coroutine_yield()) end end) else coroutine.resume(co, f) end return co end local function dispatch_wakeup() local co = next(wakeup_session) if co then wakeup_session[co] = nil local session = sleep_session[co] if session then session_id_coroutine[session] = "BREAK" return suspend(co, coroutine.resume(co, true)) end end end -- suspend is local function function suspend(co, result, command, param, size) if not result then local session = session_coroutine_id[co] local addr = session_coroutine_address[co] if session and session ~= 0 then c.send(addr, skynet.PTYPE_ERROR, session, "") end session_coroutine_id[co] = nil session_coroutine_address[co] = nil error(debug.traceback(co,tostring(command))) end if command == "CALL" then session_id_coroutine[param] = co elseif command == "SLEEP" then session_id_coroutine[param] = co sleep_session[co] = param elseif command == "RETURN" then local co_session = session_coroutine_id[co] local co_address = session_coroutine_address[co] if param == nil then error(debug.traceback(co)) end c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) return suspend(co, coroutine.resume(co)) elseif command == "EXIT" then -- coroutine exit session_coroutine_id[co] = nil session_coroutine_address[co] = nil else error("Unknown command : " .. command .. "\n" .. debug.traceback(co)) end dispatch_wakeup() dispatch_error_queue() end function skynet.timeout(ti, func) local session = c.command("TIMEOUT",tostring(ti)) assert(session) session = tonumber(session) local co = co_create(func) assert(session_id_coroutine[session] == nil) session_id_coroutine[session] = co end function skynet.sleep(ti) local session = c.command("TIMEOUT",tostring(ti)) assert(session) session = tonumber(session) local succ, ret = coroutine_yield("SLEEP", session) sleep_session[coroutine.running()] = nil if ret == true then return "BREAK" end end function skynet.yield() return skynet.sleep("0") end function skynet.wait() local session = c.genid() coroutine_yield("SLEEP", session) local co = coroutine.running() sleep_session[co] = nil session_id_coroutine[session] = nil end function skynet.register(name) c.command("REG", name) end function skynet.name(name, handle) c.command("NAME", name .. " " .. handle) end local self_handle function skynet.self() if self_handle then return self_handle end self_handle = string_to_handle(c.command("REG")) return self_handle end function skynet.localname(name) return string_to_handle(c.command("QUERY", name)) end function skynet.launch(...) local addr = c.command("LAUNCH", table.concat({...}," ")) if addr then return string_to_handle(addr) end end function skynet.now() return tonumber(c.command("NOW")) end function skynet.starttime() return tonumber(c.command("STARTTIME")) end function skynet.exit() skynet.send(".launcher","lua","REMOVE",skynet.self()) c.command("EXIT") end function skynet.kill(name) if type(name) == "number" then skynet.send(".launcher","lua","REMOVE",name) name = skynet.address(name) end c.command("KILL",name) end function skynet.getenv(key) return c.command("GETENV",key) end function skynet.setenv(key, value) c.command("SETENV",key .. " " ..value) 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 skynet.genid = assert(c.genid) skynet.redirect = function(dest,source,typename,...) return c.redirect(dest, source, proto[typename].id, ...) end skynet.pack = assert(c.pack) skynet.unpack = assert(c.unpack) skynet.tostring = assert(c.tostring) local function yield_call(service, session) watching_session[session] = service local succ, msg, sz = coroutine_yield("CALL", session) watching_session[session] = nil assert(succ, "Capture an error") return msg,sz 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)) end return p.unpack(yield_call(addr, session)) end function skynet.blockcall(addr, typename , ...) local p = proto[typename] c.command("LOCK") local session = c.send(addr, p.id , nil , p.pack(...)) if session == nil then c.command("UNLOCK") error("call to invalid address " .. skynet.address(addr)) end return p.unpack(yield_call(addr, session)) end function skynet.rawcall(addr, typename, msg, sz) local p = proto[typename] 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) end function skynet.retpack(...) return skynet.ret(skynet.pack(...)) end function skynet.wakeup(co) if sleep_session[co] and wakeup_session[co] == nil then wakeup_session[co] = true return true end end function skynet.dispatch(typename, func) local p = assert(proto[typename],tostring(typename)) assert(p.dispatch == nil, tostring(typename)) p.dispatch = func end local function unknown_request(session, address, msg, sz) print("Unknown request :" , c.tostring(msg,sz)) error(string.format("Unknown session : %d from %x", session, address)) end function skynet.dispatch_unknown_request(unknown) local prev = unknown_request unknown_request = unknown return prev end local function unknown_response(session, address, msg, sz) print("Response message :" , c.tostring(msg,sz)) error(string.format("Unknown session : %d from %x", session, address)) end function skynet.dispatch_unknown_response(unknown) local prev = unknown_response unknown_response = unknown return prev end local fork_queue = {} local tunpack = table.unpack function skynet.fork(func,...) local args = { ... } local co = co_create(function() func(tunpack(args)) end) table.insert(fork_queue, co) end local function raw_dispatch_message(prototype, msg, sz, session, source, ...) -- skynet.PTYPE_RESPONSE = 1, read skynet.h if prototype == 1 then local co = session_id_coroutine[session] if co == "BREAK" then session_id_coroutine[session] = nil elseif co == nil then unknown_response(session, source, msg, sz) else session_id_coroutine[session] = nil suspend(co, coroutine.resume(co, true, msg, sz)) end else local p = assert(proto[prototype], prototype) local f = p.dispatch if f then local co = co_create(f) session_coroutine_id[co] = session session_coroutine_address[co] = source suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz, ...))) else unknown_request(session, source, msg, sz) end end end local function dispatch_message(...) local succ, err = pcall(raw_dispatch_message,...) while true do local key,co = next(fork_queue) if co == nil then break end fork_queue[key] = nil local fork_succ, fork_err = pcall(suspend,co,coroutine.resume(co)) if not fork_succ then if succ then succ = false err = tostring(fork_err) else err = tostring(err) .. "\n" .. tostring(fork_err) end end end assert(succ, tostring(err)) end function skynet.newservice(name, ...) local param = table.concat({"snlua", name, ...}, " ") local handle = skynet.call(".launcher", "text" , param) if handle == "" then return nil else return string_to_handle(handle) end end function skynet.uniqueservice(global, ...) local handle if global == true then handle = skynet.call("SERVICE", "lua", "LAUNCH", ...) else handle = skynet.call(".service", "lua", "LAUNCH", global, ...) end assert(handle , "Unique service launch failed") return handle end function skynet.queryservice(global, ...) local handle if global == true then handle = skynet.call("SERVICE", "lua", "QUERY", ...) else handle = skynet.call(".service", "lua", "QUERY", global, ...) end assert(handle , "Unique service query failed") return handle end local function group_command(cmd, handle, address) if address then return string.format("%s %d :%x",cmd, handle, address) else return string.format("%s %d",cmd,handle) end end function skynet.address(addr) if type(addr) == "number" then return string.format(":%x",addr) else return tostring(addr) end end function skynet.harbor(addr) return c.harbor(addr) end function skynet.error(...) local t = {...} for i=1,#t do t[i] = tostring(t[i]) end return c.error(table.concat(t, " ")) end ----- debug local internal_info_func function skynet.info_func(func) internal_info_func = func end local dbgcmd = {} function dbgcmd.MEM() local kb, bytes = collectgarbage "count" skynet.ret(skynet.pack(kb,bytes)) end function dbgcmd.GC() coroutine_pool = {} collectgarbage "collect" end function dbgcmd.STAT() local stat = {} stat.mqlen = skynet.mqlen() skynet.ret(skynet.pack(stat)) end function dbgcmd.INFO() if internal_info_func then skynet.ret(skynet.pack(internal_info_func())) else skynet.ret(skynet.pack(nil)) end end local function _debug_dispatch(session, address, cmd, ...) local f = dbgcmd[cmd] assert(f, cmd) f(...) end ----- register protocol do local REG = skynet.register_protocol REG { name = "text", id = skynet.PTYPE_TEXT, pack = function (...) local n = select ("#" , ...) if n == 0 then return "" elseif n == 1 then return tostring(...) else return table.concat({...}," ") end end, unpack = c.tostring } REG { name = "lua", id = skynet.PTYPE_LUA, pack = skynet.pack, unpack = skynet.unpack, } REG { name = "response", id = skynet.PTYPE_RESPONSE, } REG { name = "debug", id = skynet.PTYPE_DEBUG, pack = skynet.pack, unpack = skynet.unpack, dispatch = _debug_dispatch, } REG { name = "error", id = skynet.PTYPE_ERROR, pack = skynet.pack, unpack = skynet.unpack, dispatch = _error_dispatch, } end local init_func = {} function skynet.init(f, name) assert(type(f) == "function") if init_func == nil then f() else if name == nil then table.insert(init_func, f) else assert(init_func[name] == nil) init_func[name] = f end end end local function init_all() local funcs = init_func init_func = nil for k,v in pairs(funcs) do v() end end local function init_template(start) init_all() init_func = {} start() init_all() end local function init_service(start) local ok, err = xpcall(init_template, debug.traceback, start) if not ok then print("init service failed:", err) skynet.send(".launcher","text", "ERROR") skynet.exit() else skynet.send(".launcher","text", "") end end function skynet.start(start_func) c.callback(dispatch_message) skynet.timeout(0, function() init_service(start_func) end) end function skynet.filter(f ,start_func) c.callback(function(...) dispatch_message(f(...)) end) skynet.timeout(0, function() init_service(start_func) end) end function skynet.endless() return c.command("ENDLESS")~=nil end function skynet.abort() c.command("ABORT") end function skynet.monitor(service, query) local monitor if query then monitor = skynet.queryservice(true, service) else monitor = skynet.uniqueservice(true, service) end assert(monitor, "Monitor launch failed") c.command("MONITOR", string.format(":%08x", monitor)) end function skynet.mqlen() return tonumber(c.command "MQLEN") end return skynet