diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index 39b8107f..bd5cc5b9 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -567,8 +567,9 @@ lsendlow(lua_State *L) { int id = luaL_checkinteger(L, 1); int sz = 0; void *buffer = get_buffer(L, 2, &sz); - skynet_socket_send_lowpriority(ctx, id, buffer, sz); - return 0; + int err = skynet_socket_send_lowpriority(ctx, id, buffer, sz); + lua_pushboolean(L, !err); + return 1; } static int diff --git a/lualib/snax/gateserver.lua b/lualib/snax/gateserver.lua index bf62e2ce..f22176bc 100644 --- a/lualib/snax/gateserver.lua +++ b/lualib/snax/gateserver.lua @@ -113,7 +113,7 @@ function gateserver.start(handler) function MSG.error(fd, msg) if fd == socket then socketdriver.close(fd) - skynet.error(msg) + skynet.error("gateserver close listen socket, accpet error:",msg) else if handler.error then handler.error(fd, msg) diff --git a/lualib/socket.lua b/lualib/socket.lua index d7741b54..f9644313 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -138,19 +138,17 @@ end local function default_warning(id, size) local s = socket_pool[id] - local last = s.warningsize or 0 - if last + 64 < size then -- if size increase 64K - s.warningsize = size - skynet.error(string.format("WARNING: %d K bytes need to send out (fd = %d)", size, id)) - end - s.warningsize = size + if not s then + return + end + skynet.error(string.format("WARNING: %d K bytes need to send out (fd = %d)", size, id)) end -- SKYNET_SOCKET_TYPE_WARNING socket_message[7] = function(id, size) local s = socket_pool[id] if s then - local warning = s.warning or default_warning + local warning = s.on_warning or default_warning warning(id, size) end end @@ -442,7 +440,7 @@ socket.udp_address = assert(driver.udp_address) function socket.warning(id, callback) local obj = socket_pool[id] assert(obj) - obj.warning = callback + obj.on_warning = callback end return socket diff --git a/lualib/socketchannel.lua b/lualib/socketchannel.lua index 89290ec0..fc3c246e 100644 --- a/lualib/socketchannel.lua +++ b/lualib/socketchannel.lua @@ -376,6 +376,12 @@ end local socket_write = socket.write local socket_lwrite = socket.lwrite +local function sock_err(self) + close_channel_socket(self) + wakeup_all(self) + error(socket_error) +end + function channel:request(request, response, padding) assert(block_connect(self, true)) -- connect once local fd = self.__sock[1] @@ -383,16 +389,18 @@ function channel:request(request, response, padding) if padding then -- padding may be a table, to support multi part request -- multi part request use low priority socket write - -- socket_lwrite returns nothing - socket_lwrite(fd , request) + -- now socket_lwrite returns as socket_write + if not socket_lwrite(fd , request) then + sock_err(self) + end for _,v in ipairs(padding) do - socket_lwrite(fd, v) + if not socket_lwrite(fd, v) then + sock_err(self) + end end else if not socket_write(fd , request) then - close_channel_socket(self) - wakeup_all(self) - error(socket_error) + sock_err(self) end end diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index af793d35..ef255060 100644 --- a/skynet-src/skynet_socket.c +++ b/skynet-src/skynet_socket.c @@ -98,6 +98,9 @@ skynet_socket_poll() { case SOCKET_UDP: forward_message(SKYNET_SOCKET_TYPE_UDP, false, &result); break; + case SOCKET_WARNING: + forward_message(SKYNET_SOCKET_TYPE_WARNING, false, &result); + break; default: skynet_error(NULL, "Unknown socket message type %d.",type); return -1; @@ -108,31 +111,14 @@ skynet_socket_poll() { return 1; } -static int -check_wsz(struct skynet_context *ctx, int id, void *buffer, int64_t wsz) { - if (wsz < 0) { - return -1; - } else if (wsz > 1024 * 1024) { - struct skynet_socket_message tmp; - tmp.type = SKYNET_SOCKET_TYPE_WARNING; - tmp.id = id; - tmp.ud = (int)(wsz / 1024); - tmp.buffer = NULL; - skynet_send(ctx, 0, skynet_context_handle(ctx), PTYPE_SOCKET, 0 , &tmp, sizeof(tmp)); -// skynet_error(ctx, "%d Mb bytes on socket %d need to send out", (int)(wsz / (1024 * 1024)), id); - } - return 0; +int +skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) { + return socket_server_send(SOCKET_SERVER, id, buffer, sz); } int -skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) { - int64_t wsz = socket_server_send(SOCKET_SERVER, id, buffer, sz); - return check_wsz(ctx, id, buffer, wsz); -} - -void skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz) { - socket_server_send_lowpriority(SOCKET_SERVER, id, buffer, sz); + return socket_server_send_lowpriority(SOCKET_SERVER, id, buffer, sz); } int @@ -189,8 +175,7 @@ skynet_socket_udp_connect(struct skynet_context *ctx, int id, const char * addr, int skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz) { - int64_t wsz = socket_server_udp_send(SOCKET_SERVER, id, (const struct socket_udp_address *)address, buffer, sz); - return check_wsz(ctx, id, (void *)buffer, wsz); + return socket_server_udp_send(SOCKET_SERVER, id, (const struct socket_udp_address *)address, buffer, sz); } const char * diff --git a/skynet-src/skynet_socket.h b/skynet-src/skynet_socket.h index 55b98595..21dfcef4 100644 --- a/skynet-src/skynet_socket.h +++ b/skynet-src/skynet_socket.h @@ -24,7 +24,7 @@ void skynet_socket_free(); int skynet_socket_poll(); int skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz); -void skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz); +int skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz); int skynet_socket_listen(struct skynet_context *ctx, const char *host, int port, int backlog); int skynet_socket_connect(struct skynet_context *ctx, const char *host, int port); int skynet_socket_bind(struct skynet_context *ctx, int fd); diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index ea90aab4..4f381280 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -53,6 +53,8 @@ #define AGAIN_WOULDBLOCK EAGAIN #endif +#define WARNING_SIZE (1024*1024) + struct write_buffer { struct write_buffer * next; void *buffer; @@ -79,6 +81,7 @@ struct socket { int id; uint16_t protocol; uint16_t type; + int64_t warn_size; union { int size; uint8_t udp_address[UDP_ADDRESS_SIZE]; @@ -391,6 +394,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque, s->p.size = MIN_READ_BUFFER; s->opaque = opaque; s->wb_size = 0; + s->warn_size = 0; check_wb_list(&s->high); check_wb_list(&s->low); return s; @@ -598,6 +602,11 @@ raise_uncomplete(struct socket * s) { high->head = high->tail = tmp; } +static inline int +send_buffer_empty(struct socket *s) { + return (s->high.head == NULL && s->low.head == NULL); +} + /* Each socket has two write buffer list, high priority and low priority. @@ -622,15 +631,24 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r // step 3 if (list_uncomplete(&s->low)) { raise_uncomplete(s); + return -1; } - } else { - // step 4 - sp_write(ss->event_fd, s->fd, s, false); + } + // step 4 + assert(send_buffer_empty(s) && s->wb_size == 0); + sp_write(ss->event_fd, s->fd, s, false); - if (s->type == SOCKET_TYPE_HALFCLOSE) { + if (s->type == SOCKET_TYPE_HALFCLOSE) { force_close(ss, s, result); return SOCKET_CLOSE; - } + } + if(s->warn_size > 0){ + s->warn_size = 0; + result->opaque = s->opaque; + result->id = s->id; + result->ud = 0; + result->data = NULL; + return SOCKET_WARNING; } } @@ -677,10 +695,6 @@ append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_ s->wb_size += buf->sz; } -static inline int -send_buffer_empty(struct socket *s) { - return (s->high.head == NULL && s->low.head == NULL); -} /* When send a package , we can assign the priority : PRIORITY_HIGH or PRIORITY_LOW @@ -757,6 +771,14 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock append_sendbuffer_udp(ss,s,priority,request,udp_address); } } + if (s->wb_size >= WARNING_SIZE && s->wb_size >= s->warn_size) { + s->warn_size = s->warn_size == 0 ? WARNING_SIZE *2 : s->warn_size*2; + result->opaque = s->opaque; + result->id = s->id; + result->ud = s->wb_size%1024 == 0 ? s->wb_size/1024 : s->wb_size/1024 + 1; + result->data = NULL; + return SOCKET_WARNING; + } return -1; } @@ -794,7 +816,8 @@ close_socket(struct socket_server *ss, struct request_close *request, struct soc } if (!send_buffer_empty(s)) { int type = send_buffer(ss,s,result); - if (type != -1) + // 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)) { @@ -1223,6 +1246,9 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int ss->event_index = 0; if (ss->event_n <= 0) { ss->event_n = 0; + if (errno == EINTR) { + continue; + } return -1; } } @@ -1334,8 +1360,8 @@ free_buffer(struct socket_server *ss, const void * buffer, int sz) { so.free_func((void *)buffer); } -// return -1 when error -int64_t +// return -1 when error, 0 when success +int socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz) { struct socket * s = &ss->slot[HASH_ID(id)]; if (s->id != id || s->type == SOCKET_TYPE_INVALID) { @@ -1349,15 +1375,16 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz request.u.send.buffer = (char *)buffer; send_request(ss, &request, 'D', sizeof(request.u.send)); - return s->wb_size; + return 0; } -void +// return -1 when error, 0 when success +int socket_server_send_lowpriority(struct socket_server *ss, int id, const void * buffer, int sz) { struct socket * s = &ss->slot[HASH_ID(id)]; if (s->id != id || s->type == SOCKET_TYPE_INVALID) { free_buffer(ss, buffer, sz); - return; + return -1; } struct request_package request; @@ -1366,6 +1393,7 @@ socket_server_send_lowpriority(struct socket_server *ss, int id, const void * bu request.u.send.buffer = (char *)buffer; send_request(ss, &request, 'P', sizeof(request.u.send)); + return 0; } void @@ -1546,7 +1574,7 @@ socket_server_udp(struct socket_server *ss, uintptr_t opaque, const char * addr, return id; } -int64_t +int socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp_address *addr, const void *buffer, int sz) { struct socket * s = &ss->slot[HASH_ID(id)]; if (s->id != id || s->type == SOCKET_TYPE_INVALID) { @@ -1576,7 +1604,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp memcpy(request.u.send_udp.address, udp_address, addrsz); send_request(ss, &request, 'A', sizeof(request.u.send_udp.send)+addrsz); - return s->wb_size; + return 0; } int diff --git a/skynet-src/socket_server.h b/skynet-src/socket_server.h index 31b3c9de..622b086f 100644 --- a/skynet-src/socket_server.h +++ b/skynet-src/socket_server.h @@ -10,6 +10,7 @@ #define SOCKET_ERR 4 #define SOCKET_EXIT 5 #define SOCKET_UDP 6 +#define SOCKET_WARNING 7 struct socket_server; @@ -30,8 +31,8 @@ void socket_server_shutdown(struct socket_server *, uintptr_t opaque, int id); void socket_server_start(struct socket_server *, uintptr_t opaque, int id); // return -1 when error -int64_t socket_server_send(struct socket_server *, int id, const void * buffer, int sz); -void socket_server_send_lowpriority(struct socket_server *, int id, const void * buffer, int sz); +int socket_server_send(struct socket_server *, int id, const void * buffer, int sz); +int socket_server_send_lowpriority(struct socket_server *, int id, const void * buffer, int sz); // ctrl command below returns id int socket_server_listen(struct socket_server *, uintptr_t opaque, const char * addr, int port, int backlog); @@ -50,7 +51,7 @@ int socket_server_udp(struct socket_server *, uintptr_t opaque, const char * add int socket_server_udp_connect(struct socket_server *, int id, const char * addr, int port); // If the socket_udp_address is NULL, use last call socket_server_udp_connect address instead // You can also use socket_server_send -int64_t socket_server_udp_send(struct socket_server *, int id, const struct socket_udp_address *, const void *buffer, int sz); +int socket_server_udp_send(struct socket_server *, int id, const struct socket_udp_address *, const void *buffer, int sz); // extract the address of the message, struct socket_message * should be SOCKET_UDP const struct socket_udp_address * socket_server_udp_address(struct socket_server *, struct socket_message *, int *addrsz);