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

[PACKET_IO] finish pcap live mode

Browse Source
This commit is contained in:
liuwentan
2022-08-05 18:32:54 +08:00
parent 9cf895fa07
commit 7442cb09d5
10 changed files with 317 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
CMakeLists.txt
View File

@@ -23,8 +23,4 @@ add_subdirectory(sdk/example)
#add gtest
enable_testing()
add_subdirectory(test)
add_subdirectory(test)

4
sdk/include/util_errors.h
View File

@@ -27,6 +27,10 @@ typedef enum {
ST_ERR_PCAP_FILE_DELETE_FAILED,
ST_ERR_PCAP_DISPATCH,
ST_ERR_PCAP_FILE_COLLECT_FAILED,
ST_ERR_PCAP_SET_SNAPLEN,
ST_ERR_PCAP_SET_PROMISC,
ST_ERR_PCAP_SET_TIMEOUT,
ST_ERR_PCAP_ACTIVATE_HANDLE,
ST_ERR_FOPEN,
ST_ERR_BPF,
ST_ERR_MAX

9
src/common/global_var.h
View File

@@ -51,12 +51,13 @@ struct packet_io_config {
/* delete after the pcap file is read */
bool should_delete;
/* loop read pcap file in directory */
bool should_loop;
time_t delay;
time_t poll_interval;
/* snapshot length */
int snaplen;
/* promiscuous value */
int promisc;
};
struct lib_config {

8
src/common/packet_queue.h
View File

@@ -12,6 +12,14 @@
#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;

14
src/packet_io/packet_io.cpp
View File

@@ -154,4 +154,18 @@ int packet_io_device_tx(struct packet_io_device *pdev, uint32_t txq_id, struct p
void packet_io_pkts_free(struct packet_io_instance *pinst, uint32_t qid, struct packet **pkts, int nr_pkts)
{
}
static int packet_copy_data_offset(struct packet *p, uint32_t offset, const uint8_t *data, uint32_t data_len)
{
memcpy(GET_PKT_DIRECT_DATA(p) + 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);
}

4
src/packet_io/packet_io.h
View File

@@ -142,4 +142,6 @@ int packet_io_device_tx(struct packet_io_device *pdev, uint32_t txq_id, struct p
/*
* @brief packet_io free packet buff
**/
void packet_io_pkts_free(struct packet_io_device *pdev, uint32_t qid, struct packet **pkts, int nr_pkts);
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);

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

@@ -17,17 +17,11 @@
#include "pio_pcap_file.h"
#include "../packet_io.h"
#include "../../sdk/include/utils.h"
#include "../../sdk/include/util_errors.h"
#include "../../sdk/include/logger.h"
#include "../../../sdk/include/utils.h"
#include "../../../sdk/include/util_errors.h"
#include "../../../sdk/include/logger.h"
#include "../../common/time_helper.h"
#define SET_PKT_LEN(p, len) do { \
(p)->pkt_len = (len); \
} while(0)
#define GET_PKT_DIRECT_DATA(p) (uint8_t *)((p) + 1)
/*
* @brief validate path is a valid plain file or directory
*
@@ -168,9 +162,7 @@ static int pcap_directory_file_init(struct pio_pcap_file_device_context *pfile_d
/* TODO: get conf and assign pdir_info */
strncpy(pdir_info->dir_name, dir_name, strlen(dir_name));
//pdir_info->should_loop = false;
//pdir_info->delay = 30;
//pdir_info->poll_interval = 5;
pdir_info->shared = &pfile_dev_ctx->shared;
pdir_info->directory = directory;
TAILQ_INIT(&pdir_info->file_queue_head);
@@ -304,18 +296,7 @@ int pio_pcap_file_device_close(struct packet_io_device *pdev)
return 0;
}
int packet_copy_data_offset(struct packet *p, uint32_t offset, const uint8_t *data, uint32_t data_len) {
memcpy(GET_PKT_DIRECT_DATA(p) + 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);
}
static void pcap_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_hdr, u_char *pkt, uint32_t nr_rxq, uint32_t rxq_id)
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);
@@ -333,9 +314,10 @@ static void pcap_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_hdr, u
hash_id = decode_packet(p) % nr_rxq;
packet_enqueue(&pfile_dev_ctx->pkt_queues[hash_id], p);
*/
/*
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);
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,
@@ -344,8 +326,8 @@ static int pcap_file_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx
if (pfile_dev_ctx->entity.file->first_pkt_hdr != nullptr) {
pthread_mutex_lock(&pfile_dev_ctx->entity.file->handle_mutex);
if (pfile_dev_ctx->entity.file->first_pkt_hdr != nullptr) {
pcap_pkt_callback_oneshot((char *)pfile_dev_ctx, pfile_dev_ctx->entity.file->first_pkt_hdr,
(u_char *)pfile_dev_ctx->entity.file->first_pkt_data, nr_rxq, rxq_id);
pcap_file_pkt_callback_oneshot((char *)pfile_dev_ctx, pfile_dev_ctx->entity.file->first_pkt_hdr,
(u_char *)pfile_dev_ctx->entity.file->first_pkt_data);
pfile_dev_ctx->entity.file->first_pkt_hdr = nullptr;
pfile_dev_ctx->entity.file->first_pkt_data = nullptr;
}
@@ -357,7 +339,7 @@ static int pcap_file_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx
pthread_mutex_lock(&pfile_dev_ctx->entity.file->handle_mutex);
res = pcap_dispatch(pfile_dev_ctx->entity.file->pcap_handle, packet_q_len,
(pcap_handler)pcap_pkt_callback_oneshot, (u_char *)pfile_dev_ctx);
(pcap_handler)pcap_file_pkt_callback_oneshot, (u_char *)pfile_dev_ctx);
pthread_mutex_unlock(&pfile_dev_ctx->entity.file->handle_mutex);
if (res < 0) {
log_error(ST_ERR_PCAP_DISPATCH, "error code %d %s for %s", res,
@@ -644,6 +626,10 @@ int pio_pcap_file_instance_create(struct packet_io_instance *pinst, __unused int
void pio_pcap_file_instance_destroy(struct packet_io_instance *pinst)
{
if (nullptr == pinst) {
return;
}
FREE(pinst->entity.pcap_file_inst_ctx);
for (uint32_t i = 0; i < pinst->dev_cnt; i++) {

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

@@ -17,8 +17,6 @@
#include "../../common/global_var.h"
#include "../../common/packet_queue.h"
#define PKT_QUEUE_MAX_NUM 256
struct pio_pcap_file_instance_context {
};
@@ -70,12 +68,7 @@ struct pcap_file_directory_info {
struct pcap_plain_file_info *current_file;
/* whether to loop through the pcap files in the directory */
//bool should_loop;
time_t delay;
/* poll pcap file interval */
//time_t poll_interval;
/* the pending files queue for the specific directory */
TAILQ_HEAD(pending_file_queue, pending_file) file_queue_head;

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

