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

add api skynet_socket_pause

Browse Source
This commit is contained in:
Cloud Wu
2020-08-10 10:57:59 +08:00
parent c35c8f173b
commit a4dc6b9094
6 changed files with 64 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
lualib-src/lua-socket.c
View File

@@ -611,6 +611,14 @@ lstart(lua_State *L) {
return 0; return 0;
} }
static int
lpause(lua_State *L) {
struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
int id = luaL_checkinteger(L, 1);
skynet_socket_pause(ctx,id);
return 0;
}
static int static int
lnodelay(lua_State *L) { lnodelay(lua_State *L) {
struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
@@ -794,6 +802,7 @@ luaopen_skynet_socketdriver(lua_State *L) {
{ "lsend", lsendlow }, { "lsend", lsendlow },
{ "bind", lbind }, { "bind", lbind },
{ "start", lstart }, { "start", lstart },
{ "pause", lpause },
{ "nodelay", lnodelay }, { "nodelay", lnodelay },
{ "udp", ludp }, { "udp", ludp },
{ "udp_connect", ludp_connect }, { "udp_connect", ludp_connect },

2
lualib/skynet/socket.lua
View File

@@ -113,7 +113,7 @@ socket_message[5] = function(id, _, err)
return return
end end
if s.callback then if s.callback then
skynet.error("socket: accpet error:", err) skynet.error("socket: accpet error:", err)
return return
end end
if s.connected then if s.connected then

7
skynet-src/skynet_socket.c
View File

@@ -162,6 +162,13 @@ skynet_socket_start(struct skynet_context *ctx, int id) {
socket_server_start(SOCKET_SERVER, source, id); socket_server_start(SOCKET_SERVER, source, id);
} }
void
skynet_socket_pause(struct skynet_context *ctx, int id) {
uint32_t source = skynet_context_handle(ctx);
socket_server_pause(SOCKET_SERVER, source, id);
}
void void
skynet_socket_nodelay(struct skynet_context *ctx, int id) { skynet_socket_nodelay(struct skynet_context *ctx, int id) {
socket_server_nodelay(SOCKET_SERVER, id); socket_server_nodelay(SOCKET_SERVER, id);

1
skynet-src/skynet_socket.h
View File

@@ -35,6 +35,7 @@ int skynet_socket_bind(struct skynet_context *ctx, int fd);
void skynet_socket_close(struct skynet_context *ctx, int id); void skynet_socket_close(struct skynet_context *ctx, int id);
void skynet_socket_shutdown(struct skynet_context *ctx, int id); void skynet_socket_shutdown(struct skynet_context *ctx, int id);
void skynet_socket_start(struct skynet_context *ctx, int id); void skynet_socket_start(struct skynet_context *ctx, int id);
void skynet_socket_pause(struct skynet_context *ctx, int id);
void skynet_socket_nodelay(struct skynet_context *ctx, int id); void skynet_socket_nodelay(struct skynet_context *ctx, int id);
int skynet_socket_udp(struct skynet_context *ctx, const char * addr, int port); int skynet_socket_udp(struct skynet_context *ctx, const char * addr, int port);

56
skynet-src/socket_server.c
View File

@@ -37,6 +37,10 @@
#define PRIORITY_HIGH 0 #define PRIORITY_HIGH 0
#define PRIORITY_LOW 1 #define PRIORITY_LOW 1
#define READING_PAUSE 0
#define READING_RESUME 1
#define READING_CLOSE 2
#define HASH_ID(id) (((unsigned)id) % MAX_SOCKET) #define HASH_ID(id) (((unsigned)id) % MAX_SOCKET)
#define ID_TAG16(id) ((id>>MAX_SOCKET_P) & 0xffff) #define ID_TAG16(id) ((id>>MAX_SOCKET_P) & 0xffff)
@@ -95,7 +99,8 @@ struct socket {
int id; int id;
uint8_t protocol; uint8_t protocol;
uint8_t type; uint8_t type;
uint16_t udpconnecting; uint8_t reading;
int udpconnecting;
int64_t warn_size; int64_t warn_size;
union { union {
int size; int size;
@@ -166,7 +171,7 @@ struct request_bind {
uintptr_t opaque; uintptr_t opaque;
}; };
struct request_start { struct request_resumepause {
int id; int id;
uintptr_t opaque; uintptr_t opaque;
}; };
@@ -212,7 +217,7 @@ struct request_package {
struct request_close close; struct request_close close;
struct request_listen listen; struct request_listen listen;
struct request_bind bind; struct request_bind bind;
struct request_start start; struct request_resumepause resumepause;
struct request_setopt setopt; struct request_setopt setopt;
struct request_udp udp; struct request_udp udp;
struct request_setudp set_udp; struct request_setudp set_udp;
@@ -477,9 +482,10 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_lock *l, s
assert(s->type != SOCKET_TYPE_RESERVE); assert(s->type != SOCKET_TYPE_RESERVE);
free_wb_list(ss,&s->high); free_wb_list(ss,&s->high);
free_wb_list(ss,&s->low); free_wb_list(ss,&s->low);
if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) { if (s->reading == READING_RESUME) {
sp_del(ss->event_fd, s->fd); sp_del(ss->event_fd, s->fd);
} }
s->reading = READING_CLOSE;
socket_lock(l); socket_lock(l);
if (s->type != SOCKET_TYPE_BIND) { if (s->type != SOCKET_TYPE_BIND) {
if (close(s->fd) < 0) { if (close(s->fd) < 0) {
@@ -538,6 +544,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
s->id = id; s->id = id;
s->fd = fd; s->fd = fd;
s->reading = READING_PAUSE;
s->sending = ID_TAG16(id) << 16 | 0; s->sending = ID_TAG16(id) << 16 | 0;
s->protocol = protocol; s->protocol = protocol;
s->p.size = MIN_READ_BUFFER; s->p.size = MIN_READ_BUFFER;
@@ -1046,7 +1053,7 @@ bind_socket(struct socket_server *ss, struct request_bind *request, struct socke
} }
static int static int
start_socket(struct socket_server *ss, struct request_start *request, struct socket_message *result) { resume_socket(struct socket_server *ss, struct request_resumepause *request, struct socket_message *result) {
int id = request->id; int id = request->id;
result->id = id; result->id = id;
result->opaque = request->opaque; result->opaque = request->opaque;
@@ -1059,12 +1066,15 @@ start_socket(struct socket_server *ss, struct request_start *request, struct soc
} }
struct socket_lock l; struct socket_lock l;
socket_lock_init(s, &l); socket_lock_init(s, &l);
if (s->type == SOCKET_TYPE_PACCEPT || s->type == SOCKET_TYPE_PLISTEN) { if (s->reading == READING_PAUSE) {
if (sp_add(ss->event_fd, s->fd, s)) { if (sp_add(ss->event_fd, s->fd, s)) {
s->reading = READING_RESUME;
force_close(ss, s, &l, result); force_close(ss, s, &l, result);
result->data = strerror(errno); result->data = strerror(errno);
return SOCKET_ERR; return SOCKET_ERR;
} }
}
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;
result->data = "start"; result->data = "start";
@@ -1079,6 +1089,19 @@ start_socket(struct socket_server *ss, struct request_start *request, struct soc
return -1; return -1;
} }
static void
pause_socket(struct socket_server *ss, struct request_resumepause *request) {
int id = request->id;
struct socket *s = &ss->slot[HASH_ID(id)];
if (s->type == SOCKET_TYPE_INVALID || s->id !=id) {
return;
}
if (s->reading == READING_RESUME) {
sp_del(ss->event_fd, s->fd);
s->reading = READING_PAUSE;
}
}
static void static void
setopt_socket(struct socket_server *ss, struct request_setopt *request) { setopt_socket(struct socket_server *ss, struct request_setopt *request) {
int id = request->id; int id = request->id;
@@ -1210,8 +1233,11 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
block_readpipe(fd, buffer, len); block_readpipe(fd, buffer, len);
// ctrl command only exist in local fd, so don't worry about endian. // ctrl command only exist in local fd, so don't worry about endian.
switch (type) { switch (type) {
case 'R':
return resume_socket(ss,(struct request_resumepause *)buffer, result);
case 'S': case 'S':
return start_socket(ss,(struct request_start *)buffer, result); pause_socket(ss,(struct request_resumepause *)buffer);
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':
@@ -1841,12 +1867,20 @@ socket_server_bind(struct socket_server *ss, uintptr_t opaque, int fd) {
return id; return id;
} }
void void
socket_server_start(struct socket_server *ss, uintptr_t opaque, int id) { socket_server_start(struct socket_server *ss, uintptr_t opaque, int id) {
struct request_package request; struct request_package request;
request.u.start.id = id; request.u.resumepause.id = id;
request.u.start.opaque = opaque; request.u.resumepause.opaque = opaque;
send_request(ss, &request, 'S', sizeof(request.u.start)); send_request(ss, &request, 'R', sizeof(request.u.resumepause));
}
void
socket_server_pause(struct socket_server *ss, uintptr_t opaque, int id) {
struct request_package request;
request.u.resumepause.id = id;
request.u.resumepause.opaque = opaque;
send_request(ss, &request, 'S', sizeof(request.u.resumepause));
} }
void void

1
skynet-src/socket_server.h
View File

@@ -32,6 +32,7 @@ void socket_server_exit(struct socket_server *);
void socket_server_close(struct socket_server *, uintptr_t opaque, int id); void socket_server_close(struct socket_server *, uintptr_t opaque, int id);
void socket_server_shutdown(struct socket_server *, uintptr_t opaque, int id); void socket_server_shutdown(struct socket_server *, uintptr_t opaque, int id);
void socket_server_start(struct socket_server *, uintptr_t opaque, int id); void socket_server_start(struct socket_server *, uintptr_t opaque, int id);
void socket_server_pause(struct socket_server *, uintptr_t opaque, int id);
// return -1 when error // return -1 when error
int socket_server_send(struct socket_server *, struct socket_sendbuffer *buffer); int socket_server_send(struct socket_server *, struct socket_sendbuffer *buffer);