allow monitoring the queue length between prods and cons.

14 files changed, 183 insertions, 90 deletions

diff --git a/app/src/marsio.c b/app/src/marsio.c
index fc999b0..995ea33 100644
--- a/app/src/marsio.c
+++ b/app/src/marsio.c
@@ -254,12 +254,12 @@ static int mrapp_ctrlmsg_init(struct mr_instance * instance)
     if (env_ctrlmsg_addr != NULL)
     {
         MR_INFO("MRZCPD_CTRLMSG_LISTEN_ADDR is %s", env_ctrlmsg_addr);
-        strncpy(str_ctrlmsg_addr, env_ctrlmsg_addr, sizeof(str_ctrlmsg_addr));
+        snprintf(str_ctrlmsg_addr, sizeof(str_ctrlmsg_addr), "%s", env_ctrlmsg_addr);
     }
     else
     {
-        MR_WARNING("MRZCPD_CTRLMSG_LISTEN_ADDR is not set, default is 127.0.0.1.");
-        strncpy(str_ctrlmsg_addr, "127.0.0.1", sizeof(str_ctrlmsg_addr));
+        MR_WARNING("MRZCPD_CTRLMSG_LISTEN_ADDR is not set, default is " CTRLMSG_DEFAULT_ADDR);
+        snprintf(str_ctrlmsg_addr, sizeof(str_ctrlmsg_addr), "%s", CTRLMSG_DEFAULT_ADDR);
     }
 
     MESA_load_profile_uint_def(instance->g_cfgfile_path, "ctrlmsg", "listen_port", &ctrlmsg_port, CTRLMSG_DEFAULT_PORT);
@@ -315,6 +315,7 @@ static int mrapp_distributer_init(struct mr_instance * instance)
     return RT_SUCCESS;
 }
 