@@ -8,40 +8,256 @@
***********************************************************************************************
*/
#include <inttypes.h>
#include <string.h>
#include <pthread.h>
#include "pio_pcap_live.h"
#include "../packet_io.h"
#include "../../sdk/include/utils.h"
#include "../../../sdk/include/logger.h"
#include "../../../sdk/include/utils.h"
#include "../../../sdk/include/util_errors.h"
int pio_pcap_live_device_open(struct packet_io_device *pdev) {
#define DEFAULT_MAX_PACKET_SIZE 65535
#define TIMEOUT_MS 500
return 0;
}
int pio_pcap_live_device_close(struct packet_io_device *pdev) {
return 0;
}
int pio_pcap_live_device_receive(struct packet_io_device *pdev, uint32_t rxq_id, struct packet **pkts, int nr_pkts) {
return 0;
}
int pio_pcap_live_device_send(struct packet_io_device *pdev, uint32_t txq_id, struct packet **pkts, int nr_pkts) {
return 0;
}
void pio_pcap_live_device_pkt_free(struct packet_io_device *pdev, uint32_t qid, struct packet **pkts, int nr_pkts)
static int pcap_live_init(struct pio_pcap_live_device_context *pcap_live_dev_ctx, const char *dev_name, uint32_t nr_rxq)
{
if (nullptr == pcap_live_dev_ctx) {
return -1;
}
char errbuf[PCAP_ERRBUF_SIZE];
pcap_live_dev_ctx->pcap_handle = pcap_create(dev_name, errbuf);
if (nullptr == pcap_live_dev_ctx->pcap_handle) {
log_error(ST_ERR_PIO_PCAP_LIVE_DEVICE, "could not create pcap handle for %s, error %s",
dev_name, errbuf);
return -1;
}
if (g_engine_instance.config.packet_io.snaplen == 0) {
pcap_live_dev_ctx->pcap_snaplen = DEFAULT_MAX_PACKET_SIZE;
} else {
pcap_live_dev_ctx->pcap_snaplen = g_engine_instance.config.packet_io.snaplen;
}
/* set snaplen */
int res = pcap_set_snaplen(pcap_live_dev_ctx->pcap_handle, pcap_live_dev_ctx->pcap_snaplen);
if (res != 0) {
log_error(ST_ERR_PCAP_SET_SNAPLEN, "could not set snaplen, error:%s",
pcap_geterr(pcap_live_dev_ctx->pcap_handle));
return -1;
}
log_info("set snaplen to %d for %s", pcap_live_dev_ctx->pcap_snaplen, dev_name);
/* set promisc */
res = pcap_set_promisc(pcap_live_dev_ctx->pcap_handle, g_engine_instance.config.packet_io.promisc);
if (res != 0) {
log_error(ST_ERR_PCAP_SET_PROMISC, "could not set promisc mode, error:%s",
pcap_geterr(pcap_live_dev_ctx->pcap_handle));
return -1;
}
/* set timeout */
res = pcap_set_timeout(pcap_live_dev_ctx->pcap_handle, TIMEOUT_MS);
if (res != 0) {
log_error(ST_ERR_PCAP_SET_TIMEOUT, "could not set timeout, error:%s",
pcap_geterr(pcap_live_dev_ctx->pcap_handle));
return -1;
}
res = pcap_activate(pcap_live_dev_ctx->pcap_handle);
if (res != 0) {
log_error(ST_ERR_PCAP_ACTIVATE_HANDLE, "could not activate the pcap handle, error:%s",
pcap_geterr(pcap_live_dev_ctx->pcap_handle));
return -1;
}
pcap_live_dev_ctx->pcap_state = PCAP_STATE_UP;
/* set bpf filter */
if (strlen(g_engine_instance.config.packet_io.bpf_string) != 0) {
strncpy(pcap_live_dev_ctx->bpf_string, g_engine_instance.config.packet_io.bpf_string,
strlen(g_engine_instance.config.packet_io.bpf_string));
if (pcap_compile(pcap_live_dev_ctx->pcap_handle, &pcap_live_dev_ctx->filter,
pcap_live_dev_ctx->bpf_string, 1, 0) < 0) {
log_error(ST_ERR_BPF, "bpf compliation error %s",
pcap_geterr(pcap_live_dev_ctx->pcap_handle));
return -1;
}
if (pcap_setfilter(pcap_live_dev_ctx->pcap_handle, &pcap_live_dev_ctx->filter) < 0) {
log_error(ST_ERR_BPF, "could not set bpf filter %s",
pcap_geterr(pcap_live_dev_ctx->pcap_handle));
return -1;
}
}
pcap_live_dev_ctx->data_link = pcap_datalink(pcap_live_dev_ctx->pcap_handle);
return res;
}
int pio_pcap_live_instance_create(struct packet_io_instance *pinst, __unused int wrk_thread_num) {
int pio_pcap_live_device_open(struct packet_io_device *pdev)
{
int res = -1;
if (nullptr == pdev) {
log_error(ST_ERR_PIO_PCAP_LIVE_DEVICE, "invalid packet_io_device pointer.");
return res;
}
pdev->entity.pcap_live_dev_ctx = CALLOC(struct pio_pcap_live_device_context, 1);
if (nullptr == pdev->entity.pcap_live_dev_ctx) {
log_error(ST_ERR_PIO_PCAP_LIVE_DEVICE, "alloc pcap_live_dev_ctx failed.");
return res;
}
pthread_mutex_init(&pdev->entity.pcap_live_dev_ctx->handle_mutex, nullptr);
res = pcap_live_init(pdev->entity.pcap_live_dev_ctx, pdev->dev_name, pdev->rxq_num);
if (res < 0) {
log_error(ST_ERR_PIO_PCAP_LIVE_DEVICE, "init pcap live failed.");
FREE(pdev->entity.pcap_live_dev_ctx);
return res;
}
return res;
}
int pio_pcap_live_device_close(struct packet_io_device *pdev)
{
if (nullptr == pdev) {
log_error(ST_ERR_PIO_PCAP_FILE_DEVICE, "invalid pdev pointer, so close pcap live device failed.");
return -1;
}
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]);
}
}
FREE(pdev->entity.pcap_live_dev_ctx);
return 0;
}
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);
if (nullptr == p) {
return;
}
if (packet_copy_data(p, (uint8_t *)pkt, pkt_hdr->caplen)) {
FREE(p);
return;
}
/*
hash to specific queue id and enqueue
hash_id = decode_packet(p) % nr_rxq;
packet_enqueue(&pfile_dev_ctx->pkt_queues[hash_id], p);
*/
/*
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); */
}
int pio_pcap_live_device_receive(struct packet_io_device *pdev, uint32_t rxq_id, struct packet **pkts, int nr_pkts)
{
int res = -1;
struct pio_pcap_live_device_context *plive_dev_ctx = pdev->entity.pcap_live_dev_ctx;
if (nullptr == plive_dev_ctx) {
log_error(ST_ERR_PIO_PCAP_LIVE_DEVICE, "invalid pcap_live_dev_ctx failed.");
return res;
}
int packet_q_len = nr_pkts;
pthread_mutex_lock(&plive_dev_ctx->handle_mutex);
res = pcap_dispatch(plive_dev_ctx->pcap_handle, packet_q_len,
(pcap_handler)pcap_live_pkt_callback_oneshot, (u_char *)plive_dev_ctx);
pthread_mutex_unlock(&plive_dev_ctx->handle_mutex);
if (res < 0) {
log_error(ST_ERR_PCAP_DISPATCH, "error code %" PRId32 " %s",
res, pcap_geterr(plive_dev_ctx->pcap_handle));
} else if (res == 0) {
} else {
struct 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;
i++;
} while (p != nullptr && (i < nr_pkts));
pthread_mutex_unlock(&plive_dev_ctx->pkt_queues[rxq_id].mutex_q);
if (nullptr == p) {
res = i - 1;
} else {
res = nr_pkts;
}
}
return res;
}
int pio_pcap_live_device_send(struct packet_io_device *pdev, uint32_t txq_id, struct packet **pkts, int nr_pkts)
{
int res = -1;
struct pio_pcap_live_device_context *plive_dev_ctx = pdev->entity.pcap_live_dev_ctx;
if (nullptr == plive_dev_ctx) {
log_error(ST_ERR_PIO_PCAP_LIVE_DEVICE, "invalid pcap_live_dev_ctx failed.");
return res;
}
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));
}
pthread_mutex_unlock(&plive_dev_ctx->handle_mutex);
return 0;
}
void pio_pcap_live_instance_destroy(struct packet_io_instance *pinst) {
void pio_pcap_live_device_pkt_free(struct packet_io_device *pdev, uint32_t qid, struct packet **pkts, int nr_pkts)
{
}
int pio_pcap_live_instance_create(struct packet_io_instance *pinst, __unused int wrk_thread_num)
{
if (nullptr == pinst) {
log_error(ST_ERR_PIO_PCAP_LIVE_INSTANCE, "invalid pcap live instance pointer.");
return -1;
}
pinst->entity.pcap_live_inst_ctx = CALLOC(struct pio_pcap_live_instance_context, 1);
if (nullptr == pinst->entity.pcap_live_inst_ctx) {
log_error(ST_ERR_PIO_PCAP_LIVE_INSTANCE, "alloc pcap_live_inst_ctx failed.");
return -1;
}
return 0;
}
void pio_pcap_live_instance_destroy(struct packet_io_instance *pinst)
{
if (nullptr == pinst) {
return;
}
FREE(pinst->entity.pcap_live_inst_ctx);
for (uint32_t i = 0; i < pinst->dev_cnt; i++) {
pio_pcap_live_device_close(pinst->devices[i]);
FREE(pinst->devices[i]);
}
}

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

@@ -13,6 +13,12 @@
#include <stdint.h>
#include <pcap/pcap.h>
#include "../../common/global_var.h"
#include "../../common/packet_queue.h"
#define PCAP_STATE_UP 1
#define PCAP_STATE_DOWN 0
struct pio_pcap_live_instance_context {
};
@@ -22,6 +28,30 @@ struct pio_pcap_live_instance_context {
**/
struct pio_pcap_live_device_context {
pcap_t *pcap_handle;
pthread_mutex_t handle_mutex;
/* handle state */
uint8_t pcap_state;
/* bpf filter string, such as "tcp and port 25" */
char bpf_string[STR_MAX_LEN];
struct bpf_program filter;
int data_link;
/* counters */
uint64_t pkts;
uint64_t bytes;
uint64_t errs;
/* pcap buffer size */
int pcap_buffer_size;
int pcap_snaplen;
/* rx packet queue */
struct packet_queue pkt_queues[PKT_QUEUE_MAX_NUM];
};
/*