From 9618dd47ce594ebc22a8ed35b48aa464de98714c Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Thu, 8 Jun 2017 20:02:39 +0800
Subject: [PATCH 01/11] direct write if send queue is empty

---
 skynet-src/socket_server.c | 158 +++++++++++++++++++++++++++----------
 1 file changed, 117 insertions(+), 41 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index 3841f233..cceb7577 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -3,6 +3,7 @@
 #include "socket_server.h"
 #include "socket_poll.h"
 #include "atomic.h"
+#include "spinlock.h"
 
 #include <sys/types.h>
 #include <sys/socket.h>
@@ -86,6 +87,10 @@ struct socket {
 		int size;
 		uint8_t udp_address[UDP_ADDRESS_SIZE];
 	} p;
+	struct spinlock dw_lock;
+	int dw_offset;
+	const void * dw_buffer;
+	size_t dw_size;
 };
 
 struct socket_server {
@@ -331,6 +336,13 @@ free_wb_list(struct socket_server *ss, struct wb_list *list) {
 	list->tail = NULL;
 }
 
+static void
+free_buffer(struct socket_server *ss, const void * buffer, int sz) {
+	struct send_object so;
+	send_object_init(ss, &so, (void *)buffer, sz);
+	so.free_func((void *)buffer);
+}
+
 static void
 force_close(struct socket_server *ss, struct socket *s, struct socket_message *result) {
 	result->id = s->id;
@@ -346,12 +358,18 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 	if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
 		sp_del(ss->event_fd, s->fd);
 	}
+	spinlock_lock(&s->dw_lock);
 	if (s->type != SOCKET_TYPE_BIND) {
 		if (close(s->fd) < 0) {
 			perror("close socket:");
 		}
 	}
 	s->type = SOCKET_TYPE_INVALID;
+	if (s->dw_buffer) {
+		free_buffer(ss, s->dw_buffer, s->dw_size);
+		s->dw_buffer = NULL;
+	}
+	spinlock_unlock(&s->dw_lock);
 }
 
 void 
@@ -397,6 +415,9 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
 	s->warn_size = 0;
 	check_wb_list(&s->high);
 	check_wb_list(&s->low);
+	spinlock_init(&s->dw_lock);
+	s->dw_buffer = NULL;
+	s->dw_size = 0;
 	return s;
 }
 
