[PACKET_IO] add pio_packet structure for pcap_live/pcap_file mode

17 files changed, 157 insertions, 360 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 8ef7c5c..0cca510 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -14,6 +14,7 @@ pkg_check_modules(SYSTEMD REQUIRED systemd)
 
 
 include_directories(${CMAKE_SOURCE_DIR}/sdk/include)
+include_directories(/opt/mrzcpd/include)
 
 #add vendor and source directory
 add_subdirectory(vendor)
diff --git a/sdk/include/marsio.h b/sdk/include/marsio.h
deleted file mode 100644
index 8ecd844..0000000
--- a/sdk/include/marsio.h
+++ /dev/null
@@ -1,268 +0,0 @@
-/*
- *	\brief MARSIO Userspace ZeroCopy Driver Version 4
- *
- *	This is the user api header file of MARSIOv4 ZeroCopy Driver
- *
- *	\author	Qiuwen Lu<[email protected]>
- *			Institute of Information Engineering, Chinese Academy of Sciences
- *
- *	\date	2016-12-01
- */
-
-#pragma once
-
-typedef enum
-{
-    /* 数据面线程数，没有缺省值 */
-    MARSIO_OPT_THREAD_NUM,
-    /* 数据面线程绑定掩码，没有缺省值，设置该掩码后，数据面线程数选项将被忽略。*/
-    MARSIO_OPT_THREAD_MASK,
-    /* 线程绑定模式，类型：uint32_t */
-    MARSIO_OPT_THREAD_AFFINITY_MODE,
-
-    /* 处理SIG信号，缺省值：类型：uint32_t，关闭（0）
-     * 启用该选项后，将自动处理SIGINT、SIGTERM信号
-     */
-    MARSIO_OPT_EXIT_WHEN_ERR,
-
-    /* 拓展数据面线程绑定掩码，适应大于64个核处理器的硬件平台
-     * 设置该掩码后，数据面线程数选项将被忽略
-     */
-    MARSIO_OPT_THREAD_MASK_IN_CPUSET,
-
-} marsio_opt_type_t;
-
-typedef enum
-{
-    /* 发送后不释放数据包，由应用自行释放数据包 */
-    MARSIO_SEND_OPT_NO_FREE = 1 << 0,
-    /* 发送时计算发包哈希值，用于分流 */
-    MARSIO_SEND_OPT_REHASH = 1 << 1,
-    /* 快速报文路径 */
-    MARSIO_SEND_OPT_FAST = 1 << 2,
-    /* 报文追踪标记 */
-    MARSIO_SEND_OPT_TRACE = 1 << 3,
-    /* 控制报文标记 */
-    MARSIO_SEND_OPT_CTRL = 1 << 4
-
-} marsio_opt_send_t;
-
-enum mr_sendpath_type
-{
-    /* 去往特定虚设备的发包路径 */
-    MR_SENDPATH_VDEV,
-    /* 路由查表确定发包路径 */
-    MR_SENDPATH_ROUTE_NORMAL,
-    /* 特定设备路由查表确定发包路径 */
-    MR_SENDPATH_ROUTE_SPEC_DEV,
-    /* MAX标记 */
-    MR_SENDPATH_MAX
-};
-
-enum mr_sendpath_option
-{
-    /* 构建四层报文头 */
-    MR_SENDPATH_OPT_BUILD_L4 = 0,
-    /* 构建三层报文头 */
-    MR_SENDPATH_OPT_BUILD_L3 = 1,
-    /* 构建二层报文头 */
-    MR_SENDPATH_OPT_BUILD_L2 = 2,
-    /* 构建前Hook点回调 */
-    MR_SENDPATH_OPT_HOOK_PREBUILD = 50,
-    /* 构建后Hook点回调 */
-    MR_SENDPATH_OPT_HOOK_POSTBUILD = 51,
-};
-
-enum mr_clone_options
-{
-    /* 拷贝区域 */
-    MR_BUFF_CLONE_DATA = 1 << 0,
-    MR_BUFF_CLONE_BUFF = 1 << 1,
-    MR_BUFF_CLONE_CTRLZONE = 1 << 2,
-};
-
-enum mr_thread_affinity_mode
-{
-    /* 禁用线程亲和性设置 */
-    MR_THREAD_AFFINITY_DISABLE = 0,
-    /* 自动线程亲和性设置 */
-    MR_THREAD_AFFINITY_AUTO = 1,
-    /* 自定义线程亲和性设置 */
-    MR_THREAD_AFFINITY_USER = 255
-};
-
-enum mr_timestamp_type
-{
-    /* 从网卡收取时或报文缓冲区申请时的时间戳 */
-    MR_TIMESTAMP_RX_OR_ALLOC = 0,
-};
-
-enum mr_buff_metadata_type
-{
-    /* 分流哈希值，会根据该值计算分流队列 */
-    MR_BUFF_REHASH_INDEX = 0,
-    /* 网卡剥离的VLAN标签 */
-    MR_BUFF_METADATA_VLAN_TCI = 1,
-};
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <netinet/in.h>
-#include <stddef.h>
-#include <stdint.h>
-#include <stdio.h>
-
-#ifndef MARSIO_SOCKET_ID_ANY
-#define MARSIO_SOCKET_ID_ANY -1
-#endif
-
-#ifndef MARSIO_LCORE_ID_ANY
-#define MARSIO_LCORE_ID_ANY -1
-#endif
-
-typedef uint32_t device_id_t;
-typedef uint32_t port_id_t;
-typedef uint32_t queue_id_t;
-typedef uint32_t thread_id_t;
-typedef void marsio_buff_t;
-
-struct mr_sendpath;
-struct mr_instance;
-struct mr_vdev;
-
-struct mr_instance * marsio_create();
-struct mr_instance * marsio_current();
-
-typedef int (*fn_sendpath_hook_t)(struct mr_sendpath * sendpath, marsio_buff_t * mbuf[], unsigned int nr_mbuf,
-                                  void * arg);
-
-int marsio_option_get(struct mr_instance * instance, int opt_type, void * out_opt, size_t out_opt_buffer);
-int marsio_option_set(struct mr_instance * instance, marsio_opt_type_t opt_type, void * opt, size_t sz_opt);
-int marsio_init(struct mr_instance * instance, const char * appsym);
-int marsio_thread_init(struct mr_instance * instance);
-int marsio_destory(struct mr_instance * instance);
-
-struct mr_vdev * marsio_open_device(struct mr_instance * instance, const char * devsym, unsigned int nr_rxstream,
-                                    unsigned int nr_txstream);
-
-void marsio_close_device(struct mr_vdev * vdev);
-
-struct mr_vdev * marsio_device_lookup(struct mr_instance * instance, const char * devsym);
-
-int marsio_recv_burst(struct mr_vdev * vdev, queue_id_t qid, marsio_buff_t * mbufs[], int nr_mbufs);
-int marsio_recv_all_burst(struct mr_instance * instance, queue_id_t qid, marsio_buff_t * mbufs[], int nr_mbufs);
-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);
-
-struct mr_sendpath * marsio_sendpath_create_by_droute(struct mr_vdev * dest_device, struct in_addr addr);
-struct mr_sendpath * marsio_sendpath_create_by_route(struct mr_instance * instance, struct in_addr addr);
-struct mr_sendpath * marsio_sendpath_create_by_vdev(struct mr_vdev * dest_device);
-
-struct mr_sendpath * marsio_sendpath_create(struct mr_instance * instance, int type, ...);
-int marsio_sendpath_option_set(struct mr_instance * instance, struct mr_sendpath * sendpath, int opt, ...);
-int marsio_sendpath_option_get(struct mr_instance * instance, struct mr_sendpath * sendpath, int opt, ...);
-void marsio_sendpath_destory(struct mr_sendpath * sendpath);
-
-// simply get control zone data ptr.
-void * marsio_buff_ctrlzone(marsio_buff_t * m, uint8_t id);
-
-// a safe way to read control zone data.
-void * marsio_buff_ctrlzone_data(marsio_buff_t * m, uint8_t id, uint8_t * size);
-
-// a safe way to set control zone data.
-void marsio_buff_ctrlzone_set(marsio_buff_t * m, uint8_t id, void * ptr_data, uint8_t size);
-
-void marsio_buff_reset(marsio_buff_t * m);
-
-marsio_buff_t * marsio_buff_getnext_seg(marsio_buff_t * m);
-marsio_buff_t * marsio_buff_getnext_pkt(marsio_buff_t * m);
-
-void marsio_buff_append_pkt(marsio_buff_t * head, marsio_buff_t * next);
-void marsio_buff_append_seg(marsio_buff_t * head, marsio_buff_t * next);
-void marsio_buff_chain_pkt(marsio_buff_t * pkt, marsio_buff_t * next);
-
-uint16_t marsio_buff_headroom(const marsio_buff_t * m);
-uint16_t marsio_buff_tailroom(const marsio_buff_t * m);
-
-marsio_buff_t * marsio_buff_getnext_seg(marsio_buff_t * m);
-marsio_buff_t * marsio_buff_getnext_pkt(marsio_buff_t * m);
-
-char * marsio_buff_mtod(marsio_buff_t * m);
-
-uint32_t marsio_buff_buflen(marsio_buff_t * m);
-
-uint32_t marsio_buff_datalen(marsio_buff_t * m);
-
-char * marsio_buff_prepend(marsio_buff_t * m, uint16_t len);
-
-char * marsio_buff_append(marsio_buff_t * m, uint16_t len);
-
-char * marsio_buff_adj(marsio_buff_t * m, uint16_t len);
-
-int marsio_buff_trim(marsio_buff_t * m, uint16_t len);
-
-char * marsio_buff_offset_set(marsio_buff_t * m, off_t offset, int is_relative);
-
-off_t marsio_buff_offset_get(marsio_buff_t * m);
-
-/* 支持写时复制的报文修改裁剪函数 */
-marsio_buff_t * marsio_buff_prepend_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out);
-
-marsio_buff_t * marsio_buff_append_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out);
-
-marsio_buff_t * marsio_buff_adj_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out);
-
-marsio_buff_t * marsio_buff_trim_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out);
-
-uint16_t marsio_buff_headroom(const marsio_buff_t * m);
-
-uint16_t marsio_buff_tailroom(const marsio_buff_t * m);
-
-uint32_t marsio_get_pkt_type(marsio_buff_t * m);
-
-void marsio_pktmbuf_dump(FILE * f, const marsio_buff_t * m, unsigned dump_len);
-
-marsio_buff_t * marsio_buff_clone_deep(struct mr_instance * instance, marsio_buff_t * md, int socket_id, int thread_id);
-
-marsio_buff_t * marsio_buff_clone_with_options(struct mr_instance * instance, marsio_buff_t * md, int socket_id,
-                                               int thread_id, uint16_t options);
-
-int marsio_buff_malloc_device(struct mr_vdev * vdev, marsio_buff_t * marsio_buff[], unsigned int nr_mbufs,
-                              int socket_id, int thread_id);
-
-int marsio_buff_malloc_global(struct mr_instance * instance, marsio_buff_t * marsio_buff[], unsigned int nr_mbufs,
-                              int socket_id, int thread_id);
-
-void marsio_buff_free(struct mr_instance * instance, marsio_buff_t * marsio_buff[], unsigned int nr_mbufs,
-                      int socket_id, int thread_id);
-
-int marsio_buff_is_ctrlbuf(marsio_buff_t * m);
-
-void marsio_buff_set_ctrlbuf(marsio_buff_t * m);
-
-marsio_buff_t * marsio_buff_malloc_smartoffload(struct mr_vdev * vdev, const char * pkt, unsigned int pkt_len);
-
-uint64_t marsio_buff_get_timestamp(marsio_buff_t * m);
-void marsio_buff_set_timestamp(marsio_buff_t * m, uint64_t timestamp);
-int marsio_buff_get_timestamp_ex(marsio_buff_t * m, enum mr_timestamp_type ts_type, struct timespec * ts);
-
-void marsio_buff_set_rehash_index(marsio_buff_t * m, uint32_t hash);
-uint32_t marsio_buff_get_rehash_index(marsio_buff_t * m);
-
-int marsio_buff_get_metadata(marsio_buff_t * m, enum mr_buff_metadata_type type, void * data, unsigned int sz_data);
-int marsio_buff_set_metadata(marsio_buff_t * m, enum mr_buff_metadata_type type, void * data, unsigned int sz_data);
-int marsio_buff_unset_metadata(marsio_buff_t * m, enum mr_buff_metadata_type type);
-
-void * marsio_shared_mem_alloc(struct mr_instance * instance, size_t len);
-void marsio_shared_mem_free(struct mr_instance * instance, void * mem);
-
-#ifdef __cplusplus
-}
-#endif
diff --git a/sdk/include/packet.h b/sdk/include/packet.h
index ba6493e..d8768ed 100644
--- a/sdk/include/packet.h
+++ b/sdk/include/packet.h
@@ -1,12 +1,5 @@
 #pragma once
 
