lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-22 02:53:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

enable error handle

Browse Source
This commit is contained in:
Cloud Wu
2020-10-13 12:04:08 +08:00
parent e75d033066
commit 77311c7243
4 changed files with 69 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
skynet-src/socket_epoll.h
View File

@@ -41,12 +41,15 @@ sp_del(int efd, int sock) {
epoll_ctl(efd, EPOLL_CTL_DEL, sock , NULL); epoll_ctl(efd, EPOLL_CTL_DEL, sock , NULL);
} }
static void static int
sp_enable(int efd, int sock, void *ud, bool read_enable, bool write_enable) { sp_enable(int efd, int sock, void *ud, bool read_enable, bool write_enable) {
struct epoll_event ev; struct epoll_event ev;
ev.events = (read_enable ? EPOLLIN : 0) | (write_enable ? EPOLLOUT : 0); ev.events = (read_enable ? EPOLLIN : 0) | (write_enable ? EPOLLOUT : 0);
ev.data.ptr = ud; ev.data.ptr = ud;
epoll_ctl(efd, EPOLL_CTL_MOD, sock, &ev); if (epoll_ctl(efd, EPOLL_CTL_MOD, sock, &ev) == -1) {
return 1;
}
return 0;
} }
static int static int

8
skynet-src/socket_kqueue.h
View File

@@ -55,17 +55,19 @@ sp_add(int kfd, int sock, void *ud) {
return 0; return 0;
} }
static void static int
sp_enable(int kfd, int sock, void *ud, bool read_enable, bool write_enable) { sp_enable(int kfd, int sock, void *ud, bool read_enable, bool write_enable) {
int ret = 0;
struct kevent ke; struct kevent ke;
EV_SET(&ke, sock, EVFILT_READ, read_enable ? EV_ENABLE : EV_DISABLE, 0, 0, ud); EV_SET(&ke, sock, EVFILT_READ, read_enable ? EV_ENABLE : EV_DISABLE, 0, 0, ud);
if (kevent(kfd, &ke, 1, NULL, 0, NULL) == -1 || ke.flags & EV_ERROR) { if (kevent(kfd, &ke, 1, NULL, 0, NULL) == -1 || ke.flags & EV_ERROR) {
// todo: check error ret |= 1;
} }
EV_SET(&ke, sock, EVFILT_WRITE, write_enable ? EV_ENABLE : EV_DISABLE, 0, 0, ud); EV_SET(&ke, sock, EVFILT_WRITE, write_enable ? EV_ENABLE : EV_DISABLE, 0, 0, ud);
if (kevent(kfd, &ke, 1, NULL, 0, NULL) == -1 || ke.flags & EV_ERROR) { if (kevent(kfd, &ke, 1, NULL, 0, NULL) == -1 || ke.flags & EV_ERROR) {
// todo: check error ret |= 1;
} }
return ret;
} }
static int static int

2
skynet-src/socket_poll.h
View File

@@ -18,7 +18,7 @@ static poll_fd sp_create();
static void sp_release(poll_fd fd); static void sp_release(poll_fd fd);
static int sp_add(poll_fd fd, int sock, void *ud); static int sp_add(poll_fd fd, int sock, void *ud);
static void sp_del(poll_fd fd, int sock); static void sp_del(poll_fd fd, int sock);
static void sp_enable(poll_fd, int sock, void *ud, bool read_enable, bool write_enable); static int sp_enable(poll_fd, int sock, void *ud, bool read_enable, bool write_enable);
static int sp_wait(poll_fd, struct event *e, int max); static int sp_wait(poll_fd, struct event *e, int max);
static void sp_nonblocking(int sock); static void sp_nonblocking(int sock);

75
skynet-src/socket_server.c
View File

