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, PTYPE_CLIENT = 3, PTYPE_SYSTEM = 4, PTYPE_HARBOR = 5, PTYPE_SOCKET = 6, PTYPE_ERROR = 7, PTYPE_QUEUE = 8, PTYPE_DEBUG = 9, PTYPE_LUA = 10, PTYPE_SNAX = 11, } -- 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, 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 for session, srv in pairs(watching_session) do if srv == error_source 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 -- 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 .. " " .. skynet.address(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) local ret = c.command("GETENV",key) if ret == "" then return else return ret end 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, debug.traceback()) 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.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 handle = skynet.tostring(skynet.rawcall(".launcher", "lua" , skynet.pack("LAUNCH", "snlua", name, ...))) if handle == "" then return nil else return string_to_handle(handle) end end function skynet.uniqueservice(global, ...) if global == true then return assert(skynet.call(".service", "lua", "GLAUNCH", ...)) else return assert(skynet.call(".service", "lua", "LAUNCH", global, ...)) end end function skynet.queryservice(global, ...) if global == true then return assert(skynet.call(".service", "lua", "GQUERY", ...)) else return assert(skynet.call(".service", "lua", "QUERY", global, ...)) 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 ----- register protocol do local REG = skynet.register_protocol REG { name = "lua", id = skynet.PTYPE_LUA, pack = skynet.pack, unpack = skynet.unpack, } REG { name = "response", id = skynet.PTYPE_RESPONSE, } REG { name = "error", id = skynet.PTYPE_ERROR, unpack = function(...) return ... end, 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","lua", "ERROR") skynet.exit() else skynet.send(".launcher","lua", "LAUNCHOK") 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 -- Inject internal debug framework local debug = require "skynet.debug" debug(skynet) return skynet