Merge pull request #631 from thomasheep/dev_socket_opt2

optimize SOCKET_WARNING msg

lualib-src/lua-socket.c

@@ -567,8 +567,9 @@ lsendlow(lua_State *L) {
 	int id = luaL_checkinteger(L, 1);
 	int sz = 0;
 	void *buffer = get_buffer(L, 2, &sz);
-	skynet_socket_send_lowpriority(ctx, id, buffer, sz);
-	return 0;
+	int err = skynet_socket_send_lowpriority(ctx, id, buffer, sz);
+	lua_pushboolean(L, !err);
+	return 1;
 }
 
 static int

lualib/snax/gateserver.lua

@@ -113,7 +113,7 @@ function gateserver.start(handler)
 	function MSG.error(fd, msg)
 		if fd == socket then
 			socketdriver.close(fd)
-			skynet.error(msg)
+			skynet.error("gateserver close listen socket, accpet error:",msg)
 		else
 			if handler.error then
 				handler.error(fd, msg)

lualib/socket.lua

@@ -138,19 +138,17 @@ end
 
 local function default_warning(id, size)
 	local s = socket_pool[id]
-		local last = s.warningsize or 0
-		if last + 64 < size then	-- if size increase 64K
-			s.warningsize = size
-			skynet.error(string.format("WARNING: %d K bytes need to send out (fd = %d)", size, id))
-		end
-		s.warningsize = size
+	if not s then
+		return
+	end
+	skynet.error(string.format("WARNING: %d K bytes need to send out (fd = %d)", size, id))
 end
 
 -- SKYNET_SOCKET_TYPE_WARNING
 socket_message[7] = function(id, size)
 	local s = socket_pool[id]
 	if s then
-		local warning = s.warning or default_warning
+		local warning = s.on_warning or default_warning
 		warning(id, size)
 	end
 end
@@ -442,7 +440,7 @@ socket.udp_address = assert(driver.udp_address)
 function socket.warning(id, callback)
 	local obj = socket_pool[id]
 	assert(obj)
-	obj.warning = callback
+	obj.on_warning = callback
 end
 
 return socket

lualib/socketchannel.lua

@@ -376,6 +376,12 @@ end
 local socket_write = socket.write
 local socket_lwrite = socket.lwrite
 
+local function sock_err(self)
+	close_channel_socket(self)
+	wakeup_all(self)
+	error(socket_error)
+end
+
 function channel:request(request, response, padding)
 	assert(block_connect(self, true))	-- connect once
 	local fd = self.__sock[1]
@@ -383,16 +389,18 @@ function channel:request(request, response, padding)
 	if padding then
 		-- padding may be a table, to support multi part request
 		-- multi part request use low priority socket write
-		-- socket_lwrite returns nothing
-		socket_lwrite(fd , request)
+		-- now socket_lwrite returns as socket_write
+		if not socket_lwrite(fd , request) then
+			sock_err(self)
+		end
 		for _,v in ipairs(padding) do
-			socket_lwrite(fd, v)
+			if not socket_lwrite(fd, v) then
+				sock_err(self)
+			end
 		end
 	else
 		if not socket_write(fd , request) then
-			close_channel_socket(self)
-			wakeup_all(self)
-			error(socket_error)
+			sock_err(self)
 		end
 	end
 

skynet-src/skynet_socket.c

@@ -98,6 +98,9 @@ skynet_socket_poll() {
 	case SOCKET_UDP:
 		forward_message(SKYNET_SOCKET_TYPE_UDP, false, &result);
 		break;
+	case SOCKET_WARNING:
+		forward_message(SKYNET_SOCKET_TYPE_WARNING, false, &result);
+		break;
 	default:
 		skynet_error(NULL, "Unknown socket message type %d.",type);
 		return -1;
@@ -108,31 +111,14 @@ skynet_socket_poll() {
 	return 1;
 }
 
-static int
-check_wsz(struct skynet_context *ctx, int id, void *buffer, int64_t wsz) {
-	if (wsz < 0) {
-		return -1;
-	} else if (wsz > 1024 * 1024) {
-		struct skynet_socket_message tmp;
-		tmp.type = SKYNET_SOCKET_TYPE_WARNING;
-		tmp.id = id;
-		tmp.ud = (int)(wsz / 1024);
-		tmp.buffer = NULL;
-		skynet_send(ctx, 0, skynet_context_handle(ctx), PTYPE_SOCKET, 0 , &tmp, sizeof(tmp));
-//		skynet_error(ctx, "%d Mb bytes on socket %d need to send out", (int)(wsz / (1024 * 1024)), id);
-	}
-	return 0;
+int
+skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) {
+	return socket_server_send(SOCKET_SERVER, id, buffer, sz);
 }
 
 int
-skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) {
-	int64_t wsz = socket_server_send(SOCKET_SERVER, id, buffer, sz);
-	return check_wsz(ctx, id, buffer, wsz);
-}
-
-void
 skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz) {
-	socket_server_send_lowpriority(SOCKET_SERVER, id, buffer, sz);
+	return socket_server_send_lowpriority(SOCKET_SERVER, id, buffer, sz);
 }
 
 int 
