Conflicts:

	tunnat/include/tunnat.h

16 files changed, 161 insertions, 86 deletions

diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index dad4418..2fc6d87 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -39,11 +39,10 @@ build-debug:
     stage: build
     extends: .build_mrzcpd
     tags:
-    - mrzcpd
-    - centos-7.4.1708
+    - share
     variables:
         BUILD_TYPE: "Debug"
-        BUILD_PREFIX: "/opt/dpdk-17.05-debug"
+        BUILD_PREFIX: "/opt/dpdk-17.11.6-debug"
     except:
     - tags
 
@@ -51,11 +50,10 @@ build-release:
     stage: build
     extends: .build_mrzcpd
     tags:
-    - mrzcpd
-    - centos-7.4.1708
+    - share
     variables:
         BUILD_TYPE: "RelWithDebInfo"
-        BUILD_PREFIX: "/opt/dpdk-17.05-rel"
+        BUILD_PREFIX: "/opt/dpdk-17.11.6-rel"
     except:
     - tags
 
@@ -63,11 +61,10 @@ package-debug:
     stage: package
     extends: .package_mrzcpd
     tags:
-    - mrzcpd
-    - centos-7.4.1708
+    - share
     variables:
         BUILD_TYPE: "Debug"
-        BUILD_PREFIX: "/opt/dpdk-17.05-debug"
+        BUILD_PREFIX: "/opt/dpdk-17.11.6-debug"
     artifacts:
         name: "mrzcpd-debug-$CI_COMMIT_REF_NAME-binary"
         paths:
@@ -79,40 +76,13 @@ package-release:
     stage: package
     extends: .package_mrzcpd
     tags:
-    - mrzcpd
-    - centos-7.4.1708
+    - share
     variables:
         BUILD_TYPE: "RelWithDebInfo"
-        BUILD_PREFIX: "/opt/dpdk-17.05-rel"
+        BUILD_PREFIX: "/opt/dpdk-17.11.6-rel"
     artifacts:
         name: "mrzcpd-$CI_COMMIT_REF_NAME-binary"
         paths:
         - build/*.rpm
     only:
     - tags
-
-deploy-debug:
-    stage: deploy
-    tags:
-    - package-server
-    dependencies:
-    - package-debug
-    variables:
-        GIT_STRATEGY: "none"
-        RPM_REPO_PATH: "/data/ci/repos/mrzcpd"
-    extends: .deploy_to_package_server
-    only:
-    - tags
-
-deploy_release:
-    stage: deploy
-    tags:
-    - package-server
-    dependencies:
-    - package-release
-    variables:
-        GIT_STRATEGY: "none"
-        RPM_REPO_PATH: "/data/ci/repos/mrzcpd"
-    extends: .deploy_to_package_server
-    only:
-    - tags
diff --git a/app/include/mrapp.h b/app/include/mrapp.h
index 7d42dc2..c6687c3 100644
--- a/app/include/mrapp.h
+++ b/app/include/mrapp.h
@@ -113,7 +113,7 @@ struct mr_instance
     /* 是否发送免费ARP */
     unsigned int nr_gratuitous_arp_send;
     /* 核心掩码 */
-    cpu_mask_t cpu_mask;
+    cpu_set_t cpu_set;
     /* 线程初始化锁 */
     pthread_mutex_t lock_thread_init;
     /* 当前待分配的线程ID */
@@ -131,4 +131,4 @@ int mrapp_monit_loop(struct mr_instance * instance);
 
 int mrapp_packet_send_burst(struct vdev_instance * vdi, queue_id_t qid, struct rte_mbuf * mbufs[], int nr_mbufs);
 
-int mrapp_packet_fast_send_burst(struct vdev_instance * vdi, queue_id_t qid, struct rte_mbuf * mbufs[], int nr_mbufs);
\ No newline at end of file
+int mrapp_packet_fast_send_burst(struct vdev_instance * vdi, queue_id_t qid, struct rte_mbuf * mbufs[], int nr_mbufs);
diff --git a/app/src/marsio.c b/app/src/marsio.c
index 211b341..d0eade2 100644
--- a/app/src/marsio.c
+++ b/app/src/marsio.c
@@ -737,6 +737,14 @@ void marsio_close_device(struct mr_vdev * vdev)
     return;
 }
 
