完善基于Json的状态监测、统计

6 files changed, 131 insertions, 26 deletions

diff --git a/core/include/mr_rtdev.h b/core/include/mr_rtdev.h
index 66c242c..f70cd32 100644
--- a/core/include/mr_rtdev.h
+++ b/core/include/mr_rtdev.h
@@ -53,6 +53,24 @@ struct rtdev_desc
 	struct rtdev_app_desc_list app_desc_list;
 };
 
+struct rtdev_stat_info
+{
+	/* RX: from service to app */
+	uint64_t rx_on_line[MR_SID_MAX];
+	uint64_t rx_deliver[MR_SID_MAX];
+	uint64_t rx_missed[MR_SID_MAX];
+
+	/* TX: from app to service */
+	uint64_t tx_on_line[MR_SID_MAX];
+	uint64_t tx_deliver[MR_SID_MAX];
+	uint64_t tx_missed[MR_SID_MAX];
+
+	/* FTX: fast tunnel from app to serivce */
+	uint64_t ftx_on_line[MR_SID_MAX];
+	uint64_t ftx_deliver[MR_SID_MAX];
+	uint64_t ftx_missed[MR_SID_MAX];
+};
+
 TAILQ_HEAD(rtdev_desc_list, rtdev_desc);
 
 // ����һ������ʱ�����豸
@@ -66,4 +84,7 @@ int mr_rt_device_destory(struct rtdev_desc * desc);
 struct rtdev_app_desc * mr_rt_device_open(struct mr_core_instance * instance, const char * devsym, const char * appsym,
 	unsigned int nr_rxstream, unsigned int nr_txstream);
 
-int mr_rt_device_close(struct rtdev_app_desc * desc);
\ No newline at end of file
+int mr_rt_device_close(struct rtdev_app_desc * desc);
+
+// ����ʱ�豸��Ϣͳ��
+int mr_rtdev_stats_get(struct rtdev_desc * dev_desc, struct rtdev_stat_info * stat_info);
\ No newline at end of file
diff --git a/core/include/mr_vnode.h b/core/include/mr_vnode.h
index 3723ef1..f650102 100644
--- a/core/include/mr_vnode.h
+++ b/core/include/mr_vnode.h
@@ -75,7 +75,7 @@ struct vnode_prod
 	struct tunnel_block_list block_list;
 	unsigned int nr_block;
 	unsigned int cur_attach;
-	struct vnode_prod_stat stat[MR_GSID_MAX];
+	struct vnode_prod_stat stat[MR_SID_MAX];
 };
 
 /* Virtual Data-Node Consumer Queue Structure */
@@ -158,6 +158,10 @@ unsigned int prodq, void * objects[], uint32_t hash[], int nr_objects);
 int vnode_dequeue_burst(struct vnode_cons * cons, struct vnode_ops * ops,
  unsigned int consq, void * objects[], int nr_max_objects);
 