@@ -189,8 +175,7 @@ skynet_socket_udp_connect(struct skynet_context *ctx, int id, const char * addr,
 
 int 
 skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz) {
-	int64_t wsz = socket_server_udp_send(SOCKET_SERVER, id, (const struct socket_udp_address *)address, buffer, sz);
-	return check_wsz(ctx, id, (void *)buffer, wsz);
+	return socket_server_udp_send(SOCKET_SERVER, id, (const struct socket_udp_address *)address, buffer, sz);
 }
 
 const char *

skynet-src/skynet_socket.h

@@ -24,7 +24,7 @@ void skynet_socket_free();
 int skynet_socket_poll();
 
 int skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz);
-void skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz);
+int skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz);
 int skynet_socket_listen(struct skynet_context *ctx, const char *host, int port, int backlog);
 int skynet_socket_connect(struct skynet_context *ctx, const char *host, int port);
 int skynet_socket_bind(struct skynet_context *ctx, int fd);

skynet-src/socket_server.c

@@ -53,6 +53,8 @@
 #define AGAIN_WOULDBLOCK EAGAIN
 #endif
 
+#define WARNING_SIZE (1024*1024)
+
 struct write_buffer {
 	struct write_buffer * next;
 	void *buffer;
@@ -79,6 +81,7 @@ struct socket {
 	int id;
 	uint16_t protocol;
 	uint16_t type;
+	int64_t warn_size;
 	union {
 		int size;
 		uint8_t udp_address[UDP_ADDRESS_SIZE];
@@ -391,6 +394,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
 	s->p.size = MIN_READ_BUFFER;
 	s->opaque = opaque;
 	s->wb_size = 0;
+	s->warn_size = 0;
 	check_wb_list(&s->high);
 	check_wb_list(&s->low);
 	return s;
@@ -598,6 +602,11 @@ raise_uncomplete(struct socket * s) {
 	high->head = high->tail = tmp;
 }
 
+static inline int
+send_buffer_empty(struct socket *s) {
+	return (s->high.head == NULL && s->low.head == NULL);
+}
+
 /*
 	Each socket has two write buffer list, high priority and low priority.
 
@@ -622,15 +631,24 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 			// step 3
 			if (list_uncomplete(&s->low)) {
 				raise_uncomplete(s);
+				return -1;
 			}
-		} else {
-			// step 4
-			sp_write(ss->event_fd, s->fd, s, false);
+		} 
+		// step 4
+		assert(send_buffer_empty(s) && s->wb_size == 0);
+		sp_write(ss->event_fd, s->fd, s, false);			
 
-			if (s->type == SOCKET_TYPE_HALFCLOSE) {
+		if (s->type == SOCKET_TYPE_HALFCLOSE) {
 				force_close(ss, s, result);
 				return SOCKET_CLOSE;
-			}
+		}
+		if(s->warn_size > 0){
+				s->warn_size = 0;
+				result->opaque = s->opaque;
+				result->id = s->id;
+				result->ud = 0;
+				result->data = NULL;
+				return SOCKET_WARNING;
 		}
 	}
 
@@ -677,10 +695,6 @@ append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_
 	s->wb_size += buf->sz;
 }
 
-static inline int
-send_buffer_empty(struct socket *s) {
-	return (s->high.head == NULL && s->low.head == NULL);
-}
 
 /*
 	When send a package , we can assign the priority : PRIORITY_HIGH or PRIORITY_LOW
@@ -757,6 +771,14 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 			append_sendbuffer_udp(ss,s,priority,request,udp_address);
 		}
 	}
+	if (s->wb_size >= WARNING_SIZE && s->wb_size >= s->warn_size) {
+		s->warn_size = s->warn_size == 0 ? WARNING_SIZE *2 : s->warn_size*2;
+		result->opaque = s->opaque;
+		result->id = s->id;
+		result->ud = s->wb_size%1024 == 0 ? s->wb_size/1024 : s->wb_size/1024 + 1;
+		result->data = NULL;
+		return SOCKET_WARNING;
+	}
 	return -1;
 }
 
@@ -794,7 +816,8 @@ close_socket(struct socket_server *ss, struct request_close *request, struct soc
 	}
 	if (!send_buffer_empty(s)) { 
 		int type = send_buffer(ss,s,result);
-		if (type != -1)
+		// type : -1 or SOCKET_WARNING or SOCKET_CLOSE, SOCKET_WARNING means send_buffer_empty
+		if (type != -1 && type != SOCKET_WARNING)
 			return type;
 	}
 	if (request->shutdown || send_buffer_empty(s)) {
@@ -1223,6 +1246,9 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 			ss->event_index = 0;
 			if (ss->event_n <= 0) {
 				ss->event_n = 0;
+				if (errno == EINTR) {
+					continue;
+				}
 				return -1;
 			}
 		}
@@ -1334,8 +1360,8 @@ free_buffer(struct socket_server *ss, const void * buffer, int sz) {
 	so.free_func((void *)buffer);
 }
 
-// return -1 when error
-int64_t 
+// return -1 when error, 0 when success
+int 
 socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz) {
 	struct socket * s = &ss->slot[HASH_ID(id)];
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
@@ -1349,15 +1375,16 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 	request.u.send.buffer = (char *)buffer;
 
 	send_request(ss, &request, 'D', sizeof(request.u.send));
-	return s->wb_size;
+	return 0;
 }
 
-void 
+// return -1 when error, 0 when success
+int 
 socket_server_send_lowpriority(struct socket_server *ss, int id, const void * buffer, int sz) {
 	struct socket * s = &ss->slot[HASH_ID(id)];
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
 		free_buffer(ss, buffer, sz);
-		return;
+		return -1;
 	}
 
 	struct request_package request;
@@ -1366,6 +1393,7 @@ socket_server_send_lowpriority(struct socket_server *ss, int id, const void * bu
 	request.u.send.buffer = (char *)buffer;
 
 	send_request(ss, &request, 'P', sizeof(request.u.send));
+	return 0;
 }
 
 void
@@ -1546,7 +1574,7 @@ socket_server_udp(struct socket_server *ss, uintptr_t opaque, const char * addr,
 	return id;
 }
 
-int64_t 
+int 
 socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp_address *addr, const void *buffer, int sz) {
 	struct socket * s = &ss->slot[HASH_ID(id)];
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
@@ -1576,7 +1604,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 	memcpy(request.u.send_udp.address, udp_address, addrsz);	
 
 	send_request(ss, &request, 'A', sizeof(request.u.send_udp.send)+addrsz);
-	return s->wb_size;
+	return 0;
 }
 
 int

skynet-src/socket_server.h

@@ -10,6 +10,7 @@
 #define SOCKET_ERR 4
 #define SOCKET_EXIT 5
 #define SOCKET_UDP 6
+#define SOCKET_WARNING 7
 
 struct socket_server;
 
@@ -30,8 +31,8 @@ void socket_server_shutdown(struct socket_server *, uintptr_t opaque, int id);
 void socket_server_start(struct socket_server *, uintptr_t opaque, int id);
 
 // return -1 when error
-int64_t socket_server_send(struct socket_server *, int id, const void * buffer, int sz);
-void socket_server_send_lowpriority(struct socket_server *, int id, const void * buffer, int sz);
+int socket_server_send(struct socket_server *, int id, const void * buffer, int sz);
+int socket_server_send_lowpriority(struct socket_server *, int id, const void * buffer, int sz);
 
 // ctrl command below returns id
 int socket_server_listen(struct socket_server *, uintptr_t opaque, const char * addr, int port, int backlog);
@@ -50,7 +51,7 @@ int socket_server_udp(struct socket_server *, uintptr_t opaque, const char * add
 int socket_server_udp_connect(struct socket_server *, int id, const char * addr, int port);
 // If the socket_udp_address is NULL, use last call socket_server_udp_connect address instead
 // You can also use socket_server_send 
-int64_t socket_server_udp_send(struct socket_server *, int id, const struct socket_udp_address *, const void *buffer, int sz);
+int socket_server_udp_send(struct socket_server *, int id, const struct socket_udp_address *, const void *buffer, int sz);
 // extract the address of the message, struct socket_message * should be SOCKET_UDP
 const struct socket_udp_address * socket_server_udp_address(struct socket_server *, struct socket_message *, int *addrsz);