cluster support

HISTORY.md

@@ -1,6 +1,7 @@
 Dev version
 -----------
 * Bugfix: update lua-bson (signed 32bit int bug)
+* cluster support
 
 v0.2.1 (2014-5-19)
 -----------

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
 

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)

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)

examples/clustername.lua

@@ -0,0 +1 @@
+db = "127.0.0.1:2528"

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"

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"

lualib-src/lua-cluster.c

@@ -0,0 +1,205 @@
+#include <lua.h>
+#include <lauxlib.h>
+#include <string.h>
+
+#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;
+}

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 {

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

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)