From 8e4a1751550da9dad68c588a741afe6b973e36af Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Thu, 13 Nov 2014 11:19:58 +0800
Subject: [PATCH] use fork to dispatch more socket message

---
 lualib-src/lua-netpack.c   | 41 ++++++++++++--------------------------
 lualib/snax/gateserver.lua | 24 ++++++++++++++++------
 2 files changed, 31 insertions(+), 34 deletions(-)

diff --git a/lualib-src/lua-netpack.c b/lualib-src/lua-netpack.c
index e424f90c..f42e21ca 100644
--- a/lualib-src/lua-netpack.c
+++ b/lualib-src/lua-netpack.c
@@ -240,19 +240,12 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
 		buffer += need;
 		size -= need;
 		if (size == 0) {
-			if (q == NULL || q->head == q->tail ) {
-				lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
-				lua_pushinteger(L, fd);
-				lua_pushlightuserdata(L, uc->pack.buffer);
-				lua_pushinteger(L, uc->pack.size);
-				skynet_free(uc);
-				return 5;
-			}
-			else{
-				push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
-				skynet_free(uc);
-				return 1;
-			}
+			lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
+			lua_pushinteger(L, fd);
+			lua_pushlightuserdata(L, uc->pack.buffer);
+			lua_pushinteger(L, uc->pack.size);
+			skynet_free(uc);
+			return 5;
 		}
 		// more data
 		push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
@@ -281,21 +274,13 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
 		}
 		if (size == pack_size) {
 			// just one package
-			if ( q == NULL || q->head == q->tail) {
-				lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
-				lua_pushinteger(L, fd);
-				void * result = skynet_malloc(pack_size);
-				memcpy(result, buffer, size);
-				lua_pushlightuserdata(L, result);
-				lua_pushinteger(L, size);
-				return 5;
-			}
-			else{
-				push_data(L, fd, buffer, pack_size, 1);
-				buffer += pack_size;
-				size -= pack_size;
-				return 1;
-			}
+			lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
+			lua_pushinteger(L, fd);
+			void * result = skynet_malloc(pack_size);
+			memcpy(result, buffer, size);
+			lua_pushlightuserdata(L, result);
+			lua_pushinteger(L, size);
+			return 5;
 		}
 		// more data
 		push_data(L, fd, buffer, pack_size, 1);
diff --git a/lualib/snax/gateserver.lua b/lualib/snax/gateserver.lua
index 6cce5266..e68efc12 100644
--- a/lualib/snax/gateserver.lua
+++ b/lualib/snax/gateserver.lua
@@ -53,20 +53,32 @@ function gateserver.start(handler)
 
 	local MSG = {}
 
-	function MSG.data(fd, msg, sz)
+	local function dispatch_msg(fd, msg, sz)
 		if connection[fd] then
 			handler.message(fd, msg, sz)
+		else
+			skynet.error(string.format("Drop message from fd (%d) : %s", fd, netpack.tostring(msg,sz)))
 		end
 	end
 
-	function MSG.more()
-		for fd, msg, sz in netpack.pop, queue do
-			if connection[fd] then
-				handler.message(fd, msg, sz)
+	MSG.data = dispatch_msg
+
+	local function dispatch_queue()
+		local fd, msg, sz = netpack.pop(queue)
+		if fd then
+			-- may dispatch even the handler.message blocked
+			-- If the handler.message never block, the queue should be empty, so only fork once and then exit.
+			skynet.fork(dispatch_queue)
+			dispatch_msg(fd, msg, sz)
+
+			for fd, msg, sz in netpack.pop, queue do
+				dispatch_msg(fd, msg, sz)
 			end
 		end
 	end
 
+	MSG.more = dispatch_queue
+
 	function MSG.open(fd, msg)
 		if client_number >= maxclient then
 			socketdriver.close(fd)
@@ -128,4 +140,4 @@ function gateserver.start(handler)
 	end)
 end
 
-return gateserver
\ No newline at end of file
+return gateserver