+static void mask_to_cpuset(uint64_t mask, cpu_set_t * cpusetp)
+{
+    for (unsigned long bit_iter = 0; bit_iter < sizeof(mask) * 8; bit_iter++)
+    {
+        if ((mask & (1ULL << bit_iter))) CPU_SET(bit_iter, cpusetp);
+    }
+}
+
 int marsio_option_set(struct mr_instance * instance, marsio_opt_type_t opt_type, void * opt, size_t sz_opt)
 {
 
@@ -755,7 +763,13 @@ int marsio_option_set(struct mr_instance * instance, marsio_opt_type_t opt_type,
 
     case MARSIO_OPT_THREAD_MASK:
         __CHECK_USER_PARAM(uint64_t);
-        instance->cpu_mask = *(uint64_t *)opt;
+        uint64_t mask = *(uint64_t *)opt;
+        mask_to_cpuset(mask, &instance->cpu_set);
+        ret = 0; break;
+
+    case MARSIO_OPT_THREAD_MASK_IN_CPUSET:
+        __CHECK_USER_PARAM(cpu_set_t);
+        instance->cpu_set = *(cpu_set_t *)opt;
         ret = 0; break;
 
     case MARSIO_OPT_EXIT_WHEN_ERR:
@@ -806,8 +820,10 @@ int marsio_init(struct mr_instance * instance, const char * appsym)
     pthread_mutex_init(&instance->lock_thread_init, NULL);
 
     /* 根据CPU_MASK计算线程数 */
-    if (instance->nr_dataplane_thread == 0 && instance->cpu_mask != 0)
-        instance->nr_dataplane_thread = mask_popcnt(instance->cpu_mask);
+    if (instance->nr_dataplane_thread == 0 && CPU_COUNT(&instance->cpu_set) != 0)
+    {
+        instance->nr_dataplane_thread = CPU_COUNT(&instance->cpu_set);
+    }
 
     if (mrapp_gconf_init(instance) != RT_SUCCESS)
     {
@@ -890,7 +906,7 @@ int marsio_thread_init(struct mr_instance * instance)
         return RT_ERR;
     }
 
-    if (instance->cpu_mask == 0)
+    if (CPU_COUNT(&instance->cpu_set) == 0)
     {
         MR_DEBUG("CPU mask is zero, thread affinity is not allowed.");
         return RT_SUCCESS;
@@ -900,7 +916,7 @@ int marsio_thread_init(struct mr_instance * instance)
     pthread_mutex_lock(&instance->lock_thread_init);
 
     /* 线程绑定 */
-    cpu_id_t cpu_id = mask_location(instance->cpu_mask, instance->to_suppose_tid);
+    cpu_id_t cpu_id = cpu_set_location(&instance->cpu_set, instance->to_suppose_tid);
     if (cpu_id < 0)
     {
         MR_ERROR("Too many threads call thread init, supposed tid is %d", instance->to_suppose_tid);
@@ -939,4 +955,4 @@ out:
 int marsio_destory(struct mr_instance * instance)
 {
     return 0;
-}
\ No newline at end of file
+}
diff --git a/app/src/rawio.c b/app/src/rawio.c
index 1423bc3..871a41e 100644
--- a/app/src/rawio.c
+++ b/app/src/rawio.c
@@ -231,6 +231,12 @@ err:
     return RT_ERR;
 }
 
+void marsio_send_burst_flush(struct mr_sendpath * sendpath, queue_id_t sid)
+{
+    vnode_mirror_flush(sendpath->target_vdi->vnode_tx_prod, sid);
+    vnode_mirror_flush(sendpath->target_vdi->vnode_ftx_prod, sid);
+}
+
 int marsio_idle_poll(struct mr_instance * instance)
 {
     return 0;
@@ -279,4 +285,4 @@ int marsio_ipv4_header_construct(marsio_buff_t * buff, uint32_t s_ip, uint32_t d
     ip_hdr->hdr_checksum = 0;
     ip_hdr->hdr_checksum = rte_ipv4_cksum(ip_hdr);
     return 0;
-}
\ No newline at end of file
+}
diff --git a/app/src/version.map b/app/src/version.map
index 7b4e91d..b526474 100644
--- a/app/src/version.map
+++ b/app/src/version.map
@@ -13,6 +13,7 @@ global:
 	marsio_recv_all_burst;
 	marsio_send_burst;
 	marsio_send_burst_with_options;
+	marsio_send_burst_flush;
 	marsio_buff_malloc_device;
 	marsio_buff_malloc_global;
 	marsio_buff_free;
@@ -71,4 +72,4 @@ global:
 	marsio_buff_get_timestamp_ex;
 	
 local:	*;
-};
\ No newline at end of file
+};
diff --git a/cmake/FindDPDK.cmake b/cmake/FindDPDK.cmake
index 60539d3..6b037c6 100644
--- a/cmake/FindDPDK.cmake
+++ b/cmake/FindDPDK.cmake
@@ -106,7 +106,7 @@ endif()
 
 # Additional library
 file(GLOB DPDK_LIBRARY_GEN "${DPDK_LIBRARY_DIR}/librte*.a" "${DPDK_LIBRARY_DIR}/libeth*.a")