@@ -481,7 +502,7 @@ send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list,
 	while (list->head) {
 		struct write_buffer * tmp = list->head;
 		for (;;) {
-			int sz = write(s->fd, tmp->ptr, tmp->sz);
+			ssize_t sz = write(s->fd, tmp->ptr, tmp->sz);
 			if (sz < 0) {
 				switch(errno) {
 				case EINTR:
@@ -616,7 +637,7 @@ send_buffer_empty(struct socket *s) {
 	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) {
+send_buffer_(struct socket_server *ss, struct socket *s, struct socket_message *result) {
 	assert(!list_uncomplete(&s->low));
 	// step 1
 	if (send_list(ss,s,&s->high,result) == SOCKET_CLOSE) {
@@ -655,13 +676,39 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 	return -1;
 }
 
+static int
+send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
+	spinlock_lock(&s->dw_lock);
+	if (s->dw_buffer) {
+		// add direct write buffer before high.head
+		struct write_buffer * buf = MALLOC(SIZEOF_TCPBUFFER);
+		struct send_object so;
+		buf->userobject = send_object_init(ss, &so, (void *)s->dw_buffer, s->dw_size);
+		buf->ptr = (char*)so.buffer+s->dw_offset;
+		buf->sz = so.sz - s->dw_offset;
+		buf->buffer = (void *)s->dw_buffer;
+		if (s->high.head == NULL) {
+			s->high.head = s->high.tail = buf;
+			buf->next = NULL;
+		} else {
+			buf->next = s->high.head;
+			s->high.head = buf;
+		}
+		s->dw_buffer = NULL;
+	}
+	int r = send_buffer_(ss,s,result);
+	spinlock_unlock(&s->dw_lock);
+
+	return r;
+}
+
 static struct write_buffer *
-append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_send * request, int size, int n) {
+append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_send * request, int size) {
 	struct write_buffer * buf = MALLOC(size);
 	struct send_object so;
 	buf->userobject = send_object_init(ss, &so, request->buffer, request->sz);
-	buf->ptr = (char*)so.buffer+n;
-	buf->sz = so.sz - n;
+	buf->ptr = (char*)so.buffer;
+	buf->sz = so.sz;
 	buf->buffer = request->buffer;
 	buf->next = NULL;
 	if (s->head == NULL) {
@@ -678,20 +725,20 @@ append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_s
 static inline void
 append_sendbuffer_udp(struct socket_server *ss, struct socket *s, int priority, struct request_send * request, const uint8_t udp_address[UDP_ADDRESS_SIZE]) {
 	struct wb_list *wl = (priority == PRIORITY_HIGH) ? &s->high : &s->low;
-	struct write_buffer *buf = append_sendbuffer_(ss, wl, request, SIZEOF_UDPBUFFER, 0);
+	struct write_buffer *buf = append_sendbuffer_(ss, wl, request, SIZEOF_UDPBUFFER);
 	memcpy(buf->udp_address, udp_address, UDP_ADDRESS_SIZE);
 	s->wb_size += buf->sz;
 }
 
 static inline void
-append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request, int n) {
-	struct write_buffer *buf = append_sendbuffer_(ss, &s->high, request, SIZEOF_TCPBUFFER, n);
+append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request) {
+	struct write_buffer *buf = append_sendbuffer_(ss, &s->high, request, SIZEOF_TCPBUFFER);
 	s->wb_size += buf->sz;
 }
 
 static inline void
 append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_send * request) {
-	struct write_buffer *buf = append_sendbuffer_(ss, &s->low, request, SIZEOF_TCPBUFFER, 0);
+	struct write_buffer *buf = append_sendbuffer_(ss, &s->low, request, SIZEOF_TCPBUFFER);
 	s->wb_size += buf->sz;
 }
 
@@ -722,25 +769,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 	}
 	if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) {
 		if (s->protocol == PROTOCOL_TCP) {
-			int n = write(s->fd, so.buffer, so.sz);
-			if (n<0) {
-				switch(errno) {
-				case EINTR:
-				case AGAIN_WOULDBLOCK:
-					n = 0;
-					break;
-				default:
-					fprintf(stderr, "socket-server: write to %d (fd=%d) error :%s.\n",id,s->fd,strerror(errno));
-					force_close(ss,s,result);
-					so.free_func(request->buffer);
-					return SOCKET_CLOSE;
-				}
-			}
-			if (n == so.sz) {
-				so.free_func(request->buffer);
-				return -1;
-			}
-			append_sendbuffer(ss, s, request, n);	// add to high priority list, even priority == PRIORITY_LOW
+			append_sendbuffer(ss, s, request);	// add to high priority list, even priority == PRIORITY_LOW
 		} else {
 			// udp
 			if (udp_address == NULL) {
@@ -762,7 +791,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 			if (priority == PRIORITY_LOW) {
 				append_sendbuffer_low(ss, s, request);
 			} else {
-				append_sendbuffer(ss, s, request, 0);
+				append_sendbuffer(ss, s, request);
 			}
 		} else {
 			if (udp_address == NULL) {
@@ -1327,7 +1356,7 @@ send_request(struct socket_server *ss, struct request_package *request, char typ
 	request->header[6] = (uint8_t)type;
 	request->header[7] = (uint8_t)len;
 	for (;;) {
-		int n = write(ss->sendctrl_fd, &request->header[6], len+2);
+		ssize_t n = write(ss->sendctrl_fd, &request->header[6], len+2);
 		if (n<0) {
 			if (errno != EINTR) {
 				fprintf(stderr, "socket-server : send ctrl command error %s.\n", strerror(errno));
@@ -1368,11 +1397,9 @@ socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * a
 	return request.u.open.id;
 }
 
-static void
-free_buffer(struct socket_server *ss, const void * buffer, int sz) {
-	struct send_object so;
-	send_object_init(ss, &so, (void *)buffer, sz);
-	so.free_func((void *)buffer);
+static inline int
+can_direct_write(struct socket *s, int id) {
+	return s->id == id && send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED && s->dw_buffer == NULL;
 }
 
 // return -1 when error, 0 when success
@@ -1384,6 +1411,35 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 		return -1;
 	}
 
+	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+		// may be we can send directly, double check
+		if (can_direct_write(s,id)) {
+			// send directly
+			struct send_object so;
+			send_object_init(ss, &so, (void *)buffer, sz);
+			ssize_t n = write(s->fd, so.buffer, so.sz);
+			if (n<0) {
+				// ignore error, let socket thread try again
+				n = 0;
+			}
+			if (n == so.sz) {
+				// write done
+				spinlock_unlock(&s->dw_lock);
+				so.free_func((void *)buffer);
+				return 0;
+			}
+			// write failed, put buffer into s->dw_* , and let socket thread send it. see send_buffer()
+			s->dw_buffer = buffer;
+			s->dw_size = sz;
+			s->dw_offset = n;
+			sp_write(ss->event_fd, s->fd, s, true);
+			spinlock_unlock(&s->dw_lock);
+			return 0;
+		} else {
+			spinlock_unlock(&s->dw_lock);
+		}
+	}
+
 	struct request_package request;
 	request.u.send.id = id;
 	request.u.send.sz = sz;
@@ -1597,11 +1653,6 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 		return -1;
 	}
 
-	struct request_package request;
-	request.u.send_udp.send.id = id;
-	request.u.send_udp.send.sz = sz;
-	request.u.send_udp.send.buffer = (char *)buffer;
-
 	const uint8_t *udp_address = (const uint8_t *)addr;
 	int addrsz;
 	switch (udp_address[0]) {
@@ -1616,7 +1667,32 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 		return -1;
 	}
 
-	memcpy(request.u.send_udp.address, udp_address, addrsz);	
+	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+		// may be we can send directly, double check
+		if (can_direct_write(s,id)) {
+			// send directly
+			struct send_object so;
+			send_object_init(ss, &so, (void *)buffer, sz);
+			union sockaddr_all sa;
+			socklen_t sasz = udp_socket_address(s, udp_address, &sa);
+			int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
+			if (n >= 0) {
+				// sendto succ
+				spinlock_unlock(&s->dw_lock);
+				so.free_func((void *)buffer);
+				return 0;
+			}
+			spinlock_unlock(&s->dw_lock);
+			// let socket thread try again, udp doesn't care the order
+		}
+	}
+
+	struct request_package request;
+	request.u.send_udp.send.id = id;
+	request.u.send_udp.send.sz = sz;
+	request.u.send_udp.send.buffer = (char *)buffer;
+
+	memcpy(request.u.send_udp.address, udp_address, addrsz);
 
 	send_request(ss, &request, 'A', sizeof(request.u.send_udp.send)+addrsz);
 	return 0;

From 8ca47f76bf961330677072a05bf3d8023890a311 Mon Sep 17 00:00:00 2001
From: hongl <hongling-0@qq.com>
Date: Fri, 16 Jun 2017 17:01:58 +0800
Subject: [PATCH 02/11] send crash

---
 skynet-src/socket_server.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index cceb7577..f33c7e9a 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -687,6 +687,7 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 		buf->ptr = (char*)so.buffer+s->dw_offset;
 		buf->sz = so.sz - s->dw_offset;
 		buf->buffer = (void *)s->dw_buffer;
+		s->wb_size+=buf->sz;
 		if (s->high.head == NULL) {
 			s->high.head = s->high.tail = buf;
 			buf->next = NULL;

From cc138adda9ce2082bfb4a196741569a0891ed90c Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Mon, 19 Jun 2017 11:48:17 +0800
Subject: [PATCH 03/11] bugfix: lock bug & udp direct write

---
 skynet-src/socket_server.c | 39 ++++++++++++++++++++++++++++++--------
 1 file changed, 31 insertions(+), 8 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index cceb7577..09f1baa1 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -80,8 +80,9 @@ struct socket {
 	int64_t wb_size;
 	int fd;
 	int id;
-	uint16_t protocol;
-	uint16_t type;
+	uint8_t protocol;
+	uint8_t type;
+	uint16_t udpconnecting;
 	int64_t warn_size;
 	union {
 		int size;
@@ -262,6 +263,9 @@ reserve_id(struct socket_server *ss) {
 		if (s->type == SOCKET_TYPE_INVALID) {
 			if (ATOM_CAS(&s->type, SOCKET_TYPE_INVALID, SOCKET_TYPE_RESERVE)) {
 				s->id = id;
+				// socket_server_udp_connect may inc s->udpconncting directly (from other thread, before new_fd), 
+				// so reset it to 0 here rather than in new_fd.
+				s->udpconnecting = 0;
 				s->fd = -1;
 				return id;
 			} else {
@@ -991,6 +995,7 @@ set_udp_address(struct socket_server *ss, struct request_setudp *request, struct
 	} else {
 		memcpy(s->p.udp_address, request->address, 1+2+16);	// 1 type, 2 port, 16 ipv6
 	}
+	ATOM_DEC(&s->udpconnecting);
 	return -1;
 }
 
@@ -1399,7 +1404,7 @@ socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * a
 
 static inline int
 can_direct_write(struct socket *s, int id) {
-	return s->id == id && send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED && s->dw_buffer == NULL;
+	return s->id == id && send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED && s->dw_buffer == NULL && s->udpconnecting == 0;
 }
 
 // return -1 when error, 0 when success
@@ -1417,7 +1422,14 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			// send directly
 			struct send_object so;
 			send_object_init(ss, &so, (void *)buffer, sz);
-			ssize_t n = write(s->fd, so.buffer, so.sz);
+			ssize_t n;
+			if (s->protocol == PROTOCOL_TCP) {
+				n = write(s->fd, so.buffer, so.sz);
+			} else {
+				union sockaddr_all sa;
+				socklen_t sasz = udp_socket_address(s, s->p.udp_address, &sa);
+				n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
+			}
 			if (n<0) {
 				// ignore error, let socket thread try again
 				n = 0;
@@ -1435,9 +1447,8 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			sp_write(ss->event_fd, s->fd, s, true);
 			spinlock_unlock(&s->dw_lock);
 			return 0;
-		} else {
-			spinlock_unlock(&s->dw_lock);
 		}
+		spinlock_unlock(&s->dw_lock);
 	}
 
 	struct request_package request;
@@ -1682,9 +1693,9 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 				so.free_func((void *)buffer);
 				return 0;
 			}
-			spinlock_unlock(&s->dw_lock);
-			// let socket thread try again, udp doesn't care the order
 		}
+		spinlock_unlock(&s->dw_lock);
+		// let socket thread try again, udp doesn't care the order
 	}
 
 	struct request_package request;
@@ -1700,6 +1711,18 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 
 int
 socket_server_udp_connect(struct socket_server *ss, int id, const char * addr, int port) {
+	struct socket * s = &ss->slot[HASH_ID(id)];
+	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
+		return -1;
+	}
+	spinlock_lock(&s->dw_lock);
+	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
+		spinlock_unlock(&s->dw_lock);
+		return -1;
+	}
+	ATOM_INC(&s->udpconnecting);
+	spinlock_unlock(&s->dw_lock);
+
 	int status;
 	struct addrinfo ai_hints;
 	struct addrinfo *ai_list = NULL;

From c53da829d3b09c2a8615e1798362bb43ff5f1cb6 Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Thu, 8 Jun 2017 20:02:39 +0800
Subject: [PATCH 04/11] direct write if send queue is empty

---
 skynet-src/socket_server.c | 158 +++++++++++++++++++++++++++----------
 1 file changed, 117 insertions(+), 41 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index 3841f233..cceb7577 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -3,6 +3,7 @@
 #include "socket_server.h"
 #include "socket_poll.h"
 #include "atomic.h"
+#include "spinlock.h"
 
 #include <sys/types.h>
 #include <sys/socket.h>
@@ -86,6 +87,10 @@ struct socket {
 		int size;
 		uint8_t udp_address[UDP_ADDRESS_SIZE];
 	} p;
+	struct spinlock dw_lock;
+	int dw_offset;
+	const void * dw_buffer;
+	size_t dw_size;
 };
 
 struct socket_server {
@@ -331,6 +336,13 @@ free_wb_list(struct socket_server *ss, struct wb_list *list) {
 	list->tail = NULL;
 }
 
+static void
+free_buffer(struct socket_server *ss, const void * buffer, int sz) {
+	struct send_object so;
+	send_object_init(ss, &so, (void *)buffer, sz);
+	so.free_func((void *)buffer);
+}
+
 static void
 force_close(struct socket_server *ss, struct socket *s, struct socket_message *result) {
 	result->id = s->id;
@@ -346,12 +358,18 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 	if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
 		sp_del(ss->event_fd, s->fd);
 	}
+	spinlock_lock(&s->dw_lock);
 	if (s->type != SOCKET_TYPE_BIND) {
 		if (close(s->fd) < 0) {
 			perror("close socket:");
 		}
 	}
 	s->type = SOCKET_TYPE_INVALID;
+	if (s->dw_buffer) {
+		free_buffer(ss, s->dw_buffer, s->dw_size);
+		s->dw_buffer = NULL;
+	}
+	spinlock_unlock(&s->dw_lock);
 }
 
 void 
@@ -397,6 +415,9 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
 	s->warn_size = 0;
 	check_wb_list(&s->high);
 	check_wb_list(&s->low);
+	spinlock_init(&s->dw_lock);
+	s->dw_buffer = NULL;
+	s->dw_size = 0;
 	return s;
 }
 
@@ -481,7 +502,7 @@ send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list,
 	while (list->head) {
 		struct write_buffer * tmp = list->head;
 		for (;;) {
-			int sz = write(s->fd, tmp->ptr, tmp->sz);
+			ssize_t sz = write(s->fd, tmp->ptr, tmp->sz);
 			if (sz < 0) {
 				switch(errno) {
 				case EINTR:
@@ -616,7 +637,7 @@ send_buffer_empty(struct socket *s) {
 	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) {
+send_buffer_(struct socket_server *ss, struct socket *s, struct socket_message *result) {
 	assert(!list_uncomplete(&s->low));
 	// step 1
 	if (send_list(ss,s,&s->high,result) == SOCKET_CLOSE) {
@@ -655,13 +676,39 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 	return -1;
 }
 
+static int
+send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
+	spinlock_lock(&s->dw_lock);
+	if (s->dw_buffer) {
+		// add direct write buffer before high.head
+		struct write_buffer * buf = MALLOC(SIZEOF_TCPBUFFER);
+		struct send_object so;
+		buf->userobject = send_object_init(ss, &so, (void *)s->dw_buffer, s->dw_size);
+		buf->ptr = (char*)so.buffer+s->dw_offset;
+		buf->sz = so.sz - s->dw_offset;
+		buf->buffer = (void *)s->dw_buffer;
+		if (s->high.head == NULL) {
+			s->high.head = s->high.tail = buf;
+			buf->next = NULL;
+		} else {
+			buf->next = s->high.head;
+			s->high.head = buf;
+		}
+		s->dw_buffer = NULL;
+	}
+	int r = send_buffer_(ss,s,result);
+	spinlock_unlock(&s->dw_lock);
+
+	return r;
+}
+
 static struct write_buffer *
-append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_send * request, int size, int n) {
+append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_send * request, int size) {
 	struct write_buffer * buf = MALLOC(size);
 	struct send_object so;
 	buf->userobject = send_object_init(ss, &so, request->buffer, request->sz);
-	buf->ptr = (char*)so.buffer+n;
-	buf->sz = so.sz - n;
+	buf->ptr = (char*)so.buffer;
+	buf->sz = so.sz;
 	buf->buffer = request->buffer;
 	buf->next = NULL;
 	if (s->head == NULL) {
@@ -678,20 +725,20 @@ append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_s
 static inline void
 append_sendbuffer_udp(struct socket_server *ss, struct socket *s, int priority, struct request_send * request, const uint8_t udp_address[UDP_ADDRESS_SIZE]) {
 	struct wb_list *wl = (priority == PRIORITY_HIGH) ? &s->high : &s->low;
-	struct write_buffer *buf = append_sendbuffer_(ss, wl, request, SIZEOF_UDPBUFFER, 0);
+	struct write_buffer *buf = append_sendbuffer_(ss, wl, request, SIZEOF_UDPBUFFER);
 	memcpy(buf->udp_address, udp_address, UDP_ADDRESS_SIZE);
 	s->wb_size += buf->sz;
 }
 
 static inline void
-append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request, int n) {
-	struct write_buffer *buf = append_sendbuffer_(ss, &s->high, request, SIZEOF_TCPBUFFER, n);
+append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request) {
+	struct write_buffer *buf = append_sendbuffer_(ss, &s->high, request, SIZEOF_TCPBUFFER);
 	s->wb_size += buf->sz;
 }
 
 static inline void
 append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_send * request) {
-	struct write_buffer *buf = append_sendbuffer_(ss, &s->low, request, SIZEOF_TCPBUFFER, 0);
+	struct write_buffer *buf = append_sendbuffer_(ss, &s->low, request, SIZEOF_TCPBUFFER);
 	s->wb_size += buf->sz;
 }
 
@@ -722,25 +769,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 	}
 	if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) {
 		if (s->protocol == PROTOCOL_TCP) {
-			int n = write(s->fd, so.buffer, so.sz);
-			if (n<0) {
-				switch(errno) {
-				case EINTR:
-				case AGAIN_WOULDBLOCK:
-					n = 0;
-					break;
-				default:
-					fprintf(stderr, "socket-server: write to %d (fd=%d) error :%s.\n",id,s->fd,strerror(errno));
-					force_close(ss,s,result);
-					so.free_func(request->buffer);
-					return SOCKET_CLOSE;
-				}
-			}
-			if (n == so.sz) {
-				so.free_func(request->buffer);
-				return -1;
-			}
-			append_sendbuffer(ss, s, request, n);	// add to high priority list, even priority == PRIORITY_LOW
+			append_sendbuffer(ss, s, request);	// add to high priority list, even priority == PRIORITY_LOW
 		} else {
 			// udp
 			if (udp_address == NULL) {
@@ -762,7 +791,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 			if (priority == PRIORITY_LOW) {
 				append_sendbuffer_low(ss, s, request);
 			} else {
-				append_sendbuffer(ss, s, request, 0);
+				append_sendbuffer(ss, s, request);
 			}
 		} else {
 			if (udp_address == NULL) {
@@ -1327,7 +1356,7 @@ send_request(struct socket_server *ss, struct request_package *request, char typ
 	request->header[6] = (uint8_t)type;
 	request->header[7] = (uint8_t)len;
 	for (;;) {
-		int n = write(ss->sendctrl_fd, &request->header[6], len+2);
+		ssize_t n = write(ss->sendctrl_fd, &request->header[6], len+2);
 		if (n<0) {
 			if (errno != EINTR) {
 				fprintf(stderr, "socket-server : send ctrl command error %s.\n", strerror(errno));
@@ -1368,11 +1397,9 @@ socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * a
 	return request.u.open.id;
 }
 
-static void
-free_buffer(struct socket_server *ss, const void * buffer, int sz) {
-	struct send_object so;
-	send_object_init(ss, &so, (void *)buffer, sz);
-	so.free_func((void *)buffer);
+static inline int
+can_direct_write(struct socket *s, int id) {
+	return s->id == id && send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED && s->dw_buffer == NULL;
 }
 
 // return -1 when error, 0 when success
@@ -1384,6 +1411,35 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 		return -1;
 	}
 
+	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+		// may be we can send directly, double check
+		if (can_direct_write(s,id)) {
+			// send directly
+			struct send_object so;
+			send_object_init(ss, &so, (void *)buffer, sz);
+			ssize_t n = write(s->fd, so.buffer, so.sz);
+			if (n<0) {
+				// ignore error, let socket thread try again
+				n = 0;
+			}
+			if (n == so.sz) {
+				// write done
+				spinlock_unlock(&s->dw_lock);
+				so.free_func((void *)buffer);
+				return 0;
+			}
+			// write failed, put buffer into s->dw_* , and let socket thread send it. see send_buffer()
+			s->dw_buffer = buffer;
+			s->dw_size = sz;
+			s->dw_offset = n;
+			sp_write(ss->event_fd, s->fd, s, true);
+			spinlock_unlock(&s->dw_lock);
+			return 0;
+		} else {
+			spinlock_unlock(&s->dw_lock);
+		}
+	}
+
 	struct request_package request;
 	request.u.send.id = id;
 	request.u.send.sz = sz;
@@ -1597,11 +1653,6 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 		return -1;
 	}
 
-	struct request_package request;
-	request.u.send_udp.send.id = id;
-	request.u.send_udp.send.sz = sz;
-	request.u.send_udp.send.buffer = (char *)buffer;
-
 	const uint8_t *udp_address = (const uint8_t *)addr;
 	int addrsz;
 	switch (udp_address[0]) {
@@ -1616,7 +1667,32 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 		return -1;
 	}
 
-	memcpy(request.u.send_udp.address, udp_address, addrsz);	
+	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+		// may be we can send directly, double check
+		if (can_direct_write(s,id)) {
+			// send directly
+			struct send_object so;
+			send_object_init(ss, &so, (void *)buffer, sz);
+			union sockaddr_all sa;
+			socklen_t sasz = udp_socket_address(s, udp_address, &sa);
+			int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
+			if (n >= 0) {
+				// sendto succ
+				spinlock_unlock(&s->dw_lock);
+				so.free_func((void *)buffer);
+				return 0;
+			}
+			spinlock_unlock(&s->dw_lock);
+			// let socket thread try again, udp doesn't care the order
+		}
+	}
+
+	struct request_package request;
+	request.u.send_udp.send.id = id;
+	request.u.send_udp.send.sz = sz;
+	request.u.send_udp.send.buffer = (char *)buffer;
+
+	memcpy(request.u.send_udp.address, udp_address, addrsz);
 
 	send_request(ss, &request, 'A', sizeof(request.u.send_udp.send)+addrsz);
 	return 0;

From 626ad544b7bfbbe2f96422093695bfdb37253d27 Mon Sep 17 00:00:00 2001
From: hongl <hongling-0@qq.com>
Date: Fri, 16 Jun 2017 17:01:58 +0800
Subject: [PATCH 05/11] send crash

---
 skynet-src/socket_server.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index cceb7577..f33c7e9a 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -687,6 +687,7 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 		buf->ptr = (char*)so.buffer+s->dw_offset;
 		buf->sz = so.sz - s->dw_offset;
 		buf->buffer = (void *)s->dw_buffer;
+		s->wb_size+=buf->sz;
 		if (s->high.head == NULL) {
 			s->high.head = s->high.tail = buf;
 			buf->next = NULL;

From e96b4e782d59fa8b2ce3689a9e799ba57891b581 Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Mon, 19 Jun 2017 11:48:17 +0800
Subject: [PATCH 06/11] bugfix: lock bug & udp direct write

---
 skynet-src/socket_server.c | 39 ++++++++++++++++++++++++++++++--------
 1 file changed, 31 insertions(+), 8 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index f33c7e9a..4fbed482 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -80,8 +80,9 @@ struct socket {
 	int64_t wb_size;
 	int fd;
 	int id;
-	uint16_t protocol;
-	uint16_t type;
+	uint8_t protocol;
+	uint8_t type;
+	uint16_t udpconnecting;
 	int64_t warn_size;
 	union {
 		int size;
@@ -262,6 +263,9 @@ reserve_id(struct socket_server *ss) {
 		if (s->type == SOCKET_TYPE_INVALID) {
 			if (ATOM_CAS(&s->type, SOCKET_TYPE_INVALID, SOCKET_TYPE_RESERVE)) {
 				s->id = id;
+				// socket_server_udp_connect may inc s->udpconncting directly (from other thread, before new_fd), 
+				// so reset it to 0 here rather than in new_fd.
+				s->udpconnecting = 0;
 				s->fd = -1;
 				return id;
 			} else {
@@ -992,6 +996,7 @@ set_udp_address(struct socket_server *ss, struct request_setudp *request, struct
 	} else {
 		memcpy(s->p.udp_address, request->address, 1+2+16);	// 1 type, 2 port, 16 ipv6
 	}
+	ATOM_DEC(&s->udpconnecting);
 	return -1;
 }
 
@@ -1400,7 +1405,7 @@ socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * a
 
 static inline int
 can_direct_write(struct socket *s, int id) {
-	return s->id == id && send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED && s->dw_buffer == NULL;
+	return s->id == id && send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED && s->dw_buffer == NULL && s->udpconnecting == 0;
 }
 
 // return -1 when error, 0 when success
@@ -1418,7 +1423,14 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			// send directly
 			struct send_object so;
 			send_object_init(ss, &so, (void *)buffer, sz);
-			ssize_t n = write(s->fd, so.buffer, so.sz);
+			ssize_t n;
+			if (s->protocol == PROTOCOL_TCP) {
+				n = write(s->fd, so.buffer, so.sz);
+			} else {
+				union sockaddr_all sa;
+				socklen_t sasz = udp_socket_address(s, s->p.udp_address, &sa);
+				n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
+			}
 			if (n<0) {
 				// ignore error, let socket thread try again
 				n = 0;
@@ -1436,9 +1448,8 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			sp_write(ss->event_fd, s->fd, s, true);
 			spinlock_unlock(&s->dw_lock);
 			return 0;
-		} else {
-			spinlock_unlock(&s->dw_lock);
 		}
+		spinlock_unlock(&s->dw_lock);
 	}
 
 	struct request_package request;
@@ -1683,9 +1694,9 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 				so.free_func((void *)buffer);
 				return 0;
 			}
-			spinlock_unlock(&s->dw_lock);
-			// let socket thread try again, udp doesn't care the order
 		}
+		spinlock_unlock(&s->dw_lock);
+		// let socket thread try again, udp doesn't care the order
 	}
 
 	struct request_package request;
@@ -1701,6 +1712,18 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 
 int
 socket_server_udp_connect(struct socket_server *ss, int id, const char * addr, int port) {
+	struct socket * s = &ss->slot[HASH_ID(id)];
+	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
+		return -1;
+	}
+	spinlock_lock(&s->dw_lock);
+	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
+		spinlock_unlock(&s->dw_lock);
+		return -1;
+	}
+	ATOM_INC(&s->udpconnecting);
+	spinlock_unlock(&s->dw_lock);
+
 	int status;
 	struct addrinfo ai_hints;
 	struct addrinfo *ai_list = NULL;

From aa42534208e4089bd46708da8007b0bdd0e36711 Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Wed, 21 Jun 2017 11:50:06 +0800
Subject: [PATCH 07/11] use trylock in send_buffer

---
 skynet-src/socket_server.c | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index 4fbed482..c73a70db 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -682,7 +682,8 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_message *
 
 static int
 send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
-	spinlock_lock(&s->dw_lock);
+	if (!spinlock_trylock(&s->dw_lock))
+		return -1;	// blocked by direct write, send later.
 	if (s->dw_buffer) {
 		// add direct write buffer before high.head
 		struct write_buffer * buf = MALLOC(SIZEOF_TCPBUFFER);
@@ -1445,8 +1446,9 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			s->dw_buffer = buffer;
 			s->dw_size = sz;
 			s->dw_offset = n;
-			sp_write(ss->event_fd, s->fd, s, true);
 			spinlock_unlock(&s->dw_lock);
+
+			sp_write(ss->event_fd, s->fd, s, true);
 			return 0;
 		}
 		spinlock_unlock(&s->dw_lock);

From ba59ed2e6c98a5a7643e5c1e428a8745975b3653 Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Tue, 27 Jun 2017 17:10:48 +0800
Subject: [PATCH 08/11] add nested spinlock, see issue #646

---
 skynet-src/socket_server.c | 37 +++++++++++++----------
 skynet-src/spinlock.h      | 61 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 82 insertions(+), 16 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index c73a70db..3a23f904 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -37,6 +37,9 @@
 #define PRIORITY_HIGH 0
 #define PRIORITY_LOW 1
 
+#define DIRECTWRITE_SOCKETTHREAD 1
+#define DIRECTWRITE_WORKERTHREAD 2
+
 #define HASH_ID(id) (((unsigned)id) % MAX_SOCKET)
 
 #define PROTOCOL_TCP 0
@@ -88,7 +91,7 @@ struct socket {
 		int size;
 		uint8_t udp_address[UDP_ADDRESS_SIZE];
 	} p;
-	struct spinlock dw_lock;
+	struct spinlock_nested dw_lock;
 	int dw_offset;
 	const void * dw_buffer;
 	size_t dw_size;
@@ -362,7 +365,7 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 	if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
 		sp_del(ss->event_fd, s->fd);
 	}
-	spinlock_lock(&s->dw_lock);
+	spinlock_nestedlock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD);
 	if (s->type != SOCKET_TYPE_BIND) {
 		if (close(s->fd) < 0) {
 			perror("close socket:");
@@ -373,7 +376,7 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 		free_buffer(ss, s->dw_buffer, s->dw_size);
 		s->dw_buffer = NULL;
 	}
-	spinlock_unlock(&s->dw_lock);
+	spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD);
 }
 
 void 
@@ -419,7 +422,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
 	s->warn_size = 0;
 	check_wb_list(&s->high);
 	check_wb_list(&s->low);
-	spinlock_init(&s->dw_lock);
+	spinlock_nestedinit(&s->dw_lock);
 	s->dw_buffer = NULL;
 	s->dw_size = 0;
 	return s;
@@ -682,7 +685,7 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_message *
 
 static int
 send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
-	if (!spinlock_trylock(&s->dw_lock))
+	if (!spinlock_nestedtrylock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD))
 		return -1;	// blocked by direct write, send later.
 	if (s->dw_buffer) {
 		// add direct write buffer before high.head
@@ -703,7 +706,7 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 		s->dw_buffer = NULL;
 	}
 	int r = send_buffer_(ss,s,result);
-	spinlock_unlock(&s->dw_lock);
+	spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD);
 
 	return r;
 }
@@ -1418,7 +1421,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 		return -1;
 	}
 
-	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+	if (can_direct_write(s,id) && spinlock_nestedtrylock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD)) {
 		// may be we can send directly, double check
 		if (can_direct_write(s,id)) {
 			// send directly
@@ -1438,7 +1441,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			}
 			if (n == so.sz) {
 				// write done
-				spinlock_unlock(&s->dw_lock);
+				spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
 				so.free_func((void *)buffer);
 				return 0;
 			}
@@ -1446,12 +1449,14 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			s->dw_buffer = buffer;
 			s->dw_size = sz;
 			s->dw_offset = n;
-			spinlock_unlock(&s->dw_lock);
 
 			sp_write(ss->event_fd, s->fd, s, true);
+
+			spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+
 			return 0;
 		}
-		spinlock_unlock(&s->dw_lock);
+		spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
 	}
 
 	struct request_package request;
@@ -1681,7 +1686,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 		return -1;
 	}
 