+#if 0
 static unsigned __table_strip(char * str, unsigned len)
 {
     int newlen = len;
@@ -402,6 +403,7 @@ static int __table_split_line(const char * buffer, char str_tokens[MR_TOKENS_MAX
     free(__buffer);
     return total_nr_tokens;
 }
+#endif
 
 #if 0
 static int mrapp_neigh_init(struct mr_instance * instance)
@@ -641,13 +643,13 @@ static int mrapp_gconf_init(struct mr_instance * instance)
         goto j_parse_error;
 
     /* 全局配置文件路径 */
-    strncpy(instance->g_cfgfile_path, g_cfg_file, sizeof(instance->g_cfgfile_path));
+    snprintf(instance->g_cfgfile_path, sizeof(instance->g_cfgfile_path), "%s", g_cfg_file);
 
     /* 本地配置文件路径 */
-    char __str_cfgfile[MR_STRING_MAX] = {0};
-    strncpy(__str_cfgfile, instance->g_cfgfile_path, sizeof(__str_cfgfile));
+    char str_cfgfile[MR_STRING_MAX] = {0};
+    snprintf(str_cfgfile, sizeof(str_cfgfile), "%s", g_cfg_file);
 
-    snprintf(instance->app_cfgfile_path, sizeof(instance->app_cfgfile_path), "%s/mrapp.%s.conf", dirname(__str_cfgfile),
+    snprintf(instance->app_cfgfile_path, sizeof(instance->app_cfgfile_path), "%s/mrapp.%s.conf", dirname(str_cfgfile),
              instance->appsym);
 
     return RT_SUCCESS;
diff --git a/app/src/monit.c b/app/src/monit.c
index 3be3ce3..225c6c2 100644
--- a/app/src/monit.c
+++ b/app/src/monit.c
@@ -16,6 +16,9 @@ int vdev_instance_stats_get(struct vdev_instance * vdi, struct vdev_stat_info *
     struct vnode_prod_stat * st_prod_ftx = NULL;
     struct vnode_prod_stat * st_prod_ltx = NULL;
 
+    stat_info->nr_rxstream = vdi->nr_rxstream;
+    stat_info->nr_txstream = vdi->nr_txstream;
+
     if(vdi->nr_rxstream > 0)
     {
         st_cons_rx = vnode_mirror_cons_stat_get(vdi->vnode_rx_cons);
@@ -28,15 +31,15 @@ int vdev_instance_stats_get(struct vdev_instance * vdi, struct vdev_stat_info *
         st_prod_ltx = vnode_mirror_prod_stat_get(vdi->vnode_ltx_prod);
     }
 
-    stat_info->nr_rxstream = vdi->nr_rxstream;
-    stat_info->nr_txstream = vdi->nr_txstream;
-
     for (int i = 0; i < vdi->nr_rxstream; i++)
     {
         stat_info->rx_on_line[i] = VNODE_STAT_READ(&st_cons_rx[i].on_line);
         stat_info->rx_deliver[i] = VNODE_STAT_READ(&st_cons_rx[i].deliver);
         stat_info->rx_missed[i] = VNODE_STAT_READ(&st_cons_rx[i].missed);
         stat_info->rx_total_len[i] = VNODE_STAT_READ(&st_cons_rx[i].total_len);
+
+        stat_info->rx_q_len_max[i] = VNODE_STAT_READ(&st_cons_rx[i].q_len_max);
+        stat_info->rx_q_len_avg_max[i] = VNODE_STAT_READ(&st_cons_rx[i].q_len_avg_max);
     }
 
     for (int i = 0; i < vdi->nr_txstream; i++)
@@ -109,6 +112,8 @@ static cJSON * __create_vdev_stats(struct vdev_instance * vdi)
     struct cJSON * j_vdev_value = cJSON_CreateObject();
     /* 瞬时值 */
     struct cJSON * j_vdev_speed = cJSON_CreateObject();
+    /* Gauge */
+    struct cJSON * j_vdev_gauge = cJSON_CreateObject();
 
 #define __JOIN_VDEV_VALUE_STATS_ITEM(item, streams)		do {                    \
 	cJSON_AddItemToObject(j_vdev_value, #item,                                  \
@@ -184,8 +189,25 @@ static cJSON * __create_vdev_stats(struct vdev_instance * vdi)
     __JOIN_VDEV_SPEED_STATS_ITEM(ltx_missed, nr_tx_stream);
     __JOIN_VDEV_SPEED_BYTES_TO_BITS_STAT_ITEM(ltx_total_len, nr_tx_stream);
 
+    cJSON * rx_q_len_array = cJSON_CreateArray();
+    for(unsigned int i = 0; i < nr_rx_stream; i++)
+    {
+        cJSON_AddItemToArray(rx_q_len_array, cJSON_CreateNumber(_stat_info.rx_q_len_max[i]));
+    }
+    cJSON_AddItemToObject(j_vdev_gauge, "rx_q_len", rx_q_len_array);
+
+    cJSON * rx_q_avg_len_array = cJSON_CreateArray();
+    for(unsigned int i = 0; i < nr_rx_stream; i++)
+    {
+        cJSON_AddItemToArray(rx_q_avg_len_array, cJSON_CreateNumber(_stat_info.rx_q_len_avg_max[i]));
+    }
+
+    cJSON_AddItemToObject(j_vdev_gauge, "rx_q_len_avg", rx_q_avg_len_array);
+
+    /*************************************************************************/
     cJSON_AddItemToObject(j_vdev_stats, "accumulative", j_vdev_value);
     cJSON_AddItemToObject(j_vdev_stats, "speed", j_vdev_speed);
+    cJSON_AddItemToObject(j_vdev_stats, "gauge", j_vdev_gauge);
     return j_vdev_stats;
 }
 
diff --git a/include/internal/vdev_define.h b/include/internal/vdev_define.h
index 4112a88..e94ac43 100644
--- a/include/internal/vdev_define.h
+++ b/include/internal/vdev_define.h
@@ -68,6 +68,10 @@ struct vdev_stat_info
     /* Batch Size */
     uint64_t batch_size_total[MR_SID_MAX];
     uint64_t batch_size_count[MR_SID_MAX];
+
+    /* Queue Len */
+    float rx_q_len_max[MR_SID_MAX];
+    float rx_q_len_avg_max[MR_SID_MAX];
 };
 
 /* 虚拟设备信息 */
diff --git a/infra/include/vnode.h b/infra/include/vnode.h
index e7d69ba..557fb76 100644
--- a/infra/include/vnode.h
+++ b/infra/include/vnode.h
@@ -8,10 +8,6 @@ extern "C" {
 #define VNODE_STAT_ENABLE					1
 #endif
 
-#ifndef VNODE_STAT_BY_ATOMIC
-#define VNODE_STAT_BY_ATOMIC                0
-#endif
-
 #include <rte_mbuf.h>
 #include <common.h>
 
@@ -57,41 +53,17 @@ struct vnode_cons_notify * vnode_##_type##_notify_ctx_cons(struct vnode_cons * c
 
 struct vnode_cons_stat
 {
-#if VNODE_STAT_BY_ATOMIC
-	rte_atomic64_t on_line;
-	rte_atomic64_t deliver;
-	rte_atomic64_t missed;
-	rte_atomic64_t total_len;
-#else
 	volatile uint64_t on_line;
 	volatile uint64_t deliver;
 	volatile uint64_t missed;
 	volatile uint64_t total_len;
-#endif
+
+    volatile unsigned int q_len_max;
+    volatile float q_len_avg_max;
 } __rte_cache_aligned;
 
 struct vnode_prod_stat
 {
-#if VNODE_STAT_BY_ATOMIC
-	rte_atomic64_t on_line;
-	rte_atomic64_t deliver;
-	rte_atomic64_t missed;
-	rte_atomic64_t total_len;
-
-    /* clone used */
-    rte_atomic64_t cloned;
-    rte_atomic64_t cloned_fail;
-
-    /* notify */
-    rte_atomic64_t notify_state_waiting;
-    rte_atomic64_t notify_state_running;
-    rte_atomic64_t notify_state_ready;
-
-    /* batch size */
-    rte_atomic64_t batch_size_total;
-    rte_atomic64_t batch_size_count;
-
-#else
 	volatile uint64_t on_line;
 	volatile uint64_t deliver;
 	volatile uint64_t missed;
@@ -108,7 +80,6 @@ struct vnode_prod_stat
     /* batch size */
     volatile uint64_t batch_size_total;
     volatile uint64_t batch_size_count;
-#endif
 
 } __rte_cache_aligned;
 
@@ -126,11 +97,7 @@ struct vnode_cons_notify
     volatile int cons_running_status;
 };
 
-#if VNODE_STAT_BY_ATOMIC
-#define VNODE_STAT_READ(_addr)		rte_atomic64_read(_addr)
-#else
 #define VNODE_STAT_READ(_addr)		*(_addr)
-#endif
 
 enum _vnode_type
 {
@@ -149,7 +116,7 @@ int vnode_mirror_dequeue_burst(struct vnode_cons * cons,
 
 struct vnode * vnode_mirror_create(const char * sym, unsigned int sz_exclusive, unsigned int sz_shared,
                                    unsigned int sz_buffer, unsigned int notify_cons_when_rx,
-                                   unsigned int batch_interval_us);
+                                   unsigned int batch_interval_us, unsigned int en_q_len_monitor);
 
 int vnode_mirror_delete(struct vnode * vnode);
 
diff --git a/infra/src/vnode_common.c b/infra/src/vnode_common.c
index 1c8abce..06a8513 100644
--- a/infra/src/vnode_common.c
+++ b/infra/src/vnode_common.c
@@ -7,29 +7,27 @@
    TODO: Try to use RCU in thread safe mode.
 */
 
-#include <rte_config.h>
 #include <rte_log.h>
 #include <rte_malloc.h>
-#include <rte_errno.h>
 #include <rte_branch_prediction.h>
 #include <rte_ring.h>
 #include <rte_cycles.h>
-#include <rte_version.h>
 
 #include <string.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <assert.h>
-#include <unistd.h> //usleep
-#include <fcntl.h>
-#include <sys/queue.h>
+#include <unistd.h>
 #include <sys/eventfd.h>
+#include <math.h>
+#include <fcntl.h>
 
 #include <common.h>
 #include <rte_mbuf.h>
 
-#include <protect.h>
 #include "vnode_common.h"
+#include <protect.h>
+#include <sys/ioctl.h>
 
 /* What is a tunnel ?
  * tunnel is a fifo with fixed size. we use rte_ring as tunnel.
@@ -251,6 +249,8 @@ struct vnode * __vnode_common_create(const char * sym, unsigned int sz_tunnel, u
     object->sz_tunnel_buffer = sz_tunnel_buffer;
 
     object->notify_cons_when_rx = notify_cons_when_rx;
+    object->en_q_len_monitor = 0;
+
     rte_spinlock_init(&object->lock);
     return object;
 }
@@ -285,6 +285,18 @@ err:
     return NULL;
 }
 
+struct variable_monitor_ioctl_cmd_args
+{
+    pid_t task_id;               // current process id
+    char name[16];
+    void * var_ptr;                  // virtual address
+    int var_len;             // byte
+    long long threshold;         // threshold value
+    unsigned char unsigned_flag; // unsigned flag (true: unsigned, false: signed)
+    unsigned char greater_flag;  // reverse flag (true: >, false: <)
+    unsigned long time_ns;       // timer interval (ns)
+};
+
 struct vnode_cons * __vnode_common_create_cons(struct vnode * vnode, const char * symbol, int nr_consq)
 {
     struct vnode_cons * cons = ZMALLOC(sizeof(struct vnode_cons));
@@ -311,6 +323,35 @@ struct vnode_cons * __vnode_common_create_cons(struct vnode * vnode, const char
                 goto err;
             }
         }
+
+        if (vnode->en_q_len_monitor)
+        {
+            struct variable_monitor_ioctl_cmd_args ioctl_cmd_args = {};
+            ioctl_cmd_args.task_id = getpid();
+
+            snprintf(ioctl_cmd_args.name, sizeof(ioctl_cmd_args.name), "%s-%d", symbol, qid);
+            ioctl_cmd_args.var_ptr = (void *)&cons->stat[qid].q_len_max;
+            ioctl_cmd_args.var_len = sizeof(cons->stat[qid].q_len_max);
+
+            ioctl_cmd_args.threshold = floor(vnode->sz_tunnel * 0.8);
+            ioctl_cmd_args.unsigned_flag = 1;
+            ioctl_cmd_args.greater_flag = 1;
+
+            /* write the cmds by ioctl to the variable monitor modules */
+            int variable_monitor_fd = open("/dev/variable_monitor", 0);
+            if (unlikely(variable_monitor_fd < 0))
+            {
+                MR_ERROR("failed at open /dev/variable_monitor: %s", strerror(errno));
+                goto err;
+            }
+
+            int ioctl_ret = ioctl(variable_monitor_fd, 1, &ioctl_cmd_args);
+            if (unlikely(ioctl_ret < 0))
+            {
+                MR_ERROR("failed at ioctl /dev/variable_monitor: %s", strerror(errno));
+                goto err;
+            }
+        }
     }
 
     rte_spinlock_lock(&vnode->lock);
@@ -481,6 +522,9 @@ struct vnode_cons_stat * __vnode_common_cons_stat_get(struct vnode_cons * cons)
         uint64_t consq_total_len = 0;
         uint64_t consq_missed = 0;
 
+        unsigned int consq_q_len_max = 0;
+        float consq_q_len_avg_max = 0;
+
         for(unsigned int prodq = 0; prodq < prod->nr_prodq; prodq++)
         {
             struct tunnel_block * block = ACCESS_ONCE(cons->block);
@@ -496,12 +540,19 @@ struct vnode_cons_stat * __vnode_common_cons_stat_get(struct vnode_cons * cons)
             consq_deliver += tunnel_desc->deliver;
             consq_total_len += tunnel_desc->total_len;
             consq_missed += tunnel_desc->missed;
+
+            consq_q_len_max = RTE_MAX(tunnel_desc->q_len, consq_q_len_max);
+            consq_q_len_avg_max = RTE_MAX(tunnel_desc->q_len_avg, consq_q_len_avg_max);
         }
 
         cons_stat[consq].on_line = consq_on_line;
         cons_stat[consq].deliver = consq_deliver;
         cons_stat[consq].total_len = consq_total_len;
         cons_stat[consq].missed = consq_missed;
+
+        /* q_len, use the max q value */
+        cons_stat[consq].q_len_max = consq_q_len_max;
+        cons_stat[consq].q_len_avg_max = consq_q_len_avg_max;
     }
 
     return cons->stat;
diff --git a/infra/src/vnode_common.h b/infra/src/vnode_common.h
index 46ebc78..ee5606d 100644
--- a/infra/src/vnode_common.h
+++ b/infra/src/vnode_common.h
@@ -69,6 +69,10 @@ struct tunnel_desc
     uint64_t deliver;
     uint64_t missed;
     uint64_t total_len;
+
+    /* queue length */
+    unsigned int q_len;
+    float q_len_avg;
 };
 
 struct tunnel_block
@@ -85,6 +89,12 @@ struct tunnel_block
     struct tunnel_desc * descs[0];
 };
 
