Add low priority list to socket send buffer

lualib-src/lua-socket.c

@@ -402,27 +402,43 @@ llisten(lua_State *L) {
 	return 1;
 }
 
-static int
-lsend(lua_State *L) {
-	struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
-	int id = luaL_checkinteger(L, 1);
+static void *
+get_buffer(lua_State *L, int *sz) {
 	void *buffer;
-	int sz;
 	if (lua_isuserdata(L,2)) {
 		buffer = lua_touserdata(L,2);
-		sz = luaL_checkinteger(L,3);
+		*sz = luaL_checkinteger(L,3);
 	} else {
 		size_t len = 0;
 		const char * str =  luaL_checklstring(L, 2, &len);
 		buffer = malloc(len);
 		memcpy(buffer, str, len);
-		sz = (int)len;
+		*sz = (int)len;
 	}
+	return buffer;
+}
+
+static int
+lsend(lua_State *L) {
+	struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
+	int id = luaL_checkinteger(L, 1);
+	int sz = 0;
+	void *buffer = get_buffer(L, &sz);
 	int err = skynet_socket_send(ctx, id, buffer, sz);
 	lua_pushboolean(L, !err);
 	return 1;
 }
 
+static int
+lsendlow(lua_State *L) {
+	struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
+	int id = luaL_checkinteger(L, 1);
+	int sz = 0;
+	void *buffer = get_buffer(L, &sz);
+	skynet_socket_send_lowpriority(ctx, id, buffer, sz);
+	return 0;
+}
+
 static int
 lbind(lua_State *L) {
 	struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
@@ -462,6 +478,7 @@ luaopen_socketdriver(lua_State *L) {
 		{ "close", lclose },
 		{ "listen", llisten },
 		{ "send", lsend },
+		{ "lsend", lsendlow },
 		{ "bind", lbind },
 		{ "start", lstart },
 		{ NULL, NULL },

lualib/socket.lua

@@ -264,6 +264,7 @@ function socket.block(id)
 end
 
 socket.write = assert(driver.send)
+socket.lwrite = assert(driver.lsend)
 
 function socket.invalid(id)
 	return socket_pool[id] == nil

skynet-src/skynet_socket.c

@@ -115,6 +115,11 @@ skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) {
 	return 0;
 }
 
+void
+skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz) {
+	socket_server_send_lowpriority(SOCKET_SERVER, id, buffer, sz);
+}
+
 int 
 skynet_socket_listen(struct skynet_context *ctx, const char *host, int port, int backlog) {
 	uint32_t source = skynet_context_handle(ctx);

skynet-src/skynet_socket.h

@@ -22,6 +22,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_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_block_connect(struct skynet_context *ctx, const char *host, int port);

skynet-src/socket_server.c

@@ -29,6 +29,9 @@
 
 #define MAX_SOCKET (1<<MAX_SOCKET_P)
 
+#define PRIORITY_HIGH 0
+#define PRIORITY_LOW 1
+
 struct write_buffer {
 	struct write_buffer * next;
 	char *ptr;
@@ -36,6 +39,11 @@ struct write_buffer {
 	void *buffer;
 };
 
+struct wb_list {
+	struct write_buffer * head;
+	struct write_buffer * tail;
+};
+
 struct socket {
 	int fd;
 	int id;
@@ -43,8 +51,8 @@ struct socket {
 	int size;
 	int64_t wb_size;
 	uintptr_t opaque;
-	struct write_buffer * head;
-	struct write_buffer * tail;
+	struct wb_list high;
+	struct wb_list low;
 };
 
 struct socket_server {
@@ -147,6 +155,12 @@ reserve_id(struct socket_server *ss) {
 	return -1;
 }
 
+static inline void
+clear_wb_list(struct wb_list *list) {
+	list->head = NULL;
+	list->tail = NULL;
+}
+
 struct socket_server * 
 socket_server_create() {
 	int i;
@@ -179,8 +193,8 @@ socket_server_create() {
 	for (i=0;i<MAX_SOCKET;i++) {
 		struct socket *s = &ss->slot[i];
 		s->type = SOCKET_TYPE_INVALID;
-		s->head = NULL;
-		s->tail = NULL;
+		clear_wb_list(&s->high);
+		clear_wb_list(&s->low);
 	}
 	ss->alloc_id = 0;
 	ss->event_n = 0;
@@ -191,6 +205,19 @@ socket_server_create() {
 	return ss;
 }
 
+static void
+free_wb_list(struct wb_list *list) {
+	struct write_buffer *wb = list->head;
+	while (wb) {
+		struct write_buffer *tmp = wb;
+		wb = wb->next;
+		FREE(tmp->buffer);
+		FREE(tmp);
+	}
+	list->head = NULL;
+	list->tail = NULL;
+}
+
 static void
 force_close(struct socket_server *ss, struct socket *s, struct socket_message *result) {
 	result->id = s->id;
@@ -201,14 +228,8 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 		return;
 	}
 	assert(s->type != SOCKET_TYPE_RESERVE);
-	struct write_buffer *wb = s->head;
-	while (wb) {
-		struct write_buffer *tmp = wb;
-		wb = wb->next;
-		FREE(tmp->buffer);
-		FREE(tmp);
-	}
-	s->head = s->tail = NULL;
+	free_wb_list(&s->high);
+	free_wb_list(&s->low);
 	if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
 		sp_del(ss->event_fd, s->fd);
 	}
@@ -234,6 +255,12 @@ socket_server_release(struct socket_server *ss) {
 	FREE(ss);
 }
 
+static inline void
+check_wb_list(struct wb_list *s) {
+	assert(s->head == NULL);
+	assert(s->tail == NULL);
+}
+
 static struct socket *
 new_fd(struct socket_server *ss, int id, int fd, uintptr_t opaque, bool add) {
 	struct socket * s = &ss->slot[id % MAX_SOCKET];
@@ -251,8 +278,8 @@ new_fd(struct socket_server *ss, int id, int fd, uintptr_t opaque, bool add) {
 	s->size = MIN_READ_BUFFER;
 	s->opaque = opaque;
 	s->wb_size = 0;
-	assert(s->head == NULL);
-	assert(s->tail == NULL);
+	check_wb_list(&s->high);
+	check_wb_list(&s->low);
 	return s;
 }
 
@@ -335,9 +362,9 @@ _failed:
 }
 
 static int
-send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
-	while (s->head) {
-		struct write_buffer * tmp = s->head;
+send_list(struct socket_server *ss, struct socket *s, struct wb_list *list, struct socket_message *result) {
+	while (list->head) {
+		struct write_buffer * tmp = list->head;
 		for (;;) {
 			int sz = write(s->fd, tmp->ptr, tmp->sz);
 			if (sz < 0) {
@@ -358,11 +385,17 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 			}
 			break;
 		}
-		s->head = tmp->next;
+		list->head = tmp->next;
 		FREE(tmp->buffer);
 		FREE(tmp);
 	}
-	s->tail = NULL;
+	list->tail = NULL;
+
+	return -1;
+}
+
+static int
+check_close(struct socket_server *ss, struct socket *s, struct socket_message *result) {
 	sp_write(ss->event_fd, s->fd, s, false);
 
 	if (s->type == SOCKET_TYPE_HALFCLOSE) {
@@ -373,14 +406,73 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 	return -1;
 }
 
+static inline int
+list_uncomplete(struct wb_list *s) {
+	struct write_buffer *wb = s->head;
+	if (wb == NULL)
+		return 0;
+	
+	return (void *)wb->ptr != wb->buffer;
+}
+
 static void
-append_sendbuffer(struct socket *s, struct request_send * request, int n) {
+raise_uncomplete(struct socket * s) {
+	struct wb_list *low = &s->low;
+	struct write_buffer *tmp = low->head;
+	assert(list_uncomplete(&s->low));
+	assert(!list_uncomplete(&s->high));
+	struct wb_list *high = &s->high;
+	low->head = tmp->next;
+	if (low->head == NULL) {
+		low->tail = NULL;
+	}
+	tmp->next = high->head;
+	high->head = tmp;
+	if (tmp->next == NULL) {
+		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.
+
+	1. send high list as far as possible.
+	2. If high list is empty, try to send low list.
+	3. If low list head is uncomplete (send a part before), move the head of low list to the head of high list (call raise_uncomplete) .
+	4. If two lists are both empty, turn off the event. (call check_close)
+ */
+static int
+send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
+	assert(!list_uncomplete(&s->low));
+	int r = send_list(ss,s,&s->high,result);
+	if (list_uncomplete(&s->high)) {
+		return r;
+	}
+	if (s->high.head == NULL) {
+		r = send_list(ss,s,&s->low,result);
+		if (list_uncomplete(&s->low)) {
+			raise_uncomplete(s);
+		}
+	}
+	if (send_buffer_empty(s)) {
+		return check_close(ss, s, result);
+	}
+
+	return r;
+}
+
+static int
+append_sendbuffer_(struct wb_list *s, struct request_send * request, int n) {
 	struct write_buffer * buf = MALLOC(sizeof(*buf));
 	buf->ptr = request->buffer+n;
 	buf->sz = request->sz - n;
 	buf->buffer = request->buffer;
 	buf->next = NULL;
-	s->wb_size += buf->sz;
 	if (s->head == NULL) {
 		s->head = s->tail = buf;
 	} else {
@@ -389,10 +481,28 @@ append_sendbuffer(struct socket *s, struct request_send * request, int n) {
 		s->tail->next = buf;
 		s->tail = buf;
 	}
+	return buf->sz;
 }
 
+static inline void
+append_sendbuffer(struct socket *s, struct request_send * request, int n) {
+	s->wb_size += append_sendbuffer_(&s->high, request, n);
+}
+
+static inline void
+append_sendbuffer_low(struct socket *s, struct request_send * request) {
+	s->wb_size += append_sendbuffer_(&s->low, request, 0);
+}
+
+/*
+	When send a package , we can assign the priority : PRIORITY_HIGH or PRIORITY_LOW
+
+	If socket buffer is empty, write to fd directly.
+		If write a part, append the rest part to high list. (Even priority is PRIORITY_LOW)
+	Else append package to high (PRIORITY_HIGH) or low (PRIORITY_LOW) list.
+ */
 static int
-send_socket(struct socket_server *ss, struct request_send * request, struct socket_message *result) {
+send_socket(struct socket_server *ss, struct request_send * request, struct socket_message *result, int priority) {
 	int id = request->id;
 	struct socket * s = &ss->slot[id % MAX_SOCKET];
 	if (s->type == SOCKET_TYPE_INVALID || s->id != id 
@@ -402,7 +512,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 		return -1;
 	}
 	assert(s->type != SOCKET_TYPE_PLISTEN && s->type != SOCKET_TYPE_LISTEN);
-	if (s->head == NULL) {
+	if (send_buffer_empty(s)) {
 		int n = write(s->fd, request->buffer, request->sz);
 		if (n<0) {
 			switch(errno) {
@@ -420,10 +530,14 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 			FREE(request->buffer);
 			return -1;
 		}
-		append_sendbuffer(s, request, n);
+		append_sendbuffer(s, request, n);	// add to high priority list, even priority == PRIORITY_LOW
 		sp_write(ss->event_fd, s->fd, s, true);
 	} else {
-		append_sendbuffer(s, request, 0);
+		if (priority == PRIORITY_LOW) {
+			append_sendbuffer_low(s, request);
+		} else {
+			append_sendbuffer(s, request, 0);
+		}
 	}
 	return -1;
 }
@@ -460,12 +574,12 @@ close_socket(struct socket_server *ss, struct request_close *request, struct soc
 		result->data = NULL;
 		return SOCKET_CLOSE;
 	}
-	if (s->head) { 
+	if (!send_buffer_empty(s)) { 
 		int type = send_buffer(ss,s,result);
 		if (type != -1)
 			return type;
 	}
-	if (s->head == NULL) {
+	if (send_buffer_empty(s)) {
 		force_close(ss,s,result);
 		result->id = id;
 		result->opaque = request->opaque;
@@ -581,7 +695,9 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
 		result->data = NULL;
 		return SOCKET_EXIT;
 	case 'D':
-		return send_socket(ss, (struct request_send *)buffer, result);
+		return send_socket(ss, (struct request_send *)buffer, result, PRIORITY_HIGH);
+	case 'P':
+		return send_socket(ss, (struct request_send *)buffer, result, PRIORITY_LOW);
 	default:
 		fprintf(stderr, "socket-server: Unknown ctrl %c.\n",type);
 		return -1;
@@ -835,6 +951,22 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 	return s->wb_size;
 }
 
+void 
+socket_server_send_lowpriority(struct socket_server *ss, int id, const void * buffer, int sz) {
+	struct socket * s = &ss->slot[id % MAX_SOCKET];
+	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
+		return;
+	}
+	assert(s->type != SOCKET_TYPE_RESERVE);
+
+	struct request_package request;
+	request.u.send.id = id;
+	request.u.send.sz = sz;
+	request.u.send.buffer = (char *)buffer;
+
+	send_request(ss, &request, 'P', sizeof(request.u.send));
+}
+
 void
 socket_server_exit(struct socket_server *ss) {
 	struct request_package request;

skynet-src/socket_server.h

@@ -29,6 +29,7 @@ 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);
 
 // ctrl command below returns id
 int socket_server_listen(struct socket_server *, uintptr_t opaque, const char * addr, int port, int backlog);