-set(DPDK_LIBRARY -Wl,--whole-archive ${DPDK_LIBRARY_GEN} -Wl,--no-whole-archive rt m dl pcap)
+set(DPDK_LIBRARY -Wl,--whole-archive ${DPDK_LIBRARY_GEN} -Wl,--no-whole-archive rt m dl pcap numa)
 
 # Force Include
 set(DPDK_FORCE_INCLUDE "-include ${DPDK_INCLUDE_DIR}/rte_config.h")
@@ -121,4 +121,4 @@ mark_as_advanced(
     DPDK_LIBRARY
 	DPDK_C_PREDEFINED
 	DPDK_CXX_PREDEFINED
-)
\ No newline at end of file
+)
diff --git a/include/external/marsio.h b/include/external/marsio.h
index f18e416..71db1b6 100644
--- a/include/external/marsio.h
+++ b/include/external/marsio.h
@@ -28,6 +28,11 @@ typedef enum
 	 */
 	MARSIO_OPT_EXIT_WHEN_ERR,
 
+	/* 拓展数据面线程绑定掩码，适应大于64个核处理器的硬件平台
+	 * 设置该掩码后，数据面线程数选项将被忽略
+	 */
+    MARSIO_OPT_THREAD_MASK_IN_CPUSET,
+
 } marsio_opt_type_t;
 
 typedef enum
@@ -146,6 +151,7 @@ int marsio_recv_all_burst(struct mr_instance * instance, queue_id_t qid, marsio_
 int marsio_send_burst(struct mr_sendpath * sendpath,  queue_id_t qid, marsio_buff_t * mbufs[], int nr_mbufs);
 int marsio_send_burst_with_options(struct mr_sendpath * sendpath, queue_id_t sid, marsio_buff_t * mbufs[],
     int nr_mbufs, uint16_t options);
+void marsio_send_burst_flush(struct mr_sendpath * sendpath, queue_id_t sid);
 
 int marsio_udp_header_construct(marsio_buff_t * buff, uint16_t s_port, uint16_t d_port);
 int marsio_ipv4_header_construct(marsio_buff_t * buff, uint32_t s_ip, uint32_t d_ip, uint8_t proto);
@@ -252,4 +258,4 @@ void marsio_shared_mem_free(struct mr_instance * instance, void * mem);
 
 #ifdef __cplusplus
 }
-#endif
\ No newline at end of file
+#endif
diff --git a/infra/include/common.h b/infra/include/common.h
index f01f6c9..5125f29 100644
--- a/infra/include/common.h
+++ b/infra/include/common.h
@@ -242,19 +242,19 @@ static unsigned int inline mask_is_set(mask_t mask, unsigned int id)
     return !!(mask & (1ULL << id));
 }
 
-static int inline mask_location(mask_t mask, unsigned int id)
+static int inline cpu_set_location(cpu_set_t * cpu_set_p, unsigned int id)
 {
-	unsigned int bit_count = 0;
-	for (unsigned bit_iter = 0; bit_iter < sizeof(mask) * 8; bit_iter++)
-	{
-		if ((mask & (1ULL << bit_iter)))
-		{
-			if (bit_count == id) return bit_iter;
-			else bit_count++;
-		}
-	}
+    unsigned int bit_count = 0;
+    for (unsigned bit_iter = 0; bit_iter < CPU_SETSIZE; bit_iter++)
+    {
+        if (CPU_ISSET(bit_iter, cpu_set_p))
+        {
+            if (bit_count == id) return bit_iter;
+            else bit_count++;
+        }
+    }
 
-	return -1;
+    return -1;
 }
 
 static int inline str_to_mask(const char * str, mask_t * out_mask)
