diff --git a/connection/main.c b/connection/main.c index 8edaa8c5..75fd959b 100644 --- a/connection/main.c +++ b/connection/main.c @@ -116,20 +116,22 @@ _poll(struct connection_server * server) { connection_del(server->pool, c->fd); free(buffer); buffer = NULL; - skynet_send(server->ctx, 0, c->address, SESSION_CLIENT, NULL, 0, DONTCOPY); + // todo: support user defined type + skynet_send(server->ctx, 0, c->address, PTYPE_CLIENT | PTYPE_TAG_DONTCOPY, 0, NULL, 0); } else { - skynet_send(server->ctx, 0, c->address, SESSION_CLIENT, buffer, size, DONTCOPY); + skynet_send(server->ctx, 0, c->address, PTYPE_CLIENT | PTYPE_TAG_DONTCOPY, 0, buffer, size); buffer = NULL; } } } static int -_main(struct skynet_context * ctx, void * ud, int session, uint32_t source, const void * msg, size_t sz) { - if (msg == NULL) { +_main(struct skynet_context * ctx, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { + if (type == PTYPE_RESPONSE) { _poll(ud); return 0; } + assert(type == PTYPE_TEXT); const char * param = (const char *)msg + 4; if (memcmp(msg, "ADD ", 4)==0) { char * endptr; diff --git a/gate/main.c b/gate/main.c index 1c1fa3a9..ff284372 100644 --- a/gate/main.c +++ b/gate/main.c @@ -24,6 +24,7 @@ struct gate { int id_index; int cap; int max_connection; + int client_tag; struct connection ** agent; struct connection * map; }; @@ -124,24 +125,23 @@ _report(struct gate *g, struct skynet_context * ctx, const char * data, ...) { int n = vsnprintf(tmp, sizeof(tmp), data, ap); va_end(ap); - skynet_send(ctx, 0, g->watchdog, 0, tmp, n, 0); + skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT, 0, tmp, n); } static void _forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) { if (g->broker) { - skynet_send(ctx, 0, g->broker, SESSION_CLIENT, data, len, 0); + skynet_send(ctx, 0, g->broker, g->client_tag, 0, data, len); return; } struct connection * agent = _id_to_agent(g,uid); if (agent->agent) { - // client package has not session , send 0x7fffffff - skynet_send(ctx, agent->client, agent->agent, SESSION_CLIENT, data, len, 0); + skynet_send(ctx, agent->client, agent->agent, g->client_tag, 0 , data, len); } else if (g->watchdog) { char * tmp = malloc(len + 32); int n = snprintf(tmp,len+32,"%d data ",uid); memcpy(tmp+n,data,len); - skynet_send(ctx, 0, g->watchdog, 0, tmp, len + n, DONTCOPY); + skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT | PTYPE_TAG_DONTCOPY, 0, tmp, len + n); } } @@ -172,12 +172,13 @@ _remove_id(struct gate *g, int uid) { } static int -_cb(struct skynet_context * ctx, void * ud, int session, uint32_t source, const void * msg, size_t sz) { +_cb(struct skynet_context * ctx, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { struct gate *g = ud; - if (msg) { + if (type == PTYPE_TEXT) { _ctrl(ctx, g , msg , (int)sz); return 0; } + assert(type == PTYPE_RESPONSE); struct mread_pool * m = g->pool; int connection_id = mread_poll(m,100); // timeout : 100ms if (connection_id < 0) { @@ -228,11 +229,15 @@ gate_init(struct gate *g , struct skynet_context * ctx, char * parm) { int sz = strlen(parm)+1; char watchdog[sz]; char binding[sz]; - int n = sscanf(parm, "%s %s %d %d",watchdog, binding,&max,&buffer); - if (n!=4) { + int client_tag = 0; + int n = sscanf(parm, "%s %s %d %d %d",watchdog, binding,&client_tag , &max,&buffer); + if (n<3) { skynet_error(ctx, "Invalid gate parm %s",parm); return 1; } + if (client_tag == 0) { + client_tag = PTYPE_CLIENT; + } char * portstr = strchr(binding,':'); uint32_t addr = INADDR_ANY; if (portstr == NULL) { @@ -273,6 +278,7 @@ gate_init(struct gate *g , struct skynet_context * ctx, char * parm) { g->cap = cap; g->max_connection = max; g->id_index = 0; + g->client_tag = client_tag; g->agent = malloc(cap * sizeof(struct connection *)); memset(g->agent, 0, cap * sizeof(struct connection *)); diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index d19c8181..a2a0ffcb 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -8,7 +8,7 @@ #include static int -_cb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { +_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 top = lua_gettop(L); @@ -19,23 +19,13 @@ _cb(struct skynet_context * context, void * ud, int session, uint32_t source, co lua_pushvalue(L,2); } - int r; - if (msg == NULL) { - if (source == 0) { - lua_pushinteger(L, session); - r = lua_pcall(L, 1, 0 , trace); - } else { - lua_pushinteger(L, session); - lua_pushnumber(L, source); - r = lua_pcall(L, 2, 0 , trace); - } - } else { - lua_pushinteger(L, session); - lua_pushnumber(L, source); - lua_pushlightuserdata(L,(void *)msg); - lua_pushinteger(L,sz); - r = lua_pcall(L, 4, 0 , trace); - } + lua_pushinteger(L, type); + lua_pushlightuserdata(L, (void *)msg); + lua_pushinteger(L,sz); + lua_pushinteger(L, session); + lua_pushnumber(L, source); + + int r = lua_pcall(L, 5, 0 , trace); if (r == LUA_OK) return 0; const char * self = skynet_command(context, "REG", NULL); @@ -98,45 +88,60 @@ _command(lua_State *L) { static int _genid(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); - int session = skynet_send(context, 0, 0, -1, NULL, 0 , 0); + int session = skynet_send(context, 0, 0, PTYPE_TAG_ALLOCSESSION , 0 , NULL, 0); lua_pushinteger(L, session); return 1; } +// copy from _send + static int _sendname(lua_State *L, struct skynet_context * context, const char * dest) { + int type = luaL_checkinteger(L, 2); int session = 0; - int index = 0; - - if (lua_type(L,2) == LUA_TNUMBER) { - session = lua_tointeger(L,2); - ++index; - } - if (lua_gettop(L) == index + 1) { - session = skynet_sendname(context, dest, session , NULL, 0, 0); + if (lua_isnil(L,3)) { + type |= PTYPE_TAG_ALLOCSESSION; } else { - int type = lua_type(L,index+2); - if (type == LUA_TSTRING) { - size_t len = 0; - void * msg = (void *)lua_tolstring(L,index+2,&len); - session = skynet_sendname(context, dest, session , msg, len, 0); - } else if (type == LUA_TNIL) { - session = skynet_sendname(context, dest, session , NULL, 0, 0); - } else { - luaL_checktype(L,index+2, LUA_TLIGHTUSERDATA); - void * msg = lua_touserdata(L,index+2); - int size = luaL_checkinteger(L,index+3); - session = skynet_sendname(context, dest, session, msg, size, DONTCOPY); - } + session = luaL_checkinteger(L,3); + } + + int mtype = lua_type(L,4); + switch (mtype) { + case LUA_TSTRING: { + size_t len = 0; + void * msg = (void *)lua_tolstring(L,4,&len); + session = skynet_sendname(context, dest, type, session , msg, len); + break; + } + case LUA_TNIL : + session = skynet_sendname(context, dest, type, session , NULL, 0); + break; + case LUA_TLIGHTUSERDATA: { + luaL_checktype(L, 4, LUA_TLIGHTUSERDATA); + void * msg = lua_touserdata(L,4); + int size = luaL_checkinteger(L,5); + session = skynet_sendname(context, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size); + break; + } + default: + luaL_error(L, "skynet.send invalid param %s", lua_type(L,4)); } lua_pushinteger(L,session); return 1; } +/* + unsigned address + string address + integer type + integer session + string message + lightuserdata message_ptr + integer len + */ static int _send(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); - int session = 0; int addr_type = lua_type(L,1); uint32_t dest = 0; switch(addr_type) { @@ -159,30 +164,33 @@ _send(lua_State *L) { return luaL_error(L, "address must be number or string, got %s",lua_typename(L,addr_type)); } - int index = 0; - if (lua_type(L,2) == LUA_TNUMBER) { - session = lua_tointeger(L,2); - ++index; - } - if (lua_gettop(L) == index + 1) { - session = skynet_send(context, 0, dest, session , NULL, 0, 0); + int type = luaL_checkinteger(L, 2); + int session = 0; + if (lua_isnil(L,3)) { + type |= PTYPE_TAG_ALLOCSESSION; } else { - int type = lua_type(L,index+2); - if (type == LUA_TSTRING) { - size_t len = 0; - void * msg = (void *)lua_tolstring(L,index+2,&len); - session = skynet_send(context, 0, dest, session , msg, len, 0); - } else if (type == LUA_TNIL) { - session = skynet_send(context, 0, dest, session , NULL, 0, 0); - } else { - luaL_checktype(L,index+2, LUA_TLIGHTUSERDATA); - void * msg = lua_touserdata(L,index+2); - int size = luaL_checkinteger(L,index+3); - session = skynet_send(context, 0, dest, session, msg, size, DONTCOPY); - } + session = luaL_checkinteger(L,3); } - if (session < 0) { - return luaL_error(L, "skynet.send drop the message to %s", dest); + + int mtype = lua_type(L,4); + switch (mtype) { + case LUA_TSTRING: { + size_t len = 0; + void * msg = (void *)lua_tolstring(L,4,&len); + if (len == 0) { + msg = NULL; + } + session = skynet_send(context, 0, dest, type, session , msg, len); + break; + } + case LUA_TLIGHTUSERDATA: { + void * msg = lua_touserdata(L,4); + int size = luaL_checkinteger(L,5); + session = skynet_send(context, 0, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size); + break; + } + default: + luaL_error(L, "skynet.send invalid param %s", lua_type(L,4)); } lua_pushinteger(L,session); return 1; @@ -193,26 +201,29 @@ _redirect(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); uint32_t dest = luaL_checkunsigned(L,1); uint32_t source = luaL_checkunsigned(L,2); - int session = luaL_checkinteger(L,3); + int type = luaL_checkinteger(L,3); + int session = luaL_checkinteger(L,4); - if (lua_gettop(L) == 3) { - session = skynet_send(context, source, dest, session , NULL, 0, 0); - } else { - int type = lua_type(L,4); - if (type == LUA_TSTRING) { - size_t len = 0; - void * msg = (void *)lua_tolstring(L,4,&len); - skynet_send(context, source, dest, session , msg, len, 0); - } else if (type == LUA_TNIL) { - session = skynet_send(context, source, dest, session , NULL, 0, 0); - } else { - luaL_checktype(L, 4, LUA_TLIGHTUSERDATA); - void * msg = lua_touserdata(L,4); - int size = luaL_checkinteger(L,5); - skynet_send(context, source, dest, session, msg, size, DONTCOPY); + int mtype = lua_type(L,5); + switch (mtype) { + case LUA_TSTRING: { + size_t len = 0; + void * msg = (void *)lua_tolstring(L,5,&len); + if (len == 0) { + msg = NULL; } + session = skynet_send(context, source, dest, type, session , msg, len); + break; + } + case LUA_TLIGHTUSERDATA: { + void * msg = lua_touserdata(L,5); + int size = luaL_checkinteger(L,6); + session = skynet_send(context, source, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size); + break; + } + default: + luaL_error(L, "skynet.redirect invalid param %s", lua_type(L,5)); } - return 0; } diff --git a/lualib/logger.lua b/lualib/logger.lua index 48fbffe8..219b7692 100644 --- a/lualib/logger.lua +++ b/lualib/logger.lua @@ -22,7 +22,7 @@ logger.CRITICAL = 50 logger.FATAL = 60 local function log_to_disk(...) - Skynet.send(".lualog" , Skynet.pack(...)) + Skynet.send(".lualog" , "lua" , ...) end local function dump_log_to_disk(t) diff --git a/lualib/mcgroup.lua b/lualib/mcgroup.lua index d7c09321..1a441d82 100644 --- a/lualib/mcgroup.lua +++ b/lualib/mcgroup.lua @@ -5,7 +5,7 @@ local SERVICE = "GROUPMGR" local group = {} function group.create() - return skynet.call(SERVICE, skynet.unpack, skynet.pack("NEW" , skynet.self())) + return skynet.call(SERVICE, "lua", "NEW" , skynet.self()) end function group.address(id) @@ -15,16 +15,16 @@ end function group.enter(id, handle) handle = handle or skynet.self() - skynet.send(SERVICE, 0 , skynet.pack("ENTER", handle, id)) + skynet.send(SERVICE, "lua" , "ENTER", handle, id) end function group.leave(id, handle) handle = handle or skynet.self() - skynet.send(SERVICE, 0 , skynet.pack("LEAVE", handle, id)) + skynet.send(SERVICE, "lua" , "LEAVE", handle, id) end function group.release(id) - skynet.send(SERVICE, 0 , skynet.pack("CLEAR", skynet.self(), id)) + skynet.send(SERVICE, "lua" , "CLEAR", skynet.self(), id) end return group \ No newline at end of file diff --git a/lualib/redis.lua b/lualib/redis.lua index 688b9a2d..2b06edd2 100644 --- a/lualib/redis.lua +++ b/lualib/redis.lua @@ -10,7 +10,7 @@ local command = {} setmetatable(command, { __index = function(t,k) local cmd = string.upper(k) local f = function(self, ...) - local err, result = skynet.call( self.__handle, skynet.unpack, skynet.pack(cmd, ...)) + local err, result = skynet.call( self.__handle, "lua", cmd, ...) assert(err, result) return result end @@ -19,7 +19,7 @@ setmetatable(command, { __index = function(t,k) end}) function command:exists(key) - local result , exists = skynet.call( self.__handle, skynet.unpack, skynet.pack("EXISTS", key)) + local result , exists = skynet.call( self.__handle, "lua" , "EXISTS", key) assert(result, exists) exists = exists ~= 0 return exists @@ -30,7 +30,7 @@ local meta = { } function redis.connect(dbname) - local handle = skynet.call(".redis-manager",skynet.unpack, skynet.pack(dbname)) + local handle = skynet.call(".redis-manager", "lua", dbname) assert(handle ~= nil) return setmetatable({ __handle = handle } , meta) end diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 01a98c02..a19fbf9d 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -5,7 +5,18 @@ local coroutine = coroutine local assert = assert local pairs = pairs +local proto = {} local skynet = {} + +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 = {} @@ -41,7 +52,11 @@ function suspend(co, result, command, param, size) elseif command == "RETURN" then local co_session = session_coroutine_id[co] local co_address = session_coroutine_address[co] - c.send(co_address, co_session, param, size) + -- PTYPE_RESPONSE = 1 , see skynet.h + if param == nil then + error(debug.traceback(co)) + end + c.send(co_address, 1, co_session, param, size) return suspend(co, coroutine.resume(co)) elseif command == nil then -- coroutine exit @@ -74,7 +89,9 @@ function skynet.sleep(ti) assert(session) local ret = coroutine.yield("SLEEP", tonumber(session)) sleep_session[coroutine.running()] = nil - return ret + if ret == true then + return "BREAK" + end end function skynet.yield() @@ -136,61 +153,30 @@ function skynet.setenv(key, value) c.command("SETENV",key .. " " ..value) end -skynet.send = assert(c.send) +function skynet.send(addr, typename, ...) + local p = proto[typename] + return c.send(addr, p.id, 0 , p.pack(...)) +end + skynet.genid = assert(c.genid) -skynet.redirect = assert(c.redirect) + +skynet.redirect = function(dest,source,typename,...) + return c.redirect(dest, source, proto[typename].id, ...) +end + skynet.pack = assert(c.pack) -skynet.tostring = assert(c.tostring) skynet.unpack = assert(c.unpack) +skynet.tostring = assert(c.tostring) -function skynet.call(addr, deseri , ...) - if deseri == nil then - local session = c.send(addr, -1, ...) - return coroutine.yield("CALL", session) - end - local t = type(deseri) - if t == "function" then - local session = c.send(addr, -1, ...) - return deseri(coroutine.yield("CALL", session)) - else - assert(t=="string") - local session = c.send(addr, -1, deseri) - return c.tostring(coroutine.yield("CALL", session)) - end +function skynet.call(addr, typename, ...) + local p = proto[typename] + local session = c.send(addr, p.id , nil , p.pack(...)) + return p.unpack(coroutine.yield("CALL", session)) end -function skynet.ret(...) - coroutine.yield("RETURN", ...) -end - -local function default_dispatch(f, unknown) - unknown = unknown or function (session, msg, sz) - local self = skynet.self() - print(self, session,msg,sz) - error(string.format("%s Unknown session %d", self, session)) - end - return function(session, address , msg, sz) - if session == nil then - return - end - if session <= 0 then - session = - session - co = coroutine.create(f) - session_coroutine_id[co] = session - session_coroutine_address[co] = address - suspend(co, coroutine.resume(co, msg, sz, session, address)) - else - local co = session_id_coroutine[session] - if co == "BREAK" then - session_id_coroutine[session] = nil - elseif co == nil then - unknown(session,msg,sz) - else - session_id_coroutine[session] = nil - suspend(co, coroutine.resume(co, msg, sz)) - end - end - end +function skynet.ret(msg, sz) + msg = msg or "" + coroutine.yield("RETURN", msg, sz) end function skynet.wakeup(co) @@ -200,30 +186,54 @@ function skynet.wakeup(co) end end -function skynet.dispatch(f,unknown) - c.callback(default_dispatch(f,unknown)) +function skynet.dispatch(typename, func) + local p = assert(proto[typename],tostring(typename)) + assert(p.dispatch == nil, tostring(typename)) + p.dispatch = func end -function skynet.filter(filter, f, unknown) - local func = default_dispatch(f, unknown) - c.callback(function (...) - func(filter(...)) +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 function start_skynet() + c.callback(function(prototype, msg, sz, session, source) + -- 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, msg, sz)) + end + else + local p = assert(proto[prototype], prototype) + local f = p.dispatch + if f then + local co = coroutine.create(f) + session_coroutine_id[co] = session + session_coroutine_address[co] = source + suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz))) + else + print("Unknown request :" , p.unpack(msg,sz)) + error(string.format("Can't dispatch type %s : ", p.name)) + end + end end) end -function skynet.start(f) - local session = c.command("TIMEOUT","0") - local co = coroutine.create( - function(...) - f(...) - skynet.send(".launcher",0) - end - ) - session_id_coroutine[tonumber(session)] = co -end - function skynet.newservice(name, ...) - local handle = skynet.call(".launcher", skynet.unpack, skynet.pack("snlua", name, ...)) + local handle = skynet.call(".launcher", "lua" , "snlua", name, ...) return handle end @@ -262,6 +272,8 @@ end ------ remote object -------- do + -- proto is 11 + local remote_query, remote_alloc, remote_bind = c.remote_init(skynet.self()) local weak_meta = { __mode = "kv" } local meta = getmetatable(c.unpack(c.pack({ __remote = 0 }))) @@ -279,7 +291,8 @@ do if f == nil then f = function(...) local addr = remote_query(t.__remote) - local session = _send(addr, -1, _pack(t,method,...)) + -- the proto is 11 (lua is 10) + local session = _send(addr, 11 , nil, _pack(t,method,...)) local msg, sz = _yield("CALL", session) return select(2,assert(_unpack(msg,sz))) end @@ -321,16 +334,59 @@ do return _yield("RETURN", _pack(pcall(f,...))) end - function skynet.remote_service(unknown) - local function f(msg,sz) - return remote_call(_unpack(msg,sz)) - end - c.callback(default_dispatch(f,unknown)) - end - function skynet.remote_root() return skynet.remote_bind(0) end + + skynet.register_protocol { + name = "remoteobj", + id = 11, + unpack = c.unpack, + dispatch = function (session, source, ...) + remote_call(...) + end + } +end + +----- register protocol +do + local REG = skynet.register_protocol + + REG { + name = "text", + id = 0, + 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 = 10, + pack = skynet.pack, + unpack = skynet.unpack, + } + + REG { + name = "response", + id = 1, + } +end + +function skynet.start(f) + start_skynet() + skynet.timeout(0, function() + f() + skynet.send(".launcher","lua", nil) + end) end return skynet diff --git a/lualib/socket.lua b/lualib/socket.lua index 7453b162..b248f7b8 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -12,9 +12,7 @@ function socket.connect(addr) if fd == nil then return true end - print("connect to " .. addr) - local command = "ADD "..fd.." ".. skynet.address(skynet.self()) - skynet.send(".connection", command ) + skynet.send(".connection", "text", "ADD", fd , skynet.address(skynet.self())) object = c.new() end @@ -47,7 +45,7 @@ end function socket.close() if fd then c.close(fd) - skynet.send(".connection","DEL "..fd) + skynet.send(".connection","text", "DEL", fd) fd = nil end end diff --git a/service-src/service_client.c b/service-src/service_client.c index b4b6c929..08f61215 100644 --- a/service-src/service_client.c +++ b/service-src/service_client.c @@ -9,7 +9,7 @@ #include static int -_cb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { +_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { assert(sz <= 65535); int fd = (int)(intptr_t)ud; diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 6bc7ead8..5089d548 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -40,6 +40,10 @@ struct hashmap { struct keyvalue *node[HASH_SIZE]; }; +/* + message type (8bits) is in destination high 8bits + harbor id (8bits) is also in that place , but remote message doesn't need harbor id. + */ struct remote_message_header { uint32_t source; uint32_t destination; @@ -354,7 +358,7 @@ _dispatch_queue(struct harbor *h, struct skynet_context * context, struct msg_qu struct msg * m = _pop_queue(queue); while (m) { struct remote_message_header * cookie = (struct remote_message_header *)(m->buffer + m->size - sizeof(*cookie)); - cookie->destination = handle; + cookie->destination |= (handle & 0xffffff); _header_to_message(cookie, (uint32_t *)cookie); int err = _send_package(fd, m->buffer, m->size); if (err) { @@ -411,12 +415,12 @@ _request_master(struct harbor *h, struct skynet_context * context, const char na */ static int -_remote_send_handle(struct harbor *h, struct skynet_context * context, uint32_t source, uint32_t destination, int session, const char * msg, size_t sz) { +_remote_send_handle(struct harbor *h, struct skynet_context * context, uint32_t source, uint32_t destination, int type, int session, const char * msg, size_t sz) { int harbor_id = destination >> HANDLE_REMOTE_SHIFT; assert(harbor_id != 0); if (harbor_id == h->id) { // local message - skynet_send(context, source, destination , session, (void *)msg, sz, DONTCOPY); + skynet_send(context, source, destination , type | PTYPE_TAG_DONTCOPY, session, (void *)msg, sz); return 1; } @@ -424,7 +428,7 @@ _remote_send_handle(struct harbor *h, struct skynet_context * context, uint32_t if (fd >= 0) { struct remote_message_header cookie; cookie.source = source; - cookie.destination = destination; + cookie.destination = (destination & 0xffffff) | ((uint32_t)type << 24); cookie.session = (uint32_t)session; int err = _send_remote(fd, msg,sz,&cookie); if (err) { @@ -456,7 +460,7 @@ _remote_register_name(struct harbor *h, struct skynet_context * context, const c } static int -_remote_send_name(struct harbor *h, struct skynet_context * context, uint32_t source, const char name[GLOBALNAME_LENGTH], int session, const char * msg, size_t sz) { +_remote_send_name(struct harbor *h, struct skynet_context * context, uint32_t source, const char name[GLOBALNAME_LENGTH], int type, int session, const char * msg, size_t sz) { struct keyvalue * node = _hash_search(h->map, name); if (node == NULL) { node = _hash_insert(h->map, name); @@ -467,14 +471,14 @@ _remote_send_name(struct harbor *h, struct skynet_context * context, uint32_t so } struct remote_message_header header; header.source = source; - header.destination = 0; + header.destination = type << 24; header.session = (uint32_t)session; _push_queue(node->queue, msg, sz, &header); // 0 for request _remote_register_name(h, context, name, 0); return 1; } else { - return _remote_send_handle(h, context, source, node->value, session, msg, sz); + return _remote_send_handle(h, context, source, node->value, type, session, msg, sz); } } @@ -485,9 +489,11 @@ _report_local_address(struct harbor *h, struct skynet_context * context, const c } static int -_mainloop(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { +_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { struct harbor * h = ud; - if (session == SESSION_CLIENT) { + switch (type) { + case PTYPE_HARBOR: { + // remote message in const char * cookie = msg; cookie += sz - 12; struct remote_message_header header; @@ -511,28 +517,37 @@ _mainloop(struct skynet_context * context, void * ud, int session, uint32_t sour _update_remote_name(h, context, msg, header.destination); } } else { - skynet_send(context, header.source, header.destination, (int)header.session, (void *)msg, sz-12, DONTCOPY); + uint32_t destination = header.destination; + int type = (destination >> 24) | PTYPE_TAG_DONTCOPY; + destination = (destination & 0xffffff) | ((uint32_t)h->id << 24); + skynet_send(context, header.source, destination, type, (int)header.session, (void *)msg, sz-12); return 1; } - } else { + return 0; + } + case PTYPE_SYSTEM: { + // register name message + const struct remote_message *rmsg = msg; + assert (sz == sizeof(rmsg->destination)); + _remote_register_name(h, context, rmsg->destination.name, rmsg->destination.handle); + return 0; + } + default: { + // remote message out const struct remote_message *rmsg = msg; - if (sz == sizeof(rmsg->destination)) { - _remote_register_name(h, context, rmsg->destination.name, rmsg->destination.handle); - return 0; - } - assert(sz == sizeof(*rmsg)); if (rmsg->destination.handle == 0) { - if (_remote_send_name(h, context, source , rmsg->destination.name, session, rmsg->message, rmsg->sz)) { + if (_remote_send_name(h, context, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz)) { return 0; } } else { - if (_remote_send_handle(h, context, source , rmsg->destination.handle, session, rmsg->message, rmsg->sz)) { + if (_remote_send_handle(h, context, source , rmsg->destination.handle, type, session, rmsg->message, rmsg->sz)) { return 0; } } free((void *)rmsg->message); + return 0; + } } - return 0; } int @@ -553,7 +568,7 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) { h->master_fd = master_fd; char tmp[128]; - sprintf(tmp,"gate ! %s %d 0",local_addr,REMOTE_MAX); + sprintf(tmp,"gate ! %s %d %d 0",local_addr, PTYPE_HARBOR, REMOTE_MAX); const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp); if (gate_addr == NULL) { skynet_error(ctx, "Harbor : launch gate failed"); @@ -566,8 +581,8 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) { } const char * self_addr = skynet_command(ctx, "REG", NULL); int n = sprintf(tmp,"broker %s",self_addr); - skynet_send(ctx, 0, gate, 0, tmp, n, 0); - skynet_send(ctx, 0, gate, 0, "start", 5, 0); + skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, tmp, n); + skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, "start", 5); h->id = harbor_id; skynet_callback(ctx, h, _mainloop); diff --git a/service-src/service_master.c b/service-src/service_master.c index a6120555..09dc0749 100644 --- a/service-src/service_master.c +++ b/service-src/service_master.c @@ -242,9 +242,12 @@ _update_address(struct skynet_context * context, struct master *m, int harbor_id */ static int -_mainloop(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { +_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { + if (type != PTYPE_HARBOR) { + skynet_error(context, "None harbor message recv from %x (type = %d)", source, type); + return 0; + } struct master *m = ud; - assert(session == SESSION_CLIENT); uint32_t handle = 0; memcpy(&handle, msg, 4); handle = ntohl(handle); @@ -266,7 +269,7 @@ _mainloop(struct skynet_context * context, void * ud, int session, uint32_t sour int master_init(struct master *m, struct skynet_context *ctx, const char * args) { char tmp[strlen(args) + 32]; - sprintf(tmp,"gate ! %s %d 0",args,REMOTE_MAX); + sprintf(tmp,"gate ! %s %d %d 0",args,PTYPE_HARBOR,REMOTE_MAX); const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp); if (gate_addr == NULL) { skynet_error(ctx, "Master : launch gate failed"); @@ -279,8 +282,8 @@ master_init(struct master *m, struct skynet_context *ctx, const char * args) { } const char * self_addr = skynet_command(ctx, "REG", NULL); int n = sprintf(tmp,"broker %s",self_addr); - skynet_send(ctx, 0, gate, 0, tmp, n, 0); - skynet_send(ctx, 0, gate, 0, "start", 5, 0); + skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, tmp, n); + skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, "start", 5); skynet_callback(ctx, m, _mainloop); diff --git a/service-src/service_multicast.c b/service-src/service_multicast.c index a2155935..c2941acd 100644 --- a/service-src/service_multicast.c +++ b/service-src/service_multicast.c @@ -16,9 +16,9 @@ multicast_release(struct skynet_multicast_group *g) { } static int -_maincb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { +_maincb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { struct skynet_multicast_group *g = ud; - if (source == 0) { + if (type == PTYPE_SYSTEM) { char cmd = '\0'; uint32_t handle = 0; sscanf(msg,"%c %x",&cmd,&handle); @@ -42,6 +42,7 @@ _maincb(struct skynet_context * context, void * ud, int session, uint32_t source } return 0; } else { + sz |= type << 24; struct skynet_multicast_message * mc = skynet_multicast_create(msg, sz, source); skynet_multicast_castgroup(context, g, mc); return 1; diff --git a/service-src/service_tunnel.c b/service-src/service_tunnel.c index 9fb6a92d..c2aae7fb 100644 --- a/service-src/service_tunnel.c +++ b/service-src/service_tunnel.c @@ -2,7 +2,7 @@ #include static int -_cb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { +_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { uint32_t dest = (uint32_t)(uintptr_t)ud; skynet_forward(context, dest); return 0; diff --git a/service/agent.lua b/service/agent.lua index ba5bdfab..e51e0d04 100644 --- a/service/agent.lua +++ b/service/agent.lua @@ -1,23 +1,19 @@ local skynet = require "skynet" local client = ... -local session_id = 0 -skynet.filter(function (session, address , msg, sz) - if session == 0x7fffffff then - skynet.send("LOG", "client message :" .. skynet.tostring(msg,sz)) +skynet.register_protocol { + name = "client", + id = 3, + pack = function(...) return ... end, + unpack = skynet.tostring, + dispatch = function (session, address, text) -- It's client, there is no session - session_id = session_id + 1 - session = - session_id - else - skynet.send("LOG", "skynet message") + skynet.send("LOG", "text", "client message :" .. text) + local result = skynet.call("SIMPLEDB", "text", text) + skynet.ret(result) end - return session, address , msg, sz -end, function (msg,sz) - local message = skynet.tostring(msg,sz) - local result = skynet.call("SIMPLEDB",message) - skynet.ret(result) -end) +} skynet.start(function() - skynet.send(client,"Welcome to skynet") + skynet.send(client,"text","Welcome to skynet") end) diff --git a/service/console.lua b/service/console.lua index 70e22b12..95c3d5a2 100644 --- a/service/console.lua +++ b/service/console.lua @@ -1,7 +1,5 @@ local skynet = require "skynet" -skynet.dispatch() - skynet.start(function() while true do local cmd = io.read() diff --git a/service/globallog.lua b/service/globallog.lua index f506b6e4..ec431cce 100644 --- a/service/globallog.lua +++ b/service/globallog.lua @@ -1,8 +1,8 @@ local skynet = require "skynet" -skynet.dispatch(function(msg, sz , session , from) - local message = skynet.tostring(msg,sz) - print("[GLOBALLOG]", skynet.address(from),message) +skynet.start(function() + skynet.dispatch("text", function(session, address, text) + print("[GLOBALLOG]", skynet.address(address),text) + end) + skynet.register "LOG" end) - -skynet.register "LOG" \ No newline at end of file diff --git a/service/group_agent.lua b/service/group_agent.lua index 3dcfa84b..386a7785 100644 --- a/service/group_agent.lua +++ b/service/group_agent.lua @@ -46,13 +46,11 @@ function command.CLEAR(id) end end -skynet.dispatch(function (msg,sz) - local cmd , id , handle = skynet.unpack(msg,sz) - local f = command[cmd] - assert(f, cmd) - f(id,handle) -end) - skynet.start(function() - skynet.send("GROUPMGR" , 0, skynet.pack("MASTER", skynet.self())) + skynet.dispatch("lua", function(session, address, cmd , id , handle) + local f = command[cmd] + assert(f, cmd) + f(id,handle) + end) + skynet.send("GROUPMGR" , "lua", "MASTER", skynet.self()) end) diff --git a/service/group_mgr.lua b/service/group_mgr.lua index 34e2e68c..ba6b9bd1 100644 --- a/service/group_mgr.lua +++ b/service/group_mgr.lua @@ -12,13 +12,13 @@ end local function create_group_in(harbor, id) local local_ctrl = assert(harbor_ctrl[harbor]) local g = multicast[id] - local local_mc = skynet.call(local_ctrl, skynet.unpack, skynet.pack("CREATE", id)) + local local_mc = skynet.call(local_ctrl, "lua", "CREATE", id) for _,mc in pairs(g) do - skynet.send(local_ctrl, 0, skynet.pack("AGENT", id, mc)) + skynet.send(local_ctrl, "lua", "AGENT", id, mc) end for harbor_id,ctrl in pairs(harbor_ctrl) do if harbor_id ~= harbor then - skynet.send(ctrl,0, skynet.pack("AGENT", id, local_mc)) + skynet.send(ctrl, "lua", "AGENT", id, local_mc) end end g[harbor] = local_mc @@ -38,14 +38,14 @@ function command.ENTER(address, harbor, id) end local local_ctrl = assert(harbor_ctrl[harbor]) - skynet.send(local_ctrl, 0, skynet.pack("ENTER", id, address)) + skynet.send(local_ctrl, "lua", "ENTER", id, address) end function command.LEAVE(address, harbor, id) local g = multicast[id] assert(g,id) local local_ctrl = assert(harbor_ctrl[harbor]) - skynet.send(local_ctrl, 0, skynet.pack("LEAVE", id, address)) + skynet.send(local_ctrl, "lua", "LEAVE", id, address) end function command.DELETE(_,_, id) @@ -54,16 +54,17 @@ function command.DELETE(_,_, id) multicast[id] = nil for harbor_id,_ in pairs(g) do - skynet.send(harbor_ctrl[harbor_id],0, skynet.pack("CLEAR", id)) + skynet.send(harbor_ctrl[harbor_id], "lua", "CLEAR", id) end end -skynet.dispatch(function (msg,sz) - local cmd , address , param = skynet.unpack(msg,sz) - local f = command[cmd] - assert(f, cmd, param) - local harbor = skynet.harbor(address) - f(address, harbor, param) +skynet.start(function() + skynet.dispatch("lua", function(_, _, cmd , address, param) + local f = command[cmd] + assert(f, cmd, param) + local harbor = skynet.harbor(address) + f(address, harbor, param) + end) + skynet.register("GROUPMGR") end) -skynet.register("GROUPMGR") \ No newline at end of file diff --git a/service/launcher.lua b/service/launcher.lua index ca42d2fa..8af6cc59 100644 --- a/service/launcher.lua +++ b/service/launcher.lua @@ -3,28 +3,26 @@ local string = string local instance = {} -local function _extract( service, ...) - return service , table.concat({...}, " ") -end +skynet.register(".launcher") -skynet.dispatch(function(msg, sz , session, address) - if session == 0 then - -- init notice - local reply = instance[address] - if reply then - skynet.send(reply[2] , reply[1], skynet.pack(address)) - instance[address] = nil - end - else - -- launch request - local service , param = _extract(skynet.unpack(msg,sz)) - local inst = skynet.launch(service, param) - if inst then - instance[inst] = { session, address } +skynet.start(function() + skynet.dispatch("lua" , function(session, address , service, ...) + if service == nil then + -- init notice + local reply = instance[address] + if reply then + skynet.redirect(reply.address , 0, "response", reply.session, skynet.pack(address)) + instance[address] = nil + end else - skynet.send(address, session, skynet.pack(nil)) + -- launch request + local param = table.concat({...}, " ") + local inst = skynet.launch(service, param) + if inst then + instance[inst] = { session = session, address = address } + else + skynet.ret(skynet.pack(nil)) + end end - end + end) end) - -skynet.register(".launcher") \ No newline at end of file diff --git a/service/lualog.lua b/service/lualog.lua index 1e60a957..be323a13 100644 --- a/service/lualog.lua +++ b/service/lualog.lua @@ -37,8 +37,9 @@ local function log(name, modname, level, timestamp, msg, src, tags) }) end -skynet.dispatch(function(msg, sz , session , from) - log(skynet.unpack(msg,sz)) +skynet.start(function() + skynet.dispatch("lua",function(session, from, ...) + log(...) + end) + skynet.register ".lualog" end) - -skynet.register ".lualog" \ No newline at end of file diff --git a/service/main.lua b/service/main.lua index a1951b8f..c67b0d4a 100644 --- a/service/main.lua +++ b/service/main.lua @@ -1,11 +1,9 @@ local skynet = require "skynet" -skynet.dispatch() - skynet.start(function() print("Server start") - local lualog = skynet.launch("snlua","lualog") local launcher = skynet.launch("snlua","launcher") + local lualog = skynet.launch("snlua","lualog") local group_mgr = skynet.launch("snlua", "group_mgr") local group_agent = skynet.launch("snlua", "group_agent") local remoteroot = skynet.launch("snlua","remote_root") @@ -15,5 +13,6 @@ skynet.start(function() local connection = skynet.launch("connection","256") local redis = skynet.launch("snlua","redis-mgr") -- skynet.launch("snlua","testgroup") + skynet.exit() end) diff --git a/service/main_log.lua b/service/main_log.lua index 4b98a112..b4243aa9 100644 --- a/service/main_log.lua +++ b/service/main_log.lua @@ -1,11 +1,9 @@ local skynet = require "skynet" -skynet.dispatch() - skynet.start(function() print("Log server start") - local lualog = skynet.launch("snlua","lualog") local launcher = skynet.launch("snlua","launcher") + local lualog = skynet.launch("snlua","lualog") local group_agent = skynet.launch("snlua", "group_agent") local console = skynet.launch("snlua","console") local log = skynet.launch("snlua","globallog") diff --git a/service/redis-cli.lua b/service/redis-cli.lua index 83ad42ec..4fedad27 100644 --- a/service/redis-cli.lua +++ b/service/redis-cli.lua @@ -27,7 +27,7 @@ local function compose_message(msg) end local function select_db(id) - local result , ok = skynet.call(skynet.self(), skynet.unpack, skynet.pack("SELECT", tostring(id))) + local result , ok = skynet.call(skynet.self(), "lua", "SELECT", tostring(id)) assert(result and ok == "OK") end @@ -48,7 +48,7 @@ end local function response(...) local reply = pop_request_queue() - skynet.send(reply[2],reply[1],skynet.pack(...)) + skynet.redirect(reply.address,0, "response", reply.session, skynet.pack(...)) end local function readline(sep) @@ -143,24 +143,28 @@ local function reconnect() split_co = coroutine.create(split_package) end -skynet.filter( - function(session, address , msg, sz) - if session == 0x7fffffff then - if msg == nil then - skynet.timeout(0, reconnect) - return - end - socket.push(msg,sz) - coroutine.resume(split_co) - elseif session < 0 then - local message = { skynet.unpack(msg,sz) } - local cmd = compose_message(message) - socket.write(cmd) - push_request_queue { -session , address, cmd } - else - return session, address, msg , sz +skynet.register_protocol { + name = "client", + id = 3, + pack = function(...) return ... end, + unpack = function(msg,sz) + if msg == nil then + skynet.timeout(0, reconnect) + return end - end -) + socket.push(msg,sz) + assert(coroutine.resume(split_co)) + end, + dispatch = function () end +} + +skynet.start(function() + skynet.dispatch("lua", function(session, address, ...) + local message = { ... } + local cmd = compose_message(message) + socket.write(cmd) + push_request_queue { session = session , address = address, cmd = cmd } + end) + init() +end) -skynet.start(init) diff --git a/service/redis-mgr.lua b/service/redis-mgr.lua index 6ce98fce..a0385947 100644 --- a/service/redis-mgr.lua +++ b/service/redis-mgr.lua @@ -6,21 +6,22 @@ local redis_conf = skynet.getenv "redis" local name = config (redis_conf) local connection = {} -skynet.dispatch(function(msg, sz , session, from) - local dbname = skynet.unpack(msg,sz) - if connection[dbname] then - skynet.ret(skynet.pack(connection[dbname])) - return - end - if name[dbname] == nil then - log.Error("Invalid db name : "..dbname) - skynet.ret(skynet.pack(nil)) - return - end +skynet.start(function() + skynet.dispatch("lua", function (session, from, dbname) + if connection[dbname] then + skynet.ret(skynet.pack(connection[dbname])) + return + end + if name[dbname] == nil then + log.Error("Invalid db name : "..dbname) + skynet.ret(skynet.pack(nil)) + return + end - local redis_cli = skynet.launch("snlua", "redis-cli", name[dbname]) - connection[dbname] = redis_cli - skynet.ret(skynet.pack(redis_cli)) + local redis_cli = skynet.launch("snlua", "redis-cli", name[dbname]) + connection[dbname] = redis_cli + skynet.ret(skynet.pack(redis_cli)) + end) + skynet.register ".redis-manager" end) -skynet.register ".redis-manager" diff --git a/service/remote_root.lua b/service/remote_root.lua index 413157f2..dcbfc2b3 100644 --- a/service/remote_root.lua +++ b/service/remote_root.lua @@ -17,4 +17,5 @@ end skynet.remote_create(root,0) -skynet.remote_service() \ No newline at end of file +skynet.start(function() end) + diff --git a/service/simpledb.lua b/service/simpledb.lua index 9a8be1d1..ff2f912c 100644 --- a/service/simpledb.lua +++ b/service/simpledb.lua @@ -13,16 +13,18 @@ function command.SET(key, value) skynet.ret(last) end -skynet.dispatch(function(msg, sz , session, from) - local message = skynet.tostring(msg,sz) - print("simpledb",message, skynet.address(from), session) - local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") - local f = command[cmd] - if f then - f(key,value) - else - skynet.ret("Invalid command : "..message) - end +skynet.start(function() + skynet.dispatch("text", function(session, address, message) + print("simpledb",message, skynet.address(address), session) + local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") + local f = command[cmd] + if f then + f(key,value) + else + skynet.ret("Invalid command : "..message) + end + end) + skynet.register "SIMPLEDB" end) -skynet.register "SIMPLEDB" + diff --git a/service/testconn.lua b/service/testconn.lua deleted file mode 100644 index 306b140c..00000000 --- a/service/testconn.lua +++ /dev/null @@ -1,25 +0,0 @@ -local skynet = require "skynet" - --- register a dummy callback function -skynet.dispatch() - -local connection = skynet.launch("connection","16") -print("connection",connection) - -skynet.start(function() - local fd = skynet.call(".connection","CONNECT 127.0.0.1:8000") - if fd == nil then - print("Connect failed") - skynet.exit() - return - end - print("connect to ",fd) - skynet.send(".connection","WRITE "..fd.." Welcome\n") - local four = skynet.call(".connection","READ "..fd .." 4") - print("READ 4", four) - local line = skynet.call(".connection","READLINE "..fd .." \n") - skynet.send(".connection","WRITE "..fd.." "..line.."\n") - skynet.send(".connection","CLOSE "..fd) - - skynet.exit() -end) diff --git a/service/testdb.lua b/service/testdb.lua index d561984b..03e711b8 100644 --- a/service/testdb.lua +++ b/service/testdb.lua @@ -1,12 +1,9 @@ local skynet = require "skynet" --- register a dummy callback function -skynet.dispatch() - skynet.start(function() - local result = skynet.call("SIMPLEDB","SET foobar hello") + local result = skynet.call("SIMPLEDB","text","SET","foobar","hello") print(result) - result = skynet.call("SIMPLEDB","GET foobar") + result = skynet.call("SIMPLEDB","text","GET","foobar") print(result) skynet.exit() end) diff --git a/service/testgroup.lua b/service/testgroup.lua index 9d4cb557..37d2dc5d 100644 --- a/service/testgroup.lua +++ b/service/testgroup.lua @@ -1,8 +1,6 @@ local skynet = require "skynet" local group = require "mcgroup" -skynet.dispatch() - skynet.start(function() local gid = group.create() local gaddr = group.address(gid) @@ -12,13 +10,6 @@ skynet.start(function() group.enter(gid , address) end skynet.sleep(1000) - skynet.send(gaddr,"Hello World") - --- local group = skynet.query_group(1) --- for i=1,10 do --- local address = skynet.newservice("testgroup_c", tostring(i)) --- skynet.enter_group(1 , address) --- end --- skynet.send(group,"Hello World") --- skynet.exit() + skynet.send(gaddr,"text","Hello World") + skynet.exit() end) diff --git a/service/testgroup_c.lua b/service/testgroup_c.lua index 3407c1a2..8309c68c 100644 --- a/service/testgroup_c.lua +++ b/service/testgroup_c.lua @@ -2,12 +2,10 @@ local skynet = require "skynet" local group = require "mcgroup" local id, g = ... -skynet.dispatch(function (msg,sz) - print("===>",id, skynet.tostring(msg,sz)) -end -) - skynet.start(function() + skynet.dispatch("text",function(session,address,text) + print("===>",id, text) + end) if g then group.enter(tonumber(g)) end diff --git a/service/testlog.lua b/service/testlog.lua index b2d042c4..f14e8066 100644 --- a/service/testlog.lua +++ b/service/testlog.lua @@ -1,7 +1,9 @@ local skynet = require "skynet" local log = require "log" -skynet.dispatch() +skynet.start(function() + log.Info("hello world") + skynet.exit() +end) -log.Info("hello world") diff --git a/service/testredis.lua b/service/testredis.lua index 264ee7e3..d02e7826 100644 --- a/service/testredis.lua +++ b/service/testredis.lua @@ -1,8 +1,6 @@ local skynet = require "skynet" local redis = require "redis" -skynet.dispatch() - skynet.start(function() local db = redis.connect "main" print(db:exists "A") diff --git a/service/testrobj.lua b/service/testrobj.lua index 32ce341a..1f69f530 100644 --- a/service/testrobj.lua +++ b/service/testrobj.lua @@ -1,7 +1,5 @@ local skynet = require "skynet" -skynet.remote_service() - local obj = { id = 0 } function obj.hello() @@ -21,4 +19,5 @@ skynet.start(function() local qobj = root.query "test" print(qobj.hello()) print(qobj:inc()) + skynet.exit() end) diff --git a/service/testtimer.lua b/service/testtimer.lua index 6b6a3a55..c6d64f36 100644 --- a/service/testtimer.lua +++ b/service/testtimer.lua @@ -1,8 +1,5 @@ local skynet = require "skynet" --- register a dummy callback function -skynet.dispatch() - local function timeout(t) print(t) end diff --git a/service/watchdog.lua b/service/watchdog.lua index 707d3d56..cdc39ae8 100644 --- a/service/watchdog.lua +++ b/service/watchdog.lua @@ -13,7 +13,7 @@ function command:open(parm) local agent = skynet.launch("snlua","agent",skynet.address(client)) if agent then agent_all[self] = { agent , client } - skynet.send(gate, "forward ".. self .. " " .. skynet.address(agent) .. " " .. skynet.address(client)) + skynet.send(gate, "text", "forward" , self, skynet.address(agent) , skynet.address(client)) end end @@ -28,27 +28,31 @@ end function command:data(data, session) local agent = agent_all[self] if agent then - -- 0x7fffffff means it's a client - skynet.redirect(agent[1], agent[2], 0x7fffffff, data) + -- PTYPE_CLIENT = 3 , read skynet.h + skynet.redirect(agent[1], agent[2], "client", 0, data) else skynet.error(string.format("agent data drop %d size=%d",self,#data)) end end -skynet.dispatch(function(msg, sz, session, address) - local message = skynet.tostring(msg,sz) - local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)") - id = tonumber(id) - local f = command[cmd] - if f then - f(id,parm,session) - else - skynet.error(string.format("[watchdog] Unknown command : %s",message)) - end -end) +skynet.register_protocol { + name = "client", + id = 3, +} skynet.start(function() - gate = skynet.launch("gate" , skynet.address(skynet.self()), port, max_agent, buffer) - skynet.send(gate,"start") + skynet.dispatch("text", function(session, from, message) + local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)") + id = tonumber(id) + local f = command[cmd] + if f then + f(id,parm,session) + else + error(string.format("[watchdog] Unknown command : %s",message)) + end + end) + -- 0 for default client tag + gate = skynet.launch("gate" , skynet.address(skynet.self()), port, 0, max_agent, buffer) + skynet.send(gate,"text", "start") skynet.register(".watchdog") end) diff --git a/skynet-src/skynet.h b/skynet-src/skynet.h index 00bda87c..1ca380f2 100644 --- a/skynet-src/skynet.h +++ b/skynet-src/skynet.h @@ -4,23 +4,27 @@ #include #include -#define DONTCOPY 1 -#define SESSION_CLIENT 0x7fffffff -#define SESSION_MULTICAST 0x7ffffffe -#define SESSION_MAX 0x7ffffff0 +#define PTYPE_TEXT 0 +#define PTYPE_RESPONSE 1 +#define PTYPE_MULTICAST 2 +#define PTYPE_CLIENT 3 +#define PTYPE_SYSTEM 4 +#define PTYPE_HARBOR 5 +#define PTYPE_TAG_DONTCOPY 0x10000 +#define PTYPE_TAG_ALLOCSESSION 0x20000 struct skynet_context; void skynet_error(struct skynet_context * context, const char *msg, ...); const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm); uint32_t skynet_queryname(struct skynet_context * context, const char * name); -int skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int session, void * msg, size_t sz, int flags); -int skynet_sendname(struct skynet_context * context, const char * destination , int session, void * msg, size_t sz, int flags); +int skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int type, int session, void * msg, size_t sz); +int skynet_sendname(struct skynet_context * context, const char * destination , int type, int session, void * msg, size_t sz); void skynet_forward(struct skynet_context *, uint32_t destination); int skynet_isremote(struct skynet_context *, uint32_t handle, int * harbor); -typedef int (*skynet_cb)(struct skynet_context * context, void *ud, int session, uint32_t source , const void * msg, size_t sz); +typedef int (*skynet_cb)(struct skynet_context * context, void *ud, int type, int session, uint32_t source , const void * msg, size_t sz); void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb); #endif diff --git a/skynet-src/skynet_error.c b/skynet-src/skynet_error.c index 3d525c52..db77eec7 100644 --- a/skynet-src/skynet_error.c +++ b/skynet-src/skynet_error.c @@ -2,7 +2,6 @@ #include "skynet_handle.h" #include "skynet_mq.h" #include "skynet_server.h" -#include "skynet_system.h" #include #include @@ -34,13 +33,13 @@ skynet_error(struct skynet_context * context, const char *msg, ...) { struct skynet_message smsg; if (context == NULL) { - smsg.source = SKYNET_SYSTEM_LOGGER; + smsg.source = 0; } else { smsg.source = skynet_context_handle(context); } smsg.session = 0; smsg.data = strdup(tmp); - smsg.sz = len; + smsg.sz = len | (PTYPE_TEXT << 24); skynet_context_push(logger, &smsg); } diff --git a/skynet-src/skynet_group.c b/skynet-src/skynet_group.c index 067e3c47..0e963f17 100644 --- a/skynet-src/skynet_group.c +++ b/skynet-src/skynet_group.c @@ -73,7 +73,7 @@ static void send_command(struct skynet_context *ctx, const char * cmd, uint32_t node) { char * tmp = malloc(16); int n = sprintf(tmp, "%s %x", cmd, node); - skynet_context_send(ctx, tmp, n+1 , 0, 0); + skynet_context_send(ctx, tmp, n+1 , 0, PTYPE_SYSTEM, 0); } void @@ -130,7 +130,7 @@ skynet_group_clear(int handle) { char * cmd = malloc(8); int n = sprintf(cmd, "C"); - skynet_context_send(ctx, cmd, n+1, 0 , 0); + skynet_context_send(ctx, cmd, n+1, 0 , PTYPE_SYSTEM, 0); *pnode = node->next; free(node); break; diff --git a/skynet-src/skynet_harbor.c b/skynet-src/skynet_harbor.c index 6c4cf80e..d14f928a 100644 --- a/skynet-src/skynet_harbor.c +++ b/skynet-src/skynet_harbor.c @@ -1,3 +1,4 @@ +#include "skynet.h" #include "skynet_harbor.h" #include "skynet_server.h" @@ -10,7 +11,10 @@ static int HARBOR = 0; void skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int session) { - skynet_context_send(REMOTE, rmsg, sizeof(*rmsg), source, session); + int type = rmsg->sz >> 24; + rmsg->sz &= 0xffffff; + assert(type != PTYPE_SYSTEM && type != PTYPE_HARBOR); + skynet_context_send(REMOTE, rmsg, sizeof(*rmsg) , source, type , session); } void @@ -25,7 +29,7 @@ skynet_harbor_register(struct remote_name *rname) { } } assert(number == 0); - skynet_context_send(REMOTE, rname, sizeof(*rname), 0, 0); + skynet_context_send(REMOTE, rname, sizeof(*rname), 0, PTYPE_SYSTEM , 0); } int diff --git a/skynet-src/skynet_logger.c b/skynet-src/skynet_logger.c index 56607eb7..068a3035 100644 --- a/skynet-src/skynet_logger.c +++ b/skynet-src/skynet_logger.c @@ -26,7 +26,7 @@ logger_release(struct logger * inst) { } static int -_logger(struct skynet_context * context, void *ud, int session, uint32_t source, const void * msg, size_t sz) { +_logger(struct skynet_context * context, void *ud, int type, int session, uint32_t source, const void * msg, size_t sz) { struct logger * inst = ud; fprintf(inst->handle, "[%x] ",source); fwrite(msg, sz , 1, inst->handle); diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index 966b7eff..569e8614 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -193,7 +193,8 @@ _drop_queue(struct message_queue *q) { int s = 0; while(!skynet_mq_pop(q, &msg)) { ++s; - if (msg.session == SESSION_MULTICAST) { + int type = msg.sz >> 24; + if (type == PTYPE_MULTICAST) { assert(msg.sz == 0); skynet_multicast_dispatch((struct skynet_multicast_message *)msg.data, NULL, NULL); } else { diff --git a/skynet-src/skynet_multicast.c b/skynet-src/skynet_multicast.c index 9f462b1d..374437be 100644 --- a/skynet-src/skynet_multicast.c +++ b/skynet-src/skynet_multicast.c @@ -214,7 +214,7 @@ skynet_multicast_castgroup(struct skynet_context * from, struct skynet_multicast int release = 0; for (i=0;inumber;i++) { struct pair * p = &group->data[i]; - skynet_context_send(p->ctx, msg, 0 , source, SESSION_MULTICAST); + skynet_context_send(p->ctx, msg, 0 , source, PTYPE_MULTICAST , 0); int ref = skynet_context_ref(p->ctx); if (ref == 1) { skynet_context_release(p->ctx); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 91f13fa3..9b40cbee 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -17,8 +17,6 @@ #define BLACKHOLE "blackhole" #define DEFAULT_MESSAGE_QUEUE 16 -#define CALLING_CHECK - #ifdef CALLING_CHECK #define CHECKCALLING_BEGIN(ctx) assert(__sync_lock_test_and_set(&ctx->calling,1) == 0); @@ -108,11 +106,6 @@ skynet_context_new(const char * name, const char *param) { int skynet_context_newsession(struct skynet_context *ctx) { int session = ++ctx->session_id; - if (session >= SESSION_MAX) { - ctx->session_id = 1; - return 1; - } - return session; } @@ -194,7 +187,9 @@ _forwarding(struct skynet_context *ctx, struct skynet_message *msg) { static void _mc(void *ud, uint32_t source, const void * msg, size_t sz) { struct skynet_context * ctx = ud; - ctx->cb(ctx, ctx->cb_ud, 0, source, msg, sz); + int type = sz >> 24; + sz &= 0xffffff; + ctx->cb(ctx, ctx->cb_ud, type, 0, source, msg, sz); if (ctx->forward) { uint32_t des = ctx->forward; ctx->forward = 0; @@ -203,7 +198,7 @@ _mc(void *ud, uint32_t source, const void * msg, size_t sz) { message.session = 0; message.data = malloc(sz); memcpy(message.data, msg, sz); - message.sz = sz; + message.sz = sz | (type << 24); _send_message(des, &message); } } @@ -212,13 +207,12 @@ static void _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { assert(ctx->init); CHECKCALLING_BEGIN(ctx) - if (msg->source == SKYNET_SYSTEM_TIMER) { - ctx->cb(ctx, ctx->cb_ud, msg->session, 0, msg->data, msg->sz); - } else if (msg->session == SESSION_MULTICAST) { - assert(msg->sz == 0); + int type = msg->sz >> 24; + size_t sz = msg->sz & 0xffffff; + if (type == PTYPE_MULTICAST) { skynet_multicast_dispatch((struct skynet_multicast_message *)msg->data, ctx, _mc); } else { - int reserve = ctx->cb(ctx, ctx->cb_ud, msg->session, msg->source, msg->data, msg->sz); + int reserve = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz); reserve |= _forwarding(ctx, msg); if (!reserve) { free(msg->data); @@ -413,7 +407,7 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return NULL; } else { _id_to_hex(context->result, inst->handle); - printf("launch %s : %x\n",param, inst->handle); + printf("[:%x] launch %s\n",inst->handle, param); return context->result; } } @@ -466,43 +460,57 @@ skynet_forward(struct skynet_context * context, uint32_t destination) { context->forward = destination; } -int -skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int session, void * data, size_t sz, int flags) { - int session_id = session; - if (source == 0) { - source = context->handle; - if (session < 0) { - session = skynet_context_newsession(context); - session_id = - session; - } - } +static void +_filter_args(struct skynet_context * context, int type, int *session, void ** data, size_t * sz) { + int dontcopy = type & PTYPE_TAG_DONTCOPY; + int allocsession = type & PTYPE_TAG_ALLOCSESSION; + type &= 0xff; + + if (allocsession) { + assert(*session == 0); + *session = skynet_context_newsession(context); + } char * msg; - if ((flags & DONTCOPY) || data == NULL) { - msg = data; + if (dontcopy || *data == NULL) { + msg = *data; } else { - msg = malloc(sz+1); - memcpy(msg, data, sz); - msg[sz] = '\0'; + msg = malloc(*sz+1); + memcpy(msg, *data, *sz); + msg[*sz] = '\0'; } + *data = msg; + + assert((*sz & 0xffffff) == *sz); + *sz |= type << 24; +} + +int +skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int type, int session, void * data, size_t sz) { + _filter_args(context, type, &session, (void **)&data, &sz); + + if (source == 0) { + source = context->handle; + } + if (destination == 0) { return session; } if (skynet_harbor_message_isremote(destination)) { struct remote_message * rmsg = malloc(sizeof(*rmsg)); rmsg->destination.handle = destination; - rmsg->message = msg; + rmsg->message = data; rmsg->sz = sz; - skynet_harbor_send(rmsg, source, session_id); + skynet_harbor_send(rmsg, source, session); } else { struct skynet_message smsg; smsg.source = source; - smsg.session = session_id; - smsg.data = msg; + smsg.session = session; + smsg.data = data; smsg.sz = sz; if (skynet_context_push(destination, &smsg)) { - free(msg); + free(data); skynet_error(NULL, "Drop message from %x to %x (size=%d)", source, destination, (int)sz); return -1; } @@ -511,67 +519,32 @@ skynet_send(struct skynet_context * context, uint32_t source, uint32_t destinati } int -skynet_sendname(struct skynet_context * context, const char * addr , int session, void * data, size_t sz, int flags) { - int session_id = session; - uint32_t source_handle = context->handle; - if (session < 0) { - session = skynet_context_newsession(context); - session_id = - session; - } - - char * msg; - if ((flags & DONTCOPY) || data == NULL) { - msg = data; - } else { - msg = malloc(sz+1); - memcpy(msg, data, sz); - msg[sz] = '\0'; - } - if (addr == NULL) { - return session; - } +skynet_sendname(struct skynet_context * context, const char * addr , int type, int session, void * data, size_t sz) { + uint32_t source = context->handle; uint32_t des = 0; if (addr[0] == ':') { des = strtoul(addr+1, NULL, 16); } else if (addr[0] == '.') { des = skynet_handle_findname(addr + 1); if (des == 0) { - free(msg); - skynet_error(context, "Drop message to %s, size = %d", addr, (int)sz); + free(data); + skynet_error(context, "Drop message to %s", addr); return session; } } else { + _filter_args(context, type, &session, (void **)&data, &sz); + struct remote_message * rmsg = malloc(sizeof(*rmsg)); _copy_name(rmsg->destination.name, addr); rmsg->destination.handle = 0; - rmsg->message = msg; + rmsg->message = data; rmsg->sz = sz; - skynet_harbor_send(rmsg, source_handle, session_id); + + skynet_harbor_send(rmsg, source, session); return session; } - assert(des > 0); - - if (skynet_harbor_message_isremote(des)) { - struct remote_message * rmsg = malloc(sizeof(*rmsg)); - rmsg->destination.handle = des; - rmsg->message = msg; - rmsg->sz = sz; - skynet_harbor_send(rmsg, source_handle, session_id); - } else { - struct skynet_message smsg; - smsg.source = source_handle; - smsg.session = session_id; - smsg.data = msg; - smsg.sz = sz; - - if (skynet_context_push(des, &smsg)) { - free(msg); - skynet_error(NULL, "Drop message from %x to %s (size=%d)", smsg.source, addr, (int)sz); - return -1; - } - } - return session; + return skynet_send(context, source, des, type, session, data, sz); } uint32_t @@ -592,12 +565,12 @@ skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb) { } void -skynet_context_send(struct skynet_context * ctx, void * msg, size_t sz, uint32_t source, int session) { +skynet_context_send(struct skynet_context * ctx, void * msg, size_t sz, uint32_t source, int type, int session) { struct skynet_message smsg; smsg.source = source; smsg.session = session; smsg.data = msg; - smsg.sz = sz; + smsg.sz = sz | type << 24; skynet_mq_push(ctx->queue, &smsg); } diff --git a/skynet-src/skynet_server.h b/skynet-src/skynet_server.h index 5297c834..07275700 100644 --- a/skynet-src/skynet_server.h +++ b/skynet-src/skynet_server.h @@ -14,7 +14,7 @@ int skynet_context_ref(struct skynet_context *); uint32_t skynet_context_handle(struct skynet_context *); void skynet_context_init(struct skynet_context *, uint32_t handle); int skynet_context_push(uint32_t handle, struct skynet_message *message); -void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int session); +void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session); int skynet_context_newsession(struct skynet_context *); int skynet_context_message_dispatch(void); // return 1 when block diff --git a/skynet-src/skynet_system.h b/skynet-src/skynet_system.h deleted file mode 100644 index 9b671977..00000000 --- a/skynet-src/skynet_system.h +++ /dev/null @@ -1,8 +0,0 @@ -#ifndef SKYNET_MESSAGE_SYSTEM_H -#define SKYNET_MESSAGE_SYSTEM_H - -#define SKYNET_SYSTEM_TIMER 1 -#define SKYNET_SYSTEM_LOGGER 2 -#define SKYNET_SYSTEM_NAME 3 - -#endif diff --git a/skynet-src/skynet_timer.c b/skynet-src/skynet_timer.c index 44a5743a..a7f76186 100644 --- a/skynet-src/skynet_timer.c +++ b/skynet-src/skynet_timer.c @@ -2,6 +2,7 @@ #include "skynet_mq.h" #include "skynet_server.h" #include "skynet_handle.h" +#include "skynet.h" #include #include @@ -111,10 +112,10 @@ timer_execute(struct timer *T) do { struct timer_event * event = (struct timer_event *)(current+1); struct skynet_message message; - message.source = SKYNET_SYSTEM_TIMER; + message.source = 0; message.session = event->session; message.data = NULL; - message.sz = 0; + message.sz = PTYPE_RESPONSE << 24; skynet_context_push(event->handle, &message); @@ -177,10 +178,10 @@ int skynet_timeout(uint32_t handle, int time, int session) { if (time == 0) { struct skynet_message message; - message.source = SKYNET_SYSTEM_TIMER; + message.source = 0; message.session = session; message.data = NULL; - message.sz = 0; + message.sz = PTYPE_RESPONSE << 24; if (skynet_context_push(handle, &message)) { return -1; diff --git a/skynet-src/skynet_timer.h b/skynet-src/skynet_timer.h index 3df0328b..4d4e2b0f 100644 --- a/skynet-src/skynet_timer.h +++ b/skynet-src/skynet_timer.h @@ -1,8 +1,6 @@ #ifndef SKYNET_TIMER_H #define SKYNET_TIMER_H -#include "skynet_system.h" - #include int skynet_timeout(uint32_t handle, int time, int session);