增加应用注册线程收发包情况统计，增加兼容MR3的DLOGREADER程序。

6 files changed, 283 insertions, 21 deletions

diff --git a/app/include/mrapp.h b/app/include/mrapp.h
index 52dbc40..db8a38a 100644
--- a/app/include/mrapp.h
+++ b/app/include/mrapp.h
@@ -10,15 +10,23 @@ struct mr_instance;
 
 struct mrapp_stat
 {
-    uint64_t t_arp_request_on_line;
-    uint64_t t_arp_request_accept;
-    uint64_t t_arp_request_ignore;
-    uint64_t t_arp_answer_on_line;
-    uint64_t t_arp_answer_send;
-    uint64_t t_arp_answer_drop;
+    uint64_t arp_request_on_line;
+    uint64_t arp_request_accept;
+    uint64_t arp_request_ignore;
+    uint64_t arp_answer_on_line;
+    uint64_t arp_answer_send;
+    uint64_t arp_answer_drop;
 
     uint64_t mbuf_alloc_count;
     uint64_t mbuf_free_count;
+
+    uint64_t packet_recv_count;
+    uint64_t packet_recv_length;
+    uint64_t packet_send_count;
+    uint64_t packet_send_length;
+
+    uint64_t packet_recv_drop;
+    uint64_t packet_send_drop;
 };
 
 /* 用户设备描述符 */
diff --git a/app/src/monit.c b/app/src/monit.c
index 88795af..03747f4 100644
--- a/app/src/monit.c
+++ b/app/src/monit.c
@@ -148,20 +148,42 @@ static cJSON * monit_app_stat(struct mr_instance * instance)
 {
     struct cJSON * j_root = cJSON_CreateObject();
 
-    uint64_t __pktmbuf_alloc_user_array[MR_SID_MAX];
-    uint64_t __pktmbuf_free_user_array[MR_SID_MAX];
-
-    for (int i = 0; i < instance->nr_dataplane_thread; i++)
-    {
-        __pktmbuf_alloc_user_array[i] = instance->stat[i].mbuf_alloc_count;
-        __pktmbuf_free_user_array[i] = instance->stat[i].mbuf_free_count;
-    }
-
-    cJSON_AddItemToObject(j_root, "pktmbuf_alloc_user",
-        __create_uint64_array(__pktmbuf_alloc_user_array, instance->nr_dataplane_thread));
-
-    cJSON_AddItemToObject(j_root, "pktmbuf_free_user",
-        __create_uint64_array(__pktmbuf_free_user_array, instance->nr_dataplane_thread));
+#define _trans_app_stat(item)                                               \
+    uint64_t __trans_##item[MR_SID_MAX];                                    \
+    for(int i = 0; i < instance->nr_dataplane_thread; i++)                  \
+    {                                                                       \
+        __trans_##item[i] = instance->stat[i].item;                         \
+    }                                                                       \
+
+#define _trans_app_stat_array(item)  __trans_##item
+
+    _trans_app_stat(packet_recv_count);
+    _trans_app_stat(packet_recv_length);
+    _trans_app_stat(packet_send_count);
+    _trans_app_stat(packet_send_length);
+    _trans_app_stat(packet_recv_drop);
+    _trans_app_stat(packet_send_drop);
+    _trans_app_stat(mbuf_alloc_count);
+    _trans_app_stat(mbuf_free_count);
+
+#define _json_generate(item)                                                \
+do {                                                                        \
+    cJSON_AddItemToObject(j_root, #item, __create_uint64_array(             \
+    _trans_app_stat_array(item), instance->nr_dataplane_thread));           \
+} while(0)                                                                  \
+        
+    _json_generate(packet_recv_count);
+    _json_generate(packet_recv_length);
+    _json_generate(packet_send_count);
+    _json_generate(packet_send_length);
+    _json_generate(packet_recv_drop);
+    _json_generate(packet_send_drop);
+    _json_generate(mbuf_alloc_count);
+    _json_generate(mbuf_free_count);
+
+#undef _trans_app_stat
+#undef _trans_app_stat_array
+#undef _json_generate
 
     return j_root;
 }
diff --git a/app/src/rawio.c b/app/src/rawio.c
index 5a81283..ec7ce32 100644
--- a/app/src/rawio.c
+++ b/app/src/rawio.c
@@ -9,6 +9,13 @@
 #include <icmp.h>
 #include <sendpath.h>
 