-	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+	if (can_direct_write(s,id) && spinlock_nestedtrylock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD)) {
 		// may be we can send directly, double check
 		if (can_direct_write(s,id)) {
 			// send directly
@@ -1692,12 +1697,12 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 			int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
 			if (n >= 0) {
 				// sendto succ
-				spinlock_unlock(&s->dw_lock);
+				spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
 				so.free_func((void *)buffer);
 				return 0;
 			}
 		}
-		spinlock_unlock(&s->dw_lock);
+		spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
 		// let socket thread try again, udp doesn't care the order
 	}
 
@@ -1718,13 +1723,13 @@ socket_server_udp_connect(struct socket_server *ss, int id, const char * addr, i
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
 		return -1;
 	}
-	spinlock_lock(&s->dw_lock);
+	spinlock_nestedlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
-		spinlock_unlock(&s->dw_lock);
+		spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
 		return -1;
 	}
 	ATOM_INC(&s->udpconnecting);
-	spinlock_unlock(&s->dw_lock);
+	spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
 
 	int status;
 	struct addrinfo ai_hints;
diff --git a/skynet-src/spinlock.h b/skynet-src/spinlock.h
index 514546f4..98a0bd4b 100644
--- a/skynet-src/spinlock.h
+++ b/skynet-src/spinlock.h
@@ -1,6 +1,8 @@
 #ifndef SKYNET_SPINLOCK_H
 #define SKYNET_SPINLOCK_H
 