-#include <sys/queue.h>
-#include "marsio.h"
-
-struct packet {
-    void *raw_pkt;
-    uint32_t pkt_len;
-    struct packet *prev;
-    struct packet *next;
-};
+#include <stdint.h>
 
+struct packet;
\ No newline at end of file
diff --git a/src/common/global_var.h b/src/common/global_var.h
index 3635346..15aaef9 100644
--- a/src/common/global_var.h
+++ b/src/common/global_var.h
@@ -58,6 +58,9 @@ struct packet_io_config {
 
     /* promiscuous value */
     int promisc;
+
+    /* marsio ctrlzone id */
+    int mr_ctrlzone_id;
 };
 
 struct lib_config {
diff --git a/src/common/packet_queue.h b/src/common/packet_queue.h
deleted file mode 100644
index a8b6529..0000000
--- a/src/common/packet_queue.h
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
-**********************************************************************************************
-*	File: packet_queue.h
-*   Description: packet queue structure and api
-*	Authors: Liu WenTan <[email protected]>
-*	Date:    2022-07-15
-*   Copyright: (c) 2018-2022 Geedge Networks, Inc. All rights reserved.
-***********************************************************************************************
-*/
-
-#pragma once
-
-#include "../../sdk/include/packet.h"
-
-#define PKT_QUEUE_MAX_NUM 256
-
-#define SET_PKT_LEN(p, len) do { \
-	(p)->pkt_len = (len); \
-	} while(0)
-
-#define GET_PKT_DIRECT_DATA(p) (uint8_t *)((p) + 1)
-
-struct packet_queue {
-	struct packet *top;
-	struct packet *bot;
-	uint32_t len;
-	pthread_mutex_t mutex_q;
-};
-
-void packet_enqueue(struct packet_queue *, struct packet *);
-struct packet *packet_dequeue(struct packet_queue *);
-void release_packet_queue(struct packet_queue *);
\ No newline at end of file
diff --git a/src/common/packet_queue.cpp b/src/common/pio_packet_queue.cpp
index 70974a7..b7bb12d 100644
--- a/src/common/packet_queue.cpp
+++ b/src/common/pio_packet_queue.cpp
@@ -9,9 +9,9 @@
 */
 
 #include "../../sdk/include/utils.h"
