gfwleak/stellar-stellar-2022
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[PACKET_IO] add pio_packet structure for pcap_live/pcap_file mode

Browse Source
This commit is contained in:
liuwentan
2022-08-09 15:58:46 +08:00
parent 7442cb09d5
commit c2d5b9cdb7
17 changed files with 157 additions and 360 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CMakeLists.txt
View File

@@ -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)

268
sdk/include/marsio.h
View File

@@ -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<luqiuwen@iie.ac.cn>
* 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

11
sdk/include/packet.h
View File

@@ -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;

3
src/common/global_var.h
View File

@@ -58,6 +58,9 @@ struct packet_io_config {
/* promiscuous value */
int promisc;
/* marsio ctrlzone id */
int mr_ctrlzone_id;
};
struct lib_config {

32
src/common/packet_queue.h
View File

@@ -1,32 +0,0 @@
/*
**********************************************************************************************
* File: packet_queue.h
* Description: packet queue structure and api
* Authors: Liu WenTan <liuwentan@geedgenetworks.com>
* 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 *);

14
src/common/packet_queue.cpp → src/common/pio_packet_queue.cpp
View File

@@ -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);
}
}

62
src/common/pio_packet_queue.h Normal file
View File

@@ -0,0 +1,62 @@
/*
**********************************************************************************************
* File: pio_packet_queue.h
* Description: pio packet queue structure and api
* Authors: Liu WenTan <liuwentan@geedgenetworks.com>
* 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 *);

14
src/main.cpp
View File

@@ -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

2
src/packet_io/CMakeLists.txt
View File

@@ -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

2
src/packet_io/marsio_mode/pio_marsio.h
View File

@@ -12,7 +12,7 @@
#include <stdint.h>
#include "marsio.h"
#include <marsio.h>
/*
* dll is short for dynamic link lib

24
src/packet_io/packet_io.cpp
View File

@@ -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);
}

2
src/packet_io/packet_io.h
View File

@@ -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);
int packet_copy_data(uint8_t *ptr, const uint8_t *pkt_data, uint32_t pkt_len);

29
src/packet_io/pcap_file_mode/pio_pcap_file.cpp
View File

@@ -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)

7
src/packet_io/pcap_file_mode/pio_pcap_file.h
View File

@@ -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;
};

36
src/packet_io/pcap_live_mode/pio_pcap_live.cpp
View File

@@ -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)

4
src/packet_io/pcap_live_mode/pio_pcap_live.h
View File

@@ -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];
};
/*

6
src/packet_io/test/gtest_packet_io.cpp
View File

@@ -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);