diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index 689e981e..6a5651c2 100644 --- a/lualib-src/lua-socket.c +++ b/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 }, diff --git a/lualib/socket.lua b/lualib/socket.lua index b041ccb8..5908c6a2 100644 --- a/lualib/socket.lua +++ b/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 diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index 7ca48233..7f6f1330 100644 --- a/skynet-src/skynet_socket.c +++ b/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); diff --git a/skynet-src/skynet_socket.h b/skynet-src/skynet_socket.h index 1d6afd45..3275521d 100644 --- a/skynet-src/skynet_socket.h +++ b/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); diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index 23a56d68..7fc47302 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -29,6 +29,9 @@ #define MAX_SOCKET (1<head = NULL; + list->tail = NULL; +} + struct socket_server * socket_server_create() { int i; @@ -179,8 +193,8 @@ socket_server_create() { for (i=0;islot[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; diff --git a/skynet-src/socket_server.h b/skynet-src/socket_server.h index 41351331..2fbb6939 100644 --- a/skynet-src/socket_server.h +++ b/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);