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

enable write in socket thread

Browse Source
This commit is contained in:
Cloud Wu
2020-10-13 14:21:38 +08:00
parent 77311c7243
commit 0971ce9912
1 changed files with 42 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
skynet-src/socket_server.c
View File

@@ -4,6 +4,7 @@
#include "socket_poll.h" #include "socket_poll.h"
#include "atomic.h" #include "atomic.h"
#include "spinlock.h" #include "spinlock.h"
#include "skynet.h"
#include <sys/types.h> #include <sys/types.h>
#include <sys/socket.h> #include <sys/socket.h>
@@ -373,17 +374,17 @@ socket_server_create(uint64_t time) {
int fd[2]; int fd[2];
poll_fd efd = sp_create(); poll_fd efd = sp_create();
if (sp_invalid(efd)) { if (sp_invalid(efd)) {
fprintf(stderr, "socket-server: create event pool failed.\n"); skynet_error(NULL, "socket-server: create event pool failed.");
return NULL; return NULL;
} }
if (pipe(fd)) { if (pipe(fd)) {
sp_release(efd); sp_release(efd);
fprintf(stderr, "socket-server: create socket pair failed.\n"); skynet_error(NULL, "socket-server: create socket pair failed.");
return NULL; return NULL;
} }
if (sp_add(efd, fd[0], NULL)) { if (sp_add(efd, fd[0], NULL)) {
// add recvctrl_fd to event poll // add recvctrl_fd to event poll
fprintf(stderr, "socket-server: can't add server fd to event pool.\n"); skynet_error(NULL, "socket-server: can't add server fd to event pool.");
close(fd[0]); close(fd[0]);
close(fd[1]); close(fd[1]);
sp_release(efd); sp_release(efd);
@@ -739,7 +740,7 @@ send_list_udp(struct socket_server *ss, struct socket *s, struct wb_list *list,
union sockaddr_all sa; union sockaddr_all sa;
socklen_t sasz = udp_socket_address(s, tmp->udp_address, &sa); socklen_t sasz = udp_socket_address(s, tmp->udp_address, &sa);
if (sasz == 0) { if (sasz == 0) {
fprintf(stderr, "socket-server : udp (%d) type mismatch.\n", s->id); skynet_error(NULL, "socket-server : udp (%d) type mismatch.", s->id);
drop_udp(ss, s, list, tmp); drop_udp(ss, s, list, tmp);
return -1; return -1;
} }
@@ -750,7 +751,7 @@ send_list_udp(struct socket_server *ss, struct socket *s, struct wb_list *list,
case AGAIN_WOULDBLOCK: case AGAIN_WOULDBLOCK:
return -1; return -1;
} }
fprintf(stderr, "socket-server : udp (%d) sendto error %s.\n",s->id, strerror(errno)); skynet_error(NULL, "socket-server : udp (%d) sendto error %s.",s->id, strerror(errno));
drop_udp(ss, s, list, tmp); drop_udp(ss, s, list, tmp);
return -1; return -1;
} }
@@ -931,6 +932,21 @@ append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_
s->wb_size += buf->sz; s->wb_size += buf->sz;
} }
static int
trigger_write(struct socket_server *ss, struct request_send * request, struct socket_message *result) {
int id = request->id;
struct socket * s = &ss->slot[HASH_ID(id)];
if (s->type == SOCKET_TYPE_INVALID || s->id != id)
return -1;
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;
}
return -1;
}
/* /*
When send a package , we can assign the priority : PRIORITY_HIGH or PRIORITY_LOW When send a package , we can assign the priority : PRIORITY_HIGH or PRIORITY_LOW
@@ -952,7 +968,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
return -1; return -1;
} }
if (s->type == SOCKET_TYPE_PLISTEN || s->type == SOCKET_TYPE_LISTEN) { if (s->type == SOCKET_TYPE_PLISTEN || s->type == SOCKET_TYPE_LISTEN) {
fprintf(stderr, "socket-server: write to listen fd %d.\n", id); skynet_error(NULL, "socket-server: write to listen fd %d.", id);
so.free_func((void *)request->buffer); so.free_func((void *)request->buffer);
return -1; return -1;
} }
@@ -968,7 +984,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
socklen_t sasz = udp_socket_address(s, udp_address, &sa); socklen_t sasz = udp_socket_address(s, udp_address, &sa);
if (sasz == 0) { if (sasz == 0) {
// udp type mismatch, just drop it. // udp type mismatch, just drop it.
fprintf(stderr, "socket-server: udp socket (%d) type mistach.\n", id); skynet_error(NULL, "socket-server: udp socket (%d) type mistach.", id);
so.free_func((void *)request->buffer); so.free_func((void *)request->buffer);
return -1; return -1;
} }
@@ -1154,7 +1170,7 @@ block_readpipe(int pipefd, void *buffer, int sz) {
if (n<0) { if (n<0) {
if (errno == EINTR) if (errno == EINTR)
continue; continue;
fprintf(stderr, "socket-server : read pipe error %s.\n",strerror(errno)); skynet_error(NULL, "socket-server : read pipe error %s.",strerror(errno));
return; return;
} }
// must atomic read from a pipe // must atomic read from a pipe
@@ -1285,6 +1301,8 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
result->ud = 0; result->ud = 0;
result->data = NULL; result->data = NULL;
return SOCKET_EXIT; return SOCKET_EXIT;
case 'W':
return trigger_write(ss, (struct request_send *)buffer, result);
case 'D': case 'D':
case 'P': { case 'P': {
int priority = (type == 'D') ? PRIORITY_HIGH : PRIORITY_LOW; int priority = (type == 'D') ? PRIORITY_HIGH : PRIORITY_LOW;
@@ -1306,7 +1324,7 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
add_udp_socket(ss, (struct request_udp *)buffer); add_udp_socket(ss, (struct request_udp *)buffer);
return -1; return -1;
default: default:
fprintf(stderr, "socket-server: Unknown ctrl %c.\n",type); skynet_error(NULL, "socket-server: Unknown ctrl %c.",type);
return -1; return -1;
}; };
@@ -1325,7 +1343,7 @@ forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_lo
case EINTR: case EINTR:
break; break;
case AGAIN_WOULDBLOCK: case AGAIN_WOULDBLOCK:
fprintf(stderr, "socket-server: EAGAIN capture.\n"); skynet_error(NULL, "socket-server: EAGAIN capture.");
break; break;
default: default:
// close when error // close when error
@@ -1590,7 +1608,7 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
break; break;
} }
case SOCKET_TYPE_INVALID: case SOCKET_TYPE_INVALID:
fprintf(stderr, "socket-server: invalid socket\n"); skynet_error(NULL, "socket-server: invalid socket");
break; break;
default: default:
if (e->read) { if (e->read) {
@@ -1654,7 +1672,7 @@ send_request(struct socket_server *ss, struct request_package *request, char typ
ssize_t 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)); skynet_error(NULL, "socket-server : send ctrl command error %s.", strerror(errno));
} }
continue; continue;
} }
@@ -1667,7 +1685,7 @@ static int
open_request(struct socket_server *ss, struct request_package *req, uintptr_t opaque, const char *addr, int port) { open_request(struct socket_server *ss, struct request_package *req, uintptr_t opaque, const char *addr, int port) {
int len = strlen(addr); int len = strlen(addr);
if (len + sizeof(req->u.open) >= 256) { if (len + sizeof(req->u.open) >= 256) {
fprintf(stderr, "socket-server : Invalid addr %s.\n",addr); skynet_error(NULL, "socket-server : Invalid addr %s.",addr);
return -1; return -1;
} }
int id = reserve_id(ss); int id = reserve_id(ss);
@@ -1723,7 +1741,7 @@ socket_server_send(struct socket_server *ss, struct socket_sendbuffer *buf) {
union sockaddr_all sa; union sockaddr_all sa;
socklen_t sasz = udp_socket_address(s, s->p.udp_address, &sa); socklen_t sasz = udp_socket_address(s, s->p.udp_address, &sa);
if (sasz == 0) { if (sasz == 0) {
fprintf(stderr, "socket-server : set udp (%d) address first.\n", id); skynet_error(NULL, "socket-server : set udp (%d) address first.", id);
socket_unlock(&l); socket_unlock(&l);
so.free_func((void *)buf->buffer); so.free_func((void *)buf->buffer);
return -1; return -1;
@@ -1745,14 +1763,18 @@ 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;
int err = enable_write(ss, s, true);
socket_unlock(&l); socket_unlock(&l);
if (err) { inc_sending_ref(s, id);
fprintf(stderr, "socket-server : enable write (%d) failed.\n", id);
return -1; struct request_package request;
} request.u.send.id = id;
request.u.send.sz = 0;
request.u.send.buffer = NULL;
// let socket thread enable write event
send_request(ss, &request, 'W', sizeof(request.u.send));
return 0; return 0;
} }
socket_unlock(&l); socket_unlock(&l);