add socket.netstat

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 },

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]

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

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);
+}

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

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);

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 <stdint.h>
+
+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

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;i<MAX_SOCKET;i++) {
+		struct socket * s = &ss->slot[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;
+}

skynet-src/socket_server.h

@@ -2,6 +2,7 @@
 #define skynet_socket_server_h
 
 #include <stdint.h>
+#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