From 61980c8a1f23a87e96cc128716680c11a38c64a8 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Fri, 7 Sep 2018 16:11:27 +0800 Subject: [PATCH] add socket.netstat --- lualib-src/lua-socket.c | 65 +++++++++++++++++ lualib/skynet/socket.lua | 1 + service/debug_console.lua | 48 +++++++++++++ skynet-src/skynet_socket.c | 12 +++- skynet-src/skynet_socket.h | 5 ++ skynet-src/skynet_start.c | 1 + skynet-src/socket_info.h | 28 ++++++++ skynet-src/socket_server.c | 142 +++++++++++++++++++++++++++++++++++-- skynet-src/socket_server.h | 6 +- 9 files changed, 299 insertions(+), 9 deletions(-) create mode 100644 skynet-src/socket_info.h diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index a1ef0295..eba3743d 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -680,6 +680,70 @@ ludp_address(lua_State *L) { return 2; } +static void +getinfo(lua_State *L, struct socket_info *si) { + lua_newtable(L); + lua_pushinteger(L, si->id); + lua_setfield(L, -2, "id"); + lua_pushinteger(L, si->opaque); + lua_setfield(L, -2, "address"); + switch(si->type) { + case SOCKET_INFO_LISTEN: + lua_pushstring(L, "LISTEN"); + lua_setfield(L, -2, "type"); + lua_pushinteger(L, si->read); + lua_setfield(L, -2, "accept"); + if (si->name[0]) { + lua_pushstring(L, si->name); + lua_setfield(L, -2, "sock"); + } + return; + case SOCKET_INFO_TCP: + lua_pushstring(L, "TCP"); + break; + case SOCKET_INFO_UDP: + lua_pushstring(L, "UDP"); + break; + case SOCKET_INFO_BIND: + lua_pushstring(L, "BIND"); + break; + default: + lua_pushstring(L, "UNKNOWN"); + lua_setfield(L, -2, "type"); + return; + } + lua_setfield(L, -2, "type"); + lua_pushinteger(L, si->read); + lua_setfield(L, -2, "read"); + lua_pushinteger(L, si->write); + lua_setfield(L, -2, "write"); + lua_pushinteger(L, si->wbuffer); + lua_setfield(L, -2, "wbuffer"); + lua_pushinteger(L, si->rtime); + lua_setfield(L, -2, "rtime"); + lua_pushinteger(L, si->wtime); + lua_setfield(L, -2, "wtime"); + if (si->name[0]) { + lua_pushstring(L, si->name); + lua_setfield(L, -2, "peer"); + } +} + +static int +linfo(lua_State *L) { + lua_newtable(L); + struct socket_info * si = skynet_socket_info(); + struct socket_info * temp = si; + int n = 0; + while (temp) { + getinfo(L, temp); + lua_seti(L, -2, ++n); + temp = temp->next; + } + socket_info_release(si); + return 1; +} + LUAMOD_API int luaopen_skynet_socketdriver(lua_State *L) { luaL_checkversion(L); @@ -693,6 +757,7 @@ luaopen_skynet_socketdriver(lua_State *L) { { "readline", lreadline }, { "str2p", lstr2p }, { "header", lheader }, + { "info", linfo }, { "unpack", lunpack }, { NULL, NULL }, diff --git a/lualib/skynet/socket.lua b/lualib/skynet/socket.lua index 357e780f..fe7e108b 100644 --- a/lualib/skynet/socket.lua +++ b/lualib/skynet/socket.lua @@ -438,6 +438,7 @@ end socket.sendto = assert(driver.udp_send) socket.udp_address = assert(driver.udp_address) +socket.netstat = assert(driver.info) function socket.warning(id, callback) local obj = socket_pool[id] diff --git a/service/debug_console.lua b/service/debug_console.lua index 7ec47453..dbde18ea 100644 --- a/service/debug_console.lua +++ b/service/debug_console.lua @@ -159,6 +159,7 @@ function COMMAND.help() ping = "ping address", call = "call address ...", trace = "trace address [proto] [on|off]", + netstat = "netstat : show netstat", } end @@ -374,3 +375,50 @@ function COMMANDX.call(cmd) local rets = table.pack(skynet.call(address, "lua", table.unpack(args, 2, args.n))) return rets end + +local function bytes(size) + if size == nil or size == 0 then + return + end + if size < 1024 then + return size + end + if size < 1024 * 1024 then + return tostring(size/1024) .. "K" + end + return tostring(size/(1024*1024)) .. "M" +end + +local function convert_stat(info) + local now = skynet.now() + local function time(t) + if t == nil then + return + end + t = now - t + if t < 6000 then + return tostring(t/100) .. "s" + end + local hour = t // (100*60*60) + t = t - hour * 100 * 60 * 60 + local min = t // (100*60) + t = t - min * 100 * 60 + local sec = t / 100 + return string.format("%s%d:%.2gs",hour == 0 and "" or (hour .. ":"),min,sec) + end + + info.address = skynet.address(info.address) + info.read = bytes(info.read) + info.write = bytes(info.write) + info.wbuffer = bytes(info.wbuffer) + info.rtime = time(info.rtime) + info.wtime = time(info.wtime) +end + +function COMMAND.netstat() + local stat = socket.netstat() + for _, info in ipairs(stat) do + convert_stat(info) + end + return stat +end \ No newline at end of file diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index ef255060..1b56024c 100644 --- a/skynet-src/skynet_socket.c +++ b/skynet-src/skynet_socket.c @@ -15,7 +15,7 @@ static struct socket_server * SOCKET_SERVER = NULL; void skynet_socket_init() { - SOCKET_SERVER = socket_server_create(); + SOCKET_SERVER = socket_server_create(skynet_now()); } void @@ -29,6 +29,11 @@ skynet_socket_free() { SOCKET_SERVER = NULL; } +void +skynet_socket_updatetime() { + socket_server_updatetime(SOCKET_SERVER, skynet_now()); +} + // mainloop thread static void forward_message(int type, bool padding, struct socket_message * result) { @@ -190,3 +195,8 @@ skynet_socket_udp_address(struct skynet_socket_message *msg, int *addrsz) { sm.data = msg->buffer; return (const char *)socket_server_udp_address(SOCKET_SERVER, &sm, addrsz); } + +struct socket_info * +skynet_socket_info() { + return socket_server_info(SOCKET_SERVER); +} diff --git a/skynet-src/skynet_socket.h b/skynet-src/skynet_socket.h index 21dfcef4..2fd38eb3 100644 --- a/skynet-src/skynet_socket.h +++ b/skynet-src/skynet_socket.h @@ -1,6 +1,8 @@ #ifndef skynet_socket_h #define skynet_socket_h +#include "socket_info.h" + struct skynet_context; #define SKYNET_SOCKET_TYPE_DATA 1 @@ -22,6 +24,7 @@ void skynet_socket_init(); void skynet_socket_exit(); void skynet_socket_free(); int skynet_socket_poll(); +void skynet_socket_updatetime(); int skynet_socket_send(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); @@ -38,4 +41,6 @@ int skynet_socket_udp_connect(struct skynet_context *ctx, int id, const char * a int skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz); const char * skynet_socket_udp_address(struct skynet_socket_message *, int *addrsz); +struct socket_info * skynet_socket_info(); + #endif diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index 2f5ddea1..266e8ea2 100644 --- a/skynet-src/skynet_start.c +++ b/skynet-src/skynet_start.c @@ -131,6 +131,7 @@ thread_timer(void *p) { skynet_initthread(THREAD_TIMER); for (;;) { skynet_updatetime(); + skynet_socket_updatetime(); CHECK_ABORT wakeup(m,m->count-1); usleep(2500); diff --git a/skynet-src/socket_info.h b/skynet-src/socket_info.h new file mode 100644 index 00000000..78facf6c --- /dev/null +++ b/skynet-src/socket_info.h @@ -0,0 +1,28 @@ +#ifndef socket_info_h +#define socket_info_h + +#define SOCKET_INFO_UNKNOWN 0 +#define SOCKET_INFO_LISTEN 1 +#define SOCKET_INFO_TCP 2 +#define SOCKET_INFO_UDP 3 +#define SOCKET_INFO_BIND 4 + +#include + +struct socket_info { + int id; + int type; + uint64_t opaque; + uint64_t read; + uint64_t write; + uint64_t rtime; + uint64_t wtime; + int64_t wbuffer; + char name[128]; + struct socket_info *next; +}; + +struct socket_info * socket_info_create(struct socket_info *last); +void socket_info_release(struct socket_info *); + +#endif diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index b74dc165..ddee2943 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -75,11 +75,19 @@ struct wb_list { struct write_buffer * tail; }; +struct socket_stat { + uint64_t rtime; + uint64_t wtime; + uint64_t read; + uint64_t write; +}; + struct socket { uintptr_t opaque; struct wb_list high; struct wb_list low; int64_t wb_size; + struct socket_stat stat; volatile uint32_t sending; int fd; int id; @@ -98,6 +106,7 @@ struct socket { }; struct socket_server { + volatile uint64_t time; int recvctrl_fd; int sendctrl_fd; int checkctrl; @@ -188,6 +197,7 @@ struct request_udp { T Set opt U Create UDP socket C set udp address + Q query info */ struct request_package { @@ -326,7 +336,7 @@ clear_wb_list(struct wb_list *list) { } struct socket_server * -socket_server_create() { +socket_server_create(uint64_t time) { int i; int fd[2]; poll_fd efd = sp_create(); @@ -349,6 +359,7 @@ socket_server_create() { } struct socket_server *ss = MALLOC(sizeof(*ss)); + ss->time = time; ss->event_fd = efd; ss->recvctrl_fd = fd[0]; ss->sendctrl_fd = fd[1]; @@ -371,6 +382,11 @@ socket_server_create() { return ss; } +void +socket_server_updatetime(struct socket_server *ss, uint64_t time) { + ss->time = time; +} + static void free_wb_list(struct socket_server *ss, struct wb_list *list) { struct write_buffer *wb = list->head; @@ -468,9 +484,22 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque, check_wb_list(&s->low); s->dw_buffer = NULL; s->dw_size = 0; + memset(&s->stat, 0, sizeof(s->stat)); return s; } +static inline void +stat_read(struct socket_server *ss, struct socket *s, int n) { + s->stat.read += n; + s->stat.rtime = ss->time; +} + +static inline void +stat_write(struct socket_server *ss, struct socket *s, int n) { + s->stat.write += n; + s->stat.wtime = ss->time; +} + // return -1 when connecting static int open_socket(struct socket_server *ss, struct request_open * request, struct socket_message *result) { @@ -563,6 +592,7 @@ send_list_tcp(struct socket_server *ss, struct socket *s, struct wb_list *list, force_close(ss,s,l,result); return SOCKET_CLOSE; } + stat_write(ss,s,(int)sz); s->wb_size -= sz; if (sz != tmp->sz) { tmp->ptr += sz; @@ -634,7 +664,7 @@ send_list_udp(struct socket_server *ss, struct socket *s, struct wb_list *list, drop_udp(ss, s, list, tmp); return -1; } - + stat_write(ss,s,tmp->sz); s->wb_size -= tmp->sz; list->head = tmp->next; write_buffer_free(ss,tmp); @@ -847,6 +877,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock if (n != so.sz) { append_sendbuffer_udp(ss,s,priority,request,udp_address); } else { + stat_write(ss,s,n); so.free_func(request->buffer); return -1; } @@ -1193,6 +1224,8 @@ forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_lo return -1; } + stat_read(ss,s,n); + if (n == sz) { s->p.size *= 2; } else if (sz > MIN_READ_BUFFER && n*2 < sz) { @@ -1242,6 +1275,8 @@ forward_message_udp(struct socket_server *ss, struct socket *s, struct socket_lo } return -1; } + stat_read(ss,s,n); + uint8_t * data; if (slen == sizeof(sa.v4)) { if (s->protocol != PROTOCOL_UDP) @@ -1298,6 +1333,20 @@ report_connect(struct socket_server *ss, struct socket *s, struct socket_lock *l } } +static int +getname(union sockaddr_all *u, char *buffer, size_t sz) { + char tmp[INET6_ADDRSTRLEN]; + void * sin_addr = (u->s.sa_family == AF_INET) ? (void*)&u->v4.sin_addr : (void *)&u->v6.sin6_addr; + int sin_port = ntohs((u->s.sa_family == AF_INET) ? u->v4.sin_port : u->v6.sin6_port); + if (inet_ntop(u->s.sa_family, sin_addr, tmp, sizeof(tmp))) { + snprintf(buffer, sz, "%s:%d", tmp, sin_port); + return 1; + } else { + buffer[0] = '\0'; + return 0; + } +} + // return 0 when failed, or -1 when file limit static int report_accept(struct socket_server *ss, struct socket *s, struct socket_message *result) { @@ -1327,17 +1376,16 @@ report_accept(struct socket_server *ss, struct socket *s, struct socket_message close(client_fd); return 0; } + // accept new one connection + stat_read(ss,s,1); + ns->type = SOCKET_TYPE_PACCEPT; result->opaque = s->opaque; result->id = s->id; result->ud = id; result->data = NULL; - void * sin_addr = (u.s.sa_family == AF_INET) ? (void*)&u.v4.sin_addr : (void *)&u.v6.sin6_addr; - int sin_port = ntohs((u.s.sa_family == AF_INET) ? u.v4.sin_port : u.v6.sin6_port); - char tmp[INET6_ADDRSTRLEN]; - if (inet_ntop(u.s.sa_family, sin_addr, tmp, sizeof(tmp))) { - snprintf(ss->buffer, sizeof(ss->buffer), "%s:%d", tmp, sin_port); + if (getname(&u, ss->buffer, sizeof(ss->buffer))) { result->data = ss->buffer; } @@ -1558,6 +1606,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz // ignore error, let socket thread try again n = 0; } + stat_write(ss,s,n); if (n == so.sz) { // write done socket_unlock(&l); @@ -1827,6 +1876,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz); if (n >= 0) { // sendto succ + stat_write(ss,s,n); socket_unlock(&l); so.free_func((void *)buffer); return 0; @@ -1915,3 +1965,81 @@ socket_server_udp_address(struct socket_server *ss, struct socket_message *msg, } return (const struct socket_udp_address *)address; } + + +struct socket_info * +socket_info_create(struct socket_info *last) { + struct socket_info *si = skynet_malloc(sizeof(*si)); + memset(si, 0 , sizeof(*si)); + si->next = last; + return si; +} + +void +socket_info_release(struct socket_info *si) { + while (si) { + struct socket_info *temp = si; + si = si->next; + skynet_free(temp); + } +} + +static int +query_info(struct socket *s, struct socket_info *si) { + union sockaddr_all u; + socklen_t slen = sizeof(u); + switch (s->type) { + case SOCKET_TYPE_BIND: + si->type = SOCKET_INFO_BIND; + si->name[0] = '\0'; + break; + case SOCKET_TYPE_LISTEN: + si->type = SOCKET_INFO_LISTEN; + if (getsockname(s->fd, &u.s, &slen) == 0) { + getname(&u, si->name, sizeof(si->name)); + } + break; + case SOCKET_TYPE_CONNECTED: + if (s->protocol == PROTOCOL_TCP) { + si->type = SOCKET_INFO_TCP; + if (getpeername(s->fd, &u.s, &slen) == 0) { + getname(&u, si->name, sizeof(si->name)); + } + } else { + si->type = SOCKET_INFO_UDP; + if (udp_socket_address(s, s->p.udp_address, &u)) { + getname(&u, si->name, sizeof(si->name)); + } + } + break; + default: + return 0; + } + si->id = s->id; + si->opaque = (uint64_t)s->opaque; + si->read = s->stat.read; + si->write = s->stat.write; + si->rtime = s->stat.rtime; + si->wtime = s->stat.wtime; + si->wbuffer = s->wb_size; + + return 1; +} + +struct socket_info * +socket_server_info(struct socket_server *ss) { + int i; + struct socket_info * si = NULL; + for (i=0;islot[i]; + int id = s->id; + struct socket_info temp; + if (query_info(s, &temp) && s->id == id) { + // socket_server_info may call in different thread, so check socket id again + si = socket_info_create(si); + temp.next = si->next; + *si = temp; + } + } + return si; +} diff --git a/skynet-src/socket_server.h b/skynet-src/socket_server.h index 622b086f..96568502 100644 --- a/skynet-src/socket_server.h +++ b/skynet-src/socket_server.h @@ -2,6 +2,7 @@ #define skynet_socket_server_h #include +#include "socket_info.h" #define SOCKET_DATA 0 #define SOCKET_CLOSE 1 @@ -21,8 +22,9 @@ struct socket_message { char * data; }; -struct socket_server * socket_server_create(); +struct socket_server * socket_server_create(uint64_t time); void socket_server_release(struct socket_server *); +void socket_server_updatetime(struct socket_server *, uint64_t time); int socket_server_poll(struct socket_server *, struct socket_message *result, int *more); void socket_server_exit(struct socket_server *); @@ -64,4 +66,6 @@ struct socket_object_interface { // if you send package sz == -1, use soi. void socket_server_userobject(struct socket_server *, struct socket_object_interface *soi); +struct socket_info * socket_server_info(struct socket_server *); + #endif