lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-24 12:20:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

direct write if send queue is empty

Browse Source
This commit is contained in:
Cloud Wu
2017-06-08 20:02:39 +08:00
parent 48e031f2fb
commit c53da829d3
1 changed files with 117 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

158
skynet-src/socket_server.c
View File

@@ -3,6 +3,7 @@
#include "socket_server.h" #include "socket_server.h"
#include "socket_poll.h" #include "socket_poll.h"
#include "atomic.h" #include "atomic.h"
#include "spinlock.h"
#include <sys/types.h> #include <sys/types.h>
#include <sys/socket.h> #include <sys/socket.h>
@@ -86,6 +87,10 @@ struct socket {
int size; int size;
uint8_t udp_address[UDP_ADDRESS_SIZE]; uint8_t udp_address[UDP_ADDRESS_SIZE];
} p; } p;
struct spinlock dw_lock;
int dw_offset;
const void * dw_buffer;
size_t dw_size;
}; };
struct socket_server { struct socket_server {
@@ -331,6 +336,13 @@ free_wb_list(struct socket_server *ss, struct wb_list *list) {
list->tail = NULL; 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 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_message *result) {
result->id = s->id; 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) { if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
sp_del(ss->event_fd, s->fd); sp_del(ss->event_fd, s->fd);
} }
spinlock_lock(&s->dw_lock);
if (s->type != SOCKET_TYPE_BIND) { if (s->type != SOCKET_TYPE_BIND) {
if (close(s->fd) < 0) { if (close(s->fd) < 0) {
perror("close socket:"); perror("close socket:");
} }
} }
s->type = SOCKET_TYPE_INVALID; 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 void
@@ -397,6 +415,9 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
s->warn_size = 0; s->warn_size = 0;
check_wb_list(&s->high); check_wb_list(&s->high);
check_wb_list(&s->low); check_wb_list(&s->low);
spinlock_init(&s->dw_lock);
s->dw_buffer = NULL;
s->dw_size = 0;
return s; return s;
} }
@@ -481,7 +502,7 @@ send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list,
while (list->head) { while (list->head) {
struct write_buffer * tmp = list->head; struct write_buffer * tmp = list->head;
for (;;) { for (;;) {
int sz = write(s->fd, tmp->ptr, tmp->sz); ssize_t sz = write(s->fd, tmp->ptr, tmp->sz);
if (sz < 0) { if (sz < 0) {
switch(errno) { switch(errno) {
case EINTR: 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) 4. If two lists are both empty, turn off the event. (call check_close)
*/ */
static int 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)); assert(!list_uncomplete(&s->low));
// step 1 // step 1
if (send_list(ss,s,&s->high,result) == SOCKET_CLOSE) { 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; 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 * 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 write_buffer * buf = MALLOC(size);
struct send_object so; struct send_object so;
buf->userobject = send_object_init(ss, &so, request->buffer, request->sz); buf->userobject = send_object_init(ss, &so, request->buffer, request->sz);
buf->ptr = (char*)so.buffer+n; buf->ptr = (char*)so.buffer;
buf->sz = so.sz - n; buf->sz = so.sz;
buf->buffer = request->buffer; buf->buffer = request->buffer;
buf->next = NULL; buf->next = NULL;
if (s->head == 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 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]) { 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 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); memcpy(buf->udp_address, udp_address, UDP_ADDRESS_SIZE);
s->wb_size += buf->sz; s->wb_size += buf->sz;
} }
static inline void static inline void
append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request, int 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, n); struct write_buffer *buf = append_sendbuffer_(ss, &s->high, request, SIZEOF_TCPBUFFER);
s->wb_size += buf->sz; s->wb_size += buf->sz;
} }
static inline void static inline void
append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_send * request) { 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; 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 (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) {
if (s->protocol == PROTOCOL_TCP) { if (s->protocol == PROTOCOL_TCP) {
int n = write(s->fd, so.buffer, so.sz); append_sendbuffer(ss, s, request); // add to high priority list, even priority == PRIORITY_LOW
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
} else { } else {
// udp // udp
if (udp_address == NULL) { if (udp_address == NULL) {
@@ -762,7 +791,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
if (priority == PRIORITY_LOW) { if (priority == PRIORITY_LOW) {
append_sendbuffer_low(ss, s, request); append_sendbuffer_low(ss, s, request);
} else { } else {
append_sendbuffer(ss, s, request, 0); append_sendbuffer(ss, s, request);
} }
} else { } else {
if (udp_address == NULL) { 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[6] = (uint8_t)type;
request->header[7] = (uint8_t)len; request->header[7] = (uint8_t)len;
for (;;) { 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 (n<0) {
if (errno != EINTR) { if (errno != EINTR) {
fprintf(stderr, "socket-server : send ctrl command error %s.\n", strerror(errno)); 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; return request.u.open.id;
} }
static void static inline int
free_buffer(struct socket_server *ss, const void * buffer, int sz) { can_direct_write(struct socket *s, int id) {
struct send_object so; return s->id == id && send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED && s->dw_buffer == NULL;
send_object_init(ss, &so, (void *)buffer, sz);
so.free_func((void *)buffer);
} }
// return -1 when error, 0 when success // 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; 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; struct request_package request;
request.u.send.id = id; request.u.send.id = id;
request.u.send.sz = sz; 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; 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; const uint8_t *udp_address = (const uint8_t *)addr;
int addrsz; int addrsz;
switch (udp_address[0]) { switch (udp_address[0]) {
@@ -1616,7 +1667,32 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
return -1; 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); send_request(ss, &request, 'A', sizeof(request.u.send_udp.send)+addrsz);
return 0; return 0;