improve cluster

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)
 -----------

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 )

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;i<sz;i++) {
+		if (buffer[i] == '\n') {
+			++n;
+			lua_pushlstring(L, (const char *)buffer, i);
+			lua_rawseti(L, 3, n);
+			buffer += i + 1;
+			sz -= i + 1;
+			return unpack_line(L, buffer, sz, n);
+		}
+	}
+_block:
+	lua_pushboolean(L, n==0 ? 0:1);
+	if (sz == 0) {
+		lua_pushnil(L);
+	} else {
+		lua_pushlstring(L, (const char *)buffer, sz);
+	}
+	return 2;
+}
+
+static int
+lreadline(lua_State *L) {
+	struct socket_buffer sb;
+	char tmp[CACHE_SIZE];
+	int ret = recv_socket(L, tmp, &sb);
+	if (ret < 0) {
+		return unpack_line(L, sb.buffer, sb.sz, 0);
+	} else {
+		return ret;
+	}
+}
+
+
 static int
 lusleep(lua_State *L) {
 	int n = luaL_checknumber(L, 1);
@@ -219,7 +278,7 @@ readline_stdin(void * arg) {
 }
 
 static int
-lreadline(lua_State *L) {
+lreadstdin(lua_State *L) {
 	struct queue *q = lua_touserdata(L, lua_upvalueindex(1));
 	LOCK(q);
 	if (q->head == 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);

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;
 }

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)

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

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)