-#include "packet_queue.h"
+#include "pio_packet_queue.h"
 
-void packet_enqueue(struct packet_queue *q, struct packet *p)
+void pio_packet_enqueue(struct pio_packet_queue *q, struct pio_packet *p)
 {
     if (nullptr == p)
         return;
@@ -32,9 +32,9 @@ void packet_enqueue(struct packet_queue *q, struct packet *p)
     q->len++;
 }
 
-struct packet *packet_dequeue(struct packet_queue *q)
+struct pio_packet *pio_packet_dequeue(struct pio_packet_queue *q)
 {
-    struct packet *p = NULL;
+    struct pio_packet *p = NULL;
 
     /* if the queue is empty there are no packets left. */
     if (q->len == 0) {
@@ -58,17 +58,19 @@ struct packet *packet_dequeue(struct packet_queue *q)
 
     p->next = nullptr;
     p->prev = nullptr;
+
     return p;
 }
 
-void release_packet_queue(struct packet_queue *q)
+void release_pio_packet_queue(struct pio_packet_queue *q)
 {
     if (nullptr == q) {
         return;
     }
 
     while (q->len != 0) {
-        struct packet *p = packet_dequeue(q);
+        struct pio_packet *p = pio_packet_dequeue(q);
+        q->len--;
         FREE(p);
     }
 }
\ No newline at end of file
diff --git a/src/common/pio_packet_queue.h b/src/common/pio_packet_queue.h
new file mode 100644
index 0000000..b719eec
--- /dev/null
+++ b/src/common/pio_packet_queue.h
@@ -0,0 +1,62 @@
+/*
+**********************************************************************************************
+*	File: pio_packet_queue.h
+*   Description: pio packet queue structure and api
+*	Authors: Liu WenTan <[email protected]>
+*	Date:    2022-07-15
+*   Copyright: (c) 2018-2022 Geedge Networks, Inc. All rights reserved.
+***********************************************************************************************
+*/
+
+#pragma once
+
+#include <stdint.h>
+
+#define PKT_QUEUE_MAX_NUM 256
+
+#define CUSTOM_ZONE_LEN 64
+#define ETHERNET_HEADER_LEN 14
+#define DEFAULT_MTU 1500
+#define DEFAULT_PACKET_SIZE (CUSTOM_ZONE_LEN + DEFAULT_MTU + ETHERNET_HEADER_LEN)
+
+#define SIZE_OF_PIO_PACKET (DEFAULT_PACKET_SIZE + sizeof(struct pio_packet))
+
+/*
+ * @brief pcap_live/pcap_file mode packet structure
+ *
+ * |<-pkt_hdr    |<-pkt_payload
+ * |             |
+ * | struct pio_packet | custom zone | L2 header | ...... |
+ *                        64bytes      14bytes    1500bytes
+ *   
+ * custom zone: user can set custom field
+ * pkt_payload: received packet data
+**/
+struct pio_packet {
+    /* pkt header pointer */
+    void *pkt_hdr;
+
+    /* pkt length */
+    uint64_t pkt_len;
+
+    /* pkt payload pointer */
+    void *pkt_payload;
+
+    /* reference counts */
+    uint64_t ref_cnt;
+
+	struct pio_packet *prev;
+	
+	struct pio_packet *next;
+};
+
+struct pio_packet_queue {
+	struct pio_packet *top;
+	struct pio_packet *bot;
+	uint32_t len;
+	pthread_mutex_t mutex_q;
+};
+
+void pio_packet_enqueue(struct pio_packet_queue *, struct pio_packet *);
+struct pio_packet *pio_packet_dequeue(struct pio_packet_queue *);
+void release_pio_packet_queue(struct pio_packet_queue *);
\ No newline at end of file
diff --git a/src/main.cpp b/src/main.cpp
index 8b8501d..5b2e558 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -29,18 +29,20 @@ void packet_io_loop(struct packet_io_loop_arg *arg)
 		int fetch_num = packet_io_device_rx(arg->dev, arg->thread_id, &rx_pkt, 1);
 		if(fetch_num > 0)
 		{
+			/*
 			event = session_manager_commit(rx_pkt);
 			while(event)
 			{
 				plugin_manager_dispatch(event);
 				event = session_manager_fetch_event();
-			}
-
+			}*/
+			printf("fetch_num:%d\n", fetch_num);
 			//clean session_manager event queue
 			packet_io_device_tx(arg->dev, arg->thread_id, &rx_pkt, 1);
 		}
 		else
 		{
+			printf("no fetch num\n");
 			//dispatch to time event
 
 			//dispatch to trigger polling event 
@@ -79,9 +81,11 @@ int main(int argc, char ** argv)
 	/* packet io init */
 	packet_io_init("stellar", g_engine_instance.config.packet_io.mode, 2);
 
-	session_manager_session_event_register(http_decoder, SESSION_TYPE_HTTP);
-	
-
+	//session_manager_session_event_register(http_decoder, SESSION_TYPE_HTTP);
+	struct packet_io_loop_arg arg;
+	while (1) {
+		//packet_io_loop();
+	}
 	//create_worker_thread
 
 	//main_loop
diff --git a/src/packet_io/CMakeLists.txt b/src/packet_io/CMakeLists.txt
index 0e45228..85bc1dd 100644
--- a/src/packet_io/CMakeLists.txt
+++ b/src/packet_io/CMakeLists.txt
@@ -1,7 +1,7 @@
 
 add_library(packet_io
 	../common/global_var.cpp
-	../common/packet_queue.cpp
+	../common/pio_packet_queue.cpp
 	../common/time_helper.cpp
 	packet_io.cpp
 	pcap_live_mode/pio_pcap_live.cpp
diff --git a/src/packet_io/marsio_mode/pio_marsio.h b/src/packet_io/marsio_mode/pio_marsio.h
index 9e77354..04a58f0 100644
--- a/src/packet_io/marsio_mode/pio_marsio.h
+++ b/src/packet_io/marsio_mode/pio_marsio.h
@@ -12,7 +12,7 @@
 
 #include <stdint.h>
 
-#include "marsio.h"
+#include <marsio.h>
 
 /*  
  * dll is short for dynamic link lib
diff --git a/src/packet_io/packet_io.cpp b/src/packet_io/packet_io.cpp
index e624834..a88fc20 100644
--- a/src/packet_io/packet_io.cpp
+++ b/src/packet_io/packet_io.cpp
@@ -63,6 +63,18 @@ struct pio_instance_operations pio_instance_ops_array[PACKET_IO_RUN_MODE_MAX] =
 struct packet_io_instance *
 packet_io_instance_create(const char *inst_name, const enum packet_io_run_mode mode, const int wrk_thread_num) 
 {
+	if (nullptr == inst_name) {
+		return nullptr;
+	}
+
+	if (mode < PACKET_IO_RUN_MODE_PCAP_FILE || mode >= PACKET_IO_RUN_MODE_MAX) {
+		return nullptr;
+	}
+
+	if (wrk_thread_num < 0) {
+		return nullptr;
+	}
+
 	struct packet_io_instance *pio_instance = CALLOC(struct packet_io_instance, 1);
 	if (nullptr == pio_instance) {
 		log_error(ST_ERR_MEM_ALLOC, "packet_io instance alloc failed.");
@@ -156,16 +168,14 @@ void packet_io_pkts_free(struct packet_io_instance *pinst, uint32_t qid, struct
 	
 }
 
-static int packet_copy_data_offset(struct packet *p, uint32_t offset, const uint8_t *data, uint32_t data_len)
+static int packet_copy_data_offset(uint8_t *ptr, uint32_t offset, const uint8_t *data, uint32_t data_len)
 {
-	memcpy(GET_PKT_DIRECT_DATA(p) + offset, data, data_len);
+	memcpy(ptr + offset, data, data_len);
 	
 	return 0;
 }
 
-int packet_copy_data(struct packet *p, const uint8_t *pkt_data, uint32_t pkt_len)
-{
-	SET_PKT_LEN(p, (size_t)pkt_len);
-    
-	return packet_copy_data_offset(p, 0, pkt_data, pkt_len);
+int packet_copy_data(uint8_t *ptr, const uint8_t *pkt_data, uint32_t pkt_len)
+{   
+	return packet_copy_data_offset(ptr, 0, pkt_data, pkt_len);
 }
\ No newline at end of file
diff --git a/src/packet_io/packet_io.h b/src/packet_io/packet_io.h
index 8f1f870..ed8bb6d 100644
--- a/src/packet_io/packet_io.h
+++ b/src/packet_io/packet_io.h
@@ -144,4 +144,4 @@ int packet_io_device_tx(struct packet_io_device *pdev, uint32_t txq_id, struct p
 **/
 void packet_io_pkts_free(struct packet_io_device *pdev, uint32_t qid, struct packet **pkts, int nr_pkts);
 
-int packet_copy_data(struct packet *p, const uint8_t *pkt_data, uint32_t pkt_len);
\ No newline at end of file
+int packet_copy_data(uint8_t *ptr, const uint8_t *pkt_data, uint32_t pkt_len);
\ No newline at end of file
diff --git a/src/packet_io/pcap_file_mode/pio_pcap_file.cpp b/src/packet_io/pcap_file_mode/pio_pcap_file.cpp
index 0f7abec..2ef7fce 100644
--- a/src/packet_io/pcap_file_mode/pio_pcap_file.cpp
+++ b/src/packet_io/pcap_file_mode/pio_pcap_file.cpp
@@ -287,7 +287,7 @@ int pio_pcap_file_device_close(struct packet_io_device *pdev)
 
     for (uint32_t i = 0; i < PKT_QUEUE_MAX_NUM; i++) {
         if (pdev->entity.pcap_file_dev_ctx->pkt_queues[i].len != 0) {
-            release_packet_queue(&pdev->entity.pcap_file_dev_ctx->pkt_queues[i]);
+            release_pio_packet_queue(&pdev->entity.pcap_file_dev_ctx->pkt_queues[i]);
         }
     }
 
@@ -299,12 +299,16 @@ int pio_pcap_file_device_close(struct packet_io_device *pdev)
 void pcap_file_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_hdr, u_char *pkt)
 {
     struct pio_pcap_file_device_context *pfile_dev_ctx = (struct pio_pcap_file_device_context *)user;
-    struct packet *p = CALLOC(struct packet, 1);
+    struct pio_packet *p = (struct pio_packet *)malloc(SIZE_OF_PIO_PACKET);
     if (nullptr == p) {
         return;
     }
+    memset(p, 0, SIZE_OF_PIO_PACKET);
 
-    if (packet_copy_data(p, pkt, pkt_hdr->caplen)) {
+    p->pkt_hdr = p;
+    p->pkt_payload = (uint8_t *)p + CUSTOM_ZONE_LEN;
+    p->pkt_len = pkt_hdr->caplen;
+    if (packet_copy_data((uint8_t *)p->pkt_payload, (uint8_t *)pkt, pkt_hdr->caplen)) {
         FREE(p);
         return;
     }
@@ -314,10 +318,10 @@ void pcap_file_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_hdr, u_c
         hash_id = decode_packet(p) % nr_rxq;
         packet_enqueue(&pfile_dev_ctx->pkt_queues[hash_id], p);
     */
-   /*
+    int rxq_id = 0;
     pthread_mutex_lock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
-    packet_enqueue(&pfile_dev_ctx->pkt_queues[rxq_id], p);
-    pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q); */
+    pio_packet_enqueue(&pfile_dev_ctx->pkt_queues[rxq_id], p);
+    pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
 }
 
 static int pcap_file_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx, uint32_t nr_rxq, uint32_t rxq_id, 
@@ -349,12 +353,12 @@ static int pcap_file_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx
         //TODO: close pcap file
     } else {
         // success
-        struct packet *p = nullptr;
+        struct pio_packet *p = nullptr;
         int i = 0;
         pthread_mutex_lock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
         do {
-            p = packet_dequeue(&pfile_dev_ctx->pkt_queues[rxq_id]);
-            pkts[i] = p;
+            p = pio_packet_dequeue(&pfile_dev_ctx->pkt_queues[rxq_id]);
+            pkts[i] = (struct packet *)p;
             i++;
         } while (p != nullptr && (i < nr_pkts));
         pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
@@ -603,9 +607,12 @@ int pio_pcap_file_device_send(struct packet_io_device *pdev, uint32_t txq_id, st
     return 0;
 }
 
-void pio_pcap_file_device_pkt_free(struct packet_io_device *pdev, uint32_t qid, struct packet **pkts, int nr_pkts) 
+void pio_pcap_file_device_pkt_free(__unused struct packet_io_device *pdev, __unused uint32_t qid, struct packet **pkts, int nr_pkts) 
 {
-
+    for (uint32_t i = 0; i < nr_pkts; i++) {
+        struct pio_packet *p = (struct pio_packet *)pkts[i];
+        FREE(p);
+    }
 }
 
 int pio_pcap_file_instance_create(struct packet_io_instance *pinst, __unused int wrk_thread_num) 
diff --git a/src/packet_io/pcap_file_mode/pio_pcap_file.h b/src/packet_io/pcap_file_mode/pio_pcap_file.h
index c42733d..228ad91 100644
--- a/src/packet_io/pcap_file_mode/pio_pcap_file.h
+++ b/src/packet_io/pcap_file_mode/pio_pcap_file.h
@@ -13,9 +13,10 @@
 #include <stdint.h>
 #include <dirent.h>
 #include <pcap/pcap.h>
+#include <sys/queue.h>
 
 #include "../../common/global_var.h"
-#include "../../common/packet_queue.h"
+#include "../../common/pio_packet_queue.h"
 
 struct pio_pcap_file_instance_context {
     
@@ -87,8 +88,8 @@ struct pio_pcap_file_device_context {
 
     bool is_dir;
 
-    /* rx packet queue */
-    struct packet_queue pkt_queues[PKT_QUEUE_MAX_NUM];
+    /* rx pio packet queue */
+    struct pio_packet_queue pkt_queues[PKT_QUEUE_MAX_NUM];
 
     struct pcap_file_shared_info shared;
 };
diff --git a/src/packet_io/pcap_live_mode/pio_pcap_live.cpp b/src/packet_io/pcap_live_mode/pio_pcap_live.cpp
index eef9a70..e8b77cb 100644
--- a/src/packet_io/pcap_live_mode/pio_pcap_live.cpp
+++ b/src/packet_io/pcap_live_mode/pio_pcap_live.cpp
@@ -133,7 +133,7 @@ int pio_pcap_live_device_close(struct packet_io_device *pdev)
 
     for (uint32_t i = 0; i < PKT_QUEUE_MAX_NUM; i++) {
         if (pdev->entity.pcap_live_dev_ctx->pkt_queues[i].len != 0) {
-            release_packet_queue(&pdev->entity.pcap_live_dev_ctx->pkt_queues[i]);
+            release_pio_packet_queue(&pdev->entity.pcap_live_dev_ctx->pkt_queues[i]);
         }
     }
 
@@ -145,12 +145,15 @@ int pio_pcap_live_device_close(struct packet_io_device *pdev)
 static void pcap_live_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_hdr, u_char *pkt)
 {
     struct pio_pcap_live_device_context *plive_dev_ctx = (struct pio_pcap_live_device_context *)user;
-    struct packet *p = CALLOC(struct packet, 1);
+    struct pio_packet *p = (struct pio_packet *)malloc(SIZE_OF_PIO_PACKET);
     if (nullptr == p) {
         return;
     }
-
-    if (packet_copy_data(p, (uint8_t *)pkt, pkt_hdr->caplen)) {
+    memset(p, 0, sizeof(SIZE_OF_PIO_PACKET));
+    p->pkt_hdr = p;
+    p->pkt_payload = (uint8_t *)p + CUSTOM_ZONE_LEN;
+    p->pkt_len = pkt_hdr->caplen;
+    if (packet_copy_data((uint8_t *)p->pkt_payload, (uint8_t *)pkt, pkt_hdr->caplen)) {
         FREE(p);
         return;
     }
@@ -160,10 +163,10 @@ static void pcap_live_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_h
         hash_id = decode_packet(p) % nr_rxq;
         packet_enqueue(&pfile_dev_ctx->pkt_queues[hash_id], p);
     */
-   /*
+    int rxq_id = 0;
     pthread_mutex_lock(&plive_dev_ctx->pkt_queues[rxq_id].mutex_q);
-    packet_enqueue(&plive_dev_ctx->pkt_queues[rxq_id], p);
-    pthread_mutex_unlock(&plive_dev_ctx->pkt_queues[rxq_id].mutex_q); */
+    pio_packet_enqueue(&plive_dev_ctx->pkt_queues[rxq_id], p);
+    pthread_mutex_unlock(&plive_dev_ctx->pkt_queues[rxq_id].mutex_q);
 }
 
 int pio_pcap_live_device_receive(struct packet_io_device *pdev, uint32_t rxq_id, struct packet **pkts, int nr_pkts)
@@ -187,12 +190,12 @@ int pio_pcap_live_device_receive(struct packet_io_device *pdev, uint32_t rxq_id,
     } else if (res == 0) {
 
     } else {
-        struct packet *p = nullptr;
+        struct pio_packet *p = nullptr;
         int i = 0;
         pthread_mutex_lock(&plive_dev_ctx->pkt_queues[rxq_id].mutex_q);
         do {
-            p = packet_dequeue(&plive_dev_ctx->pkt_queues[rxq_id]);
-            pkts[i] = p;
+            p = pio_packet_dequeue(&plive_dev_ctx->pkt_queues[rxq_id]);
+            pkts[i] = (struct packet *)p;
             i++;
         } while (p != nullptr && (i < nr_pkts));
         pthread_mutex_unlock(&plive_dev_ctx->pkt_queues[rxq_id].mutex_q);
@@ -220,16 +223,21 @@ int pio_pcap_live_device_send(struct packet_io_device *pdev, uint32_t txq_id, st
     int packet_q_len = nr_pkts;
     pthread_mutex_lock(&plive_dev_ctx->handle_mutex);
     for (uint32_t i = 0; i < nr_pkts; i++) {
-        res = pcap_sendpacket(plive_dev_ctx->pcap_handle, (u_char *)pkts[i], sizeof(struct packet));
+        struct pio_packet *p = (struct pio_packet *)pkts[i];
+        res = pcap_sendpacket(plive_dev_ctx->pcap_handle, (u_char *)p->pkt_payload, p->pkt_len);
     }
     pthread_mutex_unlock(&plive_dev_ctx->handle_mutex);
-    
+    /*TODO: when to free pio_packet? */
+
     return 0;
 }
 
-void pio_pcap_live_device_pkt_free(struct packet_io_device *pdev, uint32_t qid, struct packet **pkts, int nr_pkts)
+void pio_pcap_live_device_pkt_free(__unused struct packet_io_device *pdev, __unused uint32_t qid, struct packet **pkts, int nr_pkts)
 {
-
+    for (uint32_t i = 0; i < nr_pkts; i++) {
+        struct pio_packet *p = (struct pio_packet *)pkts[i];
+        FREE(p);
+    }
 }
 
 int pio_pcap_live_instance_create(struct packet_io_instance *pinst, __unused int wrk_thread_num)
diff --git a/src/packet_io/pcap_live_mode/pio_pcap_live.h b/src/packet_io/pcap_live_mode/pio_pcap_live.h
index 92e161f..d25be0a 100644
--- a/src/packet_io/pcap_live_mode/pio_pcap_live.h
+++ b/src/packet_io/pcap_live_mode/pio_pcap_live.h
@@ -14,7 +14,7 @@
 #include <pcap/pcap.h>
 
 #include "../../common/global_var.h"
-#include "../../common/packet_queue.h"
+#include "../../common/pio_packet_queue.h"
 
 #define PCAP_STATE_UP 1
 #define PCAP_STATE_DOWN 0
@@ -51,7 +51,7 @@ struct pio_pcap_live_device_context {
     int pcap_snaplen;
 
     /* rx packet queue */
-    struct packet_queue pkt_queues[PKT_QUEUE_MAX_NUM];
+    struct pio_packet_queue pkt_queues[PKT_QUEUE_MAX_NUM];
 };
 
 /*
diff --git a/src/packet_io/test/gtest_packet_io.cpp b/src/packet_io/test/gtest_packet_io.cpp
index 7ff08f7..31f3052 100644
--- a/src/packet_io/test/gtest_packet_io.cpp
+++ b/src/packet_io/test/gtest_packet_io.cpp
@@ -2,6 +2,12 @@
 
 #include "../packet_io.h"
 
+TEST(PACKET_IO_Test, packet_io_instance_create) {
+	struct packet_io_config ppio_config;
+	struct packet_io_instance *ppio_inst = packet_io_instance_create("stellar", PACKET_IO_RUN_MODE_PCAP_FILE, 2);
+  	EXPECT_NE(ppio_inst, nullptr);
+}
+
 TEST(PACKET_IO_Test, packet_io_open_device) {
 	struct packet_io_config ppio_config;
 	struct packet_io_instance *ppio_inst = packet_io_instance_create("stellar", PACKET_IO_RUN_MODE_PCAP_FILE, 2);