+static inline int __packet_total_len(struct rte_mbuf * mbufs[], unsigned int nr_mbufs)
+{
+    unsigned int total_len = 0;
+    for (int i = 0; i < nr_mbufs; i++) total_len += rte_pktmbuf_pkt_len(mbufs[i]);
+    return total_len;
+}
+
 int mrapp_packet_fast_send_burst(struct vdev_instance * vdi, queue_id_t qid, 
     struct rte_mbuf * mbufs[], int nr_mbufs)
 {
@@ -39,12 +46,25 @@ int marsio_recv_burst(struct mr_vdev * vdev, queue_id_t qid, marsio_buff_t * mbu
     struct vdev_instance * vdi = vdev->vdi;
 
     int ret = vnode_mirror_dequeue_burst(vdi->vnode_rx_cons, qid, (struct rte_mbuf **)mbufs, nr_mbufs);
-    if (unlikely(ret == 0)) return ret;
+    if (unlikely(ret == 0)) goto out;
 
     /* ARP */
     if (vdev->en_arp) arp_entry(vdev->instance, vdi, qid, (struct rte_mbuf **)mbufs, ret);
+    /* ICMP */
     if (vdev->en_icmp) icmp_entry(vdev->instance, vdi, qid, (struct rte_mbuf **)mbufs, ret);
+
+    /* 报文合法性检查，防止Double-Free */
     for (int i = 0; i < ret; i++) __rte_mbuf_sanity_check(mbufs[i], i);
+
+    /* 线程运行情况统计，收报情况，对于非注册线程不统计 */
+    if (thread_info.instance != NULL)
+    {
+        thread_id_t tid = thread_info.thread_id;
+        thread_info.instance->stat[tid].packet_recv_count += ret;
+        thread_info.instance->stat[tid].packet_recv_length = __packet_total_len(mbufs, ret);
+    }
+
+out:
     return ret;
 }
 
@@ -95,9 +115,26 @@ int marsio_send_burst(struct mr_sendpath * sendpath, queue_id_t sid, marsio_buff
     for (int i = 0; i < nr_mbufs; i++) hash[i] = __mbufs[i]->hash.usr;
     
     vnode_mirror_enqueue_bulk(sendpath->target_vdi->vnode_tx_prod, sid, __mbufs, hash, nr_mbufs);
+
+    /* 线程运行情况统计 */
+    if (thread_info.instance != NULL)
+    {
+        thread_id_t tid = thread_info.thread_id;
+        thread_info.instance->stat[tid].packet_send_count += nr_mbufs;
+        thread_info.instance->stat[tid].packet_send_length = __packet_total_len(__mbufs, nr_mbufs);
+    }
+
     return RT_SUCCESS;
 
 err:
+
+    /* 线程丢包情况统计 */
+    if (thread_info.instance != NULL)
+    {
+        thread_id_t tid = thread_info.thread_id;
+        thread_info.instance->stat[tid].packet_send_drop += nr_mbufs;
+    }
+
     for (int i = 0; i < nr_mbufs; i++) rte_pktmbuf_free(mbufs[i]);
     return RT_ERR;
 }
@@ -150,9 +187,25 @@ int marsio_send_burst_with_options(struct mr_sendpath * sendpath, queue_id_t sid
             __mbufs, hash, nr_mbufs); 
     }
 
+    /* 线程运行情况统计 */
+    if (thread_info.instance != NULL)
+    {
+        thread_id_t tid = thread_info.thread_id;
+        thread_info.instance->stat[tid].packet_send_count += nr_mbufs;
+        thread_info.instance->stat[tid].packet_send_length = __packet_total_len(__mbufs, nr_mbufs);
+    }
+
     return RT_SUCCESS;
 
 err:
+
+    /* 线程丢包情况统计 */
+    if (thread_info.instance != NULL)
+    {
+        thread_id_t tid = thread_info.thread_id;
+        thread_info.instance->stat[tid].packet_send_drop += nr_mbufs;
+    }
+
     for (int i = 0; i < nr_mbufs; i++) rte_pktmbuf_free(mbufs[i]);
     return RT_ERR;
 }
diff --git a/tools/CMakeLists.txt b/tools/CMakeLists.txt
index 367a907..262282b 100644
--- a/tools/CMakeLists.txt
+++ b/tools/CMakeLists.txt
@@ -1,5 +1,6 @@
 
 add_subdirectory(monit_device)
 add_subdirectory(monit_stream)
