Halfclose (#1338)

* post SOCKET_CLOSE when HALFCLOSE_READ, see #1331

* Add closing status

* Add closing stat in info, See #1333

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");

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 <stdatomic.h>
 
-#define ATOM_BYTE atomic_uchar
 #define ATOM_INT atomic_int
 #define ATOM_POINTER atomic_uintptr_t
 #define ATOM_SIZET atomic_size_t

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 <stdint.h>
 

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<<MAX_SOCKET_P)
 
@@ -94,10 +95,11 @@ struct socket {
 	ATOM_ULONG sending;
 	int fd;
 	int id;
+	ATOM_INT type;
 	uint8_t protocol;
-	ATOM_BYTE type;
 	bool reading;
 	bool writing;
+	bool closing;
 	ATOM_INT udpconnecting;
 	int64_t warn_size;
 	union {
@@ -349,7 +351,7 @@ reserve_id(struct socket_server *ss) {
 			id = ATOM_FAND(&(ss->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));
 			}