@@ -344,4 +344,4 @@ static inline int mr_thread_setname(pthread_t id, const char *name)
 
 #ifdef __cplusplus
 }
-#endif
\ No newline at end of file
+#endif
diff --git a/infra/src/vnode_mirror.c b/infra/src/vnode_mirror.c
index a1a0deb..50278c8 100644
--- a/infra/src/vnode_mirror.c
+++ b/infra/src/vnode_mirror.c
@@ -385,4 +385,4 @@ __USE_COMMON_VNODE_PROD_LOOKUP(mirror)
 __USE_COMMON_VNODE_PROD_STAT_GET(mirror)
 __USE_COMMON_VNODE_CONS_STAT_GET(mirror)
 __USE_COMMON_VNODE_PROD_ATTACH(mirror)
-__USE_COMMON_VNODE_CONS_ATTACH(mirror)
\ No newline at end of file
+__USE_COMMON_VNODE_CONS_ATTACH(mirror)
diff --git a/service/include/sc_common.h b/service/include/sc_common.h
index 5f7b3b7..f956451 100644
--- a/service/include/sc_common.h
+++ b/service/include/sc_common.h
@@ -50,7 +50,7 @@ struct sc_main
 	/* 状态监测文件 */
 	char local_monitfile[MR_STRING_MAX];
 	/* CPU掩码，本进程在这个CPU上运行 */
-	cpu_mask_t cpu_mask;
+	cpu_set_t cpu_set;
     /* Master线程的CPU亲和性设置（EAL之前） */
     cpu_set_t cpu_set_before_eal;
     /* Master线程的CPU亲和性设置（EAL之后） */
@@ -105,4 +105,4 @@ struct sc_main
 };
 
 struct sc_main * sc_main_get();
-extern unsigned int g_keep_running;
\ No newline at end of file
+extern unsigned int g_keep_running;
diff --git a/service/src/core.c b/service/src/core.c
index 01e9733..1065aa3 100644
--- a/service/src/core.c
+++ b/service/src/core.c
@@ -268,7 +268,7 @@ void sc_config_dump(struct sc_main * sc)
 {
     MR_INFO(" ");
     MR_INFO("Runtime Information");
-    MR_INFO("   Dataplane thread count         : %d", mask_popcnt(sc->cpu_mask));
+    MR_INFO("   Dataplane thread count         : %d", CPU_COUNT(&sc->cpu_set));
     MR_INFO("   Packet burst on RX direction   : %d", sc->nr_rx_burst);
     MR_INFO("   Packet burst on TX direction   : %d", sc->nr_tx_burst);
 	MR_INFO("   Packet distribute mode         : %s", ldbc_str_dist_mode(sc->dist_object));
@@ -367,12 +367,14 @@ static void sc_eal_init(struct sc_main * sc, const char * cmd)
 
     /* After EAL, copy some parameters to sc_main */
     unsigned int lcore_id = 0;
+    CPU_ZERO(&sc->cpu_set);
+
     RTE_LCORE_FOREACH(lcore_id)
     {
-        sc->cpu_mask |= 1ULL << lcore_id;
+        CPU_SET(lcore_id, &sc->cpu_set);
     }
 
-    sc->nr_serv_thread = mask_popcnt(sc->cpu_mask);
+    sc->nr_serv_thread = CPU_COUNT(&sc->cpu_set);
 
     /* EAL初始化后的线程亲和性 */
     ret = pthread_getaffinity_np(pthread_self(), sizeof(cpu_set_t), &sc->cpu_set_after_eal);
@@ -862,4 +864,4 @@ quit:
     phydev_deinit(sc);
     if(pdump_inited) rte_pdump_uninit();
     return 0;
-}
\ No newline at end of file
+}
diff --git a/service/src/fwd.c b/service/src/fwd.c
index e8978d2..b3015ac 100644
--- a/service/src/fwd.c
+++ b/service/src/fwd.c
@@ -274,7 +274,7 @@ static int forward_rule_setup(struct sc_main * sc, struct sw_forward_main * fwd_
     //TODO: 队列映射
     for (int i = 0; i < sc->nr_serv_thread; i++)
     {
-        int ret = mask_location(sc->cpu_mask, i);
+        int ret = cpu_set_location(&sc->cpu_set, i);
         fwd_rule->map_rx_queue[ret] = i;
         fwd_rule->map_tx_queue[ret] = i;
         MR_VERIFY(ret >= 0);
@@ -475,4 +475,4 @@ void sw_forward_idle_loop(struct sc_main * sc, unsigned int lcore_id)
     }
 
     return;
-}
\ No newline at end of file
+}
diff --git a/service/src/phydev.c b/service/src/phydev.c
index 4ac8703..45c84ad 100644
--- a/service/src/phydev.c
+++ b/service/src/phydev.c
@@ -7,6 +7,7 @@
 
 #include <rte_malloc.h>
 #include <rte_pci.h>
