diff --git a/HISTORY.md b/HISTORY.md index 07a4c501..aa8ebfa0 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -1,6 +1,7 @@ Dev version ----------- * Bugfix: update lua-bson (signed 32bit int bug) +* cluster support v0.2.1 (2014-5-19) ----------- diff --git a/Makefile b/Makefile index d7bb53cf..e6943b3f 100644 --- a/Makefile +++ b/Makefile @@ -41,7 +41,9 @@ jemalloc : $(MALLOC_STATICLIB) # skynet CSERVICE = snlua logger gate master harbor -LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile multicast +LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack \ + cjson clientsocket memory profile multicast \ + cluster 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 \ @@ -105,6 +107,9 @@ $(LUA_CLIB_PATH)/profile.so : lualib-src/lua-profile.c | $(LUA_CLIB_PATH) $(LUA_CLIB_PATH)/multicast.so : lualib-src/lua-multicast.c | $(LUA_CLIB_PATH) $(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ +$(LUA_CLIB_PATH)/cluster.so : lualib-src/lua-cluster.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/cluster1.lua b/examples/cluster1.lua new file mode 100644 index 00000000..8a67ebc6 --- /dev/null +++ b/examples/cluster1.lua @@ -0,0 +1,8 @@ +local skynet = require "skynet" +local cluster = require "cluster" + +skynet.start(function() + skynet.newservice("simpledb") + skynet.call("SIMPLEDB", "lua", "SET", "a", "foobar") + cluster.open(2528) +end) diff --git a/examples/cluster2.lua b/examples/cluster2.lua new file mode 100644 index 00000000..a61ff5c4 --- /dev/null +++ b/examples/cluster2.lua @@ -0,0 +1,6 @@ +local skynet = require "skynet" +local cluster = require "cluster" + +skynet.start(function() + print(cluster.call("db", "SIMPLEDB", "GET", "a")) +end) diff --git a/examples/clustername.lua b/examples/clustername.lua new file mode 100644 index 00000000..c5ed5e69 --- /dev/null +++ b/examples/clustername.lua @@ -0,0 +1 @@ +db = "127.0.0.1:2528" diff --git a/examples/config.c1 b/examples/config.c1 new file mode 100644 index 00000000..3e1307bd --- /dev/null +++ b/examples/config.c1 @@ -0,0 +1,12 @@ +thread = 8 +logger = nil +harbor = 1 +address = "127.0.0.1:2526" +master = "127.0.0.1:2013" +start = "cluster1" +bootstrap = "snlua bootstrap" -- The service for bootstrap +standalone = "0.0.0.0:2013" +luaservice = "./service/?.lua;./test/?.lua;./examples/?.lua" +lualoader = "lualib/loader.lua" +cpath = "./cservice/?.so" +cluster = "./examples/clustername.lua" \ No newline at end of file diff --git a/examples/config.c2 b/examples/config.c2 new file mode 100644 index 00000000..080ca137 --- /dev/null +++ b/examples/config.c2 @@ -0,0 +1,12 @@ +thread = 8 +logger = nil +harbor = 1 +address = "127.0.0.1:2527" +master = "127.0.0.1:2014" +start = "cluster2" +bootstrap = "snlua bootstrap" -- The service for bootstrap +standalone = "0.0.0.0:2014" +luaservice = "./service/?.lua;./test/?.lua;./examples/?.lua" +lualoader = "lualib/loader.lua" +cpath = "./cservice/?.so" +cluster = "./examples/clustername.lua" \ No newline at end of file diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c new file mode 100644 index 00000000..b3b8a9d1 --- /dev/null +++ b/lualib-src/lua-cluster.c @@ -0,0 +1,205 @@ +#include +#include +#include + +#include "skynet.h" + +/* + uint32_t/string addr + uint32_t/session session + lightuserdata msg + uint32_t sz + + return + string request + uint32_t next_session + */ + +#define TEMP_LENGTH 0x10002 + +static void +fill_uint32(uint8_t * buf, uint32_t n) { + buf[0] = n & 0xff; + buf[1] = (n >> 8) & 0xff; + buf[2] = (n >> 16) & 0xff; + buf[3] = (n >> 24) & 0xff; +} + +static void +fill_header(lua_State *L, uint8_t *buf, int sz, void *msg) { + if (sz >= 0x10000) { + skynet_free(msg); + luaL_error(L, "request message is too long %d", sz); + } + buf[0] = (sz >> 8) & 0xff; + buf[1] = sz & 0xff; +} + +static void +packreq_number(lua_State *L, int session, void * msg, size_t sz) { + uint32_t addr = lua_tounsigned(L,1); + uint8_t buf[TEMP_LENGTH]; + fill_header(L, buf, sz+9, msg); + buf[2] = 0; + fill_uint32(buf+3, addr); + fill_uint32(buf+7, (uint32_t)session); + memcpy(buf+11,msg,sz); + + lua_pushlstring(L, (const char *)buf, sz+11); +} + +static void +packreq_string(lua_State *L, int session, void * msg, size_t sz) { + size_t namelen = 0; + const char *name = lua_tolstring(L, 1, &namelen); + if (name == NULL || namelen < 1 || namelen > 255) { + skynet_free(msg); + luaL_error(L, "name is too long %s", name); + } + + uint8_t buf[TEMP_LENGTH]; + fill_header(L, buf, sz+5+namelen, msg); + buf[2] = (uint8_t)namelen; + memcpy(buf+3, name, namelen); + fill_uint32(buf+3+namelen, (uint32_t)session); + memcpy(buf+7+namelen,msg,sz); + + lua_pushlstring(L, (const char *)buf, sz+7+namelen); +} + +static int +lpackrequest(lua_State *L) { + void *msg = lua_touserdata(L,3); + if (msg == NULL) { + return luaL_error(L, "Invalid request message"); + } + size_t sz = luaL_checkunsigned(L,4); + int session = luaL_checkinteger(L,2); + if (session <= 0) { + return luaL_error(L, "Invalid request session %d", session); + } + int addr_type = lua_type(L,1); + if (addr_type == LUA_TNUMBER) { + packreq_number(L, session, msg, sz); + } else { + packreq_string(L, session, msg, sz); + } + if (++session < 0) { + session = 1; + } + skynet_free(msg); + lua_pushinteger(L, session); + return 2; +} + +/* + string packed message + return + uint32_t or string addr + int session + string msg + */ + +static inline uint32_t +unpack_uint32(const uint8_t * buf) { + return buf[0] | buf[1]<<8 | buf[2]<<16 | buf[3]<<24; +} + +static int +unpackreq_number(lua_State *L, const uint8_t * buf, size_t sz) { + if (sz < 9) { + return luaL_error(L, "Invalid cluster message"); + } + uint32_t address = unpack_uint32(buf+1); + uint32_t session = unpack_uint32(buf+5); + lua_pushunsigned(L, address); + lua_pushunsigned(L, session); + lua_pushlstring(L, (const char *)buf+9, sz-9); + + return 3; +} + +static int +unpackreq_string(lua_State *L, const uint8_t * buf, size_t sz) { + size_t namesz = buf[0]; + if (sz < namesz + 5) { + return luaL_error(L, "Invalid cluster message"); + } + lua_pushlstring(L, (const char *)buf+1, namesz); + uint32_t session = unpack_uint32(buf + namesz + 1); + lua_pushunsigned(L, session); + lua_pushlstring(L, (const char *)buf+1+namesz+4, sz - namesz - 5); + + return 3; +} + +static int +lunpackrequest(lua_State *L) { + size_t sz; + const char *msg = luaL_checklstring(L,1,&sz); + if (msg[0] == 0) { + return unpackreq_number(L, (const uint8_t *)msg, sz); + } else { + return unpackreq_string(L, (const uint8_t *)msg, sz); + } +} + +/* + int session + lightuserdata msg + int sz + return string response + */ +static int +lpackresponse(lua_State *L) { + uint32_t session = luaL_checkunsigned(L,1); + void * msg = lua_touserdata(L,2); + size_t sz = luaL_checkunsigned(L, 3); + + uint8_t buf[TEMP_LENGTH]; + fill_header(L, buf, sz+4, msg); + fill_uint32(buf+2, session); + memcpy(buf+6,msg,sz); + + skynet_free(msg); + + lua_pushlstring(L, (const char *)buf, sz+6); + + return 1; +} + +/* + string packed response + return integer session + boolean ok + string msg + */ +static int +lunpackresponse(lua_State *L) { + size_t sz; + const char * buf = luaL_checklstring(L, 1, &sz); + if (sz < 4) { + 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); + + return 3; +} + +int +luaopen_cluster_c(lua_State *L) { + luaL_Reg l[] = { + { "packrequest", lpackrequest }, + { "unpackrequest", lunpackrequest }, + { "packresponse", lpackresponse }, + { "unpackresponse", lunpackresponse }, + { NULL, NULL }, + }; + luaL_checkversion(L); + luaL_newlib(L,l); + + return 1; +} diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index 3f22956c..2428914e 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -191,7 +191,7 @@ pop_lstring(lua_State *L, struct socket_buffer *sb, int sz, int skip) { /* userdata send_buffer table pool - integer sz + integer sz or string (big endian) */ static int lpopbuffer(lua_State *L) { @@ -200,7 +200,23 @@ lpopbuffer(lua_State *L) { return luaL_error(L, "Need buffer object at param 1"); } luaL_checktype(L,2,LUA_TTABLE); - int sz = luaL_checkinteger(L,3); + int type = lua_type(L,3); + int sz; + if (type == LUA_TNUMBER) { + sz = lua_tointeger(L,3); + } else { + size_t len; + const uint8_t * s = (const uint8_t *)luaL_checklstring(L, 3, &len); + if (len > 4 || len < 1) { + return luaL_error(L, "Invalid read %s", s); + } + int i; + sz = 0; + for (i=0;i<(int)len;i++) { + sz <<= 8; + sz |= s[i]; + } + } if (sb->size < sz || sz == 0) { lua_pushnil(L); } else { diff --git a/lualib/cluster.lua b/lualib/cluster.lua new file mode 100644 index 00000000..e4f6cfa2 --- /dev/null +++ b/lualib/cluster.lua @@ -0,0 +1,19 @@ +local skynet = require "skynet" + +local clusterd +local cluster = {} + +function cluster.call(node, address, ...) + -- skynet.pack(...) will free by cluster.c.packrequest + return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...)) +end + +function cluster.open(port) + skynet.call(clusterd, "lua", "listen", "0.0.0.0", port) +end + +skynet.init(function() + clusterd = skynet.uniqueservice("clusterd") +end) + +return cluster diff --git a/service/clusterd.lua b/service/clusterd.lua new file mode 100644 index 00000000..ef6bff68 --- /dev/null +++ b/service/clusterd.lua @@ -0,0 +1,69 @@ +local skynet = require "skynet" +local sc = require "socketchannel" +local socket = require "socket" +local cluster = require "cluster.c" + +local config_name = skynet.getenv "cluster" +local node_address = {} +assert(loadfile(config_name, "t", node_address))() +local node_session = {} +local command = {} + +local function read_response(sock) + local sz = sock:read(2) + local msg = sock:read(sz) + return cluster.unpackresponse(msg) -- session, ok, data +end + +local function open_channel(t, key) + local host, port = string.match(node_address[key], "([^:]+):(.*)$") + local c = sc.channel { + host = host, + port = tonumber(port), + response = read_response, + } + assert(c:connect(true)) + t[key] = c + node_session[key] = 1 + return c +end + +local node_channel = setmetatable({}, { __index = open_channel }) + +function command.listen(source, addr, port) + local gate = skynet.newservice("gate") + skynet.call(gate, "lua", "open", { address = addr, port = port }) + skynet.ret(skynet.pack(nil)) +end + +function command.req(source, node, addr, msg, sz) + local request + local c = node_channel[node] + 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) + skynet.ret(c:request(request, session)) +end + +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) + socket.write(fd, response) + elseif subcmd == "open" then + skynet.error(string.format("socket accept from %s", msg)) + skynet.call(source, "lua", "accept", fd) + else + skynet.error(string.format("socket %s %d : %s", subcmd, fd, msg)) + end +end + +skynet.start(function() + skynet.dispatch("lua", function(_, source, cmd, ...) + local f = assert(command[cmd]) + f(source, ...) + end) +end)