From 947727e33acb7d26e500875cb17537946e481697 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Tue, 11 Aug 2015 13:21:24 +0800 Subject: [PATCH] add new socket message WARNING --- examples/watchdog.lua | 5 +++++ lualib-src/lua-netpack.c | 9 ++++++++- lualib/skynet.lua | 2 +- lualib/snax/gateserver.lua | 6 ++++++ lualib/socket.lua | 25 +++++++++++++++++++++++++ service-src/service_gate.c | 3 +++ service-src/service_harbor.c | 7 +++++++ service/gate.lua | 4 ++++ skynet-src/skynet_socket.c | 11 +++++++---- skynet-src/skynet_socket.h | 1 + 10 files changed, 67 insertions(+), 6 deletions(-) diff --git a/examples/watchdog.lua b/examples/watchdog.lua index 29111033..efee5e24 100644 --- a/examples/watchdog.lua +++ b/examples/watchdog.lua @@ -32,6 +32,11 @@ function SOCKET.error(fd, msg) close_agent(fd) end +function SOCKET.warning(fd, size) + -- size K bytes havn't send out in fd + print("socket warning", fd, size) +end + function SOCKET.data(fd, msg) end diff --git a/lualib-src/lua-netpack.c b/lualib-src/lua-netpack.c index 1839abe3..88a54d3d 100644 --- a/lualib-src/lua-netpack.c +++ b/lualib-src/lua-netpack.c @@ -19,6 +19,7 @@ #define TYPE_ERROR 3 #define TYPE_OPEN 4 #define TYPE_CLOSE 5 +#define TYPE_WARNING 6 /* Each package is uint16 + data , uint16 (serialized in big-endian) is the number of bytes comprising the data . @@ -371,6 +372,11 @@ lfilter(lua_State *L) { lua_pushinteger(L, message->id); pushstring(L, buffer, size); return 4; + case SKYNET_SOCKET_TYPE_WARNING: + lua_pushvalue(L, lua_upvalueindex(TYPE_WARNING)); + lua_pushinteger(L, message->id); + lua_pushinteger(L, message->ud); + return 4; default: // never get here return 1; @@ -537,8 +543,9 @@ luaopen_netpack(lua_State *L) { lua_pushliteral(L, "error"); lua_pushliteral(L, "open"); lua_pushliteral(L, "close"); + lua_pushliteral(L, "warning"); - lua_pushcclosure(L, lfilter, 5); + lua_pushcclosure(L, lfilter, 6); lua_setfield(L, -2, "filter"); return 1; diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 31efffcb..e4f46908 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -271,7 +271,7 @@ function skynet.sleep(ti) end function skynet.yield() - return skynet.sleep("0") + return skynet.sleep(0) end function skynet.wait() diff --git a/lualib/snax/gateserver.lua b/lualib/snax/gateserver.lua index e68efc12..dfce2fa9 100644 --- a/lualib/snax/gateserver.lua +++ b/lualib/snax/gateserver.lua @@ -114,6 +114,12 @@ function gateserver.start(handler) close_fd(fd) end + function MSG.warning(fd, size) + if handler.warning then + handler.warning(fd, size) + end + end + skynet.register_protocol { name = "socket", id = skynet.PTYPE_SOCKET, -- PTYPE_SOCKET = 6 diff --git a/lualib/socket.lua b/lualib/socket.lua index 8d98e5a3..c5fe7858 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -133,6 +133,25 @@ socket_message[6] = function(id, size, data, address) s.callback(str, address) 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 +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 + warning(id, size) + end +end + skynet.register_protocol { name = "socket", id = skynet.PTYPE_SOCKET, -- PTYPE_SOCKET = 6 @@ -404,4 +423,10 @@ end socket.sendto = assert(driver.udp_send) socket.udp_address = assert(driver.udp_address) +function socket.warning(id, callback) + local obj = socket_pool[id] + assert(obj) + obj.warning = callback +end + return socket diff --git a/service-src/service_gate.c b/service-src/service_gate.c index 9f5d98ec..ec44c722 100644 --- a/service-src/service_gate.c +++ b/service-src/service_gate.c @@ -262,6 +262,9 @@ dispatch_socket_message(struct gate *g, const struct skynet_socket_message * mes skynet_socket_start(ctx, message->ud); } break; + case SKYNET_SOCKET_TYPE_WARNING: + skynet_error(ctx, "fd (%d) send buffer (%d)K", message->id, message->ud); + break; } } diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 90d8bcc6..5ff618f1 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -662,6 +662,13 @@ mainloop(struct skynet_context * context, void * ud, int type, int session, uint case SKYNET_SOCKET_TYPE_CONNECT: // fd forward to this service break; + case SKYNET_SOCKET_TYPE_WARNING: { + int id = harbor_id(h, message->id); + if (id) { + skynet_error(context, "message havn't send to Harbor (%d) reach %d K", id, message->ud); + } + break; + } default: skynet_error(context, "recv invalid socket message type %d", type); break; diff --git a/service/gate.lua b/service/gate.lua index 9fe3a491..78fb099f 100644 --- a/service/gate.lua +++ b/service/gate.lua @@ -63,6 +63,10 @@ function handler.error(fd, msg) skynet.send(watchdog, "lua", "socket", "error", fd, msg) end +function handler.warning(fd, size) + skynet.send(watchdog, "lua", "socket", "warning", fd, size) +end + local CMD = {} function CMD.forward(source, fd, client, address) diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index 2ea26c0a..9301419f 100644 --- a/skynet-src/skynet_socket.c +++ b/skynet-src/skynet_socket.c @@ -109,10 +109,13 @@ check_wsz(struct skynet_context *ctx, int id, void *buffer, int64_t wsz) { if (wsz < 0) { return -1; } else if (wsz > 1024 * 1024) { - int kb4 = wsz / 1024 / 4; - if (kb4 % 256 == 0) { - skynet_error(ctx, "%d Mb bytes on socket %d need to send out", (int)(wsz / (1024 * 1024)), id); - } + 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; } diff --git a/skynet-src/skynet_socket.h b/skynet-src/skynet_socket.h index 5327f09f..bcdc137c 100644 --- a/skynet-src/skynet_socket.h +++ b/skynet-src/skynet_socket.h @@ -9,6 +9,7 @@ struct skynet_context; #define SKYNET_SOCKET_TYPE_ACCEPT 4 #define SKYNET_SOCKET_TYPE_ERROR 5 #define SKYNET_SOCKET_TYPE_UDP 6 +#define SKYNET_SOCKET_TYPE_WARNING 7 struct skynet_socket_message { int type;