+struct vnode_mem_alloc_hints
+{
+    unsigned int cpu_id_hints[MR_SID_MAX];
+    unsigned int numa_id_hints[MR_SID_MAX];
+};
+
 /* Virtual Data-Node Consumer Structure */
 struct vnode_cons
 {
@@ -133,6 +143,8 @@ struct vnode
     unsigned int notify_cons_when_rx;
     /* batch interval */
     unsigned int batch_interval_tsc;
+    /* q_len monitor */
+    unsigned int en_q_len_monitor;
 
     /* Guarantees one operator(consumer or producer, create or destroy) a time */
     rte_spinlock_t lock __rte_cache_aligned;
diff --git a/infra/src/vnode_mirror.c b/infra/src/vnode_mirror.c
index bf34686..0978a20 100644
--- a/infra/src/vnode_mirror.c
+++ b/infra/src/vnode_mirror.c
@@ -5,10 +5,8 @@
     Date	:	2017-01-09
 */
 
-#include <assert.h>
 #include <rte_mbuf.h>
 #include <rte_ring.h>
-#include <rte_spinlock.h>
 #include <rte_version.h>
 #include <sys/eventfd.h>
 #include <unistd.h>
@@ -18,7 +16,6 @@
 static inline void dist_tunnel_flush(struct vnode_prod * prod, struct vnode_cons * cons, unsigned int prodq,
                                      unsigned int consq, struct tunnel_desc * desc)
 {
-    struct vnode_cons_notify * cons_notify_ctx;
 
     /* nothing to send */
     if (desc->sz_en_buffer_used == 0)
@@ -77,7 +74,9 @@ static inline void dist_tunnel_flush(struct vnode_prod * prod, struct vnode_cons
         rte_cldemote(RTE_PTR_ADD(mbuf, RTE_CACHE_LINE_SIZE * 2));
     }
 
-    unsigned int n_send = rte_ring_sp_enqueue_burst(desc->tunnel_object, (void **)desc->en_buffer, n_to_send, NULL);
+    unsigned int n_free_space;
+    unsigned int n_send =
+        rte_ring_sp_enqueue_burst(desc->tunnel_object, (void **)desc->en_buffer, n_to_send, &n_free_space);
     unsigned int n_send_missed = desc->sz_en_buffer_used - n_send;
 
     /* packet is missed */
@@ -101,7 +100,7 @@ static inline void dist_tunnel_flush(struct vnode_prod * prod, struct vnode_cons
         assert(desc->shared_credict_used <= (desc->tunnel_size - desc->tunnel_exclusive_size));
     }
 
-    cons_notify_ctx = &cons->notify[consq];
+    struct vnode_cons_notify * cons_notify_ctx = &cons->notify[consq];
     if (cons_notify_ctx->enable)
     {
         /* wakeup the cons when it is waiting */
@@ -117,6 +116,10 @@ static inline void dist_tunnel_flush(struct vnode_prod * prod, struct vnode_cons
     desc->missed += n_send_missed;
     desc->total_len += n_send_len;
 
+    /* q_len */
+    desc->q_len = desc->tunnel_size - n_free_space;
+    desc->q_len_avg += 0.2F * ((float)desc->q_len - desc->q_len_avg);
+
     // clear the buffer
     desc->sz_en_buffer_used = 0;
 }
@@ -317,7 +320,7 @@ void vnode_mirror_flush(struct vnode_prod * prod, unsigned int prodq)
 
 struct vnode * vnode_mirror_create(const char * sym, unsigned int sz_exclusive, unsigned int sz_shared,
                                    unsigned int sz_buffer, unsigned int notify_cons_when_rx,
-                                   unsigned int batch_interval_us)
+                                   unsigned int batch_interval_us, unsigned int en_q_len_monitor)
 {
     struct vnode * vnode_common = __vnode_common_create(sym, sz_exclusive, sz_buffer, notify_cons_when_rx);
 
@@ -329,6 +332,7 @@ struct vnode * vnode_mirror_create(const char * sym, unsigned int sz_exclusive,
 
     vnode_common->batch_interval_tsc = batch_interval_us * rte_get_timer_cycles() / US_PER_S;
     vnode_common->sz_shared = sz_shared;
+    vnode_common->en_q_len_monitor = en_q_len_monitor;
 
     rte_atomic32_init(&vnode_common->shared_credict_counter);
     return vnode_common;
diff --git a/infra/test/TestVNode.cc b/infra/test/TestVNode.cc
index 5f43c49..6fb3dab 100644
--- a/infra/test/TestVNode.cc
+++ b/infra/test/TestVNode.cc
@@ -39,7 +39,7 @@ class TestCaseVNodeQueue : public TestCaseVNode
 
     void SetUp() override
     {
-        vnode_ = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0);
+        vnode_ = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0, 0);
         ASSERT_NE(vnode_, nullptr);
 
         assert(prod_ == nullptr);
@@ -82,7 +82,7 @@ struct rte_mempool * TestCaseVNode::pktmbuf_pool_ = nullptr;
 
 TEST_F(TestCaseVNode, CreateAndDeleteInEmptyNode)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0, 0);
     EXPECT_NE(vnode_ptr, nullptr);
 
     int ret = vnode_mirror_delete(vnode_ptr);
@@ -91,7 +91,7 @@ TEST_F(TestCaseVNode, CreateAndDeleteInEmptyNode)
 
 TEST_F(TestCaseVNode, CreateAndDelete)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -109,7 +109,7 @@ TEST_F(TestCaseVNode, CreateAndDelete)
 
 TEST_F(TestCaseVNode, CreateAndDeleteMultiThread)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     /* create multiple thread and run them at same time */
@@ -141,7 +141,7 @@ TEST_F(TestCaseVNode, CreateAndDeleteMultiThread)
 
 TEST_F(TestCaseVNode, TestVNodeProdAndConsLookup)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -164,7 +164,7 @@ TEST_F(TestCaseVNode, TestVNodeProdAndConsLookup)
 
 TEST_F(TestCaseVNode, TestVNodeEnqueue)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 1024, 0, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -212,7 +212,7 @@ TEST_F(TestCaseVNode, TestVNodeEnqueue)
 TEST_F(TestCaseVNode, TestVNodeMultipleThreadEnqueueUseSharedCredict)
 {
     /* create multiple thread */
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 32, 2048, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 32, 2048, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -293,7 +293,7 @@ TEST_F(TestCaseVNode, TestVNodeMultipleThreadEnqueueUseSharedCredict)
 
 TEST_F(TestCaseVNode, TestVNodeEnqueueAndDequeueUseSharedCredict)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 32, 512, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 32, 512, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -372,7 +372,7 @@ TEST_F(TestCaseVNode, TestVNodeEnqueueAndDequeueUseSharedCredict)
 
 TEST_F(TestCaseVNode, TestVNodeMultipleQueueUseSharedCredict)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 32, 512, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 32, 512, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -439,7 +439,7 @@ TEST_F(TestCaseVNode, TestVNodeMultipleQueueUseSharedCredict)
 
 TEST_F(TestCaseVNode, TestVNodeEnqueueUseSharedCredict)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 512, 512, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("m-vnode", 512, 512, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -493,7 +493,7 @@ TEST_F(TestCaseVNode, TestVNodeEnqueueUseSharedCredict)
 
 TEST_F(TestCaseVNodeQueue, MultQueueEnqueue)
 {
-    struct vnode * vnode_ptr = vnode_mirror_create("vnode", 1024, 0, 32, 0, 0);
+    struct vnode * vnode_ptr = vnode_mirror_create("vnode", 1024, 0, 32, 0, 0, 0);
     ASSERT_NE(vnode_ptr, nullptr);
 
     struct vnode_prod * prod;
@@ -534,6 +534,14 @@ TEST_F(TestCaseVNodeQueue, MultQueueEnqueue)
         vnode_mirror_flush(prod, i);
     }
 
+    /* queue length */
+    struct vnode_cons_stat * cons_stat = vnode_mirror_cons_stat_get(cons);
+    EXPECT_EQ(cons_stat[0].on_line, 32);
+    EXPECT_EQ(cons_stat[0].deliver, 32);
+    EXPECT_EQ(cons_stat[0].missed, 0);
+    EXPECT_EQ(cons_stat[0].q_len_max, 32);
+    EXPECT_EQ(cons_stat[0].q_len_avg_max, 32);
+
     int deq_ret = vnode_mirror_dequeue_burst(cons, 0, deq_objs, RTE_DIM(deq_objs));
     EXPECT_EQ(deq_ret, 32);
 
@@ -542,6 +550,15 @@ TEST_F(TestCaseVNodeQueue, MultQueueEnqueue)
     std::set<struct rte_mbuf *> deq_objs_set = std::set<struct rte_mbuf *>(std::begin(deq_objs), std::end(deq_objs));
     EXPECT_EQ(enq_objs_set, deq_objs_set);
 
+    /* after dequeue, the stat should be consistent. */
+    EXPECT_EQ(cons_stat[0].on_line, 32);
+    EXPECT_EQ(cons_stat[0].deliver, 32);
+    EXPECT_EQ(cons_stat[0].missed, 0);
+
+    /* after dequeue, q_len should be zero */
+    EXPECT_EQ(cons_stat[0].q_len_max, 32);
+    EXPECT_LE(cons_stat[0].q_len_avg_max, 32);
+
     rte_pktmbuf_free_bulk(deq_objs, deq_ret);
     vnode_mirror_delete(vnode_ptr);
 }
diff --git a/service/include/sc_devmgr.h b/service/include/sc_devmgr.h
index d795ea3..c8aba15 100644
--- a/service/include/sc_devmgr.h
+++ b/service/include/sc_devmgr.h
@@ -225,6 +225,9 @@ struct mr_dev_desc
     /* cores used for rx,tx */
     cpu_set_t rx_cpu_set;
     cpu_set_t tx_cpu_set;
+
+    /* q_len_monitor */
+    unsigned int en_q_len_monitor;
 };
 
 struct devmgr_main;
diff --git a/service/include/sc_vdev.h b/service/include/sc_vdev.h
index ce12e0c..f7691df 100644
--- a/service/include/sc_vdev.h
+++ b/service/include/sc_vdev.h
@@ -138,7 +138,8 @@ void vdev_stats_last_get(struct vdev * vdev, struct vdev_stat_info * stat_info_l
 
 int vdev_data_create(struct vdev_main * v_main, const char * symbol, unsigned int nr_rxstream, unsigned int nr_txstream,
                      unsigned int sz_tunnel_rx_exclusive, unsigned int sz_tunnel_rx_shared, unsigned int sz_tunnel_tx,
-                     unsigned int sz_buffer, unsigned int batch_interval_in_us, struct rte_mempool * direct_pool);
+                     unsigned int sz_buffer, unsigned int batch_interval_in_us, unsigned int en_q_len_monitor,
+                     struct rte_mempool * direct_pool);
 
 int vdev_loop_create(struct vdev_main * v_main, const char * symbol, unsigned int sz_tunnel, unsigned int sz_buffer,
         struct rte_mempool * direct_pool, struct rte_mempool * indirect_pool);
diff --git a/service/src/core.c b/service/src/core.c
index 6549969..7b158cf 100644
--- a/service/src/core.c
+++ b/service/src/core.c
@@ -490,6 +490,15 @@ static void sc_eal_init(struct sc_main * sc, const char * cmd)
     WRITE_ARG("--legacy-mem");
     devmgr_eal_args_generate(sc->devmgr_main, eal_argv, &eal_argc, MR_SERVICE_MAX_EAL_ARGC);
 
+    /* Force AVX512 */
+    unsigned int max_simd_bit_width = 0;
+    MESA_load_profile_uint_def(sc->local_cfgfile, "eal", "max_simd_bit_width", &max_simd_bit_width, 0);
+
+    if(max_simd_bit_width > 0)
+    {
+        rte_vect_set_max_simd_bitwidth(max_simd_bit_width);
+    }
+
     // DPDK和SYSTEMD的日志级别差1
     unsigned int loglevel = g_logger_level + 1;
     MESA_load_profile_uint_def(sc->local_cfgfile, "eal", "loglevel", &loglevel, loglevel);
diff --git a/service/src/devmgr.c b/service/src/devmgr.c
index 94f0247..db7a6ff 100644
--- a/service/src/devmgr.c
+++ b/service/src/devmgr.c
@@ -461,6 +461,7 @@ struct shmdev_config
 
     unsigned int sz_buffer;
     unsigned int batch_interval_in_us;
+    unsigned int en_q_len_monitor;
 };
 
 void shmdev_config_load(struct devmgr_main * devmgr_main, const char * devsym, struct shmdev_config * cfg_out)
@@ -484,15 +485,18 @@ void shmdev_config_load(struct devmgr_main * devmgr_main, const char * devsym, s
 
     unsigned int default_sz_buffer;
     unsigned int default_batch_interval_in_us;
+    unsigned int default_en_q_len_monitor;
 
     MESA_load_profile_uint_def(cfgfile, "device", "sz_buffer", &default_sz_buffer, 32);
     MESA_load_profile_uint_def(cfgfile, "device", "batch_interval_tsc", &default_batch_interval_in_us, 50);
+    MESA_load_profile_uint_def(cfgfile, "device", "en_q_len_monitor", &default_en_q_len_monitor, 0);
 
     cfg_out->sz_tun_rx_exclusive = default_sz_tun_rx_exclusive;
     cfg_out->sz_tun_tx = default_sz_tun_tx;
     cfg_out->sz_tun_rx_shared = default_sz_tun_rx_shared;
     cfg_out->sz_buffer = default_sz_buffer;
     cfg_out->batch_interval_in_us = default_batch_interval_in_us;
+    cfg_out->en_q_len_monitor = default_en_q_len_monitor;
 }
 
 int shmdev_setup_one_device(struct devmgr_main * devmgr_main, const char * devsym)
@@ -508,8 +512,6 @@ int shmdev_setup_one_device(struct devmgr_main * devmgr_main, const char * devsy
     }
 
     char sym_direct_mempool[MR_SYMBOL_MAX] = {0};
-    char sym_indirect_mempool[MR_SYMBOL_MAX] = {0};
-
     struct shmdev_config shmdev_config = {};
     shmdev_config_load(devmgr_main, devsym, &shmdev_config);
 
@@ -528,7 +530,7 @@ int shmdev_setup_one_device(struct devmgr_main * devmgr_main, const char * devsy
 
     ret = vdev_data_create(sc->vdev_main, devsym, nr_rxstream, nr_txstream, shmdev_config.sz_tun_rx_exclusive,
                            shmdev_config.sz_tun_rx_shared, shmdev_config.sz_tun_tx, shmdev_config.sz_buffer,
-                           shmdev_config.batch_interval_in_us, direct_pool);
+                           shmdev_config.batch_interval_in_us, shmdev_config.en_q_len_monitor, direct_pool);
 
     if (unlikely(ret < 0))
     {
@@ -1642,9 +1644,9 @@ int dpdk_dev_setup_from_candidate(struct devmgr_main * devmgr_main, struct dpdk_
     struct dpdk_dev * dev_dpdk = ZMALLOC(sizeof(struct dpdk_dev));
     MR_VERIFY_MALLOC(dev_dpdk);
 
-    /* copy name and pci address from candidate */
+    /* copy name and pci address from the candidate */
     dev_dpdk->port_id = port_id;
-    strncpy(dev_dpdk->symbol, dev_can->kernel_name, sizeof(dev_dpdk->symbol) - 1);
+    snprintf(dev_dpdk->symbol, sizeof(dev_dpdk->symbol) - 1, "%s", dev_can->kernel_name);
     memcpy(&dev_dpdk->pci_addr, &dev_can->pci_addr, sizeof(dev_dpdk->pci_addr));
 
     /* next, create the dev desc */
@@ -1872,7 +1874,7 @@ void devmgr_eal_args_generate(struct devmgr_main * devmgr_main, char * eal_argv[
 
     /* generate the mlx5 parameters */
     unsigned int mlx5_en_delay_drop = 0;
-    MESA_load_profile_uint_def(local_cfgfile, "eal", "mlx5_en_delay_drop", &mlx5_en_delay_drop, 1);
+    MESA_load_profile_uint_def(local_cfgfile, "eal", "mlx5_en_delay_drop", &mlx5_en_delay_drop, 0);
 
     unsigned int mlx5_en_rxq_pkt_pad = 0;
     MESA_load_profile_uint_def(local_cfgfile, "eal", "mlx5_en_rxq_pkt_pad", &mlx5_en_rxq_pkt_pad, 1);
diff --git a/service/src/node_etherfabric.c b/service/src/node_etherfabric.c
index e8fec17..5e39748 100644
--- a/service/src/node_etherfabric.c
+++ b/service/src/node_etherfabric.c
@@ -620,7 +620,7 @@ static inline int vxlan_encap_constructed_pkt(struct rte_mbuf * mbuf, struct mrb
     p_ipv4_hdr->hdr_checksum = 0;
     p_ipv4_hdr->hdr_checksum = rte_ipv4_cksum(p_ipv4_hdr);
 
-    /* outer-most ether header */
+    /* outermost ether header */
     struct rte_ether_hdr * eth_hdr = (struct rte_ether_hdr *)rte_pktmbuf_prepend(mbuf, sizeof(struct rte_ether_hdr));
 
     eth_hdr->ether_type = htons(RTE_ETHER_TYPE_IPV4);
diff --git a/service/src/vdata.c b/service/src/vdata.c
index a565fb4..62e70a8 100644
--- a/service/src/vdata.c
+++ b/service/src/vdata.c
@@ -225,11 +225,10 @@ static int vdev_data_destory(struct _vdev * _vdev)
     return 0;
 }
 
-
-
 int vdev_data_create(struct vdev_main * v_main, const char * symbol, unsigned int nr_rxstream, unsigned int nr_txstream,
                      unsigned int sz_tunnel_rx_exclusive, unsigned int sz_tunnel_rx_shared, unsigned int sz_tunnel_tx,
-                     unsigned int sz_buffer, unsigned int batch_interval_in_us, struct rte_mempool * direct_pool)
+                     unsigned int sz_buffer, unsigned int batch_interval_in_us, unsigned int en_q_len_monitor,
+                     struct rte_mempool * direct_pool)
 {
     // 检查设备是否已经存在，不允许重复创建
     struct vdev * vdev_info = vdev_lookup(v_main, symbol);
@@ -268,11 +267,11 @@ int vdev_data_create(struct vdev_main * v_main, const char * symbol, unsigned in
 
     /* 创建VNODE */
     _vdev->vnode_rx = vnode_mirror_create(vnode_sym_rx, sz_tunnel_rx_exclusive, sz_tunnel_rx_shared, sz_buffer, 1,
-                                          batch_interval_in_us);
+                                          batch_interval_in_us, en_q_len_monitor);
 
-    _vdev->vnode_tx = vnode_mirror_create(vnode_sym_tx, sz_tunnel_tx, 0, sz_buffer, 0, 0);
-    _vdev->vnode_ftx = vnode_mirror_create(vnode_sym_ftx, sz_tunnel_tx, 0, 0, 0, 0);
-    _vdev->vnode_ltx = vnode_mirror_create(vnode_sym_ltx, sz_tunnel_tx, 0, 0, 0, 0);
+    _vdev->vnode_tx = vnode_mirror_create(vnode_sym_tx, sz_tunnel_tx, 0, sz_buffer, 0, 0, 0);
+    _vdev->vnode_ftx = vnode_mirror_create(vnode_sym_ftx, sz_tunnel_tx, 0, 0, 0, 0, 0);
+    _vdev->vnode_ltx = vnode_mirror_create(vnode_sym_ltx, sz_tunnel_tx, 0, 0, 0, 0, 0);
 
 #define ERR_VERIFY(x, ...)                                                                                             \
     do                                                                                                                 \
@@ -290,10 +289,10 @@ int vdev_data_create(struct vdev_main * v_main, const char * symbol, unsigned in
     ERR_VERIFY(_vdev->vnode_ftx, "Create vdev %s fast tx vnode failed. ", vdev_info->symbol);
     ERR_VERIFY(_vdev->vnode_ltx, "Create vdev %s lock tx vnode failed. ", vdev_info->symbol);
 
-    _vdev->vnode_rx_prod = vnode_mirror_create_prod(_vdev->vnode_rx, "sv", nr_rxstream);
-    _vdev->vnode_tx_cons = vnode_mirror_create_cons(_vdev->vnode_tx, "sv", nr_txstream);
-    _vdev->vnode_ftx_cons = vnode_mirror_create_cons(_vdev->vnode_ftx, "sv", nr_txstream);
-    _vdev->vnode_ctrl_cons = vnode_mirror_create_cons(_vdev->vnode_ltx, "sv", nr_txstream);
+    _vdev->vnode_rx_prod = vnode_mirror_create_prod(_vdev->vnode_rx, "sv", (int)nr_rxstream);
+    _vdev->vnode_tx_cons = vnode_mirror_create_cons(_vdev->vnode_tx, "sv", (int)nr_txstream);
+    _vdev->vnode_ftx_cons = vnode_mirror_create_cons(_vdev->vnode_ftx, "sv", (int)nr_txstream);
+    _vdev->vnode_ctrl_cons = vnode_mirror_create_cons(_vdev->vnode_ltx, "sv", (int)nr_txstream);
 
     /* 校验，申请VNODE、VNODE生产者、消费者是否成功 */
     ERR_VERIFY(_vdev->vnode_rx_prod, "Create vdev %s rx vnode producer failed. ", vdev_info->symbol);