+#include <rte_bus_pci.h>
 #include <rte_config.h>
 #include <rte_debug.h>
 #include <rte_ethdev.h>
@@ -254,13 +255,11 @@ static int phydev_setup(struct sc_main * sc,
     unsigned nr_txq_use = 0;
 
     // TODO: 目前实现按每个服务线程处理一个队列考虑
-    dev->nr_rxq = mask_popcnt(sc->cpu_mask);
-    dev->nr_txq = mask_popcnt(sc->cpu_mask);
+    dev->nr_rxq = CPU_COUNT(&sc->cpu_set);
+    dev->nr_txq = CPU_COUNT(&sc->cpu_set);
     calc_phydev_queue(dev, &nr_rxq_use, &nr_txq_use);
 
-    retval = rte_eth_dev_configure(dev->port_id, nr_rxq_use,
-        nr_txq_use, &local_eth_conf);
-
+    retval = rte_eth_dev_configure(dev->port_id, nr_rxq_use, nr_txq_use, &local_eth_conf);
     if (retval != 0)
     {
         MR_ERROR("Physical device %s configure error: %s, errno = %d", dev->symbol,
@@ -589,7 +588,7 @@ static int phydev_scan_raw_socket(struct sc_main * sc, struct phydev_main * phyd
 
     for (int i = 0; i < nr_rawsyms; i++)
     {
-        uint8_t port_id;
+        uint16_t port_id;
         char str_attach_args[MR_STRING_MAX];
 
         snprintf(str_attach_args, sizeof(str_attach_args),
@@ -627,8 +626,7 @@ static int phydev_scan_uio(struct phydev_main * phydev_main)
 
         if (dev_info.pci_dev != NULL)
         {
-            phydev = phydev_lookup_by_pci_addr(phydev_main,
-                dev_info.pci_dev->addr);
+            phydev = phydev_lookup_by_pci_addr(phydev_main, dev_info.pci_dev->addr);
         }
 
         /* HWFILE里面没有定义，或虚拟设备 */
diff --git a/tunnat/include/tunnat.h b/tunnat/include/tunnat.h
index 2c8ab30..682fb96 100644
--- a/tunnat/include/tunnat.h
+++ b/tunnat/include/tunnat.h
@@ -22,6 +22,60 @@ extern unsigned int g_ctrlzone_id;
 
 struct TunnatThreadInstance;
 
+#ifndef MR_TUNNAT_DEFAULT_CFG_FILE
+#define MR_TUNNAT_DEFAULT_CFG_FILE                  "/etc/mrtunnat.conf"
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_MNT_FILE
+#define MR_TUNNAT_DEFAULT_MNT_FILE                  "/var/run/mrzcpd/mrmonit.tunnat"
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_APPSYM
+#define MR_TUNNAT_DEFAULT_APPSYM                    "tunnat"
+#endif
+
+#ifndef MR_TUNNAT_LCORE_MAX
+#define MR_TUNNAT_LCORE_MAX                         128
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_SESSION_MAX
+#define MR_TUNNAT_DEFAULT_SESSION_MAX               8192
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_SESSION_TUPLE4
+#define MR_TUNNAT_DEFAULT_SESSION_TUPLE4            0
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_SESSION_SLOT_MAX
+#define MR_TUNNAT_DEFAULT_SESSION_SLOT_MAX          4096
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_NR_BURST
+#define MR_TUNNAT_DEFAULT_NR_BURST                  32
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_SESSION_EXPIRE_TIME
+#define MR_TUNNAT_DEFAULT_SESSION_EXPIRE_TIME       0
+#endif
+
+#ifndef MR_TUNNAT_DEFAULT_IDLE_POLL_THRESHOLD
+#define MR_TUNNAT_DEFAULT_IDLE_POLL_THRESHOLD       1000000
+#endif
+
+enum bridge_mode
+{
+    MR_TUNNAT_BRIDGE_MODE_DISABLE,
+    MR_TUNNAT_BRIDGE_MODE_ONE_ARM,
+    MR_TUNNAT_BRIDGE_MODE_TWO_ARM,
+};
+
+enum ctrlzone_addrinfo_type
+{
+    MR_TUNNAT_CTRLZONE_ADDR_INFO_DISABLE,
+    MR_TUNNAT_CTRLZONE_ADDR_INFO_TUNNEL,
+    MR_TUNNAT_CTRLZONE_ADDR_INFO_REAL
+};
+
 /* 全局句柄 */
 struct TunnatInstance
 {
@@ -39,7 +93,12 @@ struct TunnatInstance
     /* Burst数量 */
     unsigned int nr_burst;
     /* 运行数据面线程核心掩码 */
-    cpu_mask_t coremask;
+    cpu_set_t coremask{};
+    /* IDLE POLL Threshold */
+    unsigned int idle_threshold{MR_TUNNAT_DEFAULT_IDLE_POLL_THRESHOLD};
+
+    /* 会话表使用四元组还是二元组 */
+    unsigned int sess_tb_order_by_tuple4{MR_TUNNAT_DEFAULT_SESSION_TUPLE4};
     /* 会话表最大会话数 */
     unsigned int nr_max_sessions;
     /* 会话表槽数量 */
diff --git a/tunnat/src/core.cc b/tunnat/src/core.cc
index b83a339..c365576 100644
--- a/tunnat/src/core.cc
+++ b/tunnat/src/core.cc
@@ -200,14 +200,13 @@ int tunnat_mrinstance_setup(TunnatInstance * instance)
         return RT_ERR;
     }
 
-    uint64_t coremask = 0;
+    CPU_ZERO(&instance->coremask);
     for (int i = 0; i < lcore_id_count; i++)
     {
-        coremask |= 1ULL << lcore_id_range[i];
+        CPU_SET(lcore_id_range[i], &instance->coremask);
     }
 
-    instance->coremask = coremask;
-    instance->nr_thread = mask_popcnt(instance->coremask);
+    instance->nr_thread = CPU_COUNT(&instance->coremask);
 
     // 读应用名称
     char cstr_appsym[MR_STRING_MAX];
@@ -248,6 +247,11 @@ int tunnat_mrinstance_setup(TunnatInstance * instance)
     }
 
     instance->mr_instance = _mr_instance;
+
+    // Idle threshold
+    MESA_load_profile_uint_def(instance->cfgfile.c_str(), "tunnat", "idle_threshold",
+        &instance->idle_threshold, MR_TUNNAT_DEFAULT_IDLE_POLL_THRESHOLD);
+
     return RT_SUCCESS;
 }
 
diff --git a/tunnat/src/runtime.cc b/tunnat/src/runtime.cc
index f2ce931..ee10010 100644
--- a/tunnat/src/runtime.cc
+++ b/tunnat/src/runtime.cc
@@ -479,11 +479,24 @@ void * tunnat_thread_loop(void * arg)
     auto & virtdev = instance->virtdevs[0];
     auto & phydev = instance->phydevs[0];
 
+    /* Idle Counter*/
+    unsigned long idle_counter = 0;
+
     while (g_keep_running)
     {
         __phy_to_virt_one_device(instance, th_instance, &phydev, &virtdev);
         __virt_to_phy_one_device(instance, th_instance, &phydev, &virtdev);
+
+        if(likely(instance->idle_threshold > 0) &&
+           unlikely(idle_counter >= instance->idle_threshold))
+        {
+            marsio_send_burst_flush(virtdev.vdev_sendpath, th_instance->thread_id);
+            marsio_send_burst_flush(phydev.vdev_sendpath, th_instance->thread_id);
+            idle_counter = 0;
+        }
+
+        idle_counter++;
     }
 
-    return 0;
+    return (void *)nullptr;
 }