@@ -524,20 +524,22 @@ check_wb_list(struct wb_list *s) {
assert(s->tail == NULL); assert(s->tail == NULL);
} }
static inline void static inline int
enable_write(struct socket_server *ss, struct socket *s, bool enable) { enable_write(struct socket_server *ss, struct socket *s, bool enable) {
if (s->writing != enable) { if (s->writing != enable) {
s->writing = enable; s->writing = enable;
sp_enable(ss->event_fd, s->fd, s, s->reading, enable); return sp_enable(ss->event_fd, s->fd, s, s->reading, enable);
} }
return 0;
} }
static inline void static inline int
enable_read(struct socket_server *ss, struct socket *s, bool enable) { enable_read(struct socket_server *ss, struct socket *s, bool enable) {
if (s->reading != enable) { if (s->reading != enable) {
s->reading = enable; s->reading = enable;
sp_enable(ss->event_fd, s->fd, s, enable, s->writing); return sp_enable(ss->event_fd, s->fd, s, enable, s->writing);
} }
return 0;
} }
static struct socket * static struct socket *
@@ -565,7 +567,10 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
s->dw_buffer = NULL; s->dw_buffer = NULL;
s->dw_size = 0; s->dw_size = 0;
memset(&s->stat, 0, sizeof(s->stat)); memset(&s->stat, 0, sizeof(s->stat));
enable_read(ss, s, reading); if (enable_read(ss, s, reading)) {
s->type = SOCKET_TYPE_INVALID;
return NULL;
}
return s; return s;
} }
@@ -646,7 +651,10 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock
return SOCKET_OPEN; return SOCKET_OPEN;
} else { } else {
ns->type = SOCKET_TYPE_CONNECTING; ns->type = SOCKET_TYPE_CONNECTING;
enable_write(ss, ns, true); if (enable_write(ss, ns, true)) {
result->data = "enable write failed";
goto _failed;
}
} }
freeaddrinfo( ai_list ); freeaddrinfo( ai_list );
@@ -827,12 +835,21 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_lock *l,
} }
// step 4 // step 4
assert(send_buffer_empty(s) && s->wb_size == 0); assert(send_buffer_empty(s) && s->wb_size == 0);
enable_write(ss, s, false); int err = enable_write(ss, s, false);
if (s->type == SOCKET_TYPE_HALFCLOSE) { if (s->type == SOCKET_TYPE_HALFCLOSE) {
force_close(ss, s, l, result); force_close(ss, s, l, result);
return SOCKET_CLOSE; return SOCKET_CLOSE;
} }
if (err) {
result->opaque = s->opaque;
result->id = s->id;
result->ud = 0;
result->data = "disable write failed";
return SOCKET_ERR;
}
if(s->warn_size > 0){ if(s->warn_size > 0){
s->warn_size = 0; s->warn_size = 0;
result->opaque = s->opaque; result->opaque = s->opaque;
@@ -964,7 +981,13 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
return -1; return -1;
} }
} }
enable_write(ss, s, true); if (enable_write(ss, s, true)) {
result->opaque = s->opaque;
result->id = s->id;
result->ud = 0;
result->data = "enable write failed";
return SOCKET_ERR;
}
} else { } else {
if (s->protocol == PROTOCOL_TCP) { if (s->protocol == PROTOCOL_TCP) {
if (priority == PRIORITY_LOW) { if (priority == PRIORITY_LOW) {
@@ -1077,7 +1100,10 @@ resume_socket(struct socket_server *ss, struct request_resumepause *request, str
} }
struct socket_lock l; struct socket_lock l;
socket_lock_init(s, &l); socket_lock_init(s, &l);
enable_read(ss, s, true); if (enable_read(ss, s, true)) {
result->data = "enable read failed";
return SOCKET_ERR;
}
if (s->type == SOCKET_TYPE_PACCEPT || s->type == SOCKET_TYPE_PLISTEN) { if (s->type == SOCKET_TYPE_PACCEPT || s->type == SOCKET_TYPE_PLISTEN) {
s->type = (s->type == SOCKET_TYPE_PACCEPT) ? SOCKET_TYPE_CONNECTED : SOCKET_TYPE_LISTEN; s->type = (s->type == SOCKET_TYPE_PACCEPT) ? SOCKET_TYPE_CONNECTED : SOCKET_TYPE_LISTEN;
s->opaque = request->opaque; s->opaque = request->opaque;
@@ -1093,14 +1119,21 @@ resume_socket(struct socket_server *ss, struct request_resumepause *request, str
return -1; return -1;
} }
static void static int
pause_socket(struct socket_server *ss, struct request_resumepause *request) { pause_socket(struct socket_server *ss, struct request_resumepause *request, struct socket_message *result) {
int id = request->id; int id = request->id;
struct socket *s = &ss->slot[HASH_ID(id)]; struct socket *s = &ss->slot[HASH_ID(id)];
if (s->type == SOCKET_TYPE_INVALID || s->id !=id) { if (s->type == SOCKET_TYPE_INVALID || s->id !=id) {
return; return -1;
} }
enable_read(ss, s, false); if (enable_read(ss, s, false)) {
result->id = id;
result->opaque = request->opaque;
result->ud = 0;
result->data = "enable read failed";
return SOCKET_ERR;
}
return -1;
} }
static void static void
@@ -1237,8 +1270,7 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
case 'R': case 'R':
return resume_socket(ss,(struct request_resumepause *)buffer, result); return resume_socket(ss,(struct request_resumepause *)buffer, result);
case 'S': case 'S':
pause_socket(ss,(struct request_resumepause *)buffer); return pause_socket(ss,(struct request_resumepause *)buffer, result);
return -1;
case 'B': case 'B':
return bind_socket(ss,(struct request_bind *)buffer, result); return bind_socket(ss,(struct request_bind *)buffer, result);
case 'L': case 'L':
@@ -1408,7 +1440,11 @@ report_connect(struct socket_server *ss, struct socket *s, struct socket_lock *l
result->id = s->id; result->id = s->id;
result->ud = 0; result->ud = 0;
if (nomore_sending_data(s)) { if (nomore_sending_data(s)) {
enable_write(ss, s, false); if (enable_write(ss, s, false)) {
force_close(ss,s,l, result);
result->data = "disable write failed";
return SOCKET_ERR;
}
} }
union sockaddr_all u; union sockaddr_all u;
socklen_t slen = sizeof(u); socklen_t slen = sizeof(u);
@@ -1709,9 +1745,14 @@ socket_server_send(struct socket_server *ss, struct socket_sendbuffer *buf) {
s->dw_buffer = clone_buffer(buf, &s->dw_size); s->dw_buffer = clone_buffer(buf, &s->dw_size);
s->dw_offset = n; s->dw_offset = n;
enable_write(ss, s, true); int err = enable_write(ss, s, true);
socket_unlock(&l); socket_unlock(&l);
if (err) {
fprintf(stderr, "socket-server : enable write (%d) failed.\n", id);
return -1;
}
return 0; return 0;
} }
socket_unlock(&l); socket_unlock(&l);