+#include <assert.h>
+
 #define SPIN_INIT(q) spinlock_init(&(q)->lock);
 #define SPIN_LOCK(q) spinlock_lock(&(q)->lock);
 #define SPIN_UNLOCK(q) spinlock_unlock(&(q)->lock);
@@ -75,4 +77,63 @@ spinlock_destroy(struct spinlock *lock) {
 
 #endif
 
+struct spinlock_nested {
+	struct spinlock lock;
+	int thread;
+	int count;
+};
+
+static inline void
+spinlock_nestedinit(struct spinlock_nested *ln) {
+	spinlock_init(&ln->lock);
+	ln->thread = 0;
+	ln->count = 0;
+}
+
+static inline void
+spinlock_nestedlock(struct spinlock_nested *ln, int thread) {
+	if (thread != ln->thread) {
+		spinlock_lock(&ln->lock);
+		assert(ln->count == 0);
+		ln->thread = thread;
+	}
+	++ln->count;
+}
+
+static inline int
+spinlock_nestedtrylock(struct spinlock_nested *ln, int thread) {
+	if (thread != ln->thread) {
+		if (spinlock_trylock(&ln->lock)) {
+			assert(ln->count == 0);
+			ln->thread = thread;
+			++ln->count;
+			return 1;	// lock succ
+		} else {
+			return 0;
+		}
+	} else {
+		// in the same thread
+		++ln->count;
+		return 1;
+	}
+}
+
+static inline void
+spinlock_nestedunlock(struct spinlock_nested *ln, int thread) {
+	assert(thread == ln->thread);
+	--ln->count;
+	if (ln->count > 0) {
+		return;
+	}
+	assert(ln->count == 0);
+	ln->thread = 0;
+	spinlock_unlock(&ln->lock);
+}
+
+static inline void
+spinlock_nesteddestroy(struct spinlock_nested *ln) {
+	assert(ln->thread == 0 && ln->count == 0);
+	spinlock_destroy(&ln->lock);
+}
+
 #endif

From 9320d8b28d1d2341ae7979203770366c716d003b Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Tue, 27 Jun 2017 22:08:33 +0800
Subject: [PATCH 09/11] Revert "add nested spinlock, see issue #646"

This reverts commit ba59ed2e6c98a5a7643e5c1e428a8745975b3653.
---
 skynet-src/socket_server.c | 37 ++++++++++-------------
 skynet-src/spinlock.h      | 61 --------------------------------------
 2 files changed, 16 insertions(+), 82 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index 3a23f904..c73a70db 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -37,9 +37,6 @@
 #define PRIORITY_HIGH 0
 #define PRIORITY_LOW 1
 
-#define DIRECTWRITE_SOCKETTHREAD 1
-#define DIRECTWRITE_WORKERTHREAD 2
-
 #define HASH_ID(id) (((unsigned)id) % MAX_SOCKET)
 
 #define PROTOCOL_TCP 0
@@ -91,7 +88,7 @@ struct socket {
 		int size;
 		uint8_t udp_address[UDP_ADDRESS_SIZE];
 	} p;
-	struct spinlock_nested dw_lock;
+	struct spinlock dw_lock;
 	int dw_offset;
 	const void * dw_buffer;
 	size_t dw_size;
@@ -365,7 +362,7 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 	if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
 		sp_del(ss->event_fd, s->fd);
 	}
