From a090899bce1a4ec00ccbce393c1738c7f514deb2 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sat, 12 Jul 2014 23:14:05 +0800 Subject: [PATCH] improve cluster --- HISTORY.md | 3 ++ examples/client.lua | 2 +- lualib-src/lua-clientsocket.c | 87 ++++++++++++++++++++++++++++++++--- lualib-src/lua-cluster.c | 31 +++++++++---- lualib/cluster.lua | 4 ++ lualib/skynet.lua | 3 ++ service/clusterd.lua | 27 +++++++++-- 7 files changed, 136 insertions(+), 21 deletions(-) diff --git a/HISTORY.md b/HISTORY.md index 1deddf51..52e1d4b0 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -5,6 +5,9 @@ Dev version * Add worker thread weight * Add skynet.queue * Bugfix: socketchannel +* cluster can throw error +* Add readline and writeline to clientsocket lib +* Add cluster.reload to reload config file v0.4.1 (2014-7-7) ----------- diff --git a/examples/client.lua b/examples/client.lua index 9197a1c3..278922e6 100644 --- a/examples/client.lua +++ b/examples/client.lua @@ -39,7 +39,7 @@ end while true do dispatch() - local cmd = socket.readline() + local cmd = socket.readstdin() if cmd then local args = {} string.gsub(cmd, '[^ ]+', function(v) table.insert(args, v) end ) diff --git a/lualib-src/lua-clientsocket.c b/lualib-src/lua-clientsocket.c index 1e499f2d..2676c267 100644 --- a/lualib-src/lua-clientsocket.c +++ b/lualib-src/lua-clientsocket.c @@ -125,14 +125,19 @@ _block: boolean (true: data, false: block, nil: close) string last */ + +struct socket_buffer { + void * buffer; + int sz; +}; + static int -lrecv(lua_State *L) { +recv_socket(lua_State *L, char *tmp, struct socket_buffer *result) { int fd = luaL_checkinteger(L,1); size_t sz = 0; const char * last = lua_tolstring(L,2,&sz); luaL_checktype(L, 3, LUA_TTABLE); - char tmp[CACHE_SIZE]; char * buffer; int r = recv(fd, tmp, CACHE_SIZE, 0); if (r == 0) { @@ -163,10 +168,64 @@ lrecv(lua_State *L) { lua_pushnil(L); lua_rawseti(L, 3, i); } - - return unpack(L, (uint8_t *)buffer, r+sz, 0); + result->buffer = buffer; + result->sz = r + sz; + return -1; } +static int +lrecv(lua_State *L) { + struct socket_buffer sb; + char tmp[CACHE_SIZE]; + int ret = recv_socket(L, tmp, &sb); + if (ret < 0) { + return unpack(L, sb.buffer, sb.sz, 0); + } else { + return ret; + } +} + +static int +unpack_line(lua_State *L, uint8_t *buffer, int sz, int n) { + if (sz == 0) + goto _block; + if (buffer[0] == '\n') { + return unpack_line(L, buffer+1, sz-1, n); + } + int i; + for (i=1;ihead == q->tail) { @@ -236,6 +295,18 @@ lreadline(lua_State *L) { return 1; } +static int +lwriteline(lua_State *L) { + size_t sz = 0; + int fd = luaL_checkinteger(L,1); + const char * msg = luaL_checklstring(L, 2, &sz); + block_send(L, fd, msg, sz); + char nl[1] = { '\n' }; + block_send(L, fd, nl, 1); + + return 0; +} + int luaopen_clientsocket(lua_State *L) { luaL_checkversion(L); @@ -245,14 +316,16 @@ luaopen_clientsocket(lua_State *L) { { "send", lsend }, { "close", lclose }, { "usleep", lusleep }, + { "readline", lreadline }, + { "writeline", lwriteline }, { NULL, NULL }, }; luaL_newlib(L, l); struct queue * q = lua_newuserdata(L, sizeof(*q)); memset(q, 0, sizeof(*q)); - lua_pushcclosure(L, lreadline, 1); - lua_setfield(L, -2, "readline"); + lua_pushcclosure(L, lreadstdin, 1); + lua_setfield(L, -2, "readstdin"); pthread_t pid ; pthread_create(&pid, NULL, readline_stdin, q); diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index 92134ced..29ebb08a 100644 --- a/lualib-src/lua-cluster.c +++ b/lualib-src/lua-cluster.c @@ -15,7 +15,7 @@ uint32_t next_session */ -#define TEMP_LENGTH 0x10002 +#define TEMP_LENGTH 0x10007 static void fill_uint32(uint8_t * buf, uint32_t n) { @@ -146,6 +146,7 @@ lunpackrequest(lua_State *L) { /* int session + boolean ok lightuserdata msg int sz return string response @@ -155,15 +156,27 @@ lpackresponse(lua_State *L) { uint32_t session = luaL_checkunsigned(L,1); // clusterd.lua:command.socket call lpackresponse, // and the msg/sz is return by skynet.rawcall , so don't free(msg) - void * msg = lua_touserdata(L,2); - size_t sz = luaL_checkunsigned(L, 3); + int ok = lua_toboolean(L,2); + void * msg; + size_t sz; + + if (lua_type(L,3) == LUA_TSTRING) { + msg = (void *)lua_tolstring(L, 3, &sz); + if (sz > 0x1000) { + sz = 0x1000; + } + } else { + msg = lua_touserdata(L,3); + sz = luaL_checkunsigned(L, 4); + } uint8_t buf[TEMP_LENGTH]; - fill_header(L, buf, sz+4, msg); + fill_header(L, buf, sz+5, msg); fill_uint32(buf+2, session); - memcpy(buf+6,msg,sz); + buf[6] = ok; + memcpy(buf+7,msg,sz); - lua_pushlstring(L, (const char *)buf, sz+6); + lua_pushlstring(L, (const char *)buf, sz+7); return 1; } @@ -178,13 +191,13 @@ static int lunpackresponse(lua_State *L) { size_t sz; const char * buf = luaL_checklstring(L, 1, &sz); - if (sz < 4) { + if (sz < 5) { return 0; } uint32_t session = unpack_uint32((const uint8_t *)buf); lua_pushunsigned(L, session); - lua_pushboolean(L, 1); - lua_pushlstring(L, buf+4, sz-4); + lua_pushboolean(L, buf[4]); + lua_pushlstring(L, buf+5, sz-5); return 3; } diff --git a/lualib/cluster.lua b/lualib/cluster.lua index bc4312b3..d53e6d9b 100644 --- a/lualib/cluster.lua +++ b/lualib/cluster.lua @@ -16,6 +16,10 @@ function cluster.open(port) end end +function cluster.reload() + skynet.call(clusterd, "lua", "reload") +end + skynet.init(function() clusterd = skynet.uniqueservice("clusterd") end) diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 7eba476c..d9c93616 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -325,6 +325,9 @@ end function skynet.rawcall(addr, typename, msg, sz) local p = proto[typename] + if watching_service[addr] == false then + error("Service is dead") + end local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address") return yield_call(addr, session) end diff --git a/service/clusterd.lua b/service/clusterd.lua index 405e28ab..039f7b77 100644 --- a/service/clusterd.lua +++ b/service/clusterd.lua @@ -5,7 +5,14 @@ local cluster = require "cluster.c" local config_name = skynet.getenv "cluster" local node_address = {} -assert(loadfile(config_name, "t", node_address))() + +local function loadconfig() + local f = assert(io.open(config_name)) + local source = f:read "*a" + f:close() + assert(load(source, "@"..config_name, "t", node_address))() +end + local node_session = {} local command = {} @@ -30,6 +37,11 @@ end local node_channel = setmetatable({}, { __index = open_channel }) +function command.reload() + loadconfig() + skynet.ret(skynet.pack(nil)) +end + function command.listen(source, addr, port) local gate = skynet.newservice("gate") if port == nil then @@ -45,6 +57,7 @@ function command.req(source, node, addr, msg, sz) local session = node_session[node] -- msg is a local pointer, cluster.packrequest will free it request, node_session[node] = cluster.packrequest(addr, session , msg, sz) + local ok, result = pcall(c.request, c, request, session) skynet.ret(c:request(request, session)) end @@ -53,8 +66,13 @@ local request_fd = {} function command.socket(source, subcmd, fd, msg) if subcmd == "data" then local addr, session, msg = cluster.unpackrequest(msg) - local msg, sz = skynet.rawcall(addr, "lua", msg) - local response = cluster.packresponse(session, msg, sz) + local ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg) + local response + if ok then + response = cluster.packresponse(session, true, msg, sz) + else + response = cluster.packresponse(session, false, msg) + end socket.write(fd, response) elseif subcmd == "open" then skynet.error(string.format("socket accept from %s", msg)) @@ -65,7 +83,8 @@ function command.socket(source, subcmd, fd, msg) end skynet.start(function() - skynet.dispatch("lua", function(_, source, cmd, ...) + loadconfig() + skynet.dispatch("lua", function(session , source, cmd, ...) local f = assert(command[cmd]) f(source, ...) end)