complete cluster.send

examples/cluster2.lua

@@ -11,9 +11,11 @@ skynet.start(function()
 	print(skynet.call(proxy, "lua", "SET", largekey, largevalue))
 	local v = skynet.call(proxy, "lua", "GET", largekey)
 	assert(largevalue == v)
+	skynet.send(proxy, "lua", "PING", "proxy")
 
 	print(cluster.call("db", sdb, "GET", "a"))
 	print(cluster.call("db2", sdb, "GET", "b"))
+	cluster.send("db2", sdb, "PING", "db2:longstring" .. largevalue)
 
 	-- test snax service
 	local pingserver = cluster.snax("db", "pingserver")

examples/simpledb.lua

@@ -16,7 +16,17 @@ end
 
 skynet.start(function()
 	skynet.dispatch("lua", function(session, address, cmd, ...)
-		local f = command[string.upper(cmd)]
+		cmd = cmd:upper()
+		if cmd == "PING" then
+			assert(session == 0)
+			local str = (...)
+			if #str > 20 then
+				str = str:sub(1,20) .. "...(" .. #str .. ")"
+			end
+			skynet.error(string.format("%s ping %s", skynet.address(address), str))
+			return
+		end
+		local f = command[cmd]
 		if f then
 			skynet.ret(skynet.pack(f(...)))
 		else

lualib-src/lua-cluster.c

@@ -46,7 +46,7 @@ fill_header(lua_State *L, uint8_t *buf, int sz) {
 		PADDING msg(sz)
 	size > 0x8000 and address is id
 		DWORD 13
-		BYTE 1	; multireq	
+		BYTE 1	; multireq	, 0x41: multi push
 		DWORD addr
 		DWORD session
 		DWORD sz
@@ -60,7 +60,7 @@ fill_header(lua_State *L, uint8_t *buf, int sz) {
 		PADDING msg(sz)
 	size > 0x8000 and address is string
 		DWORD 10 + namelen
-		BYTE 0x81
+		BYTE 0x81	; 0xc1 : multi push
 		BYTE namelen
 		STRING name
 		DWORD session
@@ -73,14 +73,14 @@ fill_header(lua_State *L, uint8_t *buf, int sz) {
 		PADDING msgpart(sz)
  */
 static int
-packreq_number(lua_State *L, int session, void * msg, uint32_t sz) {
+packreq_number(lua_State *L, int session, void * msg, uint32_t sz, int is_push) {
 	uint32_t addr = (uint32_t)lua_tointeger(L,1);
 	uint8_t buf[TEMP_LENGTH];
 	if (sz < MULTI_PART) {
 		fill_header(L, buf, sz+9);
 		buf[2] = 0;
 		fill_uint32(buf+3, addr);
-		fill_uint32(buf+7, (uint32_t)session);
+		fill_uint32(buf+7, is_push ? 0 : (uint32_t)session);
 		memcpy(buf+11,msg,sz);
 
 		lua_pushlstring(L, (const char *)buf, sz+11);
@@ -88,7 +88,7 @@ packreq_number(lua_State *L, int session, void * msg, uint32_t sz) {
 	} else {
 		int part = (sz - 1) / MULTI_PART + 1;
 		fill_header(L, buf, 13);
-		buf[2] = 1;
+		buf[2] = is_push ? 0x41 : 1;	// multi push or request
 		fill_uint32(buf+3, addr);
 		fill_uint32(buf+7, (uint32_t)session);
 		fill_uint32(buf+11, sz);
@@ -98,7 +98,7 @@ packreq_number(lua_State *L, int session, void * msg, uint32_t sz) {
 }
 
 static int
-packreq_string(lua_State *L, int session, void * msg, uint32_t sz) {
+packreq_string(lua_State *L, int session, void * msg, uint32_t sz, int is_push) {
 	size_t namelen = 0;
 	const char *name = lua_tolstring(L, 1, &namelen);
 	if (name == NULL || namelen < 1 || namelen > 255) {
@@ -112,7 +112,7 @@ packreq_string(lua_State *L, int session, void * msg, uint32_t sz) {
 		buf[2] = 0x80;
 		buf[3] = (uint8_t)namelen;
 		memcpy(buf+4, name, namelen);
-		fill_uint32(buf+4+namelen, (uint32_t)session);
+		fill_uint32(buf+4+namelen, is_push ? 0 : (uint32_t)session);
 		memcpy(buf+8+namelen,msg,sz);
 
 		lua_pushlstring(L, (const char *)buf, sz+8+namelen);
@@ -120,7 +120,7 @@ packreq_string(lua_State *L, int session, void * msg, uint32_t sz) {
 	} else {
 		int part = (sz - 1) / MULTI_PART + 1;
 		fill_header(L, buf, 10+namelen);
-		buf[2] = 0x81;
+		buf[2] = is_push ? 0xc1 : 0x81;	// multi push or request
 		buf[3] = (uint8_t)namelen;
 		memcpy(buf+4, name, namelen);
 		fill_uint32(buf+4+namelen, (uint32_t)session);
@@ -157,7 +157,7 @@ packreq_multi(lua_State *L, int session, void * msg, uint32_t sz) {
 }
 
 static int
-lpackrequest(lua_State *L) {
+packrequest(lua_State *L, int is_push) {
 	void *msg = lua_touserdata(L,3);
 	if (msg == NULL) {
 		return luaL_error(L, "Invalid request message");
@@ -171,9 +171,9 @@ lpackrequest(lua_State *L) {
 	int addr_type = lua_type(L,1);
 	int multipak;
 	if (addr_type == LUA_TNUMBER) {
-		multipak = packreq_number(L, session, msg, sz);
+		multipak = packreq_number(L, session, msg, sz, is_push);
 	} else {
-		multipak = packreq_string(L, session, msg, sz);
+		multipak = packreq_string(L, session, msg, sz, is_push);
 	}
 	int current_session = session;
 	if (++session < 0) {
@@ -191,6 +191,16 @@ lpackrequest(lua_State *L) {
 	}
 }
 
+static int
+lpackrequest(lua_State *L) {
+	return packrequest(L, 0);
+}
+
+static int
+lpackpush(lua_State *L) {
+	return packrequest(L, 1);
+}
+
 /*
 	string packed message
 	return 	
@@ -215,12 +225,17 @@ unpackreq_number(lua_State *L, const uint8_t * buf, int sz) {
 	lua_pushinteger(L, address);
 	lua_pushinteger(L, session);
 	lua_pushlstring(L, (const char *)buf+9, sz-9);
+	if (session == 0) {
+		lua_pushnil(L);
+		lua_pushboolean(L,1);	// is_push, no reponse
+		return 5;
+	}
 
 	return 3;
 }
 
 static int
-unpackmreq_number(lua_State *L, const uint8_t * buf, int sz) {
+unpackmreq_number(lua_State *L, const uint8_t * buf, int sz, int is_push) {
 	if (sz != 13) {
 		return luaL_error(L, "Invalid cluster message size %d (multi req must be 13)", sz);
 	}
@@ -231,8 +246,9 @@ unpackmreq_number(lua_State *L, const uint8_t * buf, int sz) {
 	lua_pushinteger(L, session);
 	lua_pushinteger(L, size);
 	lua_pushboolean(L, 1);	// padding multi part
+	lua_pushboolean(L, is_push);
 
-	return 4;
+	return 5;
 }
 
 static int
@@ -263,12 +279,17 @@ unpackreq_string(lua_State *L, const uint8_t * buf, int sz) {
 	uint32_t session = unpack_uint32(buf + namesz + 2);
 	lua_pushinteger(L, (uint32_t)session);
 	lua_pushlstring(L, (const char *)buf+2+namesz+4, sz - namesz - 6);
+	if (session == 0) {
+		lua_pushnil(L);
+		lua_pushboolean(L,1);	// is_push, no reponse
+		return 5;
+	}
 
 	return 3;
 }
 
 static int
-unpackmreq_string(lua_State *L, const uint8_t * buf, int sz) {
+unpackmreq_string(lua_State *L, const uint8_t * buf, int sz, int is_push) {
 	if (sz < 2) {
 		return luaL_error(L, "Invalid cluster message (size=%d)", sz);
 	}
@@ -282,8 +303,9 @@ unpackmreq_string(lua_State *L, const uint8_t * buf, int sz) {
 	lua_pushinteger(L, session);
 	lua_pushinteger(L, size);
 	lua_pushboolean(L, 1);	// padding multipart
+	lua_pushboolean(L, is_push);
 
-	return 4;
+	return 5;
 }
 
 static int
@@ -295,14 +317,18 @@ lunpackrequest(lua_State *L) {
 	case 0:
 		return unpackreq_number(L, (const uint8_t *)msg, sz);
 	case 1:
-		return unpackmreq_number(L, (const uint8_t *)msg, sz);
+		return unpackmreq_number(L, (const uint8_t *)msg, sz, 0);	// request
+	case '\x41':
+		return unpackmreq_number(L, (const uint8_t *)msg, sz, 1);	// push
 	case 2:
 	case 3:
 		return unpackmreq_part(L, (const uint8_t *)msg, sz);
 	case '\x80':
 		return unpackreq_string(L, (const uint8_t *)msg, sz);
 	case '\x81':
-		return unpackmreq_string(L, (const uint8_t *)msg, sz);
+		return unpackmreq_string(L, (const uint8_t *)msg, sz, 0 );	// request
+	case '\xc1':
+		return unpackmreq_string(L, (const uint8_t *)msg, sz, 1 );	// push
 	default:
 		return luaL_error(L, "Invalid req package type %d", msg[0]);
 	}
@@ -481,6 +507,7 @@ LUAMOD_API int
 luaopen_cluster_core(lua_State *L) {
 	luaL_Reg l[] = {
 		{ "packrequest", lpackrequest },
+		{ "packpush", lpackpush },
 		{ "unpackrequest", lunpackrequest },
 		{ "packresponse", lpackresponse },
 		{ "unpackresponse", lunpackresponse },

lualib/cluster.lua

@@ -9,8 +9,8 @@ function cluster.call(node, address, ...)
 end
 
 function cluster.send(node, address, ...)
-	-- skynet.pack(...) will free by cluster.core.packrequest
+	-- push is the same with req, but no response
-	skynet.send(clusterd, "lua", "req", node, address, skynet.pack(...))
+	skynet.send(clusterd, "lua", "push", node, address, skynet.pack(...))
 end
 
 function cluster.open(port)

lualib/skynet.lua

@@ -362,6 +362,11 @@ function skynet.send(addr, typename, ...)
 	return c.send(addr, p.id, 0 , p.pack(...))
 end
 
+function skynet.rawsend(addr, typename, msg, sz)
+	local p = proto[typename]
+	return c.send(addr, p.id, 0 , msg, sz)
+end
+
 skynet.genid = assert(c.genid)
 
 skynet.redirect = function(dest,source,typename,...)

service/clusterd.lua

@@ -91,6 +91,21 @@ function command.req(...)
 	end
 end
 
+function command.push(source, node, addr, msg, sz)
+	local session = node_session[node] or 1
+	local request, new_session, padding = cluster.packpush(addr, session, msg, sz)
+	if padding then	-- is multi push
+		node_session[node] = new_session
+	end
+
+	-- node_channel[node] may yield or throw error
+	local c = node_channel[node]
+
+	c:request(request, nil, padding)
+
+	-- notice: push may fail where the channel is disconnected or broken.
+end
+
 local proxy = {}
 
 function command.proxy(source, node, name)
@@ -121,9 +136,9 @@ local large_request = {}
 function command.socket(source, subcmd, fd, msg)
 	if subcmd == "data" then
 		local sz
-		local addr, session, msg, padding = cluster.unpackrequest(msg)
+		local addr, session, msg, padding, is_push = cluster.unpackrequest(msg)
 		if padding then
-			local req = large_request[session] or { addr = addr }
+			local req = large_request[session] or { addr = addr , is_push = is_push }
 			large_request[session] = req
 			table.insert(req, msg)
 			return
@@ -134,6 +149,7 @@ function command.socket(source, subcmd, fd, msg)
 				table.insert(req, msg)
 				msg,sz = cluster.concat(req)
 				addr = req.addr
+				is_push = req.is_push
 			end
 			if not msg then
 				local response = cluster.packresponse(session, false, "Invalid large req")
@@ -152,6 +168,9 @@ function command.socket(source, subcmd, fd, msg)
 				ok = false
 				msg = "name not found"
 			end
+		elseif is_push then
+			skynet.rawsend(addr, "lua", msg, sz)
+			return	-- no response
 		else
 			ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz)
 		end

service/clusterproxy.lua

@@ -23,6 +23,10 @@ skynet.forward_type( forward_map ,function()
 		address = n
 	end
 	skynet.dispatch("system", function (session, source, msg, sz)
-		skynet.ret(skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz)))
+		if session == 0 then
+			skynet.send(clusterd, "lua", "push", node, address, msg, sz)
+		else
+			skynet.ret(skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz)))
+		end
 	end)
 end)