diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index 0da565a9..0f735300 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -742,6 +742,9 @@ getinfo(lua_State *L, struct socket_info *si) { case SOCKET_INFO_BIND: lua_pushstring(L, "BIND"); break; + case SOCKET_INFO_CLOSING: + lua_pushstring(L, "CLOSING"); + break; default: lua_pushstring(L, "UNKNOWN"); lua_setfield(L, -2, "type"); diff --git a/skynet-src/atomic.h b/skynet-src/atomic.h index d327bcdd..a7b60534 100644 --- a/skynet-src/atomic.h +++ b/skynet-src/atomic.h @@ -3,7 +3,6 @@ #ifdef __STDC_NO_ATOMICS__ -#define ATOM_BYTE unsigned char #define ATOM_INT int #define ATOM_POINTER void * #define ATOM_SIZET size_t @@ -23,7 +22,6 @@ #include -#define ATOM_BYTE atomic_uchar #define ATOM_INT atomic_int #define ATOM_POINTER atomic_uintptr_t #define ATOM_SIZET atomic_size_t diff --git a/skynet-src/socket_info.h b/skynet-src/socket_info.h index 739536e8..a22511ab 100644 --- a/skynet-src/socket_info.h +++ b/skynet-src/socket_info.h @@ -6,6 +6,7 @@ #define SOCKET_INFO_TCP 2 #define SOCKET_INFO_UDP 3 #define SOCKET_INFO_BIND 4 +#define SOCKET_INFO_CLOSING 5 #include diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index aee51575..bb38a330 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -29,9 +29,10 @@ #define SOCKET_TYPE_LISTEN 3 #define SOCKET_TYPE_CONNECTING 4 #define SOCKET_TYPE_CONNECTED 5 -#define SOCKET_TYPE_HALFCLOSE 6 -#define SOCKET_TYPE_PACCEPT 7 -#define SOCKET_TYPE_BIND 8 +#define SOCKET_TYPE_HALFCLOSE_READ 6 +#define SOCKET_TYPE_HALFCLOSE_WRITE 7 +#define SOCKET_TYPE_PACCEPT 8 +#define SOCKET_TYPE_BIND 9 #define MAX_SOCKET (1<alloc_id), 0x7fffffff) & 0x7fffffff; } struct socket *s = &ss->slot[HASH_ID(id)]; - unsigned char type_invalid = ATOM_LOAD(&s->type); + int type_invalid = ATOM_LOAD(&s->type); if (type_invalid == SOCKET_TYPE_INVALID) { if (ATOM_CAS(&s->type, type_invalid, SOCKET_TYPE_RESERVE)) { s->id = id; @@ -564,6 +566,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque, s->fd = fd; s->reading = true; s->writing = false; + s->closing = false; ATOM_INIT(&s->sending , ID_TAG16(id) << 16 | 0); s->protocol = protocol; s->p.size = MIN_READ_BUFFER; @@ -674,6 +677,17 @@ _failed_getaddrinfo: return SOCKET_ERR; } +static int +close_write(struct socket_server *ss, struct socket *s, struct socket_lock *l, struct socket_message *result) { + if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE_READ) { + force_close(ss,s,l,result); + return SOCKET_CLOSE; + } else { + ATOM_STORE(&s->type, SOCKET_TYPE_HALFCLOSE_WRITE); + return -1; + } +} + static int send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list, struct socket_lock *l, struct socket_message *result) { while (list->head) { @@ -687,8 +701,7 @@ send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list, case AGAIN_WOULDBLOCK: return -1; } - force_close(ss,s,l,result); - return SOCKET_CLOSE; + return close_write(ss, s, l, result); } stat_write(ss,s,(int)sz); s->wb_size -= sz; @@ -827,12 +840,18 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_lock *l, if (send_list(ss,s,&s->high,l,result) == SOCKET_CLOSE) { return SOCKET_CLOSE; } + if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE_WRITE) { + return -1; + } if (s->high.head == NULL) { // step 2 if (s->low.head != NULL) { if (send_list(ss,s,&s->low,l,result) == SOCKET_CLOSE) { return SOCKET_CLOSE; } + if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE_WRITE) { + return -1; + } // step 3 if (list_uncomplete(&s->low)) { raise_uncomplete(s); @@ -843,13 +862,14 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_lock *l, } // step 4 assert(send_buffer_empty(s) && s->wb_size == 0); - int err = enable_write(ss, s, false); - if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE) { - force_close(ss, s, l, result); - return SOCKET_CLOSE; + if (s->closing) { + force_close(ss, s, l, result); + return SOCKET_CLOSE; } + int err = enable_write(ss, s, false); + if (err) { result->opaque = s->opaque; result->id = s->id; @@ -859,12 +879,12 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_lock *l, } 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; + s->warn_size = 0; + result->opaque = s->opaque; + result->id = s->id; + result->ud = 0; + result->data = NULL; + return SOCKET_WARNING; } } @@ -970,8 +990,9 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock send_object_init(ss, &so, request->buffer, request->sz); uint8_t type = ATOM_LOAD(&s->type); if (type == SOCKET_TYPE_INVALID || s->id != id - || type == SOCKET_TYPE_HALFCLOSE - || type == SOCKET_TYPE_PACCEPT) { + || type == SOCKET_TYPE_HALFCLOSE_WRITE + || type == SOCKET_TYPE_PACCEPT + || s->closing) { so.free_func((void *)request->buffer); return -1; } @@ -992,7 +1013,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock socklen_t sasz = udp_socket_address(s, udp_address, &sa); if (sasz == 0) { // udp type mismatch, just drop it. - skynet_error(NULL, "socket-server: udp socket (%d) type mistach.", id); + skynet_error(NULL, "socket-server: udp socket (%d) type mismatch.", id); so.free_func((void *)request->buffer); return -1; } @@ -1060,7 +1081,8 @@ _failed: static inline int nomore_sending_data(struct socket *s) { - return send_buffer_empty(s) && s->dw_buffer == NULL && (ATOM_LOAD(&s->sending) & 0xffff) == 0; + return (send_buffer_empty(s) && s->dw_buffer == NULL && (ATOM_LOAD(&s->sending) & 0xffff) == 0) + || (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE_WRITE); } static int @@ -1076,20 +1098,15 @@ close_socket(struct socket_server *ss, struct request_close *request, struct soc } struct socket_lock l; socket_lock_init(s, &l); - if (!nomore_sending_data(s)) { - enable_read(ss,s,true); - int type = send_buffer(ss,s,&l,result); - // type : -1 or SOCKET_WARNING or SOCKET_CLOSE, SOCKET_WARNING means nomore_sending_data - if (type != -1 && type != SOCKET_WARNING) - return type; - } + if (request->shutdown || nomore_sending_data(s)) { force_close(ss,s,&l,result); - result->id = id; - result->opaque = request->opaque; return SOCKET_CLOSE; } - ATOM_STORE(&s->type , SOCKET_TYPE_HALFCLOSE); + + ATOM_STORE(&s->type , SOCKET_TYPE_HALFCLOSE_READ); + s->closing = true; + enable_read(ss,s,true); shutdown(s->fd, SHUT_RD); return -1; @@ -1142,7 +1159,7 @@ resume_socket(struct socket_server *ss, struct request_resumepause *request, str result->data = "transfer"; return SOCKET_OPEN; } - // if s->type == SOCKET_TYPE_HALFCLOSE , SOCKET_CLOSE message will send later + // if s->type == SOCKET_TYPE_HALFCLOSE_* , SOCKET_CLOSE message will send later return -1; } @@ -1368,15 +1385,14 @@ forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_lo FREE(buffer); if (nomore_sending_data(s)) { force_close(ss,s,l,result); - return SOCKET_CLOSE; } else { - ATOM_STORE(&s->type , SOCKET_TYPE_HALFCLOSE); + ATOM_STORE(&s->type , SOCKET_TYPE_HALFCLOSE_READ); shutdown(s->fd, SHUT_RD); - return -1; } + return SOCKET_CLOSE; } - if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE) { + if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE_READ) { // discard recv data FREE(buffer); return -1; @@ -1737,7 +1753,7 @@ int socket_server_send(struct socket_server *ss, struct socket_sendbuffer *buf) { int id = buf->id; struct socket * s = &ss->slot[HASH_ID(id)]; - if (socket_invalid(s, id)) { + if (socket_invalid(s, id) || s->closing) { free_buffer(ss, buf); return -1; } @@ -2167,6 +2183,7 @@ static int query_info(struct socket *s, struct socket_info *si) { union sockaddr_all u; socklen_t slen = sizeof(u); + int closing = 0; switch (ATOM_LOAD(&s->type)) { case SOCKET_TYPE_BIND: si->type = SOCKET_INFO_BIND; @@ -2178,9 +2195,12 @@ query_info(struct socket *s, struct socket_info *si) { getname(&u, si->name, sizeof(si->name)); } break; + case SOCKET_TYPE_HALFCLOSE_READ: + case SOCKET_TYPE_HALFCLOSE_WRITE: + closing = 1; case SOCKET_TYPE_CONNECTED: if (s->protocol == PROTOCOL_TCP) { - si->type = SOCKET_INFO_TCP; + si->type = closing ? SOCKET_INFO_CLOSING : SOCKET_INFO_TCP; if (getpeername(s->fd, &u.s, &slen) == 0) { getname(&u, si->name, sizeof(si->name)); }