-	spinlock_nestedlock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD);
+	spinlock_lock(&s->dw_lock);
 	if (s->type != SOCKET_TYPE_BIND) {
 		if (close(s->fd) < 0) {
 			perror("close socket:");
@@ -376,7 +373,7 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 		free_buffer(ss, s->dw_buffer, s->dw_size);
 		s->dw_buffer = NULL;
 	}
-	spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD);
+	spinlock_unlock(&s->dw_lock);
 }
 
 void 
@@ -422,7 +419,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
 	s->warn_size = 0;
 	check_wb_list(&s->high);
 	check_wb_list(&s->low);
-	spinlock_nestedinit(&s->dw_lock);
+	spinlock_init(&s->dw_lock);
 	s->dw_buffer = NULL;
 	s->dw_size = 0;
 	return s;
@@ -685,7 +682,7 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_message *
 
 static int
 send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
-	if (!spinlock_nestedtrylock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD))
+	if (!spinlock_trylock(&s->dw_lock))
 		return -1;	// blocked by direct write, send later.
 	if (s->dw_buffer) {
 		// add direct write buffer before high.head
@@ -706,7 +703,7 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 		s->dw_buffer = NULL;
 	}
 	int r = send_buffer_(ss,s,result);
-	spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_SOCKETTHREAD);
+	spinlock_unlock(&s->dw_lock);
 
 	return r;
 }