+struct vnode_prod_stat * vnode_prod_stat_get(struct vnode_prod * prod);
+
+struct vnode_cons_stat * vnode_cons_stat_get(struct vnode_cons * cons);
+
 /**
  * \brief Attach Operation for Vnode's Coumser
  * \param node VNode Structure
diff --git a/core/src/rtdev.c b/core/src/rtdev.c
index 746abcb..24402c1 100644
--- a/core/src/rtdev.c
+++ b/core/src/rtdev.c
@@ -76,6 +76,30 @@ struct rtdev_app_desc * mr_rtdev_app_lookup(struct rtdev_desc * dev_desc, const
 	return __rtdev_app_lookup_unsafe(dev_desc, appsym);
 }
 
+int mr_rtdev_stats_get(struct rtdev_desc * dev_desc, struct rtdev_stat_info * stat_info)
+{
+	struct vnode_prod_stat * st_prod_rx = vnode_prod_stat_get(dev_desc->vnode_prod_rx);
+	struct vnode_cons_stat * st_cons_tx = vnode_cons_stat_get(dev_desc->vnode_cons_tx);
+	struct vnode_cons_stat * st_cons_ftx = vnode_cons_stat_get(dev_desc->vnode_cons_ftx);
+	
+	for(int i = 0; i < dev_desc->nr_serv_stream; i++)
+	{
+		stat_info->rx_on_line[i] = rte_atomic64_read(&st_prod_rx[i].on_line);
+		stat_info->rx_deliver[i] = rte_atomic64_read(&st_prod_rx[i].sent);
+		stat_info->rx_missed[i] = rte_atomic64_read(&st_prod_rx[i].missed);
+
+		stat_info->tx_on_line[i] = rte_atomic64_read(&st_cons_tx[i].on_line);
+		stat_info->tx_deliver[i] = rte_atomic64_read(&st_cons_tx[i].recieved);
+		stat_info->tx_missed[i] = rte_atomic64_read(&st_cons_tx[i].missed);
+
+		stat_info->ftx_on_line[i] = rte_atomic64_read(&st_cons_ftx[i].on_line);
+		stat_info->ftx_deliver[i] = rte_atomic64_read(&st_cons_ftx[i].recieved);
+		stat_info->ftx_missed[i] = rte_atomic64_read(&st_cons_ftx[i].missed);
+	}
+
+	return 0;
+}
+
  // ����һ������ʱ�����豸
 struct rtdev_desc * mr_rtdev_create(struct mr_core_instance * instance, const char * devsym, 
 	struct rte_mempool * pool, unsigned int nr_serv_thread, unsigned int sz_tunnel, unsigned int sz_buffer)
diff --git a/core/src/vnode.c b/core/src/vnode.c
index 1d147da..402d043 100644
--- a/core/src/vnode.c
+++ b/core/src/vnode.c
@@ -708,6 +708,16 @@ int vnode_prod_attach(struct vnode * node, struct vnode_prod* prod)
 	return ret;
 }
 
+struct vnode_prod_stat * vnode_prod_stat_get(struct vnode_prod * prod)
+{
+	return prod->stat;
+}
+
+struct vnode_cons_stat * vnode_cons_stat_get(struct vnode_cons * cons)
+{
+	return cons->stat;
+}
+
 int vnode_delete_prod(struct vnode_prod * prod, struct vnode_ops * ops)
 {
 	assert(prod != NULL && prod->vnode != NULL && ops != NULL);
diff --git a/service/src/monit.c b/service/src/monit.c
index 14489c3..4de0cc7 100644
--- a/service/src/monit.c
+++ b/service/src/monit.c
@@ -11,6 +11,62 @@
 

 #include "cJSON.h"

 

+static cJSON * __create_uint64_array(const uint64_t * value, int nr_value)

+{

+	struct cJSON * uint64_array = cJSON_CreateArray();

+	for (int i = 0; i < nr_value; i++)

+		cJSON_AddItemToArray(uint64_array, cJSON_CreateNumber(value[i]));

+	return uint64_array;

+}

+

+static cJSON * __create_raw_device_stats(struct rtdev_desc * dev_desc)

+{

+	struct rtdev_stat_info _stat_info;

+	mr_rtdev_stats_get(dev_desc, &_stat_info);

+	

+	struct cJSON * j_raw_device_stats = cJSON_CreateObject();

+	unsigned int nr_serv_thread = dev_desc->nr_serv_stream;

+

+#define __JOIN_RAW_DEVICE_STATS_ITEM(item)		do {				\

+	cJSON_AddItemToObject(j_raw_device_stats, #item,				\

+	__create_uint64_array(_stat_info.item, nr_serv_thread));		\

+} while(0)

+

+	__JOIN_RAW_DEVICE_STATS_ITEM(rx_on_line);

+	__JOIN_RAW_DEVICE_STATS_ITEM(rx_deliver);

+	__JOIN_RAW_DEVICE_STATS_ITEM(rx_missed);

+	__JOIN_RAW_DEVICE_STATS_ITEM(tx_on_line);

+	__JOIN_RAW_DEVICE_STATS_ITEM(tx_deliver);

+	__JOIN_RAW_DEVICE_STATS_ITEM(tx_missed);

+	__JOIN_RAW_DEVICE_STATS_ITEM(ftx_on_line);

+	__JOIN_RAW_DEVICE_STATS_ITEM(ftx_deliver);

+	__JOIN_RAW_DEVICE_STATS_ITEM(ftx_missed);

+	

+	return j_raw_device_stats;

+}

+

+

+// ����ʱԭʼ�����豸ͳ�Ƽ���

+static cJSON * monit_raw_device(struct sc_instance * instance)

+{

+	struct cJSON * j_raw_device_array = cJSON_CreateArray();

+	struct sc_device * sc_device_iter = NULL;

+

+	TAILQ_FOREACH(sc_device_iter, &instance->device_list, next)

+	{

+		cJSON * j_raw_device = cJSON_CreateObject();

+		struct rtdev_desc * dev_desc = sc_device_iter->rt_dev;

+		if (dev_desc == NULL) continue;

+

+		cJSON_AddStringToObject(j_raw_device, "symbol", dev_desc->symbol);

+		cJSON_AddNumberToObject(j_raw_device, "streams", dev_desc->nr_serv_stream);

+		cJSON_AddItemToObject(j_raw_device, "stats", __create_raw_device_stats(dev_desc));

+		cJSON_AddItemToArray(j_raw_device_array, j_raw_device);

+	}

+	

+	return j_raw_device_array;

+}

+

 // ��������ͳ�Ƽ���

 static cJSON * __create_device_stats(struct mr_dev * dev)

 {

@@ -92,6 +148,7 @@ static cJSON * monit_root(struct sc_instance * sc_instance)
 {

 	struct cJSON * j_root = cJSON_CreateObject();

 	cJSON_AddItemToObject(j_root, "device", monit_device_info(sc_instance));

+	cJSON_AddItemToObject(j_root, "raw", monit_raw_device(sc_instance));

 	return j_root;

 }

 

diff --git a/service/src/rxtx.c b/service/src/rxtx.c
index cc038f4..d52df25 100644
--- a/service/src/rxtx.c
+++ b/service/src/rxtx.c
@@ -16,7 +16,7 @@ void __free_mbufs(struct rte_mbuf * bufs[], int nr_bufs)
 }
 
 int rt_serv_device_rx_bulk(struct rtdev_desc * dev_desc, thread_id_t sid,
-	struct rte_mbuf * mbufs_in[], struct rte_mbuf * mbufs_out[], int nr_mbufs_in, int * nr_mbufs_out)
+	struct rte_mbuf * mbufs_in[], int nr_mbufs_in)
 {
 	hash_t hash[MR_BURST_MAX];
 
@@ -26,24 +26,16 @@ int rt_serv_device_rx_bulk(struct rtdev_desc * dev_desc, thread_id_t sid,
 	int ret = vnode_enqueue_burst_with_hash(dev_desc->vnode_prod_rx, 
 		&dev_desc->ops, sid, (void **)mbufs_in, hash, nr_mbufs_in);
 
-	if (likely(ret >= 0))
-	{
-		*nr_mbufs_out = 0;
-		return 0;
-	}
-
-	// ��������������еı��Ķ�û�д�����д��OUT������
-	*nr_mbufs_out = nr_mbufs_in;
-	for (int i = 0; i < nr_mbufs_in; i++)
-		mbufs_out[i] = mbufs_in[i];
-
-	return 0;
+	if (likely(ret < 0)) return 0;
+	for (int i = 0; i < nr_mbufs_in; i++) mbufs_in[i] = NULL;
+	return nr_mbufs_in;
 }
 
 int rt_serv_device_tx_bulk(struct rtdev_desc * dev_desc, thread_id_t sid,
 	struct rte_mbuf * mbufs_out[], int nr_mbufs)
 {
-	int __nr_mbufs_out = 0; int __nr_mbufs_left = nr_mbufs;
+	int __nr_mbufs_out = 0;
+	int __nr_mbufs_left = nr_mbufs;
 
 	// ��FTXͨ��ȡ��
 	int ret = vnode_dequeue_burst(dev_desc->vnode_cons_ftx, &dev_desc->ops, sid,
@@ -102,11 +94,9 @@ static inline void __rx_loop_one_device(struct sc_instance * instance, thread_id
 	struct sc_device * sc_dev, unsigned int nr_rx_burst)
 {
 	struct rte_mbuf * rx_bufs_in[MR_BURST_MAX];
-	struct rte_mbuf * rx_bufs_out[MR_BURST_MAX];
 	hash_t hash_result[MR_BURST_MAX];
 
 	int nr_rx_bufs_in;
-	int nr_rx_bufs_out;
 	struct mr_dev_q * devq = sc_dev->rx_devq[tid];
 	assert(devq != NULL);
 	
@@ -132,18 +122,18 @@ static inline void __rx_loop_one_device(struct sc_instance * instance, thread_id
 	}
 	else
 	{
-		rt_serv_device_rx_bulk(sc_dev->rt_dev, tid, rx_bufs_in, rx_bufs_out, nr_rx_bufs_in
-			, &nr_rx_bufs_out);
+		rt_serv_device_rx_bulk(sc_dev->rt_dev, tid, rx_bufs_in, nr_rx_bufs_in);
 	}
 		
 clean:
-	for (int i = 0; i < nr_rx_bufs_out; i++)
-		rte_pktmbuf_free(rx_bufs_out[i]);
-	
+	for (int i = 0; i < nr_rx_bufs_in; i++)
+	{
+		if (rx_bufs_in[i] != NULL) rte_pktmbuf_free(rx_bufs_in[i]);
+	}
+		
 	return;
 }
 
-
 void sc_thread_rxtx_loop(struct sc_instance * instance, thread_id_t tid)
 {
 	PERF_BEGIN(sc_rx_loop_all);
@@ -171,8 +161,7 @@ void * sc_runtime_thread(void * args)
 	struct sc_instance * instance = (struct sc_instance *)args;
 	thread_id_t tid = mr_thread_id();
 	
-	MR_LOG(INFO, SERVICE, "Thread Running(ThreadID=%d, CPU=%d, SocketID=%d\n",
-		mr_thread_id(), mr_cpu_id(), mr_socket_id());
+	MR_LOG(INFO, SERVICE, "Thread %d Running...\n", mr_thread_id());
 
 	while (1)
 	{