diff --git a/HISTORY.md b/HISTORY.md new file mode 100644 index 00000000..3424167b --- /dev/null +++ b/HISTORY.md @@ -0,0 +1,32 @@ +Dev version +----------- + +* Rewrite malloc hook , use `pthread_getspecific` instead of `__thread` to get current service handle. +* Optimize global unique service query, rewrite `service_mgr` . +* Add some snax api, snax.uniqueservice (etc.) , use independent protocol `PTYPE_SNAX` . +* Add bootstrap lua script , remove some code in C . +* Use a lua loader to load lua service code (and set the lua environment), remove some code in C. +* Support preload a file before each lua serivce start. +* Add datacenter serivce. +* Add multicast api. +* Remove skynet.blockcall , simplify the implement of message queue. +* When dropping message queue (at service exit) , dispatcher will post an error back to the source of each message. +* Remove skynet.watch , monitor is not necessary for watching skynet.call . so simplemonitor.lua is move to examples. +* Remove the limit of global queue size (64K actived service limit before). + +v0.1.1 (2014-4-28) +------------------ + +* Socket channel should clear request queue when reconnect. +* Fix the issue that socket close may block the coroutine. +* Fix the issue that jemalloc api may crash on macosx (disable jemalloc on macosx). + +v0.1.0 (2014-4-23) +------------------ + +* First release version. + +First public version (2012-8-1) +------------------ + +* Make skynet from a private project to public. diff --git a/Makefile b/Makefile index 7c57002e..d7bb53cf 100644 --- a/Makefile +++ b/Makefile @@ -5,7 +5,7 @@ CSERVICE_PATH ?= cservice SKYNET_BUILD_PATH ?= . -CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS) $(SKYNET_DEFINES) +CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS) # lua @@ -41,7 +41,7 @@ jemalloc : $(MALLOC_STATICLIB) # skynet CSERVICE = snlua logger gate master harbor -LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile +LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile multicast SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \ skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \ @@ -54,7 +54,7 @@ all : \ $(foreach v, $(LUA_CLIB), $(LUA_CLIB_PATH)/$(v).so) $(SKYNET_BUILD_PATH)/skynet : $(foreach v, $(SKYNET_SRC), skynet-src/$(v)) $(LUA_LIB) $(MALLOC_STATICLIB) - $(CC) $(CFLAGS) -o $@ $^ -Iskynet-src -I$(JEMALLOC_INC) $(LDFLAGS) $(EXPORT) $(SKYNET_LIBS) + $(CC) $(CFLAGS) -o $@ $^ -Iskynet-src -I$(JEMALLOC_INC) $(LDFLAGS) $(EXPORT) $(SKYNET_LIBS) $(SKYNET_DEFINES) $(LUA_CLIB_PATH) : mkdir $(LUA_CLIB_PATH) @@ -91,7 +91,7 @@ $(LUA_CLIB_PATH)/netpack.so : lualib-src/lua-netpack.c | $(LUA_CLIB_PATH) $(CC) $(CFLAGS) $(SHARED) $^ -Iskynet-src -o $@ $(LUA_CLIB_PATH)/cjson.so : | $(LUA_CLIB_PATH) - cd 3rd/lua-cjson && $(MAKE) LUA_INCLUDE_DIR=../../$(LUA_INC) CC=$(CC) CJSON_LDFLAGS="$(SHARED)" && cp cjson.so ../../$@ + cd 3rd/lua-cjson && $(MAKE) LUA_INCLUDE_DIR=../../$(LUA_INC) CC=$(CC) CJSON_LDFLAGS="$(SHARED)" && cd ../.. && cp 3rd/lua-cjson/cjson.so $@ $(LUA_CLIB_PATH)/clientsocket.so : lualib-src/lua-clientsocket.c | $(LUA_CLIB_PATH) $(CC) $(CFLAGS) $(SHARED) $^ -o $@ -lpthread @@ -102,6 +102,9 @@ $(LUA_CLIB_PATH)/memory.so : lualib-src/lua-memory.c | $(LUA_CLIB_PATH) $(LUA_CLIB_PATH)/profile.so : lualib-src/lua-profile.c | $(LUA_CLIB_PATH) $(CC) $(CFLAGS) $(SHARED) $^ -o $@ +$(LUA_CLIB_PATH)/multicast.so : lualib-src/lua-multicast.c | $(LUA_CLIB_PATH) + $(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ + clean : rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so diff --git a/examples/config b/examples/config index f59fd8ac..f89af28b 100644 --- a/examples/config +++ b/examples/config @@ -4,8 +4,11 @@ logger = nil harbor = 1 address = "127.0.0.1:2526" master = "127.0.0.1:2013" -start = "main" +start = "main" -- main script +bootstrap = "snlua bootstrap" -- The service for bootstrap standalone = "0.0.0.0:2013" luaservice = root.."service/?.lua;"..root.."test/?.lua;"..root.."examples/?.lua" +lualoader = "lualib/loader.lua" +-- preload = "./examples/preload.lua" -- run preload.lua before every lua service run snax = root.."examples/?.lua;"..root.."test/?.lua" cpath = root.."cservice/?.so" diff --git a/examples/config.mc b/examples/config.mc new file mode 100644 index 00000000..2e019784 --- /dev/null +++ b/examples/config.mc @@ -0,0 +1,14 @@ +root = "./" +thread = 8 +logger = nil +harbor = 2 +address = "127.0.0.1:2527" +master = "127.0.0.1:2013" +start = "testmulticast2" -- main script +bootstrap = "snlua bootstrap" -- The service for bootstrap +--standalone = "0.0.0.0:2013" +luaservice = root.."service/?.lua;"..root.."test/?.lua;"..root.."examples/?.lua" +lualoader = "lualib/loader.lua" +-- preload = "./examples/preload.lua" -- run preload.lua before every lua service run +snax = root.."examples/?.lua;"..root.."test/?.lua" +cpath = root.."cservice/?.so" diff --git a/examples/config_log b/examples/config_log index e7ba2eb4..9261e435 100644 --- a/examples/config_log +++ b/examples/config_log @@ -7,3 +7,4 @@ address = "127.0.0.1:2527" master = "127.0.0.1:2013" start = "main_log" luaservice ="./service/?.lua;./test/?.lua;./examples/?.lua" +snax = "./examples/?.lua;./test/?.lua" diff --git a/examples/main.lua b/examples/main.lua index 8c459ef6..8019c2bf 100644 --- a/examples/main.lua +++ b/examples/main.lua @@ -4,10 +4,8 @@ local max_client = 64 skynet.start(function() print("Server start") - local service = skynet.newservice("service_mgr") - skynet.monitor "simplemonitor" local console = skynet.newservice("console") --- skynet.newservice("debug_console",8000) + skynet.newservice("debug_console",8000) skynet.newservice("simpledb") local watchdog = skynet.newservice("watchdog") skynet.call(watchdog, "lua", "start", { diff --git a/examples/preload.lua b/examples/preload.lua new file mode 100644 index 00000000..f4f31034 --- /dev/null +++ b/examples/preload.lua @@ -0,0 +1,5 @@ +-- This file will execute before every lua service start +-- See config + +print("PRELOAD", ...) + diff --git a/service/simplemonitor.lua b/examples/simplemonitor.lua similarity index 89% rename from service/simplemonitor.lua rename to examples/simplemonitor.lua index 56653d0c..1528bb50 100644 --- a/service/simplemonitor.lua +++ b/examples/simplemonitor.lua @@ -12,11 +12,10 @@ skynet.register_protocol { local w = service_map[address] if w then for watcher in pairs(w) do - skynet.send(watcher, "error", address) + skynet.redirect(watcher, address, "error", "") end service_map[address] = false end - print(string.format("[:%x] exit", address)) end } diff --git a/lualib-src/lua-mongo.c b/lualib-src/lua-mongo.c index 166e6683..a0d312fd 100644 --- a/lualib-src/lua-mongo.c +++ b/lualib-src/lua-mongo.c @@ -55,7 +55,7 @@ little_endian(uint32_t v) { typedef void * document; static inline uint32_t -get_length(const document buffer) { +get_length(document buffer) { union { uint32_t v; uint8_t b[4]; @@ -243,7 +243,7 @@ op_reply(lua_State *L) { lua_pushlightuserdata(L, (void *)doc); lua_rawseti(L, 2, i); - int32_t doc_len = get_length((const document)doc); + int32_t doc_len = get_length((document)doc); doc += doc_len; sz -= doc_len; diff --git a/lualib-src/lua-multicast.c b/lualib-src/lua-multicast.c new file mode 100644 index 00000000..25b9d936 --- /dev/null +++ b/lualib-src/lua-multicast.c @@ -0,0 +1,167 @@ +#include "skynet.h" + +#include +#include +#include +#include + +struct mc_package { + int reference; + uint32_t size; + void *data; +}; + +static int +pack(lua_State *L, void *data, size_t size) { + struct mc_package * pack = skynet_malloc(sizeof(struct mc_package)); + pack->reference = 0; + pack->size = (uint32_t)size; + pack->data = data; + struct mc_package ** ret = skynet_malloc(sizeof(*ret)); + *ret = pack; + lua_pushlightuserdata(L, ret); + lua_pushinteger(L, sizeof(ret)); + return 2; +} + +/* + lightuserdata + integer size + + return lightuserdata, sizeof(struct mc_package *) + */ +static int +mc_packlocal(lua_State *L) { + void * data = lua_touserdata(L, 1); + size_t size = luaL_checkunsigned(L, 2); + if (size != (uint32_t)size) { + return luaL_error(L, "Size should be 32bit integer"); + } + return pack(L, data, size); +} + +/* + lightuserdata + integer size + + return lightuserdata, sizeof(struct mc_package *) + */ +static int +mc_packremote(lua_State *L) { + void * data = lua_touserdata(L, 1); + size_t size = luaL_checkunsigned(L, 2); + if (size != (uint32_t)size) { + return luaL_error(L, "Size should be 32bit integer"); + } + void * msg = skynet_malloc(size); + memcpy(msg, data, size); + return pack(L, msg, size); +} + +static int +mc_packstring(lua_State *L) { + size_t size; + const char * msg = luaL_checklstring(L, 1, &size); + if (size != (uint32_t)size) { + return luaL_error(L, "string is too long"); + } + void * data = skynet_malloc(size); + memcpy(data, msg, size); + return pack(L, data, size); +} + +/* + lightuserdata struct mc_package ** + integer size (must be sizeof(struct mc_package *) + + return package, lightuserdata, size + */ +static int +mc_unpacklocal(lua_State *L) { + struct mc_package ** pack = lua_touserdata(L,1); + int sz = luaL_checkinteger(L,2); + if (sz != sizeof(*pack)) { + return luaL_error(L, "Invalid multicast package size %d", sz); + } + lua_settop(L, 1); + lua_pushlightuserdata(L, (*pack)->data); + lua_pushunsigned(L, (*pack)->size); + return 3; +} + +/* + lightuserdata struct mc_package ** + integer reference + */ +static int +mc_bindrefer(lua_State *L) { + struct mc_package ** pack = lua_touserdata(L,1); + int ref = luaL_checkinteger(L,2); + if ((*pack)->reference != 0) { + return luaL_error(L, "Can't bind a multicast package more than once"); + } + (*pack)->reference = ref; + + return 0; +} + +/* + lightuserdata struct mc_package ** + */ +static int +mc_closelocal(lua_State *L) { + struct mc_package **ptr = lua_touserdata(L,1); + struct mc_package *pack = *ptr; + + int ref = __sync_sub_and_fetch(&pack->reference, 1); + if (ref <= 0) { + skynet_free(pack->data); + skynet_free(pack); + if (ref < 0) { + return luaL_error(L, "Invalid multicast package reference %d", ref); + } + } + + return 0; +} + +/* + lightuserdata struct mc_package ** + return lightuserdata/size + */ +static int +mc_remote(lua_State *L) { + struct mc_package **ptr = lua_touserdata(L,1); + struct mc_package *pack = *ptr; + lua_pushlightuserdata(L, pack->data); + lua_pushunsigned(L, pack->size); + skynet_free(pack); + return 2; +} + +static int +mc_nextid(lua_State *L) { + uint32_t id = luaL_checkunsigned(L, 1); + id += 256; + lua_pushunsigned(L, id); + + return 1; +} + +int +luaopen_multicast_c(lua_State *L) { + luaL_Reg l[] = { + { "pack", mc_packlocal }, + { "unpack", mc_unpacklocal }, + { "bind", mc_bindrefer }, + { "close", mc_closelocal }, + { "remote", mc_remote }, + { "packstring", mc_packstring }, + { "packremote", mc_packremote }, + { "nextid", mc_nextid }, + { NULL, NULL }, + }; + luaL_checkversion(L); + luaL_newlib(L,l); + return 1; +} diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index 502894b2..0025dd52 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -1,6 +1,5 @@ #include "skynet.h" #include "lua-seri.h" -#include "service_lua.h" #define KNRM "\x1B[0m" #define KRED "\x1B[31m" @@ -11,18 +10,36 @@ #include #include +struct snlua { + lua_State * L; + struct skynet_context * ctx; + const char * preload; +}; + +static int +traceback (lua_State *L) { + const char *msg = lua_tostring(L, 1); + if (msg) + luaL_traceback(L, L, msg, 1); + else { + lua_pushliteral(L, "(no error message)"); + } + return 1; +} + static int _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { lua_State *L = ud; int trace = 1; int r; int top = lua_gettop(L); - if (top == 1) { + if (top == 0) { + lua_pushcfunction(L, traceback); lua_rawgetp(L, LUA_REGISTRYINDEX, _cb); } else { assert(top == 2); - lua_pushvalue(L,2); } + lua_pushvalue(L,2); lua_pushinteger(L, type); lua_pushlightuserdata(L, (void *)msg); @@ -290,15 +307,14 @@ luaopen_skynet_c(lua_State *L) { { NULL, NULL }, }; - lua_getfield(L, LUA_REGISTRYINDEX, "skynet_lua"); - struct snlua *lua = lua_touserdata(L,-1); - if (lua == NULL || lua->ctx == NULL) { + luaL_newlibtable(L, l); + + lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context"); + struct skynet_context *ctx = lua_touserdata(L,-1); + if (ctx == NULL) { return luaL_error(L, "Init skynet context first"); } - assert(lua->L == L); - luaL_newlibtable(L, l); - lua_pushlightuserdata(L, lua->ctx); luaL_setfuncs(L,l,1); return 1; diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index c3ba9f3b..3f22956c 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -10,7 +10,6 @@ #include #include "skynet_socket.h" -#include "service_lua.h" #define BACKLOG 32 // 2 ** 12 == 4096 @@ -485,15 +484,12 @@ luaopen_socketdriver(lua_State *L) { { "start", lstart }, { NULL, NULL }, }; - lua_getfield(L, LUA_REGISTRYINDEX, "skynet_lua"); - struct snlua *lua = lua_touserdata(L,-1); - if (lua == NULL || lua->ctx == NULL) { + lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context"); + struct skynet_context *ctx = lua_touserdata(L,-1); + if (ctx == NULL) { return luaL_error(L, "Init skynet context first"); } - // assert(lua->L == L); - lua_pop(L,1); - lua_pushlightuserdata(L, lua->ctx); luaL_setfuncs(L,l2,1); return 1; diff --git a/lualib/datacenter.lua b/lualib/datacenter.lua new file mode 100644 index 00000000..347c140e --- /dev/null +++ b/lualib/datacenter.lua @@ -0,0 +1,14 @@ +local skynet = require "skynet" + +local datacenter = {} + +function datacenter.get(...) + return skynet.call("DATACENTER", "lua", "QUERY", ...) +end + +function datacenter.set(...) + return skynet.call("DATACENTER", "lua", "UPDATE", ...) +end + +return datacenter + diff --git a/lualib/loader.lua b/lualib/loader.lua new file mode 100644 index 00000000..05224b92 --- /dev/null +++ b/lualib/loader.lua @@ -0,0 +1,46 @@ +local args = {} +for word in string.gmatch(..., "%S+") do + table.insert(args, word) +end + +local main, pattern + +local err = {} +for pat in string.gmatch(LUA_SERVICE, "([^;]+);*") do + local filename = string.gsub(pat, "?", args[1]) + local f, msg = loadfile(filename) + if not f then + table.insert(err, msg) + else + pattern = pat + main = f + break + end +end + +if not main then + error(table.concat(err, "\n")) +end + +LUA_SERVICE = nil +package.path , LUA_PATH = LUA_PATH +package.cpath , LUA_CPATH = LUA_CPATH + +local service_path = string.match(pattern, "(.*/)[^/?]+$") + +if service_path then + service_path = string.gsub(service_path, "?", args[1]) + package.path = service_path .. "?.lua;" .. package.path + SERVICE_PATH = service_path +else + local p = string.match(pattern, "(.*/).+$") + SERVICE_PATH = p +end + +if LUA_PRELOAD then + local f = assert(loadfile(LUA_PRELOAD)) + f(table.unpack(args)) + LUA_PRELOAD = nil +end + +main(select(2, table.unpack(args))) diff --git a/lualib/multicast.lua b/lualib/multicast.lua new file mode 100644 index 00000000..502f8d21 --- /dev/null +++ b/lualib/multicast.lua @@ -0,0 +1,102 @@ +local skynet = require "skynet" +local mc = require "multicast.c" + +local multicastd +local multicast = {} +local dispatch = setmetatable({} , {__mode = "kv" }) + +local chan = {} +local chan_meta = { + __index = chan, + __gc = unsubscribe, + __tostring = function (self) + return string.format("[Multicast:%x]",self.channel) + end, +} + +local function default_conf(conf) + conf = conf or {} + conf.pack = conf.pack or skynet.pack + conf.unpack = conf.unpack or skynet.unpack + + return conf +end + +function multicast.new(conf) + assert(multicastd, "Init first") + local self = {} + conf = conf or self + self.channel = conf.channel + if self.channel == nil then + self.channel = skynet.call(multicastd, "lua", "NEW") + end + self.__pack = conf.pack or skynet.pack + self.__unpack = conf.unpack or skynet.unpack + self.__dispatch = conf.dispatch + + return setmetatable(self, chan_meta) +end + +function chan:delete() + local c = assert(self.channel) + skynet.send(multicastd, "lua", "DEL", c) + self.channel = nil + self.__subscribe = nil +end + +function chan:publish(...) + local c = assert(self.channel) + skynet.call(multicastd, "lua", "PUB", c, mc.pack(self.__pack(...))) +end + +function chan:subscribe() + local c = assert(self.channel) + if self.__subscribe then + -- already subscribe + return + end + skynet.call(multicastd, "lua", "SUB", c) + self.__subscribe = true + dispatch[c] = self +end + +function chan:unsubscribe() + if not self.__subscribe then + -- already unsubscribe + return + end + local c = assert(self.channel) + skynet.send(multicastd, "lua", "USUB", c) + self.__subscribe = nil +end + +local function dispatch_subscribe(channel, source, pack, msg, sz) + local self = dispatch[channel] + if not self then + mc.close(pack) + error ("Unknown channel " .. channel) + end + + if self.__subscribe then + local ok, err = pcall(self.__dispatch, self, source, self.__unpack(msg, sz)) + mc.close(pack) + assert(ok, err) + else + -- maybe unsubscribe first, but the message is send out. drop the message unneed + mc.close(pack) + end +end + +local function init() + multicastd = skynet.uniqueservice "multicastd" + skynet.register_protocol { + name = "multicast", + id = skynet.PTYPE_MULTICAST, + unpack = mc.unpack, + dispatch = dispatch_subscribe, + } +end + +skynet.init(init, "multicast") + +return multicast \ No newline at end of file diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 3d3135db..1d936052 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -16,7 +16,7 @@ local skynet = { -- read skynet.h PTYPE_TEXT = 0, PTYPE_RESPONSE = 1, --- PTYPE_MULTICAST = 2, -- DEPRECATED + PTYPE_MULTICAST = 2, PTYPE_CLIENT = 3, PTYPE_SYSTEM = 4, PTYPE_HARBOR = 5, @@ -24,7 +24,8 @@ local skynet = { PTYPE_ERROR = 7, PTYPE_QUEUE = 8, PTYPE_DEBUG = 9, - PTYPE_LUA = 10 + PTYPE_LUA = 10, + PTYPE_SNAX = 11, } -- code cache @@ -68,13 +69,13 @@ local function dispatch_error_queue() end end -local function _error_dispatch(error_session, monitor, service) - if service then +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[service] = false + watching_service[error_source] = false for session, srv in pairs(watching_session) do - if srv == service then + if srv == error_source then table.insert(error_queue, session) end end @@ -86,21 +87,6 @@ local function _error_dispatch(error_session, monitor, service) 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 = {} @@ -209,7 +195,7 @@ function skynet.register(name) end function skynet.name(name, handle) - c.command("NAME", name .. " " .. handle) + c.command("NAME", name .. " " .. skynet.address(handle)) end local self_handle @@ -254,7 +240,12 @@ function skynet.kill(name) end function skynet.getenv(key) - return c.command("GETENV",key) + local ret = c.command("GETENV",key) + if ret == "" then + return + else + return ret + end end function skynet.setenv(key, value) @@ -283,7 +274,7 @@ 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") + assert(succ, debug.traceback()) return msg,sz end @@ -299,17 +290,6 @@ function skynet.call(addr, typename, ...) 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") @@ -322,7 +302,7 @@ function skynet.ret(msg, sz) end function skynet.retpack(...) - return skynet.ret(skynet.pack(...)) + return skynet.ret(skynet.pack(...)) end function skynet.wakeup(co) @@ -339,14 +319,14 @@ function skynet.dispatch(typename, func) end local function unknown_request(session, address, msg, sz) - print("Unknown request :" , c.tostring(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 + local prev = unknown_request + unknown_request = unknown + return prev end local function unknown_response(session, address, msg, sz) @@ -393,7 +373,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...) session_coroutine_address[co] = source suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz, ...))) else - unknown_request(session, source, msg, sz) + unknown_request(session, source, msg, sz) end end end @@ -420,8 +400,7 @@ local function dispatch_message(...) end function skynet.newservice(name, ...) - local param = table.concat({"snlua", name, ...}, " ") - local handle = skynet.call(".launcher", "text" , param) + local handle = skynet.tostring(skynet.rawcall(".launcher", "lua" , skynet.pack("LAUNCH", "snlua", name, ...))) if handle == "" then return nil else @@ -430,32 +409,18 @@ function skynet.newservice(name, ...) end function skynet.uniqueservice(global, ...) - local handle if global == true then - handle = skynet.call("SERVICE", "lua", "LAUNCH", ...) + return assert(skynet.call(".service", "lua", "GLAUNCH", ...)) else - handle = skynet.call(".service", "lua", "LAUNCH", global, ...) + return assert(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", ...) + return assert(skynet.call(".service", "lua", "GQUERY", ...)) 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) + return assert(skynet.call(".service", "lua", "QUERY", global, ...)) end end @@ -479,66 +444,10 @@ function skynet.error(...) 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, @@ -551,19 +460,10 @@ do 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, + unpack = function(...) return ... end, dispatch = _error_dispatch, } end @@ -603,10 +503,10 @@ 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.send(".launcher","lua", "ERROR") skynet.exit() else - skynet.send(".launcher","text", "") + skynet.send(".launcher","lua", "LAUNCHOK") end end @@ -649,4 +549,8 @@ function skynet.mqlen() return tonumber(c.command "MQLEN") end +-- Inject internal debug framework +local debug = require "skynet.debug" +debug(skynet) + return skynet diff --git a/lualib/skynet/debug.lua b/lualib/skynet/debug.lua new file mode 100644 index 00000000..807ee2ae --- /dev/null +++ b/lualib/skynet/debug.lua @@ -0,0 +1,49 @@ +return function (skynet) + +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 + +skynet.register_protocol { + name = "debug", + id = assert(skynet.PTYPE_DEBUG), + pack = assert(skynet.pack), + unpack = assert(skynet.unpack), + dispatch = _debug_dispatch, +} + +end \ No newline at end of file diff --git a/lualib/snax.lua b/lualib/snax.lua index 67dafc7e..f810e392 100644 --- a/lualib/snax.lua +++ b/lualib/snax.lua @@ -1,11 +1,19 @@ local skynet = require "skynet" -local snax_interface = require "snax_interface" +local snax_interface = require "snax.interface" local snax = {} local typeclass = {} local G = { require = function() end } +skynet.register_protocol { + name = "snax", + id = skynet.PTYPE_SNAX, + pack = skynet.pack, + unpack = skynet.unpack, +} + + function snax.interface(name) if typeclass[name] then return typeclass[name] @@ -38,7 +46,7 @@ local function gen_post(type, handle) __index = function( t, k ) local id = assert(type.accept[k] , string.format("post %s no exist", k)) return function(...) - skynet_send(handle, "lua", id, ...) + skynet_send(handle, "snax", id, ...) end end }) end @@ -48,7 +56,7 @@ local function gen_req(type, handle) __index = function( t, k ) local id = assert(type.response[k] , string.format("request %s no exist", k)) return function(...) - return skynet_call(handle, "lua", id, ...) + return skynet_call(handle, "snax", id, ...) end end }) end @@ -64,16 +72,14 @@ end local handle_cache = setmetatable( {} , { __mode = "kv" } ) -function snax.newservice(name, ...) +function snax.rawnewservice(name, ...) local t = snax.interface(name) local handle = skynet.newservice("snaxd", name) assert(handle_cache[handle] == nil) if t.system.init then - skynet.call(handle, "lua", t.system.init, ...) + skynet.call(handle, "snax", t.system.init, ...) end - local ret = wrapper(handle, name, t) - handle_cache[handle] = ret - return ret + return handle end function snax.bind(handle, type) @@ -88,9 +94,42 @@ function snax.bind(handle, type) return ret end +function snax.newservice(name, ...) + local handle = snax.rawnewservice(name, ...) + return snax.bind(handle, name) +end + +local function service_name(global, name, ...) + if global == true then + return name + else + return global + end +end + +function snax.uniqueservice(name, ...) + local handle = assert(skynet.call(".service", "lua", "LAUNCH", "snaxd", name, ...)) + return snax.bind(handle, name) +end + +function snax.globalservice(name, ...) + local handle = assert(skynet.call(".service", "lua", "GLAUNCH", "snaxd", name, ...)) + return snax.bind(handle, name) +end + +function snax.queryservice(name) + local handle = assert(skynet.call(".service", "lua", "QUERY", "snaxd", name)) + return snax.bind(handle, name) +end + +function snax.queryglobal(name) + local handle = assert(skynet.call(".service", "lua", "GQUERY", "snaxd", name)) + return snax.bind(handle, name) +end + function snax.kill(obj, ...) local t = snax.interface(obj.type) - skynet_call(obj.handle, "lua", t.system.exit, ...) + skynet_call(obj.handle, "snax", t.system.exit, ...) end local function test_result(ok, ...) @@ -103,7 +142,7 @@ end function snax.hotfix(obj, source, ...) local t = snax.interface(obj.type) - return test_result(skynet_call(obj.handle, "lua", t.system.hotfix, source, ...)) + return test_result(skynet_call(obj.handle, "snax", t.system.hotfix, source, ...)) end return snax diff --git a/lualib/snax_hotfix.lua b/lualib/snax/hotfix.lua similarity index 98% rename from lualib/snax_hotfix.lua rename to lualib/snax/hotfix.lua index f42ab0c6..e0cc2a7f 100644 --- a/lualib/snax_hotfix.lua +++ b/lualib/snax/hotfix.lua @@ -1,4 +1,4 @@ -local si = require "snax_interface" +local si = require "snax.interface" local io = io local hotfix = {} diff --git a/lualib/snax_interface.lua b/lualib/snax/interface.lua similarity index 100% rename from lualib/snax_interface.lua rename to lualib/snax/interface.lua diff --git a/platform.mk b/platform.mk index 975cf293..a112f0f2 100644 --- a/platform.mk +++ b/platform.mk @@ -36,6 +36,5 @@ linux freebsd : SKYNET_LIBS += -lrt macosx : MALLOC_STATICLIB := macosx : SKYNET_DEFINES :=-DNOUSE_JEMALLOC - linux macosx freebsd : $(MAKE) all PLAT=$@ SKYNET_LIBS="$(SKYNET_LIBS)" SHARED="$(SHARED)" EXPORT="$(EXPORT)" MALLOC_STATICLIB="$(MALLOC_STATICLIB)" SKYNET_DEFINES="$(SKYNET_DEFINES)" diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 4c5fe48b..b881acb2 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -402,9 +402,8 @@ _remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int _send_remote(context, fd, msg,sz,&cookie); } else { // throw an error return to source - if (session != 0) { - skynet_send(context, destination, source, PTYPE_RESERVED_ERROR, session, NULL, 0); - } + // report the destination is dead + skynet_send(context, destination, source, PTYPE_ERROR, 0 , NULL, 0); skynet_error(context, "Drop message to harbor %d from %x to %x (session = %d, msgsz = %d)",harbor_id, source, destination,session,(int)sz); } return 0; @@ -593,6 +592,8 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) { fprintf(stderr, "Harbor: Connect to master failed\n"); exit(1); } + skynet_harbor_start(ctx); + h->local_addr = skynet_strdup(local_addr); _launch_gate(ctx, local_addr); diff --git a/service-src/service_lua.h b/service-src/service_lua.h deleted file mode 100644 index 1daff429..00000000 --- a/service-src/service_lua.h +++ /dev/null @@ -1,10 +0,0 @@ -#ifndef SKYNET_SERVICE_LUA_H -#define SKYNET_SERVICE_LUA_H - -struct snlua { - lua_State * L; - struct skynet_context * ctx; - int (*init)(struct snlua *l, struct skynet_context *ctx, const char * args); -}; - -#endif diff --git a/service-src/service_snlua.c b/service-src/service_snlua.c index c162a528..33e0a5ff 100644 --- a/service-src/service_snlua.c +++ b/service-src/service_snlua.c @@ -4,13 +4,17 @@ #include #include #include -#include "service_lua.h" #include #include #include #include +struct snlua { + lua_State * L; + struct skynet_context * ctx; +}; + // LUA_CACHELIB may defined in patched lua for shared proto #ifdef LUA_CACHELIB @@ -30,121 +34,13 @@ codecache(lua_State *L) { { NULL, NULL }, }; luaL_newlib(L,l); - lua_getglobal(L, "loadfile"); - lua_setfield(L, -2, "loadfile"); + lua_getglobal(L, "loadfile"); + lua_setfield(L, -2, "loadfile"); return 1; } #endif -// time - -#include -#define NANOSEC 1000000000 - -#if defined(__APPLE__) -#include -#include -#endif - -static void -current_time(struct timespec *ti) { -#if !defined(__APPLE__) - clock_gettime(CLOCK_THREAD_CPUTIME_ID, ti); -#else - struct task_thread_times_info aTaskInfo; - mach_msg_type_number_t aTaskInfoCount = TASK_THREAD_TIMES_INFO_COUNT; - assert(KERN_SUCCESS == task_info(mach_task_self(), TASK_THREAD_TIMES_INFO, (task_info_t )&aTaskInfo, &aTaskInfoCount)); - ti->tv_sec = aTaskInfo.user_time.seconds; - ti->tv_nsec = aTaskInfo.user_time.microseconds * 1000; -#endif -} - -static double -diff_time(struct timespec *ti) { - struct timespec end; - current_time(&end); - int diffsec = end.tv_sec - ti->tv_sec; - int diffnsec = end.tv_nsec - ti->tv_nsec; - if (diffnsec < 0) { - --diffsec; - diffnsec += NANOSEC; - } - return (double)diffsec + (double)diffnsec / NANOSEC; -} - -static int -_try_load(lua_State *L, const char * path, int pathlen, const char * name) { - int namelen = strlen(name); - char tmp[pathlen + namelen]; - int i; - for (i=0;i=0;i--) { - if (tmp[i] == '/') { - lua_pushlstring(L,tmp,i+1); - lua_setglobal(L,"SERVICE_PATH"); - break; - } - } - if (i<0) { - return 0; - } - lua_getglobal(L,"package"); - lua_getfield(L,-1,"path"); - luaL_Buffer b; - luaL_buffinit(L, &b); - luaL_addlstring(&b, tmp, i+1); - luaL_addstring(&b, "?.lua;"); - luaL_addvalue(&b); - luaL_pushresult(&b); - lua_setfield(L,-2,"path"); - lua_pop(L,1); - return 0; - } else if (r == LUA_ERRFILE) { - lua_pop(L,1); - return -1; - } - return 1; -} - -static int -_load(lua_State *L, char ** filename) { - const char * name = strsep(filename, " \r\n"); - const char * path = skynet_command(NULL, "GETENV", "luaservice"); - while (path[0]) { - int pathlen; - char * pathend = strchr(path,';'); - if (pathend) { - pathlen = pathend - path; - } else { - pathlen = strlen(path); - } - int r = _try_load(L, path, pathlen, name); - if (r >=0) { - return r; - } - path+=pathlen; - if (path[0]==';') - ++path; - } - return -1; -} - static int traceback (lua_State *L) { const char *msg = lua_tostring(L, 1); @@ -156,80 +52,70 @@ traceback (lua_State *L) { return 1; } -static void -_report_error(lua_State *L, struct skynet_context *ctx, const char *filename, int err) { - switch (err) { - case LUA_ERRRUN: - skynet_error(ctx, "lua do [%s] error : %s", filename, lua_tostring(L,-1)); - break; - case LUA_ERRMEM: - skynet_error(ctx, "lua memory error : %s",filename); - break; - case LUA_ERRERR: - skynet_error(ctx, "lua message error : %s",filename); - break; - case LUA_ERRGCMM: - skynet_error(ctx, "lua gc error : %s",filename); - break; - }; - lua_pop(L,1); -} - static void _report_launcher_error(struct skynet_context *ctx) { // sizeof "ERROR" == 5 skynet_sendname(ctx, ".launcher", PTYPE_TEXT, 0, "ERROR", 5); } +static const char * +optstring(struct skynet_context *ctx, const char *key, const char * str) { + const char * ret = skynet_command(ctx, "GETENV", key); + if (ret == NULL) { + return str; + } + return ret; +} + static int -_init(struct snlua *l, struct skynet_context *ctx, const char * args) { +_init(struct snlua *l, struct skynet_context *ctx, const char * args, size_t sz) { lua_State *L = l->L; l->ctx = ctx; lua_gc(L, LUA_GCSTOP, 0); + lua_pushboolean(L, 1); /* signal for libraries to ignore env. vars. */ + lua_setfield(L, LUA_REGISTRYINDEX, "LUA_NOENV"); luaL_openlibs(L); - lua_pushlightuserdata(L, l); - lua_setfield(L, LUA_REGISTRYINDEX, "skynet_lua"); + lua_pushlightuserdata(L, ctx); + lua_setfield(L, LUA_REGISTRYINDEX, "skynet_context"); luaL_requiref(L, "skynet.codecache", codecache , 0); lua_pop(L,1); - lua_gc(L, LUA_GCRESTART, 0); - char tmp[strlen(args)+1]; - char *parm = tmp; - strcpy(parm,args); + const char *path = optstring(ctx, "lua_path","./lualib/?.lua;./lualib/?/init.lua"); + lua_pushstring(L, path); + lua_setglobal(L, "LUA_PATH"); + const char *cpath = optstring(ctx, "lua_cpath","./luaclib/?.so"); + lua_pushstring(L, cpath); + lua_setglobal(L, "LUA_CPATH"); + const char *service = optstring(ctx, "luaservice", "./service/?.lua"); + lua_pushstring(L, service); + lua_setglobal(L, "LUA_SERVICE"); + const char *preload = skynet_command(ctx, "GETENV", "preload"); + lua_pushstring(L, preload); + lua_setglobal(L, "LUA_PRELOAD"); lua_pushcfunction(L, traceback); - int traceback_index = lua_gettop(L); - assert(traceback_index == 1); + assert(lua_gettop(L) == 1); - const char * filename = parm; - int r = _load(L, &parm); - if (r != 0) { - if (r<0) { - skynet_error(ctx, "lua parser [%s] load error", filename); - } else { - skynet_error(ctx, "lua parser [%s] error : %s", filename, lua_tostring(L,-1)); - } + const char * loader = optstring(ctx, "lualoader", "./lualib/loader.lua"); + + int r = luaL_loadfile(L,loader); + if (r != LUA_OK) { + skynet_error(ctx, "Can't load %s : %s", loader, lua_tostring(L, -1)); _report_launcher_error(ctx); return 1; } - int n=0; - while(parm) { - const char * arg = strsep(&parm, " \r\n"); - if (arg && arg[0]!='\0') { - lua_pushstring(L, arg); - ++n; - } + lua_pushlstring(L, args, sz); + r = lua_pcall(L,1,0,1); + if (r != LUA_OK) { + skynet_error(ctx, "lua loader error : %s", lua_tostring(L, -1)); + _report_launcher_error(ctx); + return 1; } - r = lua_pcall(L,n,0,traceback_index); - if (r == LUA_OK) { - r = lua_gc(L, LUA_GCCOLLECT, 0); - if (r == LUA_OK) { - return 0; - } - } - _report_error(L, ctx, filename, r); - _report_launcher_error(ctx); - return 1; + lua_settop(L,0); + + lua_gc(L, LUA_GCRESTART, 0); + + return 0; } static int @@ -237,12 +123,7 @@ _launch(struct skynet_context * context, void *ud, int type, int session, uint32 assert(type == 0 && session == 0); struct snlua *l = ud; skynet_callback(context, NULL, NULL); - struct timespec ti; - current_time(&ti); - int err = _init(l, context, msg); - double t = diff_time(&ti); - lua_pushnumber(l->L, t); - lua_setfield(l->L, LUA_REGISTRYINDEX, "skynet_boottime"); + int err = _init(l, context, msg, sz); if (err) { skynet_command(context, "EXIT", NULL); } @@ -253,13 +134,13 @@ _launch(struct skynet_context * context, void *ud, int type, int session, uint32 int snlua_init(struct snlua *l, struct skynet_context *ctx, const char * args) { int sz = strlen(args); - char * tmp = skynet_malloc(sz+1); - memcpy(tmp, args, sz+1); + char * tmp = skynet_malloc(sz); + memcpy(tmp, args, sz); skynet_callback(ctx, l , _launch); const char * self = skynet_command(ctx, "REG", NULL); uint32_t handle_id = strtoul(self+1, NULL, 16); // it must be first message - skynet_send(ctx, 0, handle_id, PTYPE_TAG_DONTCOPY,0, tmp, sz+1); + skynet_send(ctx, 0, handle_id, PTYPE_TAG_DONTCOPY,0, tmp, sz); return 0; } @@ -268,7 +149,6 @@ snlua_create(void) { struct snlua * l = skynet_malloc(sizeof(*l)); memset(l,0,sizeof(*l)); l->L = lua_newstate(skynet_lalloc, NULL); - l->init = _init; return l; } diff --git a/service/bootstrap.lua b/service/bootstrap.lua new file mode 100644 index 00000000..51a57999 --- /dev/null +++ b/service/bootstrap.lua @@ -0,0 +1,28 @@ +local skynet = require "skynet" + +skynet.start(function() + assert(skynet.launch("logger", skynet.getenv "logger")) + + local standalone = skynet.getenv "standalone" + local master_addr = skynet.getenv "master" + + if standalone then + assert(skynet.launch("master", master_addr)) + end + + local local_addr = skynet.getenv "address" + local harbor_id = skynet.getenv "harbor" + + assert(skynet.launch("harbor",master_addr, local_addr, harbor_id)) + + local launcher = assert(skynet.launch("snlua","launcher")) + skynet.name(".launcher", launcher) + + if standalone then + local datacenter = assert(skynet.newservice "datacenterd") + skynet.name("DATACENTER", datacenter) + end + assert(skynet.newservice "service_mgr") + assert(skynet.newservice(skynet.getenv "start" or "main")) + skynet.exit() +end) diff --git a/service/datacenterd.lua b/service/datacenterd.lua new file mode 100644 index 00000000..ad431ae2 --- /dev/null +++ b/service/datacenterd.lua @@ -0,0 +1,43 @@ +local skynet = require "skynet" + +local command = {} +local database = {} + +local function query(db, key, ...) + if key == nil then + return db + else + return query(db[key], ...) + end +end + +function command.QUERY(key, ...) + local d = database[key] + if d then + return query(d, ...) + end +end + +local function update(db, key, value, ...) + if select("#",...) == 0 then + local ret = db[key] + db[key] = value + return ret + else + if db[key] == nil then + db[key] = {} + end + return update(db[key], value, ...) + end +end + +function command.UPDATE(...) + return update(database, ...) +end + +skynet.start(function() + skynet.dispatch("lua", function (_, source, cmd, ...) + local f = assert(command[cmd]) + skynet.ret(skynet.pack(f(...))) + end) +end) diff --git a/service/debug_console.lua b/service/debug_console.lua index 7e743617..d49dd809 100644 --- a/service/debug_console.lua +++ b/service/debug_console.lua @@ -1,6 +1,7 @@ local skynet = require "skynet" local codecache = require "skynet.codecache" local socket = require "socket" +local snax = require "snax" local port = tonumber(...) local COMMAND = {} @@ -109,7 +110,9 @@ function COMMAND.help() mem = "mem : show memory status", gc = "gc : force every lua service do garbage collect", start = "lanuch a new lua service", + snax = "lanuch a new snax service", clearcache = "clear lua code cache", + service = "List unique service", } end @@ -126,4 +129,16 @@ function COMMAND.start(...) end end +function COMMAND.snax(...) + local s = snax.newservice(...) + if s then + local addr = s.handle + return { [skynet.address(addr)] = ... } + else + return "Failed" + end +end +function COMMAND.service() + return skynet.call("SERVICE", "lua", "LIST") +end diff --git a/service/launcher.lua b/service/launcher.lua index 6283b3fa..c6ed0c4c 100644 --- a/service/launcher.lua +++ b/service/launcher.lua @@ -2,8 +2,8 @@ local skynet = require "skynet" local string = string local services = {} - local command = {} +local instance = {} -- for confirm (function command.LAUNCH / command.ERROR / command.LAUNCHOK) local function handle_to_address(handle) return tonumber("0x" .. string.sub(handle , 2)) @@ -19,18 +19,6 @@ function command.LIST() return list end -function command.RELOAD(handle) - handle = handle_to_address(handle) - local cmd = string.match(services[handle], "snlua (.+)") - print(services[handle],cmd) - if cmd then - skynet.send(handle,"debug","RELOAD",cmd) - return {cmd} - else - return {"Support only snlua"} - end -end - function command.STAT() local list = {} for k,v in pairs(services) do @@ -40,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 @@ -50,23 +38,7 @@ function command.INFO(handle) end end -function command.TIMING(handle) - handle = handle_to_address(handle) - if services[handle] == nil then - return - else - local r = skynet.call(handle,"debug","TIMING") - local result = {} - for k,v in pairs(r) do - v.name = services[k] - v.avg = v.ti/v.n - result[skynet.address(k)] = v - end - return result - 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]) } @@ -90,48 +62,71 @@ function command.GC() return command.MEM() end -function command.REMOVE(handle) +function command.REMOVE(_,_, handle) services[handle] = nil -- don't return (skynet.ret) because the handle may exit return NORET end -local instance = {} - -skynet.dispatch("text" , function(session, address , cmd) - if cmd == "" then - -- init notice - local reply = instance[address] - if reply then - skynet.redirect(reply.address , 0, "response", reply.session, skynet.address(address)) - instance[address] = nil - end - elseif cmd == "ERROR" then - -- see serivce-src/service_lua.c - -- init failed - local reply = instance[address] - if reply then - skynet.redirect(reply.address , 0, "response", reply.session, "") - instance[address] = nil - end +function command.LAUNCH(address, session, service, ...) + local param = table.concat({...}, " ") + local inst = skynet.launch(service, param) + if inst then + services[inst] = service .. " " .. param + instance[inst] = { session = session, address = address } else - -- launch request - local service, param = string.match(cmd,"([^ ]+) (.*)") - local inst = skynet.launch(service, param) - if inst then - services[inst] = cmd - instance[inst] = { session = session, address = address } - else - skynet.ret("") - end + skynet.ret("") -- launch failed end -end) + return NORET +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, "") + instance[address] = nil + end + + return NORET +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)) + instance[address] = nil + end + + return NORET +end + +-- for historical reasons, launcher support text command (for C service) + +skynet.register_protocol { + name = "text", + id = skynet.PTYPE_TEXT, + unpack = skynet.tostring, + dispatch = function(session, address , cmd) + if cmd == "" then + command.LAUNCHOK(address) + elseif cmd == "ERROR" then + command.ERROR(address) + else + -- launch request + local service, param = string.match(cmd,"([^ ]+) (.*)") + command.LAUNCH(address, session, service, param) + end + end, +} skynet.dispatch("lua", function(session, address, cmd , ...) cmd = string.upper(cmd) local f = command[cmd] if f then - local ret = f(...) + local ret = f(address, session, ...) if ret ~= NORET then skynet.ret(skynet.pack(ret)) end diff --git a/service/multicastd.lua b/service/multicastd.lua new file mode 100644 index 00000000..66e4f1c4 --- /dev/null +++ b/service/multicastd.lua @@ -0,0 +1,194 @@ +local skynet = require "skynet" +local mc = require "multicast.c" +local datacenter = require "datacenter" + +local harbor_id = skynet.harbor(skynet.self()) + +local command = {} +local channel = {} +local channel_n = {} +local channel_remote = {} +local channel_id = harbor_id +local NORET = {} + +local function get_address(t, id) + local v = assert(datacenter.get("multicast", id)) + t[id] = v + return v +end + +local node_address = setmetatable({}, { __index = get_address }) + +-- new LOCAL channel , The low 8bit is the same with harbor_id +function command.NEW() + while channel[channel_id] do + channel_id = mc.nextid(channel_id) + end + channel[channel_id] = {} + channel_n[channel_id] = 0 + local ret = channel_id + channel_id = mc.nextid(channel_id) + return ret +end + +-- MUST call by the owner node of channel, delete a remote channel +function command.DELR(source, c) + channel[c] = nil + channel_n[c] = nil + return NORET +end + +-- delete a channel, if the channel is remote, forward the command to the owner node +-- otherwise, delete the channel, and call all the remote node, DELR +function command.DEL(source, c) + local node = c % 256 + if node ~= harbor_id then + skynet.send(node_address[node], "lua", "DEL", c) + return NORET + end + local remote = channel_remote[c] + channel[c] = nil + channel_n[c] = nil + channel_remote[c] = nil + for node in pairs(remote) do + skynet.send(node_address[node], "lua", "DELR", c) + end + return NORET +end + +-- forward multicast message to a node (channel id use the session field) +local function remote_publish(node, channel, source, ...) + skynet.redirect(node_address[node], source, "multicast", channel, ...) +end + +-- publish a message, for local node, use the message pointer (call mc.bind to add the reference) +-- for remote node, call remote_publish. (call mc.unpack and skynet.tostring to convert message pointer to string) +local function publish(c , source, pack, size) + local group = channel[c] + if group == nil then + -- dead channel, delete the pack + mc.bind(pack, 1) + mc.close(pack) + return + end + mc.bind(pack, channel_n[c]) + local msg = skynet.tostring(pack, size) + for k in pairs(group) do + skynet.redirect(k, source, "multicast", c , msg) + end + local remote = channel_remote[c] + if remote then + local _, msg, sz = mc.unpack(pack, size) + local msg = skynet.tostring(msg,sz) + for node in pairs(remote) do + remote_publish(node, c, source, msg) + end + end +end + +skynet.register_protocol { + name = "multicast", + id = skynet.PTYPE_MULTICAST, + unpack = function(msg, sz) + return mc.packremote(msg, sz) + end, + dispatch = publish, +} + +-- publish a message, if the caller is remote, forward the message to the owner node (by remote_publish) +-- If the caller is local, call publish +function command.PUB(source, c, pack, size) + assert(skynet.harbor(source) == harbor_id) + local node = c % 256 + if node ~= harbor_id then + -- remote publish + remote_publish(node, c, source, mc.remote(pack)) + else + publish(c, source, pack,size) + end +end + +-- the node (source) subscribe a channel +-- MUST call by channel owner node (assert source is not local and channel is create by self) +-- If channel is not exist, return true +-- Else set channel_remote[channel] true +function command.SUBR(source, c) + local node = skynet.harbor(source) + if not channel[c] then + -- channel none exist + return true + end + assert(node ~= harbor_id and c % 256 == harbor_id) + local group = channel_remote[c] + if group == nil then + group = {} + channel_remote[c] = group + end + group[node] = true +end + +-- the service (source) subscribe a channel +-- If the channel is remote, node subscribe it by send a SUBR to the owner . +function command.SUB(source, c) + local node = c % 256 + if node ~= harbor_id then + -- remote group + if channel[c] == nil then + if skynet.call(node_address[node], "lua", "SUBR", c) then + return + end + if channel[c] == nil then + -- double check, because skynet.call whould yield, other SUB may occur. + channel[c] = {} + channel_n[c] = 0 + end + end + end + local group = channel[c] + if group and not group[source] then + channel_n[c] = channel_n[c] + 1 + group[source] = true + end +end + +-- MUST call by a node, unsubscribe a channel +function command.USUBR(source, c) + local node = skynet.harbor(source) + assert(node ~= harbor_id) + local group = assert(channel_remote[c]) + group[node] = nil + return NORET +end + +-- Unsubscribe a channel, if the subscriber is empty and the channel is remote, send USUBR to the channel owner +function command.USUB(source, c) + local group = assert(channel[c]) + if group[source] then + group[source] = nil + channel_n[c] = channel_n[c] - 1 + if channel_n[c] == 0 then + local node = c % 256 + if node ~= harbor_id then + -- remote group + channel[c] = nil + channel_n[c] = nil + skynet.send(node_address[node], "lua", "USUBR", c) + end + end + end + return NORET +end + +skynet.start(function() + skynet.dispatch("lua", function(_,source, cmd, ...) + local f = assert(command[cmd]) + local result = f(source, ...) + if result ~= NORET then + skynet.ret(skynet.pack(result)) + end + end) + local self = skynet.self() + local id = skynet.harbor(self) + assert(datacenter.set("multicast", id, self) == nil) +end) + diff --git a/service/service_mgr.lua b/service/service_mgr.lua index 7f771f3f..f49a79f7 100644 --- a/service/service_mgr.lua +++ b/service/service_mgr.lua @@ -1,72 +1,191 @@ local skynet = require "skynet" +local snax = require "snax" local cmd = {} local service = {} -function cmd.LAUNCH(service_name, ...) - local s = service[service_name] - if type(s) == "number" then - return s +local function request(name, func, ...) + local ok, handle = pcall(func, ...) + local s = service[name] + assert(type(s) == "table") + if ok then + service[name] = handle + else + service[name] = tostring(handle) end - if s == nil then - s = { launch = true } - service[service_name] = s - elseif s.launch then - assert(type(s) == "table") - local co = coroutine.running() - table.insert(s, co) - skynet.wait() - s = service[service_name] - assert(type(s) == "number") - return s - end - - local handle = skynet.newservice(service_name, ...) for _,v in ipairs(s) do skynet.wakeup(v) end - service[service_name] = handle - - return handle + if ok then + return handle + else + error(tostring(handle)) + end end -function cmd.QUERY(service_name) - local s = service[service_name] +local function waitfor(name , func, ...) + local s = service[name] if type(s) == "number" then return s end + local co = coroutine.running() + if s == nil then s = {} - service[service_name] = s + service[name] = s + elseif type(s) == "string" then + error(s) end + assert(type(s) == "table") - local co = coroutine.running() + + if not s.launch and func then + s.launch = true + return request(name, func, ...) + end + table.insert(s, co) skynet.wait() - s = service[service_name] + s = service[name] + if type(s) == "string" then + error(s) + end assert(type(s) == "number") return s end +local function read_name(service_name) + if string.byte(service_name) == 64 then -- '@' + return string.sub(service_name , 2) + else + return service_name + end +end + +function cmd.LAUNCH(service_name, subname, ...) + local realname = read_name(service_name) + + if realname == "snaxd" then + return waitfor(service_name.."."..subname, snax.rawnewservice, subname, ...) + else + return waitfor(service_name, skynet.newservice, realname, subname, ...) + end +end + +function cmd.QUERY(service_name, subname) + local realname = read_name(service_name) + + if realname == "snaxd" then + return waitfor(service_name.."."..subname) + else + return waitfor(service_name) + end +end + +local function list_service() + local result = {} + for k,v in pairs(service) do + if type(v) == "string" then + v = "Error: " .. v + elseif type(v) == "table" then + v = "Querying" + else + v = skynet.address(v) + end + + result[k] = v + end + + return result +end + + +local function register_global() + function cmd.GLAUNCH(name, ...) + local global_name = "@" .. name + return cmd.LAUNCH(global_name, ...) + end + + function cmd.GQUERY(name, ...) + local global_name = "@" .. name + return cmd.QUERY(global_name, ...) + end + + local mgr = {} + + function cmd.REPORT(m) + mgr[m] = true + skynet.watch(m) + end + + local function add_list(all, m) + local harbor = "@" .. skynet.harbor(m) + local result = skynet.call(m, "lua", "LIST") + for k,v in pairs(result) do + all[k .. harbor] = v + end + end + + function cmd.LIST() + local result = {} + for k in pairs(mgr) do + pcall(add_list, result, k) + end + local l = list_service() + for k, v in pairs(l) do + result[k] = v + end + return result + end +end + +local function register_local() + function cmd.GLAUNCH(name, ...) + local global_name = "@" .. name + return waitfor(global_name, skynet.call, "SERVICE", "lua", "LAUNCH", global_name, ...) + end + + function cmd.GQUERY(name, ...) + local global_name = "@" .. name + return waitfor(global_name, skynet.call, "SERVICE", "lua", "QUERY", global_name, ...) + end + + function cmd.LIST() + return list_service() + end + + skynet.call("SERVICE", "lua", "REPORT", skynet.self()) +end + skynet.start(function() - skynet.dispatch("lua", function(session, address, command, service_name , ...) + skynet.dispatch("lua", function(session, address, command, ...) local f = cmd[command] if f == nil then - skynet.ret(skynet.pack(nil)) + skynet.ret(skynet.pack(nil, "Invalid command " .. command)) return end - local ok, r = pcall(f, service_name, ...) + local ok, r = pcall(f, ...) + if ok then skynet.ret(skynet.pack(r)) else - skynet.ret(skynet.pack(nil)) + skynet.ret(skynet.pack(nil, r)) end end) - skynet.register(".service") + local handle = skynet.localname ".service" + if handle ~= 0 then + skynet.error(".service is already register by ", skynet.address(handle)) + skynet.exit() + else + skynet.register(".service") + end if skynet.getenv "standalone" then skynet.register("SERVICE") + register_global() + else + register_local() end end) diff --git a/service/snaxd.lua b/service/snaxd.lua index 7783a7af..de41d6c3 100644 --- a/service/snaxd.lua +++ b/service/snaxd.lua @@ -1,7 +1,8 @@ local skynet = require "skynet" local c = require "skynet.c" -local snax_interface = require "snax_interface" +local snax_interface = require "snax.interface" local profile = require "profile" +local snax = require "snax" local func = snax_interface(tostring(...), _ENV) local mode @@ -20,8 +21,10 @@ local function update_stat(name, ti) t.time = t.time + ti end +local traceback = debug.traceback + local function do_func(f, msg) - return pcall(f, table.unpack(msg)) + return xpcall(f, traceback, table.unpack(msg)) end local function dispatch(f, ...) @@ -41,7 +44,7 @@ local function message_dispatch() if method[2] == "accept" then -- no return profile.start() - local ok, data = pcall(f, table.unpack(msg)) + local ok, data = xpcall(f, traceback, table.unpack(msg)) local ti = profile.stop() update_stat(method[3], ti) if not ok then @@ -49,7 +52,7 @@ local function message_dispatch() end else profile.start() - local ok, data, size = pcall(dispatch, f, table.unpack(msg)) + local ok, data, size = xpcall(dispatch, traceback, f, table.unpack(msg)) local ti = profile.stop() update_stat(method[3], ti) if ok then @@ -88,9 +91,9 @@ local function timing( method, ... ) profile.start() if method[2] == "accept" then -- no return - err,msg = pcall(method[4], ...) + err,msg = xpcall(method[4], traceback, ...) else - err,msg = pcall(return_f, method[4], ...) + err,msg = xpcall(return_f, traceback, method[4], ...) end local ti = profile.stop() update_stat(method[3], ti) @@ -98,13 +101,13 @@ local function timing( method, ... ) end skynet.start(function() - skynet.dispatch("lua", function ( session , source , id, ...) + skynet.dispatch("snax", function ( session , source , id, ...) local method = func[id] if method[2] == "system" then local command = method[3] if command == "hotfix" then - local hotfix = require "snax_hotfix" + local hotfix = require "snax.hotfix" skynet.ret(skynet.pack(hotfix(func, ...))) elseif command == "init" then assert(not init, "Already init") diff --git a/skynet-src/malloc_hook.c b/skynet-src/malloc_hook.c index 33dc5263..61c07392 100644 --- a/skynet-src/malloc_hook.c +++ b/skynet-src/malloc_hook.c @@ -224,9 +224,3 @@ skynet_lalloc(void *ud, void *ptr, size_t osize, size_t nsize) { return skynet_realloc(ptr, nsize); } } - -void -malloc_inithook(void) { - memset(mem_stats, 0, sizeof(mem_stats)); -} - diff --git a/skynet-src/skynet.h b/skynet-src/skynet.h index ce82b7ac..0df5ca9c 100644 --- a/skynet-src/skynet.h +++ b/skynet-src/skynet.h @@ -8,13 +8,13 @@ #define PTYPE_TEXT 0 #define PTYPE_RESPONSE 1 -#define PTYPE_MULTICAST_DEPRECATED 2 +#define PTYPE_MULTICAST 2 #define PTYPE_CLIENT 3 #define PTYPE_SYSTEM 4 #define PTYPE_HARBOR 5 #define PTYPE_SOCKET 6 // read lualib/skynet.lua lualib/simplemonitor.lua -#define PTYPE_RESERVED_ERROR 7 +#define PTYPE_ERROR 7 // read lualib/skynet.lua lualib/mqueue.lua #define PTYPE_RESERVED_QUEUE 8 #define PTYPE_RESERVED_DEBUG 9 diff --git a/skynet-src/skynet_harbor.c b/skynet-src/skynet_harbor.c index 023f1f6d..ae47cf45 100644 --- a/skynet-src/skynet_harbor.c +++ b/skynet-src/skynet_harbor.c @@ -34,6 +34,7 @@ skynet_harbor_register(struct remote_name *rname) { int skynet_harbor_message_isremote(uint32_t handle) { + assert(HARBOR != 0); int h = (handle & ~HANDLE_MASK); return h != HARBOR && h !=0; } @@ -43,16 +44,7 @@ skynet_harbor_init(int harbor) { HARBOR = (unsigned int)harbor << HANDLE_REMOTE_SHIFT; } -int -skynet_harbor_start(const char * master, const char *local) { - size_t sz = strlen(master) + strlen(local) + 32; - char args[sz]; - sprintf(args, "%s %s %d",master,local,HARBOR >> HANDLE_REMOTE_SHIFT); - struct skynet_context * inst = skynet_context_new("harbor",args); - if (inst == NULL) { - return 1; - } - REMOTE = inst; - - return 0; +void +skynet_harbor_start(void *ctx) { + REMOTE = ctx; } diff --git a/skynet-src/skynet_harbor.h b/skynet-src/skynet_harbor.h index 6ad6023a..0116a91a 100644 --- a/skynet-src/skynet_harbor.h +++ b/skynet-src/skynet_harbor.h @@ -26,6 +26,6 @@ void skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int sessio void skynet_harbor_register(struct remote_name *rname); int skynet_harbor_message_isremote(uint32_t handle); void skynet_harbor_init(int harbor); -int skynet_harbor_start(const char * master, const char *local); +void skynet_harbor_start(void * ctx); #endif diff --git a/skynet-src/skynet_imp.h b/skynet-src/skynet_imp.h index a5c7e078..81fad4c8 100644 --- a/skynet-src/skynet_imp.h +++ b/skynet-src/skynet_imp.h @@ -4,14 +4,16 @@ struct skynet_config { int thread; int harbor; - const char * logger; const char * module_path; - const char * master; - const char * local; - const char * start; - const char * standalone; + const char * bootstrap; }; +#define THREAD_WORKER 0 +#define THREAD_MAIN 1 +#define THREAD_SOCKET 2 +#define THREAD_TIMER 3 +#define THREAD_MONITOR 4 + void skynet_start(struct skynet_config * config); #endif diff --git a/skynet-src/skynet_main.c b/skynet-src/skynet_main.c index 8cebd00e..3b8cf1b2 100644 --- a/skynet-src/skynet_main.c +++ b/skynet-src/skynet_main.c @@ -2,6 +2,7 @@ #include "skynet_imp.h" #include "skynet_env.h" +#include "skynet_server.h" #include #include @@ -87,6 +88,7 @@ main(int argc, char *argv[]) { if (argc > 1) { config_file = argv[1]; } + skynet_globalinit(); skynet_env_init(); sigign(); @@ -111,24 +113,15 @@ main(int argc, char *argv[]) { printf("Skynet lua code cache enable\n"); #endif - const char *path = optstring("lua_path","./lualib/?.lua;./lualib/?/init.lua"); - setenv("LUA_PATH",path,1); - const char *cpath = optstring("lua_cpath","./luaclib/?.so"); - setenv("LUA_CPATH",cpath,1); - optstring("luaservice","./service/?.lua"); - config.thread = optint("thread",8); config.module_path = optstring("cpath","./cservice/?.so"); - config.logger = optstring("logger",NULL); config.harbor = optint("harbor", 1); - config.master = optstring("master","127.0.0.1:2012"); - config.start = optstring("start","main.lua"); - config.local = optstring("address","127.0.0.1:2525"); - config.standalone = optstring("standalone",NULL); + config.bootstrap = optstring("bootstrap","snlua bootstrap"); lua_close(L); skynet_start(&config); + skynet_globalexit(); printf("skynet exit\n"); diff --git a/skynet-src/skynet_malloc.h b/skynet-src/skynet_malloc.h index 1b3b94d3..f7e4337f 100644 --- a/skynet-src/skynet_malloc.h +++ b/skynet-src/skynet_malloc.h @@ -3,15 +3,11 @@ #include -#ifdef NOUSE_JEMALLOC - #define skynet_malloc malloc #define skynet_calloc calloc #define skynet_realloc realloc #define skynet_free free -#endif - void * skynet_malloc(size_t sz); void * skynet_calloc(size_t nmemb,size_t size); void * skynet_realloc(void *ptr, size_t size); diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index f0e48440..c5d80f0b 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -13,12 +13,8 @@ // 0 means mq is not in global mq. // 1 means mq is in global mq , or the message is dispatching. -// 2 means message is dispatching with locked session set. -// 3 means mq is not in global mq, and locked session has been set. #define MQ_IN_GLOBAL 1 -#define MQ_DISPATCHING 2 -#define MQ_LOCKED 3 struct message_queue { uint32_t handle; @@ -27,17 +23,19 @@ struct message_queue { int tail; int lock; int release; - int lock_session; int in_global; struct skynet_message *queue; + struct message_queue *next; }; struct global_queue { uint32_t head; uint32_t tail; struct message_queue ** queue; + // We use a separated flag array to ensure the mq is pushed. + // See the comments below. bool * flag; - + struct message_queue *list; }; static struct global_queue *Q = NULL; @@ -51,7 +49,22 @@ static void skynet_globalmq_push(struct message_queue * queue) { struct global_queue *q= Q; + if (q->flag[GP(q->tail)]) { + // The queue may full seldom, save queue in list + assert(queue->next == NULL); + struct message_queue * last; + do { + last = q->list; + queue->next = last; + } while(!__sync_bool_compare_and_swap(&q->list, last, queue)); + + return; + } + uint32_t tail = GP(__sync_fetch_and_add(&q->tail,1)); + // The thread would suspend here, and the q->queue[tail] is last version , + // but the queue tail is increased. + // So we set q->flag[tail] after changing q->queue[tail]. q->queue[tail] = queue; __sync_synchronize(); q->flag[tail] = true; @@ -61,11 +74,26 @@ struct message_queue * skynet_globalmq_pop() { struct global_queue *q = Q; uint32_t head = q->head; - uint32_t head_ptr = GP(head); - if (head_ptr == GP(q->tail)) { + + if (head == q->tail) { + // The queue is empty. return NULL; } + uint32_t head_ptr = GP(head); + + struct message_queue * list = q->list; + if (list) { + // If q->list is not empty, try to load it back to the queue + struct message_queue *newhead = list->next; + if (__sync_bool_compare_and_swap(&q->list, list, newhead)) { + // try load list only once, if success , push it back to the queue. + list->next = NULL; + skynet_globalmq_push(list); + } + } + + // Check the flag first, if the flag is false, the pushing may not complete. if(!q->flag[head_ptr]) { return NULL; } @@ -89,16 +117,20 @@ skynet_mq_create(uint32_t handle) { q->head = 0; q->tail = 0; q->lock = 0; + // When the queue is create (always between service create and service init) , + // set in_global flag to avoid push it to global queue . + // If the service init success, skynet_context_new will call skynet_mq_force_push to push it to global queue. q->in_global = MQ_IN_GLOBAL; q->release = 0; - q->lock_session = 0; q->queue = skynet_malloc(sizeof(struct skynet_message) * q->cap); + q->next = NULL; return q; } static void _release(struct message_queue *q) { + assert(q->next == NULL); skynet_free(q->queue); skynet_free(q); } @@ -161,82 +193,28 @@ expand_queue(struct message_queue *q) { q->queue = new_queue; } -static void -_unlock(struct message_queue *q) { - // this api use in push a unlock message, so the in_global flags must not be 0 , - // but the q is not exist in global queue. - if (q->in_global == MQ_LOCKED) { - skynet_globalmq_push(q); - q->in_global = MQ_IN_GLOBAL; - } else { - assert(q->in_global == MQ_DISPATCHING); - } - q->lock_session = 0; -} - -static void -_pushhead(struct message_queue *q, struct skynet_message *message) { - int head = q->head - 1; - if (head < 0) { - head = q->cap - 1; - } - if (head == q->tail) { - expand_queue(q); - --q->tail; - head = q->cap - 1; - } - - q->queue[head] = *message; - q->head = head; - - _unlock(q); -} - void skynet_mq_push(struct message_queue *q, struct skynet_message *message) { assert(message); LOCK(q) - - if (q->lock_session !=0 && message->session == q->lock_session) { - _pushhead(q,message); - } else { - q->queue[q->tail] = *message; - if (++ q->tail >= q->cap) { - q->tail = 0; - } - if (q->head == q->tail) { - expand_queue(q); - } + q->queue[q->tail] = *message; + if (++ q->tail >= q->cap) { + q->tail = 0; + } - if (q->lock_session == 0) { - if (q->in_global == 0) { - q->in_global = MQ_IN_GLOBAL; - skynet_globalmq_push(q); - } - } + if (q->head == q->tail) { + expand_queue(q); + } + + if (q->in_global == 0) { + q->in_global = MQ_IN_GLOBAL; + skynet_globalmq_push(q); } UNLOCK(q) } -void -skynet_mq_lock(struct message_queue *q, int session) { - LOCK(q) - assert(q->lock_session == 0); - assert(q->in_global == MQ_IN_GLOBAL); - q->in_global = MQ_DISPATCHING; - q->lock_session = session; - UNLOCK(q) -} - -void -skynet_mq_unlock(struct message_queue *q) { - LOCK(q) - _unlock(q); - UNLOCK(q) -} - void skynet_mq_init() { struct global_queue *q = skynet_malloc(sizeof(*q)); @@ -257,14 +235,8 @@ void skynet_mq_pushglobal(struct message_queue *queue) { LOCK(queue) assert(queue->in_global); - if (queue->in_global == MQ_DISPATCHING) { - // lock message queue just now. - queue->in_global = MQ_LOCKED; - } - if (queue->lock_session == 0) { - skynet_globalmq_push(queue); - queue->in_global = MQ_IN_GLOBAL; - } + skynet_globalmq_push(queue); + queue->in_global = MQ_IN_GLOBAL; UNLOCK(queue) } @@ -279,31 +251,24 @@ skynet_mq_mark_release(struct message_queue *q) { UNLOCK(q) } -static int -_drop_queue(struct message_queue *q) { - // todo: send message back to message source +static void +_drop_queue(struct message_queue *q, message_drop drop_func, void *ud) { struct skynet_message msg; - int s = 0; while(!skynet_mq_pop(q, &msg)) { - ++s; - skynet_free(msg.data); + drop_func(&msg, ud); } _release(q); - return s; } -int -skynet_mq_release(struct message_queue *q) { - int ret = 0; +void +skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud) { LOCK(q) if (q->release) { UNLOCK(q) - ret = _drop_queue(q); + _drop_queue(q, drop_func, ud); } else { skynet_mq_force_push(q); UNLOCK(q) } - - return ret; } diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index 11945ce9..5ac582a1 100644 --- a/skynet-src/skynet_mq.h +++ b/skynet-src/skynet_mq.h @@ -17,14 +17,15 @@ struct message_queue * skynet_globalmq_pop(void); struct message_queue * skynet_mq_create(uint32_t handle); void skynet_mq_mark_release(struct message_queue *q); -int skynet_mq_release(struct message_queue *q); + +typedef void (*message_drop)(struct skynet_message *, void *); + +void skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud); uint32_t skynet_mq_handle(struct message_queue *); // 0 for success int skynet_mq_pop(struct message_queue *q, struct skynet_message *message); void skynet_mq_push(struct message_queue *q, struct skynet_message *message); -void skynet_mq_lock(struct message_queue *q, int session); -void skynet_mq_unlock(struct message_queue *q); // return the length of message queue, for debug int skynet_mq_length(struct message_queue *q); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 4c879cb2..4bc7b66a 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -8,6 +8,9 @@ #include "skynet_harbor.h" #include "skynet_env.h" #include "skynet_monitor.h" +#include "skynet_imp.h" + +#include #include #include @@ -50,10 +53,10 @@ struct skynet_context { struct skynet_node { int total; uint32_t monitor_exit; + pthread_key_t handle_key; }; -static struct skynet_node G_NODE = { 0,0 }; -static __thread uint32_t handle_tls = 0xffffffff; +static struct skynet_node G_NODE; int skynet_context_total() { @@ -61,22 +64,23 @@ skynet_context_total() { } static void -_context_inc() { +context_inc() { __sync_fetch_and_add(&G_NODE.total,1); } static void -_context_dec() { +context_dec() { __sync_fetch_and_sub(&G_NODE.total,1); } uint32_t skynet_current_handle(void) { - return handle_tls; + void * handle = pthread_getspecific(G_NODE.handle_key); + return (uint32_t)(uintptr_t)handle; } static void -_id_to_hex(char * str, uint32_t id) { +id_to_hex(char * str, uint32_t id) { int i; static char hex[16] = { '0','1','2','3','4','5','6','7','8','9','A','B','C','D','E','F' }; str[0] = ':'; @@ -86,6 +90,20 @@ _id_to_hex(char * str, uint32_t id) { str[9] = '\0'; } +struct drop_t { + uint32_t handle; +}; + +static void +drop_message(struct skynet_message *msg, void *ud) { + struct drop_t *d = ud; + skynet_free(msg->data); + uint32_t source = d->handle; + assert(source); + // report error to the message source + skynet_send(NULL, source, msg->source, PTYPE_ERROR, 0, NULL, 0); +} + struct skynet_context * skynet_context_new(const char * name, const char *param) { struct skynet_module * mod = skynet_module_query(name); @@ -111,7 +129,7 @@ skynet_context_new(const char * name, const char *param) { ctx->handle = skynet_handle_register(ctx); struct message_queue * queue = ctx->queue = skynet_mq_create(ctx->handle); // init function maybe use ctx->handle, so it must init at last - _context_inc(); + context_inc(); CHECKCALLING_BEGIN(ctx) int r = skynet_module_instance_init(mod, inst, ctx, param); @@ -128,9 +146,11 @@ skynet_context_new(const char * name, const char *param) { return ret; } else { skynet_error(ctx, "FAILED launch %s", name); + uint32_t handle = ctx->handle; skynet_context_release(ctx); - skynet_handle_retire(ctx->handle); - skynet_mq_release(queue); + skynet_handle_retire(handle); + struct drop_t d = { handle }; + skynet_mq_release(queue, drop_message, &d); return NULL; } } @@ -148,17 +168,17 @@ skynet_context_grab(struct skynet_context *ctx) { } static void -_delete_context(struct skynet_context *ctx) { +delete_context(struct skynet_context *ctx) { skynet_module_instance_release(ctx->mod, ctx->instance); skynet_mq_mark_release(ctx->queue); skynet_free(ctx); - _context_dec(); + context_dec(); } struct skynet_context * skynet_context_release(struct skynet_context *ctx) { if (__sync_sub_and_fetch(&ctx->ref,1) == 0) { - _delete_context(ctx); + delete_context(ctx); return NULL; } return ctx; @@ -199,13 +219,12 @@ static void _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { assert(ctx->init); CHECKCALLING_BEGIN(ctx) - handle_tls = ctx->handle; + pthread_setspecific(G_NODE.handle_key, (void *)(uintptr_t)(ctx->handle)); int type = msg->sz >> HANDLE_REMOTE_SHIFT; size_t sz = msg->sz & HANDLE_MASK; if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) { skynet_free(msg->data); } - handle_tls = 0xffffffff; CHECKCALLING_END(ctx) } @@ -219,10 +238,8 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) { struct skynet_context * ctx = skynet_handle_grab(handle); if (ctx == NULL) { - int s = skynet_mq_release(q); - if (s>0) { - skynet_error(NULL, "Drop message queue %x (%d messages)", handle,s); - } + struct drop_t d = { handle }; + skynet_mq_release(q, drop_message, &d); return 0; } @@ -236,7 +253,6 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) { if (ctx->cb == NULL) { skynet_free(msg.data); - skynet_error(NULL, "Drop message from %x to %x without callback , size = %d",msg.source, handle, (int)msg.sz); } else { _dispatch_message(ctx, &msg); } @@ -251,7 +267,7 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) { } static void -_copy_name(char name[GLOBALNAME_LENGTH], const char * addr) { +copy_name(char name[GLOBALNAME_LENGTH], const char * addr) { int i; for (i=0;ihandle, ti, session); - sprintf(context->result, "%d", session); +// skynet command + +struct command_func { + const char *name; + const char * (*func)(struct skynet_context * context, const char * param); +}; + +static const char * +cmd_timeout(struct skynet_context * context, const char * param) { + char * session_ptr = NULL; + int ti = strtol(param, &session_ptr, 10); + int session = skynet_context_newsession(context); + skynet_timeout(context->handle, ti, session); + sprintf(context->result, "%d", session); + return context->result; +} + +static const char * +cmd_reg(struct skynet_context * context, const char * param) { + if (param == NULL || param[0] == '\0') { + sprintf(context->result, ":%x", context->handle); + return context->result; + } else if (param[0] == '.') { + return skynet_handle_namehandle(context->handle, param + 1); + } else { + assert(context->handle!=0); + struct remote_name *rname = skynet_malloc(sizeof(*rname)); + copy_name(rname->name, param); + rname->handle = context->handle; + skynet_harbor_register(rname); + return NULL; + } +} + +static const char * +cmd_query(struct skynet_context * context, const char * param) { + if (param[0] == '.') { + uint32_t handle = skynet_handle_findname(param+1); + sprintf(context->result, ":%x", handle); return context->result; } + return NULL; +} - if (strcmp(cmd,"LOCK") == 0) { - if (context->init == false) { - return NULL; - } - skynet_mq_lock(context->queue, context->session_id+1); +static const char * +cmd_name(struct skynet_context * context, const char * param) { + int size = strlen(param); + char name[size+1]; + char handle[size+1]; + sscanf(param,"%s %s",name,handle); + if (handle[0] != ':') { return NULL; } - - if (strcmp(cmd,"UNLOCK") == 0) { - if (context->init == false) { - return NULL; - } - skynet_mq_unlock(context->queue); + uint32_t handle_id = strtoul(handle+1, NULL, 16); + if (handle_id == 0) { return NULL; } - - if (strcmp(cmd,"REG") == 0) { - if (param == NULL || param[0] == '\0') { - sprintf(context->result, ":%x", context->handle); - return context->result; - } else if (param[0] == '.') { - return skynet_handle_namehandle(context->handle, param + 1); - } else { - assert(context->handle!=0); - struct remote_name *rname = skynet_malloc(sizeof(*rname)); - _copy_name(rname->name, param); - rname->handle = context->handle; - skynet_harbor_register(rname); - return NULL; - } + if (name[0] == '.') { + return skynet_handle_namehandle(handle_id, name + 1); + } else { + struct remote_name *rname = skynet_malloc(sizeof(*rname)); + copy_name(rname->name, name); + rname->handle = handle_id; + skynet_harbor_register(rname); } + return NULL; +} - if (strcmp(cmd,"QUERY") == 0) { - if (param[0] == '.') { - uint32_t handle = skynet_handle_findname(param+1); - sprintf(context->result, ":%x", handle); +static const char * +cmd_now(struct skynet_context * context, const char * param) { + uint32_t ti = skynet_gettime(); + sprintf(context->result,"%u",ti); + return context->result; +} + +static const char * +cmd_exit(struct skynet_context * context, const char * param) { + handle_exit(context, 0); + return NULL; +} + +static const char * +cmd_kill(struct skynet_context * context, const char * param) { + uint32_t handle = 0; + if (param[0] == ':') { + handle = strtoul(param+1, NULL, 16); + } else if (param[0] == '.') { + handle = skynet_handle_findname(param+1); + } else { + skynet_error(context, "Can't kill %s",param); + // todo : kill global service + } + if (handle) { + handle_exit(context, handle); + } + return NULL; +} + +static const char * +cmd_launch(struct skynet_context * context, const char * param) { + size_t sz = strlen(param); + char tmp[sz+1]; + strcpy(tmp,param); + char * args = tmp; + char * mod = strsep(&args, " \t\r\n"); + args = strsep(&args, "\r\n"); + struct skynet_context * inst = skynet_context_new(mod,args); + if (inst == NULL) { + return NULL; + } else { + id_to_hex(context->result, inst->handle); + return context->result; + } +} + +static const char * +cmd_getenv(struct skynet_context * context, const char * param) { + return skynet_getenv(param); +} + +static const char * +cmd_setenv(struct skynet_context * context, const char * param) { + size_t sz = strlen(param); + char key[sz+1]; + int i; + for (i=0;param[i] != ' ' && param[i];i++) { + key[i] = param[i]; + } + if (param[i] == '\0') + return NULL; + + key[i] = '\0'; + param += i+1; + + skynet_setenv(key,param); + return NULL; +} + +static const char * +cmd_starttime(struct skynet_context * context, const char * param) { + uint32_t sec = skynet_gettime_fixsec(); + sprintf(context->result,"%u",sec); + return context->result; +} + +static const char * +cmd_endless(struct skynet_context * context, const char * param) { + if (context->endless) { + strcpy(context->result, "1"); + context->endless = false; + return context->result; + } + return NULL; +} + +static const char * +cmd_abort(struct skynet_context * context, const char * param) { + skynet_handle_retireall(); + return NULL; +} + +static const char * +cmd_monitor(struct skynet_context * context, const char * param) { + uint32_t handle=0; + if (param == NULL || param[0] == '\0') { + if (G_NODE.monitor_exit) { + // return current monitor serivce + sprintf(context->result, ":%x", G_NODE.monitor_exit); return context->result; } return NULL; - } - - if (strcmp(cmd,"NAME") == 0) { - int size = strlen(param); - char name[size+1]; - char handle[size+1]; - sscanf(param,"%s %s",name,handle); - if (handle[0] != ':') { - return NULL; - } - uint32_t handle_id = strtoul(handle+1, NULL, 16); - if (handle_id == 0) { - return NULL; - } - if (name[0] == '.') { - return skynet_handle_namehandle(handle_id, name + 1); - } else { - struct remote_name *rname = skynet_malloc(sizeof(*rname)); - _copy_name(rname->name, name); - rname->handle = handle_id; - skynet_harbor_register(rname); - } - return NULL; - } - - if (strcmp(cmd,"NOW") == 0) { - uint32_t ti = skynet_gettime(); - sprintf(context->result,"%u",ti); - return context->result; - } - - if (strcmp(cmd,"EXIT") == 0) { - handle_exit(context, 0); - return NULL; - } - - if (strcmp(cmd,"KILL") == 0) { - uint32_t handle = 0; + } else { if (param[0] == ':') { handle = strtoul(param+1, NULL, 16); } else if (param[0] == '.') { handle = skynet_handle_findname(param+1); } else { - skynet_error(context, "Can't kill %s",param); - // todo : kill global service - } - if (handle) { - handle_exit(context, handle); - } - return NULL; - } - - if (strcmp(cmd,"LAUNCH") == 0) { - size_t sz = strlen(param); - char tmp[sz+1]; - strcpy(tmp,param); - char * args = tmp; - char * mod = strsep(&args, " \t\r\n"); - args = strsep(&args, "\r\n"); - struct skynet_context * inst = skynet_context_new(mod,args); - if (inst == NULL) { - return NULL; - } else { - _id_to_hex(context->result, inst->handle); - return context->result; + skynet_error(context, "Can't monitor %s",param); + // todo : monitor global service } } + G_NODE.monitor_exit = handle; + return NULL; +} - if (strcmp(cmd,"GETENV") == 0) { - return skynet_getenv(param); - } +static const char * +cmd_mqlen(struct skynet_context * context, const char * param) { + int len = skynet_mq_length(context->queue); + sprintf(context->result, "%d", len); + return context->result; +} - if (strcmp(cmd,"SETENV") == 0) { - size_t sz = strlen(param); - char key[sz+1]; - int i; - for (i=0;param[i] != ' ' && param[i];i++) { - key[i] = param[i]; +static struct command_func cmd_funcs[] = { + { "TIMEOUT", cmd_timeout }, + { "REG", cmd_reg }, + { "QUERY", cmd_query }, + { "NAME", cmd_name }, + { "NOW", cmd_now }, + { "EXIT", cmd_exit }, + { "KILL", cmd_kill }, + { "LAUNCH", cmd_launch }, + { "GETENV", cmd_getenv }, + { "SETENV", cmd_setenv }, + { "STARTTIME", cmd_starttime }, + { "ENDLESS", cmd_endless }, + { "ABORT", cmd_abort }, + { "MONITOR", cmd_monitor }, + { "MQLEN", cmd_mqlen }, + { NULL, NULL }, +}; + +const char * +skynet_command(struct skynet_context * context, const char * cmd , const char * param) { + struct command_func * method = &cmd_funcs[0]; + while(method->name) { + if (strcmp(cmd, method->name) == 0) { + return method->func(context, param); } - if (param[i] == '\0') - return NULL; - - key[i] = '\0'; - param += i+1; - - skynet_setenv(key,param); - return NULL; - } - - if (strcmp(cmd,"STARTTIME") == 0) { - uint32_t sec = skynet_gettime_fixsec(); - sprintf(context->result,"%u",sec); - return context->result; - } - - if (strcmp(cmd,"ENDLESS") == 0) { - if (context->endless) { - strcpy(context->result, "1"); - context->endless = false; - return context->result; - } - return NULL; - } - - if (strcmp(cmd,"ABORT") == 0) { - skynet_handle_retireall(); - return NULL; - } - - if (strcmp(cmd,"MONITOR") == 0) { - uint32_t handle=0; - if (param == NULL || param[0] == '\0') { - if (G_NODE.monitor_exit) { - // return current monitor serivce - sprintf(context->result, ":%x", G_NODE.monitor_exit); - return context->result; - } - return NULL; - } else { - if (param[0] == ':') { - handle = strtoul(param+1, NULL, 16); - } else if (param[0] == '.') { - handle = skynet_handle_findname(param+1); - } else { - skynet_error(context, "Can't monitor %s",param); - // todo : monitor global service - } - } - G_NODE.monitor_exit = handle; - return NULL; - } - - if (strcmp(cmd, "MQLEN") == 0) { - int len = skynet_mq_length(context->queue); - sprintf(context->result, "%d", len); - return context->result; + ++method; } return NULL; @@ -526,7 +575,6 @@ skynet_send(struct skynet_context * context, uint32_t source, uint32_t destinati if (skynet_context_push(destination, &smsg)) { skynet_free(data); - skynet_error(NULL, "Drop message from %x to %x (type=%d)(size=%d)", source, destination, type&0xff, (int)(sz & HANDLE_MASK)); return -1; } } @@ -543,16 +591,15 @@ skynet_sendname(struct skynet_context * context, const char * addr , int type, i des = skynet_handle_findname(addr + 1); if (des == 0) { if (type & PTYPE_TAG_DONTCOPY) { - skynet_free(data); - } - skynet_error(context, "Drop message to %s", addr); + skynet_free(data); + } return session; } } else { _filter_args(context, type, &session, (void **)&data, &sz); struct remote_message * rmsg = skynet_malloc(sizeof(*rmsg)); - _copy_name(rmsg->destination.name, addr); + copy_name(rmsg->destination.name, addr); rmsg->destination.handle = 0; rmsg->message = data; rmsg->sz = sz; @@ -590,3 +637,27 @@ skynet_context_send(struct skynet_context * ctx, void * msg, size_t sz, uint32_t skynet_mq_push(ctx->queue, &smsg); } + +void +skynet_globalinit(void) { + G_NODE.total = 0; + G_NODE.monitor_exit = 0; + if (pthread_key_create(&G_NODE.handle_key, NULL)) { + fprintf(stderr, "pthread_key_create failed"); + exit(1); + } + // set mainthread's key + skynet_initthread(THREAD_MAIN); +} + +void +skynet_globalexit(void) { + pthread_key_delete(G_NODE.handle_key); +} + +void +skynet_initthread(int m) { + uintptr_t v = (uint32_t)(-m); + pthread_setspecific(G_NODE.handle_key, (void *)v); +} + diff --git a/skynet-src/skynet_server.h b/skynet-src/skynet_server.h index 604529fe..7adc1292 100644 --- a/skynet-src/skynet_server.h +++ b/skynet-src/skynet_server.h @@ -21,4 +21,8 @@ int skynet_context_total(); void skynet_context_endless(uint32_t handle); // for monitor +void skynet_globalinit(void); +void skynet_globalexit(void); +void skynet_initthread(int m); + #endif diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index bdb8c811..d1f80e10 100644 --- a/skynet-src/skynet_start.c +++ b/skynet-src/skynet_start.c @@ -5,7 +5,6 @@ #include "skynet_handle.h" #include "skynet_module.h" #include "skynet_timer.h" -#include "skynet_harbor.h" #include "skynet_monitor.h" #include "skynet_socket.h" @@ -50,6 +49,7 @@ wakeup(struct monitor *m, int busy) { static void * _socket(void *p) { struct monitor * m = p; + skynet_initthread(THREAD_SOCKET); for (;;) { int r = skynet_socket_poll(); if (r==0) @@ -81,6 +81,7 @@ _monitor(void *p) { struct monitor * m = p; int i; int n = m->count; + skynet_initthread(THREAD_MONITOR); for (;;) { CHECK_ABORT for (i=0;iid; struct monitor *m = wp->m; struct skynet_monitor *sm = m->m[id]; + skynet_initthread(THREAD_WORKER); for (;;) { if (skynet_context_message_dispatch(sm)) { CHECK_ABORT @@ -177,12 +180,17 @@ _start(int thread) { free_monitor(m); } -static int -_start_master(const char * master) { - struct skynet_context *ctx = skynet_context_new("master", master); - if (ctx == NULL) - return 1; - return 0; +static void +bootstrap(const char * cmdline) { + int sz = strlen(cmdline); + char name[sz+1]; + char args[sz+1]; + sscanf(cmdline, "%s %s", name, args); + struct skynet_context *ctx = skynet_context_new(name, args); + if (ctx == NULL) { + skynet_error(NULL, "Bootstrap error : %s\n", cmdline); + exit(1); + } } void @@ -194,32 +202,8 @@ skynet_start(struct skynet_config * config) { skynet_timer_init(); skynet_socket_init(); - struct skynet_context *ctx; - ctx = skynet_context_new("logger", config->logger); - if (ctx == NULL) { - fprintf(stderr,"launch logger error"); - exit(1); - } - - if (config->standalone) { - if (_start_master(config->standalone)) { - fprintf(stderr, "Init fail : mater"); - return; - } - } - // harbor must be init first - if (skynet_harbor_start(config->master , config->local)) { - fprintf(stderr, "Init fail : no master"); - return; - } - - ctx = skynet_context_new("snlua", "launcher"); - if (ctx) { - skynet_command(ctx, "REG", ".launcher"); - ctx = skynet_context_new("snlua", config->start); - } + bootstrap(config->bootstrap); _start(config->thread); skynet_socket_free(); } - diff --git a/test/testdeadcall.lua b/test/testdeadcall.lua new file mode 100644 index 00000000..d348a411 --- /dev/null +++ b/test/testdeadcall.lua @@ -0,0 +1,27 @@ +local skynet = require "skynet" + +local mode = ... + +if mode == "test" then + +skynet.start(function() + skynet.dispatch("lua", function (...) + print("====>", ...) + skynet.exit() + end) +end) + +else + + skynet.start(function() + local test = skynet.newservice("testdeadcall", "test") -- launch self in test mode + + print(pcall(function() + skynet.send(test,"lua", "hello world") + skynet.send(test,"lua", "never get there") + skynet.call(test,"lua", "fake call") + end)) + + skynet.exit() + end) +end \ No newline at end of file diff --git a/test/testlog.lua b/test/testlog.lua deleted file mode 100644 index f14e8066..00000000 --- a/test/testlog.lua +++ /dev/null @@ -1,9 +0,0 @@ -local skynet = require "skynet" -local log = require "log" - -skynet.start(function() - log.Info("hello world") - skynet.exit() -end) - - diff --git a/test/testmulticast.lua b/test/testmulticast.lua new file mode 100644 index 00000000..88f79ccb --- /dev/null +++ b/test/testmulticast.lua @@ -0,0 +1,40 @@ +local skynet = require "skynet" +local mc = require "multicast" +local dc = require "datacenter" + +local mode = ... + +if mode == "sub" then + +skynet.start(function() + skynet.dispatch("lua", function (_,_, cmd, channel) + assert(cmd == "init") + local c = mc.new { + channel = channel , + dispatch = function (channel, source, ...) + print(string.format("%s <=== %s %s",skynet.address(skynet.self()),skynet.address(source), channel), ...) + end + } + print(skynet.address(skynet.self()), "sub", c) + c:subscribe() + skynet.ret(skynet.pack()) + end) +end) + +else + +skynet.start(function() + local channel = mc.new() + print("New channel", channel) + for i=1,10 do + local sub = skynet.newservice("testmulticast", "sub") + skynet.call(sub, "lua", "init", channel.channel) + end + + dc.set("MCCHANNEL", channel.channel) -- for multi node test + + print(skynet.address(skynet.self()), "===>", channel) + channel:publish("Hello World") +end) + +end \ No newline at end of file diff --git a/test/testmulticast2.lua b/test/testmulticast2.lua new file mode 100644 index 00000000..0f55b77d --- /dev/null +++ b/test/testmulticast2.lua @@ -0,0 +1,24 @@ +local skynet = require "skynet" +local dc = require "datacenter" +local mc = require "multicast" + +skynet.start(function() + print("remote start") + skynet.monitor("simplemonitor", true) + local console = skynet.newservice("console") + local channel = dc.get "MCCHANNEL" + for i=1,10 do + local sub = skynet.newservice("testmulticast", "sub") + skynet.call(sub, "lua", "init", channel) + end + local c = mc.new { + channel = channel , + dispatch = function(...) print("======>", ...) end, + } + c:subscribe() + c:publish("Remote message") + c:unsubscribe() + c:publish("Remote message2") + c:delete() + skynet.exit() +end) diff --git a/test/testping.lua b/test/testping.lua index 61a82bc0..c9f05e63 100644 --- a/test/testping.lua +++ b/test/testping.lua @@ -2,7 +2,7 @@ local skynet = require "skynet" local snax = require "snax" skynet.start(function() - local ps = snax.newservice ("pingserver", "hello world") + local ps = snax.uniqueservice ("pingserver", "hello world") print(ps.req.ping("foobar")) print(ps.post.hello()) print(pcall(ps.req.error)) diff --git a/test/testsocket.lua b/test/testsocket.lua index 3794fefb..c465079c 100644 --- a/test/testsocket.lua +++ b/test/testsocket.lua @@ -37,7 +37,8 @@ else end skynet.start(function() - local id = socket.listen("127.0.0.1", 8000) + local id = socket.listen("127.0.0.1", 8001) + print("Listen socket :", "127.0.0.1", 8001) socket.start(id , function(id, addr) print("connect from " .. addr .. " " .. id)