@@ -1421,7 +1418,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 		return -1;
 	}
 
-	if (can_direct_write(s,id) && spinlock_nestedtrylock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD)) {
+	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
 		// may be we can send directly, double check
 		if (can_direct_write(s,id)) {
 			// send directly
@@ -1441,7 +1438,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			}
 			if (n == so.sz) {
 				// write done
-				spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+				spinlock_unlock(&s->dw_lock);
 				so.free_func((void *)buffer);
 				return 0;
 			}
@@ -1449,14 +1446,12 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			s->dw_buffer = buffer;
 			s->dw_size = sz;
 			s->dw_offset = n;
+			spinlock_unlock(&s->dw_lock);
 
 			sp_write(ss->event_fd, s->fd, s, true);
-
-			spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
-
 			return 0;
 		}
-		spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+		spinlock_unlock(&s->dw_lock);
 	}
 
 	struct request_package request;
@@ -1686,7 +1681,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 		return -1;
 	}
 
-	if (can_direct_write(s,id) && spinlock_nestedtrylock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD)) {
+	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
 		// may be we can send directly, double check
 		if (can_direct_write(s,id)) {
 			// send directly
@@ -1697,12 +1692,12 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 			int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
 			if (n >= 0) {
 				// sendto succ
-				spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+				spinlock_unlock(&s->dw_lock);
 				so.free_func((void *)buffer);
 				return 0;
 			}
 		}
-		spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+		spinlock_unlock(&s->dw_lock);
 		// let socket thread try again, udp doesn't care the order
 	}
 
@@ -1723,13 +1718,13 @@ socket_server_udp_connect(struct socket_server *ss, int id, const char * addr, i
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
 		return -1;
 	}
-	spinlock_nestedlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+	spinlock_lock(&s->dw_lock);
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
-		spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+		spinlock_unlock(&s->dw_lock);
 		return -1;
 	}
 	ATOM_INC(&s->udpconnecting);
-	spinlock_nestedunlock(&s->dw_lock, DIRECTWRITE_WORKERTHREAD);
+	spinlock_unlock(&s->dw_lock);
 
 	int status;
 	struct addrinfo ai_hints;
diff --git a/skynet-src/spinlock.h b/skynet-src/spinlock.h
index 98a0bd4b..514546f4 100644
--- a/skynet-src/spinlock.h
+++ b/skynet-src/spinlock.h
@@ -1,8 +1,6 @@
 #ifndef SKYNET_SPINLOCK_H
 #define SKYNET_SPINLOCK_H
 
-#include <assert.h>
-
 #define SPIN_INIT(q) spinlock_init(&(q)->lock);
 #define SPIN_LOCK(q) spinlock_lock(&(q)->lock);
 #define SPIN_UNLOCK(q) spinlock_unlock(&(q)->lock);