+add_subdirectory(dlogreader)
 add_subdirectory(devbind)
 add_subdirectory(systemd)
\ No newline at end of file
diff --git a/tools/dlogreader/CMakeLists.txt b/tools/dlogreader/CMakeLists.txt
new file mode 100644
index 0000000..80a6fc1
--- /dev/null
+++ b/tools/dlogreader/CMakeLists.txt
@@ -0,0 +1,14 @@
+
+set(PI_DIST_PATH ${CMAKE_CURRENT_BINARY_DIR}/pi_dist)
+set(PI_BUILD_PATH ${CMAKE_CURRENT_BINARY_DIR}/pi_build)
+set(PI_SPEC_PATH ${CMAKE_CURRENT_BINARY_DIR}/pi_spec)
+
+add_custom_command(OUTPUT ${PI_DIST_PATH}/dlogreader
+                   COMMAND pyinstaller -F -y --distpath ${PI_DIST_PATH} 
+                   --workpath ${PI_BUILD_PATH} 
+                   --specpath ${PI_SPEC_PATH}
+                   --noconsole ${CMAKE_CURRENT_SOURCE_DIR}/dlogreader.py
+                   DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/dlogreader.py)
+
+add_custom_target(dlogreader ALL DEPENDS ${PI_DIST_PATH}/dlogreader)
+install(PROGRAMS ${PI_DIST_PATH}/dlogreader DESTINATION bin COMPONENT Program)
\ No newline at end of file
diff --git a/tools/dlogreader/dlogreader.py b/tools/dlogreader/dlogreader.py
new file mode 100644
index 0000000..7ec5d08
--- /dev/null
+++ b/tools/dlogreader/dlogreader.py
@@ -0,0 +1,164 @@
+#!/usr/bin/env python2
+
+import argparse
+import json
+import prettytable
+import time
+import sys
+import signal
+import os
+
+TBPS = (1 * 1000 * 1000 * 1000 * 1000)
+GBPS = (1 * 1000 * 1000 * 1000)
+MBPS = (1 * 1000 * 1000)
+KBPS = (1 * 1000)
+
+G_JSON_PATH = '/var/run/mrzcpd/mrmonit.daemon'
+G_APP_JSON_PATH = '/var/run/mrzcpd/mrmonit.app.%s'
+
+TITLE_VECTOR = ['PhyRXFrame','PhyRXBits','PhyRXMissed','PhyRXError',
+                'PhyRXNoBUF','PhyTXFrame','PhyTXBits','PhyTXError',
+                'UsrRXDrops', 'UsrTXDrops']
+
+TITLE_MAP = { 'PhyRXFrame'   : 'ipackets',
+              'PhyRXBits'    : 'ibytes',
+              'PhyRXMissed'  : 'imissed',
+              'PhyRXError'   : 'ierrors',
+              'PhyRXNoBUF'   : 'rxnombuf',
+              'PhyTXFrame'   : 'opackets',
+              'PhyTXBits'    : 'obytes',
+              'PhyTXError'   : 'oerrors',
+              'UsrRXDrops'   : 'userrxdrop',
+              'UsrTXDrops'   : 'usertxdrop'
+            }
+
+def locate_vector_by_symbol(vector, symbol):
+    return [s for s in vector if s['symbol'] == symbol]
+
+def list_all_phydev(json_fp):
+    return [ s['symbol'] for s in json_fp['device']]
+
+def phydev_value_read(json_fp, str_device, str_item):
+    phydevs = locate_vector_by_symbol(json_fp['device'], str_device)
+    return phydevs[0]['stats']['accumulative'][str_item]
+
+def phydev_speed_read(json_fp, str_device, str_item):
+    phydevs = locate_vector_by_symbol(json_fp['device'], str_device)
+    return phydevs[0]['stats']['speed'][str_item]
+
+def app_stat_read(json_fp, item):
+    return json_fp["appstat"][item]
+
+def app_json_load(appsym):
+    with open(G_APP_JSON_PATH % appsym) as json_fp:
+        return json.load(json_fp) 
+
+def trans_to_human_readable(value):
+    if value > TBPS:
+        return value * 1.0 / TBPS, 'T'
+    if value > GBPS:
+        return value * 1.0 / GBPS, 'G'
+    if value > MBPS:
+        return value * 1.0 / MBPS, 'M'
+    if value > KBPS:
+        return value * 1.0 / KBPS, 'K'
+
+    return value * 1.0, ' '
+
+def dump_physical_devices(json_fp, phydev_user_list):
+    print("------------------------------------------------------------------------------------------------------------------------");
+    print("                                           Network Ports Statistics                                                     ");
+    print("------------------------------------------------------------------------------------------------------------------------");
+
+    for devsym in phydev_user_list:
+        print("Port %s" % devsym)
+        print("RX: packets%20lu | bytes:%20lu | dropped: %10lu | fps:%s | bps: %s" %      \
+             (phydev_value_read(json_fp, devsym, 'ipackets'),
+              phydev_value_read(json_fp, devsym, 'ibytes'),
+              phydev_value_read(json_fp, devsym, 'imissed'),
+              "%7.2f%c" % trans_to_human_readable(phydev_speed_read(json_fp, devsym, 'ipackets')),
+              "%7.2f%c" % trans_to_human_readable(phydev_speed_read(json_fp, devsym, 'ibytes'))))
+        
+        print("TX: packets%20lu | bytes:%20lu | dropped: %10lu | fps:%s | bps: %s" %
+             (phydev_value_read(json_fp, devsym, 'opackets'),
+              phydev_value_read(json_fp, devsym, 'obytes'),
+              phydev_value_read(json_fp, devsym, 'usertxdrop'),
+              "%7.2f%c" % trans_to_human_readable(phydev_speed_read(json_fp, devsym, 'opackets')),
+              "%7.2f%c" % trans_to_human_readable(phydev_speed_read(json_fp, devsym, 'obytes'))))
+              
+    return
+    
+def dump_app_status(json_fp):
+
+    print("------------------------------------------------------------------------------------------------------------------------")
+    print("                                           Worker(Client Process) Statistics                                            ")
+    print("------------------------------------------------------------------------------------------------------------------------")
+    
+    nr_thread = len(app_stat_read(json_fp, 'packet_recv_count'))
+    for thread_id in range(nr_thread):
+        print("WORKER %2d | packets %20lu | bytes: %20lu | dropped: %12lu | runtime %7.2f" % \
+             (thread_id, app_stat_read(json_fp, 'packet_recv_count')[thread_id],
+              app_stat_read(json_fp, 'packet_recv_length')[thread_id],
+              app_stat_read(json_fp, 'packet_recv_drop')[thread_id] + 
+              app_stat_read(json_fp, 'packet_send_drop')[thread_id], 0))
+    return
+
+def dump_tail():
+    print("------------------------------------------------------------------------------------------------------------------------");
+
+def setup_argv_parser(phydev_list):
+
+    parser = argparse.ArgumentParser(description='Marsio ZeroCopy Tools -- DLOGREADER',
+                                     version = 'Marsio ZeroCopy Tools Suite 4.1')
+
+    parser.add_argument('-t', '--time', help = 'interval, seconds to wait between updates',
+                        type=int, default = 1)
+    parser.add_argument('-l', '--loop', help = 'print loop, exit when recv a signal',
+                        action='store_true', default = 0)
+    parser.add_argument('--clear-screen', help = 'clear screen at start of loop', 
+                        action='store_true', default = 1)
+
+    return parser.parse_args()
+
+def phydev_json_load():
+    with open(G_JSON_PATH) as json_fp:
+        return json.load(json_fp)
+
+def sigint_handler(handler, frame):
+    sys.exit(0)    
+
+def main():
+    signal.signal(signal.SIGINT, sigint_handler)
+
+    try:
+        json_fp = phydev_json_load()
+        phydev_list = list_all_phydev(json_fp)
+        r_option = setup_argv_parser(phydev_list)
+
+        while True:
+            if r_option.clear_screen:
+                os.system('clear')
+
+            json_fp = phydev_json_load()
+            dump_physical_devices(json_fp, phydev_list)
+
+            app_json_fp = app_json_load('janus')
+            dump_app_status(app_json_fp)
+            dump_tail()
+
+            if not r_option.loop:
+                break
+
+            time.sleep(r_option.time)
+
+    except KeyboardInterrupt:
+        pass
+    except ValueError as err:
+        print(("%s, perhaps mrzcpd program is not running.") % str(err))
+    except IOError as err:
+        print(("%s, perhaps mrzcpd program is not running.") % str(err))
+
+    return 0
+
+if __name__ == '__main__':
+    main()
\ No newline at end of file