@@ -77,63 +75,4 @@ spinlock_destroy(struct spinlock *lock) {
 
 #endif
 
-struct spinlock_nested {
-	struct spinlock lock;
-	int thread;
-	int count;
-};
-
-static inline void
-spinlock_nestedinit(struct spinlock_nested *ln) {
-	spinlock_init(&ln->lock);
-	ln->thread = 0;
-	ln->count = 0;
-}
-
-static inline void
-spinlock_nestedlock(struct spinlock_nested *ln, int thread) {
-	if (thread != ln->thread) {
-		spinlock_lock(&ln->lock);
-		assert(ln->count == 0);
-		ln->thread = thread;
-	}
-	++ln->count;
-}
-
-static inline int
-spinlock_nestedtrylock(struct spinlock_nested *ln, int thread) {
-	if (thread != ln->thread) {
-		if (spinlock_trylock(&ln->lock)) {
-			assert(ln->count == 0);
-			ln->thread = thread;
-			++ln->count;
-			return 1;	// lock succ
-		} else {
-			return 0;
-		}
-	} else {
-		// in the same thread
-		++ln->count;
-		return 1;
-	}
-}
-
-static inline void
-spinlock_nestedunlock(struct spinlock_nested *ln, int thread) {
-	assert(thread == ln->thread);
-	--ln->count;
-	if (ln->count > 0) {
-		return;
-	}
-	assert(ln->count == 0);
-	ln->thread = 0;
-	spinlock_unlock(&ln->lock);
-}
-
-static inline void
-spinlock_nesteddestroy(struct spinlock_nested *ln) {
-	assert(ln->thread == 0 && ln->count == 0);
-	spinlock_destroy(&ln->lock);
-}
-
 #endif

From 03acc7f5a99aa53db50d92b93ff0bc50d9b6934f Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Tue, 27 Jun 2017 22:09:41 +0800
Subject: [PATCH 10/11] sp_write should be lock

---
 skynet-src/socket_server.c | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index c73a70db..eceaf593 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -1446,9 +1446,10 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			s->dw_buffer = buffer;
 			s->dw_size = sz;
 			s->dw_offset = n;
-			spinlock_unlock(&s->dw_lock);
 
 			sp_write(ss->event_fd, s->fd, s, true);
+
+			spinlock_unlock(&s->dw_lock);
 			return 0;
 		}
 		spinlock_unlock(&s->dw_lock);

From 751517d7c1e1ac11540a7161d3e21689e6f7ba25 Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Tue, 27 Jun 2017 22:59:21 +0800
Subject: [PATCH 11/11] spinlock can be nested, see issue #646

---
 skynet-src/socket_server.c | 136 ++++++++++++++++++++++++++-----------
 1 file changed, 95 insertions(+), 41 deletions(-)

diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c
index eceaf593..5d4989e9 100644
--- a/skynet-src/socket_server.c
+++ b/skynet-src/socket_server.c
@@ -220,6 +220,44 @@ struct send_object {
 #define MALLOC skynet_malloc
 #define FREE skynet_free
 
+struct socket_lock {
+	struct spinlock *lock;
+	int count;
+};
+
+static inline void
+socket_lock_init(struct socket *s, struct socket_lock *sl) {
+	sl->lock = &s->dw_lock;
+	sl->count = 0;
+}
+
+static inline void
+socket_lock(struct socket_lock *sl) {
+	if (sl->count == 0) {
+		spinlock_lock(sl->lock);
+	}
+	++sl->count;
+}
+
+static inline int
+socket_trylock(struct socket_lock *sl) {
+	if (sl->count == 0) {
+		if (!spinlock_trylock(sl->lock))
+			return 0;	// lock failed
+	}
+	++sl->count;
+	return 1;
+}
+
+static inline void
+socket_unlock(struct socket_lock *sl) {
+	--sl->count;
+	if (sl->count <= 0) {
+		assert(sl->count == 0);
+		spinlock_unlock(sl->lock);
+	}
+}
+
 static inline bool
 send_object_init(struct socket_server *ss, struct send_object *so, void *object, int sz) {
 	if (sz < 0) {
@@ -348,7 +386,7 @@ free_buffer(struct socket_server *ss, const void * buffer, int sz) {
 }
 
 static void
-force_close(struct socket_server *ss, struct socket *s, struct socket_message *result) {
+force_close(struct socket_server *ss, struct socket *s, struct socket_lock *l, struct socket_message *result) {
 	result->id = s->id;
 	result->ud = 0;
 	result->data = NULL;
@@ -362,7 +400,7 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 	if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
 		sp_del(ss->event_fd, s->fd);
 	}
-	spinlock_lock(&s->dw_lock);
+	socket_lock(l);
 	if (s->type != SOCKET_TYPE_BIND) {
 		if (close(s->fd) < 0) {
 			perror("close socket:");
@@ -373,7 +411,7 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 		free_buffer(ss, s->dw_buffer, s->dw_size);
 		s->dw_buffer = NULL;
 	}
-	spinlock_unlock(&s->dw_lock);
+	socket_unlock(l);
 }
 
 void 
@@ -382,8 +420,10 @@ socket_server_release(struct socket_server *ss) {
 	struct socket_message dummy;
 	for (i=0;i<MAX_SOCKET;i++) {
 		struct socket *s = &ss->slot[i];
+		struct socket_lock l;
+		socket_lock_init(s, &l);
 		if (s->type != SOCKET_TYPE_RESERVE) {
-			force_close(ss, s , &dummy);
+			force_close(ss, s, &l, &dummy);
 		}
 	}
 	close(ss->sendctrl_fd);
@@ -502,7 +542,7 @@ _failed:
 }
 
 static int
-send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list, struct socket_message *result) {
+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) {
 		struct write_buffer * tmp = list->head;
 		for (;;) {
@@ -514,7 +554,7 @@ send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list,
 				case AGAIN_WOULDBLOCK:
 					return -1;
 				}
-				force_close(ss,s, result);
+				force_close(ss,s,l,result);
 				return SOCKET_CLOSE;
 			}
 			s->wb_size -= sz;
@@ -593,9 +633,9 @@ send_list_udp(struct socket_server *ss, struct socket *s, struct wb_list *list,
 }
 
 static int
-send_list(struct socket_server *ss, struct socket *s, struct wb_list *list, struct socket_message *result) {
+send_list(struct socket_server *ss, struct socket *s, struct wb_list *list, struct socket_lock *l, struct socket_message *result) {
 	if (s->protocol == PROTOCOL_TCP) {
-		return send_list_tcp(ss, s, list, result);
+		return send_list_tcp(ss, s, list, l, result);
 	} else {
 		return send_list_udp(ss, s, list, result);
 	}
@@ -641,16 +681,16 @@ send_buffer_empty(struct socket *s) {
 	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) {
+send_buffer_(struct socket_server *ss, struct socket *s, struct socket_lock *l, struct socket_message *result) {
 	assert(!list_uncomplete(&s->low));
 	// step 1
-	if (send_list(ss,s,&s->high,result) == SOCKET_CLOSE) {
+	if (send_list(ss,s,&s->high,l,result) == SOCKET_CLOSE) {
 		return SOCKET_CLOSE;
 	}
 	if (s->high.head == NULL) {
 		// step 2
 		if (s->low.head != NULL) {
-			if (send_list(ss,s,&s->low,result) == SOCKET_CLOSE) {
+			if (send_list(ss,s,&s->low,l,result) == SOCKET_CLOSE) {
 				return SOCKET_CLOSE;
 			}
 			// step 3
@@ -664,7 +704,7 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_message *
 		sp_write(ss->event_fd, s->fd, s, false);			
 
 		if (s->type == SOCKET_TYPE_HALFCLOSE) {
-				force_close(ss, s, result);
+				force_close(ss, s, l, result);
 				return SOCKET_CLOSE;
 		}
 		if(s->warn_size > 0){
@@ -681,8 +721,8 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_message *
 }
 
 static int
-send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *result) {
-	if (!spinlock_trylock(&s->dw_lock))
+send_buffer(struct socket_server *ss, struct socket *s, struct socket_lock *l, struct socket_message *result) {
+	if (!socket_trylock(l))
 		return -1;	// blocked by direct write, send later.
 	if (s->dw_buffer) {
 		// add direct write buffer before high.head
@@ -702,8 +742,8 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 		}
 		s->dw_buffer = NULL;
 	}
-	int r = send_buffer_(ss,s,result);
-	spinlock_unlock(&s->dw_lock);
+	int r = send_buffer_(ss,s,l,result);
+	socket_unlock(l);
 
 	return r;
 }
@@ -849,14 +889,16 @@ close_socket(struct socket_server *ss, struct request_close *request, struct soc
 		result->data = NULL;
 		return SOCKET_CLOSE;
 	}
+	struct socket_lock l;
+	socket_lock_init(s, &l);
 	if (!send_buffer_empty(s)) { 
-		int type = send_buffer(ss,s,result);
+		int type = send_buffer(ss,s,&l,result);
 		// 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)) {
-		force_close(ss,s,result);
+		force_close(ss,s,&l,result);
 		result->id = id;
 		result->opaque = request->opaque;
 		return SOCKET_CLOSE;
@@ -895,9 +937,11 @@ start_socket(struct socket_server *ss, struct request_start *request, struct soc
 		result->data = "invalid socket";
 		return SOCKET_ERR;
 	}
+	struct socket_lock l;
+	socket_lock_init(s, &l);
 	if (s->type == SOCKET_TYPE_PACCEPT || s->type == SOCKET_TYPE_PLISTEN) {
 		if (sp_add(ss->event_fd, s->fd, s)) {
-			force_close(ss, s, result);
+			force_close(ss, s, &l, result);
 			result->data = strerror(errno);
 			return SOCKET_ERR;
 		}
@@ -1056,7 +1100,7 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
 
 // return -1 (ignore) when error
 static int
-forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_message * result) {
+forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_lock *l, struct socket_message * result) {
 	int sz = s->p.size;
 	char * buffer = MALLOC(sz);
 	int n = (int)read(s->fd, buffer, sz);
@@ -1070,7 +1114,7 @@ forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_me
 			break;
 		default:
 			// close when error
-			force_close(ss, s, result);
+			force_close(ss, s, l, result);
 			result->data = strerror(errno);
 			return SOCKET_ERR;
 		}
@@ -1078,7 +1122,7 @@ forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_me
 	}
 	if (n==0) {
 		FREE(buffer);
-		force_close(ss, s, result);
+		force_close(ss, s, l, result);
 		return SOCKET_CLOSE;
 	}
 
@@ -1120,7 +1164,7 @@ gen_udp_address(int protocol, union sockaddr_all *sa, uint8_t * udp_address) {
 }
 
 static int
-forward_message_udp(struct socket_server *ss, struct socket *s, struct socket_message * result) {
+forward_message_udp(struct socket_server *ss, struct socket *s, struct socket_lock *l, struct socket_message * result) {
 	union sockaddr_all sa;
 	socklen_t slen = sizeof(sa);
 	int n = recvfrom(s->fd, ss->udpbuffer,MAX_UDP_PACKAGE,0,&sa.s,&slen);
@@ -1131,7 +1175,7 @@ forward_message_udp(struct socket_server *ss, struct socket *s, struct socket_me
 			break;
 		default:
 			// close when error
-			force_close(ss, s, result);
+			force_close(ss, s, l, result);
 			result->data = strerror(errno);
 			return SOCKET_ERR;
 		}
@@ -1160,12 +1204,12 @@ forward_message_udp(struct socket_server *ss, struct socket *s, struct socket_me
 }
 
 static int
-report_connect(struct socket_server *ss, struct socket *s, struct socket_message *result) {
+report_connect(struct socket_server *ss, struct socket *s, struct socket_lock *l, struct socket_message *result) {
 	int error;
 	socklen_t len = sizeof(error);  
 	int code = getsockopt(s->fd, SOL_SOCKET, SO_ERROR, &error, &len);  
 	if (code < 0 || error) {  
-		force_close(ss,s, result);
+		force_close(ss,s,l, result);
 		if (code >= 0)
 			result->data = strerror(error);
 		else
@@ -1294,9 +1338,11 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 			// dispatch pipe message at beginning
 			continue;
 		}
+		struct socket_lock l;
+		socket_lock_init(s, &l);
 		switch (s->type) {
 		case SOCKET_TYPE_CONNECTING:
-			return report_connect(ss, s, result);
+			return report_connect(ss, s, &l, result);
 		case SOCKET_TYPE_LISTEN: {
 			int ok = report_accept(ss, s, result);
 			if (ok > 0) {
@@ -1314,9 +1360,9 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 			if (e->read) {
 				int type;
 				if (s->protocol == PROTOCOL_TCP) {
-					type = forward_message_tcp(ss, s, result);
+					type = forward_message_tcp(ss, s, &l, result);
 				} else {
-					type = forward_message_udp(ss, s, result);
+					type = forward_message_udp(ss, s, &l, result);
 					if (type == SOCKET_UDP) {
 						// try read again
 						--ss->event_index;
@@ -1333,7 +1379,7 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 				return type;
 			}
 			if (e->write) {
-				int type = send_buffer(ss, s, result);
+				int type = send_buffer(ss, s, &l, result);
 				if (type == -1)
 					break;
 				return type;
@@ -1350,7 +1396,7 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 				} else {
 					result->data = "Unknown error";
 				}
-				force_close(ss, s, result);
+				force_close(ss, s, &l, result);
 				return SOCKET_ERR;
 			}
 			break;
@@ -1418,7 +1464,10 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 		return -1;
 	}
 
-	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+	struct socket_lock l;
+	socket_lock_init(s, &l);
+
+	if (can_direct_write(s,id) && socket_trylock(&l)) {
 		// may be we can send directly, double check
 		if (can_direct_write(s,id)) {
 			// send directly
@@ -1438,7 +1487,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 			}
 			if (n == so.sz) {
 				// write done
-				spinlock_unlock(&s->dw_lock);
+				socket_unlock(&l);
 				so.free_func((void *)buffer);
 				return 0;
 			}
@@ -1449,10 +1498,10 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 
 			sp_write(ss->event_fd, s->fd, s, true);
 
-			spinlock_unlock(&s->dw_lock);
+			socket_unlock(&l);
 			return 0;
 		}
-		spinlock_unlock(&s->dw_lock);
+		socket_unlock(&l);
 	}
 
 	struct request_package request;
@@ -1682,7 +1731,10 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 		return -1;
 	}
 
-	if (can_direct_write(s,id) && spinlock_trylock(&s->dw_lock)) {
+	struct socket_lock l;
+	socket_lock_init(s, &l);
+
+	if (can_direct_write(s,id) && socket_trylock(&l)) {
 		// may be we can send directly, double check
 		if (can_direct_write(s,id)) {
 			// send directly
@@ -1693,12 +1745,12 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
 			int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
 			if (n >= 0) {
 				// sendto succ
-				spinlock_unlock(&s->dw_lock);
+				socket_unlock(&l);
 				so.free_func((void *)buffer);
 				return 0;
 			}
 		}
-		spinlock_unlock(&s->dw_lock);
+		socket_unlock(&l);
 		// let socket thread try again, udp doesn't care the order
 	}
 
@@ -1719,13 +1771,15 @@ socket_server_udp_connect(struct socket_server *ss, int id, const char * addr, i
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
 		return -1;
 	}
-	spinlock_lock(&s->dw_lock);
+	struct socket_lock l;
+	socket_lock_init(s, &l);
+	socket_lock(&l);
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
-		spinlock_unlock(&s->dw_lock);
+		socket_unlock(&l);
 		return -1;
 	}
 	ATOM_INC(&s->udpconnecting);
-	spinlock_unlock(&s->dw_lock);
+	socket_unlock(&l);
 
 	int status;
 	struct addrinfo ai_hints;