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

feature: packet IO support IP reassembly

Browse Source
This commit is contained in:
luwenpeng
2024-10-23 10:01:20 +08:00
parent a7b79a0e22
commit fd3cc20554
54 changed files with 3474 additions and 4271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
infra/packet_io/CMakeLists.txt
View File

@@ -1,3 +1,3 @@
add_library(packet_io pcap_io.c marsio_io.c packet_io.c)
add_library(packet_io pcap_io.c mars_io.c packet_io.c)
target_include_directories(packet_io PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_link_libraries(packet_io marsio pcap packet_manager)
target_link_libraries(packet_io marsio pcap packet_manager ip_reassembly)

618
infra/packet_io/mars_io.c Normal file
View File

@@ -0,0 +1,618 @@
#include <sched.h>
#include <assert.h>
#include "marsio.h"
#include "mars_io.h"
#include "packet_pool.h"
#include "packet_parser.h"
#include "ip_reassembly.h"
#include "log_internal.h"
#include "utils_internal.h"
#include "packet_internal.h"
#define MARS_IO_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "mars io", format, ##__VA_ARGS__)
#define MARS_IO_LOG_INFO(format, ...) STELLAR_LOG_INFO(__thread_local_logger, "mars io", format, ##__VA_ARGS__)
struct mars_io_cfg
{
char app_symbol[64];
char dev_symbol[64];
uint64_t thread_num; // range [1, MAX_THREAD_NUM]
uint64_t cpu_mask[MAX_THREAD_NUM];
uint64_t idle_yield_ms; // range: [0, 6000] (ms)
// packet pool
uint64_t capacity; // range: [1, 4294967295]
// ip reassembly
uint64_t fail_action; // 0: bypass, 1: drop
uint64_t timeout_ms; // range: [1, 60000] (ms)
uint64_t frag_queue_num; // range: [1, 4294967295
uint64_t frag_queue_size; // range: [2, 65535]
};
struct mars_io
{
struct mars_io_cfg *cfg;
struct mr_instance *mr_ins;
struct mr_vdev *mr_dev;
struct mr_sendpath *mr_path;
struct packet_pool *pool[MAX_THREAD_NUM];
struct packet_io_stat stat[MAX_THREAD_NUM];
struct ip_reassembly *ip_reass[MAX_THREAD_NUM];
};
/******************************************************************************
* Private API
******************************************************************************/
static struct mars_io_cfg *mars_io_cfg_new(const char *toml_file)
{
struct mars_io_cfg *cfg = (struct mars_io_cfg *)calloc(1, sizeof(struct mars_io_cfg));
if (cfg == NULL)
{
return NULL;
}
int ret = 0;
int num = 0;
ret += load_toml_str_config(toml_file, "packet_io.app_symbol", cfg->app_symbol);
ret += load_toml_str_config(toml_file, "packet_io.dev_symbol", cfg->dev_symbol);
ret += load_toml_integer_config(toml_file, "packet_io.thread_num", &cfg->thread_num, 1, MAX_THREAD_NUM);
ret += load_toml_integer_config(toml_file, "packet_io.idle_yield_ms", &cfg->idle_yield_ms, 0, 60000);
num = load_toml_array_config(toml_file, "packet_io.cpu_mask", cfg->cpu_mask, MAX_THREAD_NUM);
ret += load_toml_integer_config(toml_file, "packet_io.packet_pool.capacity", &cfg->capacity, 1, 4294967295);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.fail_action", &cfg->fail_action, 0, 1);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.timeout_ms", &cfg->timeout_ms, 1, 60000);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.frag_queue_num", &cfg->frag_queue_num, 1, 4294967295);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.frag_queue_size", &cfg->frag_queue_size, 2, 65535);
if (ret != 0 || num != (int)cfg->thread_num)
{
free(cfg);
return NULL;
}
else
{
return cfg;
}
}
static void mars_io_cfg_free(struct mars_io_cfg *cfg)
{
if (cfg)
{
free(cfg);
cfg = NULL;
}
}
static void mars_io_cfg_print(const struct mars_io_cfg *cfg)
{
if (cfg)
{
MARS_IO_LOG_INFO("packet_io.app_symbol : %s", cfg->app_symbol);
MARS_IO_LOG_INFO("packet_io.dev_symbol : %s", cfg->dev_symbol);
MARS_IO_LOG_INFO("packet_io.idle_yield_ms : %lu", cfg->idle_yield_ms);
MARS_IO_LOG_INFO("packet_io.thread_num : %lu", cfg->thread_num);
for (uint64_t i = 0; i < cfg->thread_num; i++)
{
MARS_IO_LOG_INFO("packet_io.cpu_mask[%03d] : %d", i, cfg->cpu_mask[i]);
}
MARS_IO_LOG_INFO("packet_io.packet_pool.capacity : %lu", cfg->capacity);
MARS_IO_LOG_INFO("packet_io.ip_reassembly.fail_action : %lu", cfg->fail_action);
MARS_IO_LOG_INFO("packet_io.ip_reassembly.timeout_ms : %lu", cfg->timeout_ms);
MARS_IO_LOG_INFO("packet_io.ip_reassembly.frag_queue_num : %lu", cfg->frag_queue_num);
MARS_IO_LOG_INFO("packet_io.ip_reassembly.frag_queue_size : %lu", cfg->frag_queue_size);
}
}
static void packet_set_metadata(struct packet *pkt, marsio_buff_t *mbuff)
{
struct route_ctx route_ctx = {};
route_ctx.used = marsio_buff_get_metadata(mbuff, MR_BUFF_ROUTE_CTX, &route_ctx.data, sizeof(route_ctx.data));
if (route_ctx.used > 0)
{
packet_set_route_ctx(pkt, &route_ctx);
}
else
{
MARS_IO_LOG_ERROR("failed to get route ctx");
}
struct sids sids = {};
sids.used = marsio_buff_get_sid_list(mbuff, sids.sid, sizeof(sids.sid) / sizeof(sids.sid[0]));
if (sids.used > 0)
{
packet_set_sids(pkt, &sids);
}
else
{
MARS_IO_LOG_ERROR("failed to get sids");
}
uint64_t session_id = 0;
if (marsio_buff_get_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) == sizeof(session_id))
{
packet_set_session_id(pkt, session_id);
}
else
{
MARS_IO_LOG_ERROR("failed to get session id");
}
// TODO
#if 0
if (marsio_buff_get_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) == sizeof(domain))
{
packet_set_domain(pkt, domain);
}
else
{
MARS_IO_LOG_ERROR("failed to get domain id");
}
#endif
uint16_t link_id = 0;
if (marsio_buff_get_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) == sizeof(link_id))
{
packet_set_link_id(pkt, link_id);
}
else
{
MARS_IO_LOG_ERROR("failed to get link id");
}
packet_set_ctrl(pkt, marsio_buff_is_ctrlbuf(mbuff));
enum packet_direction direction = PACKET_DIRECTION_OUTGOING;
if (marsio_buff_get_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) == sizeof(direction))
{
packet_set_direction(pkt, direction);
}
else
{
MARS_IO_LOG_ERROR("failed to get direction");
}
packet_set_action(pkt, PACKET_ACTION_FORWARD);
// TODO
const struct timeval tv = {};
packet_set_timeval(pkt, &tv);
}
static void mbuff_set_metadata(marsio_buff_t *mbuff, struct packet *pkt)
{
const struct route_ctx *route_ctx = packet_get_route_ctx(pkt);
if (marsio_buff_set_metadata(mbuff, MR_BUFF_ROUTE_CTX, (void *)route_ctx->data, route_ctx->used) != 0)
{
MARS_IO_LOG_ERROR("failed to set route ctx");
}
const struct sids *sids = packet_get_sids(pkt);
if (marsio_buff_set_sid_list(mbuff, (sid_t *)sids->sid, sids->used) != 0)
{
MARS_IO_LOG_ERROR("failed to set sids");
}
uint64_t session_id = packet_get_session_id(pkt);
if (marsio_buff_set_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) != 0)
{
MARS_IO_LOG_ERROR("failed to set session id");
}
// TODO
#if 0
uint64_t domain = packet_get_domain(pkt);
if (marsio_buff_set_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) != 0)
{
MARS_IO_LOG_ERROR("failed to set domain id");
}
#endif
uint16_t link_id = packet_get_link_id(pkt);
if (marsio_buff_set_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) != 0)
{
MARS_IO_LOG_ERROR("failed to set link id");
}
if (packet_is_ctrl(pkt))
{
marsio_buff_set_ctrlbuf(mbuff);
}
enum packet_direction direction = packet_get_direction(pkt);
if (marsio_buff_set_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) != 0)
{
MARS_IO_LOG_ERROR("failed to set direction");
}
}
static int is_keepalive_packet(const char *data, int len)
{
if (data == NULL || len < (int)(sizeof(struct ethhdr)))
{
return 0;
}
struct ethhdr *eth_hdr = (struct ethhdr *)data;
if (eth_hdr->h_proto == 0xAAAA)
{
return 1;
}
else
{
return 0;
}
}
static void origin_free_cb(struct packet *pkt, void *args)
{
struct mars_io *mars_io = (struct mars_io *)args;
struct packet_origin *origin = packet_get_origin(pkt);
marsio_buff_t *mbuff = (marsio_buff_t *)origin->ctx;
struct packet_io_stat *stat = &mars_io->stat[origin->thr_idx];
stat->pkts_user_freed++;
stat->bytes_user_freed += packet_get_raw_len(pkt);
marsio_buff_free(mars_io->mr_ins, &mbuff, 1, 0, origin->thr_idx);
}
static struct packet *recv_packet(struct mars_io *mars_io, marsio_buff_t *mbuff, uint16_t thr_idx)
{
struct packet_io_stat *stat = &mars_io->stat[thr_idx];
struct ip_reassembly *ip_reass = mars_io->ip_reass[thr_idx];
struct packet_pool *pool = mars_io->pool[thr_idx];
int len = marsio_buff_datalen(mbuff);
char *data = marsio_buff_mtod(mbuff);
stat->pkts_rx++;
stat->bytes_rx += len;
if (is_keepalive_packet(data, len))
{
stat->keep_alive_pkts++;
stat->keep_alive_bytes += len;
stat->pkts_tx++;
stat->bytes_tx += len;
marsio_send_burst(mars_io->mr_path, thr_idx, &mbuff, 1);
return NULL;
}
else
{
struct packet *pkt = packet_pool_pop(pool);
assert(pkt != NULL);
struct packet_origin origin = {
.type = ORIGIN_TYPE_MR,
.ctx = mbuff,
.cb = origin_free_cb,
.args = mars_io,
.thr_idx = thr_idx,
};
packet_parse(pkt, data, len);
packet_set_metadata(pkt, mbuff);
packet_set_origin(pkt, &origin);
if (packet_is_fragment(pkt))
{
return ip_reassembly_defrag(ip_reass, pkt, clock_get_real_time_ms());
}
else
{
return pkt;
}
}
}
static void send_packet(struct mars_io *mars_io, struct packet *pkt, uint16_t thr_idx)
{
marsio_buff_t *mbuff = NULL;
struct packet_io_stat *stat = &mars_io->stat[thr_idx];
int len = packet_get_raw_len(pkt);
struct packet_origin *origin = packet_get_origin(pkt);
if (origin->type == ORIGIN_TYPE_MR)
{
mbuff = (marsio_buff_t *)origin->ctx;
mbuff_set_metadata(mbuff, pkt);
marsio_send_burst(mars_io->mr_path, thr_idx, &mbuff, 1);
packet_pool_push(mars_io->pool[thr_idx], pkt);
}
else
{
if (marsio_buff_malloc_global(mars_io->mr_ins, &mbuff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY) < 0)
{
MARS_IO_LOG_ERROR("unable to allocate marsio buffer for inject packet");
packet_free(pkt);
return;
}
else
{
stat->pkts_injected++;
stat->bytes_injected += len;
char *ptr = marsio_buff_append(mbuff, len);
memcpy(ptr, packet_get_raw_data(pkt), len);
mbuff_set_metadata(mbuff, pkt);
marsio_send_burst_with_options(mars_io->mr_path, thr_idx, &mbuff, 1, MARSIO_SEND_OPT_REHASH);
packet_free(pkt);
}
}
stat->pkts_tx++;
stat->bytes_tx += len;
}
static void drop_packet(struct mars_io *mars_io, struct packet *pkt, uint16_t thr_idx)
{
struct packet_io_stat *stat = &mars_io->stat[thr_idx];
int len = packet_get_raw_len(pkt);
struct packet_origin *origin = packet_get_origin(pkt);
stat->pkts_dropped++;
stat->bytes_dropped += len;
if (origin->type == ORIGIN_TYPE_MR)
{
marsio_buff_t *mbuff = (marsio_buff_t *)origin->ctx;
marsio_buff_free(mars_io->mr_ins, &mbuff, 1, 0, thr_idx);
packet_pool_push(mars_io->pool[thr_idx], pkt);
}
else
{
packet_free(pkt);
}
}
/******************************************************************************
* Public API
******************************************************************************/
void *mars_io_new(const char *toml_file)
{
int opt = 1;
cpu_set_t coremask;
CPU_ZERO(&coremask);
struct mars_io *mars_io = (struct mars_io *)calloc(1, sizeof(struct mars_io));
if (mars_io == NULL)
{
MARS_IO_LOG_ERROR("unable to allocate memory for mars_io");
return NULL;
}
mars_io->cfg = mars_io_cfg_new(toml_file);
if (mars_io->cfg == NULL)
{
MARS_IO_LOG_ERROR("unable to create mars_io_cfg");
goto error_out;
}
mars_io_cfg_print(mars_io->cfg);
for (uint16_t i = 0; i < mars_io->cfg->thread_num; i++)
{
CPU_SET(mars_io->cfg->cpu_mask[i], &coremask);
}
mars_io->mr_ins = marsio_create();
if (mars_io->mr_ins == NULL)
{
MARS_IO_LOG_ERROR("unable to create marsio instance");
goto error_out;
}
marsio_option_set(mars_io->mr_ins, MARSIO_OPT_THREAD_MASK_IN_CPUSET, &coremask, sizeof(cpu_set_t));
marsio_option_set(mars_io->mr_ins, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt));
if (marsio_init(mars_io->mr_ins, mars_io->cfg->app_symbol) != 0)
{
MARS_IO_LOG_ERROR("unable to init marsio instance");
goto error_out;
}
mars_io->mr_dev = marsio_open_device(mars_io->mr_ins, mars_io->cfg->dev_symbol, mars_io->cfg->thread_num, mars_io->cfg->thread_num);
if (mars_io->mr_dev == NULL)
{
MARS_IO_LOG_ERROR("unable to open marsio device");
goto error_out;
}
mars_io->mr_path = marsio_sendpath_create_by_vdev(mars_io->mr_dev);
if (mars_io->mr_path == NULL)
{
MARS_IO_LOG_ERROR("unable to create marsio sendpath");
goto error_out;
}
for (uint64_t i = 0; i < mars_io->cfg->thread_num; i++)
{
mars_io->pool[i] = packet_pool_new(mars_io->cfg->capacity);
if (mars_io->pool[i] == NULL)
{
MARS_IO_LOG_ERROR("unable to create packet pool");
goto error_out;
}
mars_io->ip_reass[i] = ip_reassembly_new(mars_io->cfg->timeout_ms, mars_io->cfg->frag_queue_num, mars_io->cfg->frag_queue_size);
if (mars_io->ip_reass[i] == NULL)
{
MARS_IO_LOG_ERROR("unable to create ip reassembly");
goto error_out;
}
}
return mars_io;
error_out:
mars_io_free(mars_io);
return NULL;
}
void mars_io_free(void *handle)
{
struct mars_io *mars_io = (struct mars_io *)handle;
if (mars_io)
{
for (uint64_t i = 0; i < mars_io->cfg->thread_num; i++)
{
ip_reassembly_free(mars_io->ip_reass[i]);
packet_pool_free(mars_io->pool[i]);
}
if (mars_io->mr_path)
{
marsio_sendpath_destory(mars_io->mr_path);
mars_io->mr_path = NULL;
}
if (mars_io->mr_dev)
{
marsio_close_device(mars_io->mr_dev);
mars_io->mr_dev = NULL;
}
if (mars_io->mr_ins)
{
marsio_destory(mars_io->mr_ins);
mars_io->mr_ins = NULL;
}
mars_io_cfg_free(mars_io->cfg);
free(mars_io);
mars_io = NULL;
}
}
int mars_io_isbreak(void *handle __attribute__((unused)))
{
return 0;
}
int mars_io_init(void *handle, uint16_t thr_idx __attribute__((unused)))
{
struct mars_io *mars_io = (struct mars_io *)handle;
if (marsio_thread_init(mars_io->mr_ins) != 0)
{
MARS_IO_LOG_ERROR("unable to init marsio thread");
return -1;
}
else
{
return 0;
}
}
int mars_io_recv(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
struct packet *pkt = NULL;
marsio_buff_t *mbuff = NULL;
marsio_buff_t *mbuffs[RX_BURST_MAX];
struct mars_io *mars_io = (struct mars_io *)handle;
int ret = 0;
int nr_recv = marsio_recv_burst(mars_io->mr_dev, thr_idx, mbuffs, MIN(RX_BURST_MAX, nr_pkts));
for (int i = 0; i < nr_recv; i++)
{
mbuff = mbuffs[i];
pkt = recv_packet(mars_io, mbuff, thr_idx);
if (pkt)
{
pkts[ret++] = pkt;
}
}
return ret;
}
void mars_io_send(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
struct packet *frag = NULL;
struct packet *pkt = NULL;
struct mars_io *mars_io = (struct mars_io *)handle;
for (int i = 0; i < nr_pkts; i++)
{
pkt = pkts[i];
if (packet_is_defraged(pkt))
{
while ((frag = packet_pop_frag(pkt)))
{
send_packet(mars_io, frag, thr_idx);
}
packet_free(pkt);
}
else
{
send_packet(mars_io, pkt, thr_idx);
}
pkts[i] = NULL;
}
}
void mars_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
struct packet *pkt = NULL;
struct packet *frag = NULL;
struct mars_io *mars_io = (struct mars_io *)handle;
for (int i = 0; i < nr_pkts; i++)
{
pkt = pkts[i];
if (packet_is_defraged(pkt))
{
while ((frag = packet_pop_frag(pkt)))
{
drop_packet(mars_io, frag, thr_idx);
}
packet_free(pkt);
}
else
{
drop_packet(mars_io, pkt, thr_idx);
}
pkts[i] = NULL;
}
}
void mars_io_yield(void *handle, uint16_t thr_idx)
{
struct mars_io *mars_io = (struct mars_io *)handle;
struct mr_vdev *vdevs[1] = {
mars_io->mr_dev,
};
marsio_poll_wait(mars_io->mr_ins, vdevs, 1, thr_idx, mars_io->cfg->idle_yield_ms);
}
void mars_io_polling(void *handle, uint16_t thr_idx)
{
struct mars_io *mars_io = (struct mars_io *)handle;
struct ip_reassembly *ip_reass = mars_io->ip_reass[thr_idx];
struct packet *pkt = NULL;
uint64_t now_ms = clock_get_real_time_ms();
while ((pkt = ip_reassembly_clean(ip_reass, now_ms)))
{
if (mars_io->cfg->fail_action == 0)
{
send_packet(mars_io, pkt, thr_idx);
}
else
{
drop_packet(mars_io, pkt, thr_idx);
}
}
// TODO
// output stat
}
struct packet_io_stat *mars_io_stat(void *handle, uint16_t thr_idx)
{
struct mars_io *mars_io = (struct mars_io *)handle;
return &mars_io->stat[thr_idx];
}

24
infra/packet_io/mars_io.h Normal file
View File

@@ -0,0 +1,24 @@
#pragma once
#ifdef __cplusplus
extern "C"
{
#endif
#include "packet_io.h"
void *mars_io_new(const char *toml_file);
void mars_io_free(void *handle);
int mars_io_isbreak(void *handle);
int mars_io_init(void *handle, uint16_t thr_idx);
int mars_io_recv(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void mars_io_send(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void mars_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void mars_io_yield(void *handle, uint16_t thr_idx);
void mars_io_polling(void *handle, uint16_t thr_idx);
struct packet_io_stat *mars_io_stat(void *handle, uint16_t thr_idx);
#ifdef __cplusplus
}
#endif

422
infra/packet_io/marsio_io.c
View File

@@ -1,422 +0,0 @@
#include <sched.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <netinet/ether.h>
#include "marsio.h"
#include "marsio_io.h"
#include "log_internal.h"
#include "packet_parser.h"
#include "packet_internal.h"
#define MARSIO_IO_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "marsio io", format, ##__VA_ARGS__)
struct marsio_io
{
struct packet_io_config cfg;
struct mr_instance *mr_ins;
struct mr_vdev *mr_dev;
struct mr_sendpath *mr_path;
struct packet_io_stat stat[MAX_THREAD_NUM];
};
/******************************************************************************
* Private API
******************************************************************************/
static void metadata_from_mbuff_to_packet(marsio_buff_t *mbuff, struct packet *pkt)
{
struct route_ctx route_ctx = {};
struct sids sids = {};
uint64_t session_id = {0};
uint16_t link_id = {0};
int is_ctrl = {0};
enum packet_direction direction = PACKET_DIRECTION_OUTGOING;
route_ctx.used = marsio_buff_get_metadata(mbuff, MR_BUFF_ROUTE_CTX, &route_ctx.data, sizeof(route_ctx.data));
if (route_ctx.used > 0)
{
packet_set_route_ctx(pkt, &route_ctx);
}
else
{
MARSIO_IO_LOG_ERROR("failed to get route ctx");
}
sids.used = marsio_buff_get_sid_list(mbuff, sids.sid, sizeof(sids.sid) / sizeof(sids.sid[0]));
if (sids.used > 0)
{
packet_set_sids(pkt, &sids);
}
else
{
MARSIO_IO_LOG_ERROR("failed to get sids");
}
if (marsio_buff_get_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) == sizeof(session_id))
{
packet_set_session_id(pkt, session_id);
}
else
{
MARSIO_IO_LOG_ERROR("failed to get session id");
}
// TODO
#if 0
if (marsio_buff_get_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) == sizeof(domain))
{
packet_set_domain(pkt, domain);
}
else
{
MARSIO_IO_LOG_ERROR("failed to get domain id");
}
#endif
if (marsio_buff_get_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) == sizeof(link_id))
{
packet_set_link_id(pkt, link_id);
}
else
{
MARSIO_IO_LOG_ERROR("failed to get link id");
}
is_ctrl = marsio_buff_is_ctrlbuf(mbuff);
packet_set_ctrl(pkt, is_ctrl);
if (marsio_buff_get_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) == sizeof(direction))
{
packet_set_direction(pkt, direction);
}
else
{
MARSIO_IO_LOG_ERROR("failed to get direction");
}
packet_set_action(pkt, PACKET_ACTION_FORWARD);
packet_set_origin_ctx(pkt, mbuff);
// TODO
const struct timeval tv = {};
packet_set_timeval(pkt, &tv);
}
static void metadata_from_packet_to_mbuff(struct packet *pkt, marsio_buff_t *mbuff)
{
const struct route_ctx *route_ctx = packet_get_route_ctx(pkt);
const struct sids *sids = packet_get_sids(pkt);
uint64_t session_id = packet_get_session_id(pkt);
// uint64_t domain = packet_get_domain(pkt);
uint16_t link_id = packet_get_link_id(pkt);
int is_ctrl = packet_is_ctrl(pkt);
enum packet_direction direction = packet_get_direction(pkt);
if (marsio_buff_set_metadata(mbuff, MR_BUFF_ROUTE_CTX, (void *)route_ctx->data, route_ctx->used) != 0)
{
MARSIO_IO_LOG_ERROR("failed to set route ctx");
}
if (marsio_buff_set_sid_list(mbuff, (sid_t *)sids->sid, sids->used) != 0)
{
MARSIO_IO_LOG_ERROR("failed to set sids");
}
if (marsio_buff_set_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) != 0)
{
MARSIO_IO_LOG_ERROR("failed to set session id");
}
// TODO
#if 0
if (marsio_buff_set_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) != 0)
{
MARSIO_IO_LOG_ERROR("failed to set domain id");
}
#endif
if (marsio_buff_set_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) != 0)
{
MARSIO_IO_LOG_ERROR("failed to set link id");
}
if (is_ctrl)
{
marsio_buff_set_ctrlbuf(mbuff);
}
if (marsio_buff_set_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) != 0)
{
MARSIO_IO_LOG_ERROR("failed to set direction");
}
}
static inline int is_keepalive_packet(const char *data, int len)
{
if (data == NULL || len < (int)(sizeof(struct ethhdr)))
{
return 0;
}
struct ethhdr *eth_hdr = (struct ethhdr *)data;
if (eth_hdr->h_proto == 0xAAAA)
{
return 1;
}
else
{
return 0;
}
}
/******************************************************************************
* Public API
******************************************************************************/
void *marsio_io_new(const struct packet_io_config *cfg)
{
int opt = 1;
cpu_set_t coremask;
CPU_ZERO(&coremask);
struct marsio_io *handle = (struct marsio_io *)calloc(1, sizeof(struct marsio_io));
if (handle == NULL)
{
MARSIO_IO_LOG_ERROR("unable to allocate memory for marsio_io");
return NULL;
}
memcpy(&handle->cfg, cfg, sizeof(struct packet_io_config));
for (uint16_t i = 0; i < handle->cfg.nr_worker_thread; i++)
{
CPU_SET(handle->cfg.cpu_mask[i], &coremask);
}
handle->mr_ins = marsio_create();
if (handle->mr_ins == NULL)
{
MARSIO_IO_LOG_ERROR("unable to create marsio instance");
goto error_out;
}
marsio_option_set(handle->mr_ins, MARSIO_OPT_THREAD_MASK_IN_CPUSET, &coremask, sizeof(cpu_set_t));
marsio_option_set(handle->mr_ins, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt));
if (marsio_init(handle->mr_ins, handle->cfg.app_symbol) != 0)
{
MARSIO_IO_LOG_ERROR("unable to init marsio instance");
goto error_out;
}
handle->mr_dev = marsio_open_device(handle->mr_ins, handle->cfg.dev_symbol, handle->cfg.nr_worker_thread, handle->cfg.nr_worker_thread);
if (handle->mr_dev == NULL)
{
MARSIO_IO_LOG_ERROR("unable to open marsio device");
goto error_out;
}
handle->mr_path = marsio_sendpath_create_by_vdev(handle->mr_dev);
if (handle->mr_path == NULL)
{
MARSIO_IO_LOG_ERROR("unable to create marsio sendpath");
goto error_out;
}
return handle;
error_out:
marsio_io_free(handle);
return NULL;
}
void marsio_io_free(void *handle)
{
struct marsio_io *mr_io = (struct marsio_io *)handle;
if (mr_io)
{
if (mr_io->mr_path)
{
marsio_sendpath_destory(mr_io->mr_path);
mr_io->mr_path = NULL;
}
if (mr_io->mr_dev)
{
marsio_close_device(mr_io->mr_dev);
mr_io->mr_dev = NULL;
}
if (mr_io->mr_ins)
{
marsio_destory(mr_io->mr_ins);
mr_io->mr_ins = NULL;
}
free(mr_io);
mr_io = NULL;
}
}
int marsio_io_isbreak(void *handle __attribute__((unused)))
{
return 0;
}
int marsio_io_init(void *handle, uint16_t thr_idx __attribute__((unused)))
{
struct marsio_io *mr_io = (struct marsio_io *)handle;
if (marsio_thread_init(mr_io->mr_ins) != 0)
{
MARSIO_IO_LOG_ERROR("unable to init marsio thread");
return -1;
}
return 0;
}
uint16_t marsio_io_ingress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
{
int len;
char *data;
uint16_t nr_packet_parsed = 0;
struct packet *pkt;
marsio_buff_t *mbuff;
marsio_buff_t *rx_buffs[RX_BURST_MAX];
struct marsio_io *mr_io = (struct marsio_io *)handle;
struct packet_io_stat *stat = &mr_io->stat[thr_idx];
int nr_packet_received = marsio_recv_burst(mr_io->mr_dev, thr_idx, rx_buffs, MIN(RX_BURST_MAX, nr_pkts));
if (nr_packet_received <= 0)
{
return nr_packet_parsed;
}
for (int i = 0; i < nr_packet_received; i++)
{
mbuff = rx_buffs[i];
data = marsio_buff_mtod(mbuff);
len = marsio_buff_datalen(mbuff);
stat->pkts_rx++;
stat->bytes_rx += len;
if (is_keepalive_packet(data, len))
{
stat->keep_alive_pkts++;
stat->keep_alive_bytes += len;
stat->pkts_tx++;
stat->bytes_tx += len;
marsio_send_burst(mr_io->mr_path, thr_idx, &mbuff, 1);
continue;
}
pkt = &pkts[nr_packet_parsed++];
packet_parse(pkt, data, len);
metadata_from_mbuff_to_packet(mbuff, pkt);
if (marsio_buff_is_ctrlbuf(mbuff))
{
stat->ctrl_pkts_rx++;
stat->ctrl_bytes_rx += len;
}
else
{
stat->raw_pkts_rx++;
stat->raw_bytes_rx += len;
}
}
return nr_packet_parsed;
}
void marsio_io_egress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
{
int is_injected_packet = 0;
int len;
char *ptr;
struct packet *pkt;
marsio_buff_t *mbuff;
struct marsio_io *mr_io = (struct marsio_io *)handle;
struct packet_io_stat *stat = &mr_io->stat[thr_idx];
for (uint16_t i = 0; i < nr_pkts; i++)
{
is_injected_packet = 0;
pkt = &pkts[i];
len = packet_get_raw_len(pkt);
mbuff = (marsio_buff_t *)packet_get_origin_ctx(pkt);
if (mbuff == NULL)
{
if (marsio_buff_malloc_global(mr_io->mr_ins, &mbuff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY) < 0)
{
MARSIO_IO_LOG_ERROR("unable to allocate marsio buffer for inject packet");
continue;
}
ptr = marsio_buff_append(mbuff, len);
memcpy(ptr, packet_get_raw_data(pkt), len);
is_injected_packet = 1;
}
metadata_from_packet_to_mbuff(pkt, mbuff);
stat->pkts_tx++;
stat->bytes_tx += len;
if (packet_is_ctrl(pkt))
{
stat->ctrl_pkts_tx++;
stat->ctrl_bytes_tx += len;
}
else
{
stat->raw_pkts_tx++;
stat->raw_bytes_tx += len;
}
if (is_injected_packet)
{
stat->pkts_injected++;
stat->bytes_injected += len;
marsio_send_burst_with_options(mr_io->mr_path, thr_idx, &mbuff, 1, MARSIO_SEND_OPT_REHASH);
}
else
{
marsio_send_burst(mr_io->mr_path, thr_idx, &mbuff, 1);
}
}
}
void marsio_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
{
struct packet *pkt;
marsio_buff_t *mbuff;
struct marsio_io *mr_io = (struct marsio_io *)handle;
struct packet_io_stat *stat = &mr_io->stat[thr_idx];
for (uint16_t i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
mbuff = (marsio_buff_t *)packet_get_origin_ctx(pkt);
if (mbuff)
{
stat->pkts_dropped++;
stat->bytes_dropped += packet_get_raw_len(pkt);
marsio_buff_free(mr_io->mr_ins, &mbuff, 1, 0, thr_idx);
}
}
}
void marsio_io_yield(void *handle, uint16_t thr_idx)
{
struct marsio_io *mr_io = (struct marsio_io *)handle;
struct mr_vdev *vdevs[1] = {
mr_io->mr_dev,
};
marsio_poll_wait(mr_io->mr_ins, vdevs, 1, thr_idx, mr_io->cfg.idle_yield_interval_ms);
}
struct packet_io_stat *marsio_io_stat(void *handle, uint16_t thr_idx)
{
struct marsio_io *mr_io = (struct marsio_io *)handle;
return &mr_io->stat[thr_idx];
}

23
infra/packet_io/marsio_io.h
View File

@@ -1,23 +0,0 @@
#pragma once
#ifdef __cplusplus
extern "C"
{
#endif
#include "packet_io.h"
void *marsio_io_new(const struct packet_io_config *cfg);
void marsio_io_free(void *handle);
int marsio_io_isbreak(void *handle);
int marsio_io_init(void *handle, uint16_t thr_idx);
uint16_t marsio_io_ingress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void marsio_io_egress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void marsio_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void marsio_io_yield(void *handle, uint16_t thr_idx);
struct packet_io_stat *marsio_io_stat(void *handle, uint16_t thr_idx);
#ifdef __cplusplus
}
#endif

283
infra/packet_io/packet_io.c
View File

@@ -1,267 +1,50 @@
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include "toml.h"
#include "pcap_io.h"
#include "marsio_io.h"
#include "mars_io.h"
#include "log_internal.h"
#include "utils_internal.h"
#define PACKET_IO_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "packet io", format, ##__VA_ARGS__)
#define PACKET_IO_LOG_INFO(format, ...) STELLAR_LOG_INFO(__thread_local_logger, "packet io", format, ##__VA_ARGS__)
struct packet_io
{
struct packet_io_config *cfg;
void *handle;
void *(*new_func)(const struct packet_io_config *cfg);
void *(*new_func)(const char *toml_file);
void (*free_func)(void *handle);
int (*isbreak_func)(void *handle);
int (*init_func)(void *handle, uint16_t thr_idx);
uint16_t (*ingress_func)(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void (*egress_func)(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void (*drop_func)(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
int (*recv_func)(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void (*send_func)(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void (*drop_func)(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void (*yield_func)(void *handle, uint16_t thr_idx);
void (*polling_func)(void *handle, uint16_t thr_idx);
struct packet_io_stat *(*stat_func)(void *handle, uint16_t thr_idx);
};
static struct packet_io_config *packet_io_config_new(const char *toml_file)
{
int ret = -1;
const char *ptr;
char *ptr_mode = NULL;
char *ptr_pcap_path = NULL;
char *ptr_app_symbol = NULL;
char *ptr_dev_symbol = NULL;
char errbuf[200];
FILE *fp = NULL;
toml_table_t *root = NULL;
toml_table_t *table = NULL;
toml_array_t *mask;
struct packet_io_config *cfg = (struct packet_io_config *)calloc(1, sizeof(struct packet_io_config));
if (cfg == NULL)
{
return NULL;
}
fp = fopen(toml_file, "r");
if (fp == NULL)
{
PACKET_IO_LOG_ERROR("config file %s open failed, %s", toml_file, strerror(errno));
goto error_out;
}
root = toml_parse_file(fp, errbuf, sizeof(errbuf));
if (root == NULL)
{
PACKET_IO_LOG_ERROR("config file %s parse failed, %s", toml_file, errbuf);
goto error_out;
}
table = toml_table_in(root, "packet_io");
if (table == NULL)
{
PACKET_IO_LOG_ERROR("config file %s missing packet_io", toml_file);
goto error_out;
}
ptr = toml_raw_in(table, "mode");
if (ptr == NULL || toml_rtos(ptr, &ptr_mode) != 0)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.mode");
goto error_out;
}
if (strcmp(ptr_mode, "pcapfile") == 0)
{
cfg->mode = PACKET_IO_PCAPFILE;
}
else if (strcmp(ptr_mode, "pcaplist") == 0)
{
cfg->mode = PACKET_IO_PCAPLIST;
}
else if (strcmp(ptr_mode, "marsio") == 0)
{
cfg->mode = PACKET_IO_MARSIO;
}
else
{
PACKET_IO_LOG_ERROR("config file invalid packet_io.mode %s", ptr);
goto error_out;
}
if (cfg->mode == PACKET_IO_PCAPFILE || cfg->mode == PACKET_IO_PCAPLIST)
{
ptr = toml_raw_in(table, "pcap_path");
if (ptr == NULL || toml_rtos(ptr, &ptr_pcap_path) != 0)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.pcap_path");
goto error_out;
}
strcpy(cfg->pcap_path, ptr_pcap_path);
}
else
{
ptr = toml_raw_in(table, "app_symbol");
if (ptr == NULL || toml_rtos(ptr, &ptr_app_symbol) != 0)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.app_symbol");
goto error_out;
}
strcpy(cfg->app_symbol, ptr_app_symbol);
ptr = toml_raw_in(table, "dev_symbol");
if (ptr == NULL || toml_rtos(ptr, &ptr_dev_symbol) != 0)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.dev_symbol");
goto error_out;
}
strcpy(cfg->dev_symbol, ptr_dev_symbol);
}
ptr = toml_raw_in(table, "nr_worker_thread");
if (ptr == NULL)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.nr_worker_thread");
goto error_out;
}
cfg->nr_worker_thread = atoi(ptr);
if (cfg->nr_worker_thread == 0 || cfg->nr_worker_thread > MAX_THREAD_NUM)
{
PACKET_IO_LOG_ERROR("config file invalid packet_io.nr_worker_thread %d, range [1, %d]", cfg->nr_worker_thread, MAX_THREAD_NUM);
goto error_out;
}
mask = toml_array_in(table, "cpu_mask");
if (mask == NULL)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.cpu_mask");
goto error_out;
}
for (uint16_t i = 0; i < cfg->nr_worker_thread; i++)
{
ptr = toml_raw_at(mask, i);
if (ptr == NULL)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.cpu_mask[%d]", i);
goto error_out;
}
cfg->cpu_mask[i] = atoi(ptr);
}
ptr = toml_raw_in(table, "idle_yield_interval_ms");
if (ptr == NULL)
{
PACKET_IO_LOG_ERROR("config file missing packet_io.idle_yield_interval_ms");
goto error_out;
}
cfg->idle_yield_interval_ms = atoll(ptr);
if (cfg->idle_yield_interval_ms > 60000)
{
PACKET_IO_LOG_ERROR("config file invalid packet_io.idle_yield_interval_ms %d, range [0, %d]", cfg->idle_yield_interval_ms, 60000);
goto error_out;
}
ret = 0;
error_out:
if (ptr_mode)
{
free(ptr_mode);
}
if (ptr_pcap_path)
{
free(ptr_pcap_path);
}
if (ptr_app_symbol)
{
free(ptr_app_symbol);
}
if (ptr_dev_symbol)
{
free(ptr_dev_symbol);
}
if (root)
{
toml_free(root);
}
if (fp)
{
fclose(fp);
}
if (ret == -1)
{
free(cfg);
return NULL;
}
else
{
return cfg;
}
}
static void packet_io_config_free(struct packet_io_config *cfg)
{
if (cfg)
{
free(cfg);
cfg = NULL;
}
}
static void packet_io_config_print(const struct packet_io_config *cfg)
{
if (cfg)
{
PACKET_IO_LOG_INFO("packet_io.mode : %s", cfg->mode == PACKET_IO_PCAPFILE ? "pcapfile" : (cfg->mode == PACKET_IO_PCAPLIST ? "pcaplist" : "marsio"));
if (cfg->mode == PACKET_IO_PCAPFILE || cfg->mode == PACKET_IO_PCAPLIST)
{
PACKET_IO_LOG_INFO("packet_io.pcap_path : %s", cfg->pcap_path);
}
else
{
PACKET_IO_LOG_INFO("packet_io.app_symbol : %s", cfg->app_symbol);
PACKET_IO_LOG_INFO("packet_io.dev_symbol : %s", cfg->dev_symbol);
}
PACKET_IO_LOG_INFO("packet_io.nr_worker_thread : %d", cfg->nr_worker_thread);
for (uint16_t i = 0; i < cfg->nr_worker_thread; i++)
{
PACKET_IO_LOG_INFO("packet_io.cpu_mask[%03d] : %d", i, cfg->cpu_mask[i]);
}
PACKET_IO_LOG_INFO("packet_io.idle_yield_interval_ms : %lu", cfg->idle_yield_interval_ms);
}
}
struct packet_io *packet_io_new(const char *toml_file)
{
char mode[64] = {0};
struct packet_io *pkt_io = (struct packet_io *)calloc(1, sizeof(struct packet_io));
if (pkt_io == NULL)
{
return NULL;
}
pkt_io->cfg = packet_io_config_new(toml_file);
if (pkt_io->cfg == NULL)
load_toml_str_config(toml_file, "packet_io.mode", mode);
if (strcmp(mode, "marsio") == 0)
{
free(pkt_io);
return NULL;
}
packet_io_config_print(pkt_io->cfg);
if (pkt_io->cfg->mode == PACKET_IO_MARSIO)
{
pkt_io->new_func = marsio_io_new;
pkt_io->free_func = marsio_io_free;
pkt_io->isbreak_func = marsio_io_isbreak;
pkt_io->init_func = marsio_io_init;
pkt_io->ingress_func = marsio_io_ingress;
pkt_io->egress_func = marsio_io_egress;
pkt_io->drop_func = marsio_io_drop;
pkt_io->yield_func = marsio_io_yield;
pkt_io->stat_func = marsio_io_stat;
pkt_io->new_func = mars_io_new;
pkt_io->free_func = mars_io_free;
pkt_io->isbreak_func = mars_io_isbreak;
pkt_io->init_func = mars_io_init;
pkt_io->recv_func = mars_io_recv;
pkt_io->send_func = mars_io_send;
pkt_io->drop_func = mars_io_drop;
pkt_io->yield_func = mars_io_yield;
pkt_io->polling_func = mars_io_polling;
pkt_io->stat_func = mars_io_stat;
}
else
{
@@ -269,14 +52,15 @@ struct packet_io *packet_io_new(const char *toml_file)
pkt_io->free_func = pcap_io_free;
pkt_io->isbreak_func = pcap_io_isbreak;
pkt_io->init_func = pcap_io_init;
pkt_io->ingress_func = pcap_io_ingress;
pkt_io->egress_func = pcap_io_egress;
pkt_io->recv_func = pcap_io_recv;
pkt_io->send_func = pcap_io_send;
pkt_io->drop_func = pcap_io_drop;
pkt_io->yield_func = pcap_io_yield;
pkt_io->polling_func = pcap_io_polling;
pkt_io->stat_func = pcap_io_stat;
}
pkt_io->handle = pkt_io->new_func(pkt_io->cfg);
pkt_io->handle = pkt_io->new_func(toml_file);
if (pkt_io->handle == NULL)
{
packet_io_free(pkt_io);
@@ -294,10 +78,6 @@ void packet_io_free(struct packet_io *pkt_io)
{
pkt_io->free_func(pkt_io->handle);
}
if (pkt_io->cfg)
{
packet_io_config_free(pkt_io->cfg);
}
free(pkt_io);
pkt_io = NULL;
}
@@ -313,17 +93,17 @@ int packet_io_init(struct packet_io *pkt_io, uint16_t thr_idx)
return pkt_io->init_func(pkt_io->handle, thr_idx);
}
uint16_t packet_io_ingress(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
int packet_io_recv(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
return pkt_io->ingress_func(pkt_io->handle, thr_idx, pkts, nr_pkts);
return pkt_io->recv_func(pkt_io->handle, thr_idx, pkts, nr_pkts);
}
void packet_io_egress(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
void packet_io_send(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
pkt_io->egress_func(pkt_io->handle, thr_idx, pkts, nr_pkts);
pkt_io->send_func(pkt_io->handle, thr_idx, pkts, nr_pkts);
}
void packet_io_drop(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
void packet_io_drop(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
pkt_io->drop_func(pkt_io->handle, thr_idx, pkts, nr_pkts);
}
@@ -333,6 +113,11 @@ void packet_io_yield(struct packet_io *pkt_io, uint16_t thr_idx)
pkt_io->yield_func(pkt_io->handle, thr_idx);
}
void packet_io_polling(struct packet_io *pkt_io, uint16_t thr_idx)
{
pkt_io->polling_func(pkt_io->handle, thr_idx);
}
struct packet_io_stat *packet_io_stat(struct packet_io *pkt_io, uint16_t thr_idx)
{
return pkt_io->stat_func(pkt_io->handle, thr_idx);

45
infra/packet_io/packet_io.h
View File

@@ -6,9 +6,8 @@ extern "C"
#endif
#include <stdint.h>
#include <limits.h>
#include "utils.h"
#include "stellar/packet.h"
struct packet_io_stat
{
@@ -23,13 +22,6 @@ struct packet_io_stat
uint64_t keep_alive_pkts;
uint64_t keep_alive_bytes;
// raw packet
uint64_t raw_pkts_rx;
uint64_t raw_bytes_rx;
uint64_t raw_pkts_tx;
uint64_t raw_bytes_tx;
// drop packet
uint64_t pkts_dropped;
uint64_t bytes_dropped;
@@ -38,43 +30,22 @@ struct packet_io_stat
uint64_t pkts_injected;
uint64_t bytes_injected;
// ctrl packet
uint64_t ctrl_pkts_rx;
uint64_t ctrl_bytes_rx;
uint64_t ctrl_pkts_tx;
uint64_t ctrl_bytes_tx;
// user freed
uint64_t pkts_user_freed;
uint64_t bytes_user_freed;
} __attribute__((aligned(64)));
enum packet_io_mode
{
PACKET_IO_PCAPFILE = 0,
PACKET_IO_PCAPLIST = 1,
PACKET_IO_MARSIO = 2,
};
struct packet_io_config
{
enum packet_io_mode mode;
char pcap_path[PATH_MAX];
char app_symbol[64];
char dev_symbol[64];
uint16_t nr_worker_thread; // range [1, MAX_THREAD_NUM]
uint16_t cpu_mask[MAX_THREAD_NUM];
uint64_t idle_yield_interval_ms; // range: [0, 6000] (ms)
};
struct packet;
struct packet_io;
struct packet_io *packet_io_new(const char *toml_file);
void packet_io_free(struct packet_io *pkt_io);
int packet_io_isbreak(struct packet_io *pkt_io);
int packet_io_init(struct packet_io *pkt_io, uint16_t thr_idx);
uint16_t packet_io_ingress(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void packet_io_egress(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void packet_io_drop(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
int packet_io_recv(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void packet_io_send(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void packet_io_drop(struct packet_io *pkt_io, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void packet_io_yield(struct packet_io *pkt_io, uint16_t thr_idx);
void packet_io_polling(struct packet_io *pkt_io, uint16_t thr_idx);
struct packet_io_stat *packet_io_stat(struct packet_io *pkt_io, uint16_t thr_idx);
#ifdef __cplusplus

606
infra/packet_io/pcap_io.c
View File

@@ -1,42 +1,23 @@
#include <pcap/pcap.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <errno.h>
#include <dirent.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <limits.h>
#include <pthread.h>
#include <pcap/pcap.h>
#include "tuple.h"
#include "utils.h"
#include "log_internal.h"
#include "pcap_io.h"
#include "packet_dump.h"
#include "packet_pool.h"
#include "packet_parser.h"
#include "ip_reassembly.h"
#include "log_internal.h"
#include "packet_internal.h"
#include "utils_internal.h"
#define PCAP_IO_LOG_FATAL(format, ...) STELLAR_LOG_FATAL(__thread_local_logger, "pcap io", format, ##__VA_ARGS__)
#define PCAP_IO_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "pcap io", format, ##__VA_ARGS__)
#define PCAP_IO_LOG_INFO(format, ...) STELLAR_LOG_INFO(__thread_local_logger, "pcap io", format, ##__VA_ARGS__)
#define MAX_PACKET_QUEUE_SIZE (4096 * 1000)
struct pcap_io
{
struct packet_io_config cfg;
pcap_t *pcap;
struct logger *logger;
struct packet_queue *queue[MAX_THREAD_NUM];
struct packet_io_stat stat[MAX_THREAD_NUM];
uint64_t io_thread_need_exit;
uint64_t io_thread_is_runing;
uint64_t io_thread_wait_exit;
uint64_t read_pcap_files;
uint64_t read_pcap_pkts;
};
#define RING_BUFFER_MAX_SIZE (4096 * 1000)
struct pcap_pkt
{
@@ -45,92 +26,186 @@ struct pcap_pkt
struct timeval ts;
};
struct pcap_io_cfg
{
char mode[16]; // pcapfile, pcaplist
char pcap_path[PATH_MAX];
uint64_t thread_num; // range [1, MAX_THREAD_NUM]
// packet pool
uint64_t capacity; // range: [1, 4294967295]
// ip reassembly
uint64_t fail_action; // 0: bypass, 1: drop
uint64_t timeout_ms; // range: [1, 60000] (ms)
uint64_t frag_queue_num; // range: [1, 4294967295
uint64_t frag_queue_size; // range: [2, 65535]
};
struct pcap_io
{
struct pcap_io_cfg *cfg;
pcap_t *pcap;
struct logger *logger;
struct ring_buffer *ring[MAX_THREAD_NUM];
struct packet_pool *pool[MAX_THREAD_NUM];
struct packet_io_stat stat[MAX_THREAD_NUM];
struct ip_reassembly *ip_reass[MAX_THREAD_NUM];
uint64_t io_thread_need_exit;
uint64_t io_thread_is_runing;
uint64_t io_thread_wait_exit;
uint64_t read_pcap_files;
uint64_t read_pcap_pkts;
};
/******************************************************************************
* Private API -- queue
* Private API -- ring
******************************************************************************/
struct packet_queue
struct ring_buffer
{
uint64_t *queue;
uint64_t *buff;
uint32_t size;
uint32_t head;
uint32_t tail;
};
static struct packet_queue *packet_queue_new(uint32_t size)
static struct ring_buffer *ring_buffer_new(uint32_t size)
{
struct packet_queue *queue = (struct packet_queue *)calloc(1, sizeof(struct packet_queue));
if (queue == NULL)
struct ring_buffer *ring = (struct ring_buffer *)calloc(1, sizeof(struct ring_buffer));
if (ring == NULL)
{
PCAP_IO_LOG_ERROR("unable to new packet queue");
PCAP_IO_LOG_ERROR("unable to new ring buffer");
return NULL;
}
queue->queue = (uint64_t *)calloc(size, sizeof(uint64_t));
if (queue->queue == NULL)
ring->buff = (uint64_t *)calloc(size, sizeof(uint64_t));
if (ring->buff == NULL)
{
PCAP_IO_LOG_ERROR("unable to new packet queue");
free(queue);
PCAP_IO_LOG_ERROR("unable to new ring buffer");
free(ring);
return NULL;
}
queue->size = size;
queue->head = 0;
queue->tail = 0;
ring->size = size;
ring->head = 0;
ring->tail = 0;
return queue;
return ring;
}
static void packet_queue_free(struct packet_queue *queue)
static void ring_buffer_free(struct ring_buffer *ring)
{
if (queue == NULL)
if (ring)
{
return;
if (ring->buff)
{
free(ring->buff);
ring->buff = NULL;
}
free(ring);
ring = NULL;
}
if (queue->queue)
{
free(queue->queue);
queue->queue = NULL;
}
free(queue);
}
static int packet_queue_push(struct packet_queue *queue, void *data)
static int ring_buffer_push(struct ring_buffer *ring, void *data)
{
if (__sync_val_compare_and_swap(&queue->queue[queue->tail], 0, data) != 0)
if (__sync_val_compare_and_swap(&ring->buff[ring->tail], 0, data) != 0)
{
PCAP_IO_LOG_ERROR("packet queue is full, retry later");
PCAP_IO_LOG_ERROR("ring buffer is full, retry later");
return -1;
}
queue->tail = (queue->tail + 1) % queue->size;
ring->tail = (ring->tail + 1) % ring->size;
return 0;
}
static void packet_queue_pop(struct packet_queue *queue, void **data)
static void ring_buffer_pop(struct ring_buffer *ring, void **data)
{
uint64_t read = ATOMIC_READ(&queue->queue[queue->head]);
uint64_t read = ATOMIC_READ(&ring->buff[ring->head]);
if (read == 0)
{
*data = NULL;
return;
}
__sync_val_compare_and_swap(&queue->queue[queue->head], read, 0);
__sync_val_compare_and_swap(&ring->buff[ring->head], read, 0);
*data = (void *)read;
queue->head = (queue->head + 1) % queue->size;
ring->head = (ring->head + 1) % ring->size;
}
/******************************************************************************
* Private API -- utils
* Private API -- config
******************************************************************************/
static struct pcap_io_cfg *pcap_io_cfg_new(const char *toml_file)
{
struct pcap_io_cfg *cfg = (struct pcap_io_cfg *)calloc(1, sizeof(struct pcap_io_cfg));
if (cfg == NULL)
{
return NULL;
}
int ret = 0;
ret += load_toml_str_config(toml_file, "packet_io.mode", cfg->mode);
ret += load_toml_str_config(toml_file, "packet_io.pcap_path", cfg->pcap_path);
ret += load_toml_integer_config(toml_file, "packet_io.thread_num", &cfg->thread_num, 1, MAX_THREAD_NUM);
ret += load_toml_integer_config(toml_file, "packet_io.packet_pool.capacity", &cfg->capacity, 1, 4294967295);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.fail_action", &cfg->fail_action, 0, 1);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.timeout_ms", &cfg->timeout_ms, 1, 60000);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.frag_queue_num", &cfg->frag_queue_num, 1, 4294967295);
ret += load_toml_integer_config(toml_file, "packet_io.ip_reassembly.frag_queue_size", &cfg->frag_queue_size, 2, 65535);
if (strcmp(cfg->mode, "pcapfile") != 0 && strcmp(cfg->mode, "pcaplist") != 0)
{
PCAP_IO_LOG_ERROR("config file invalid packet_io.mode %s", cfg->mode);
ret = -1;
}
if (ret != 0)
{
free(cfg);
return NULL;
}
else
{
return cfg;
}
}
static void pcap_io_cfg_free(struct pcap_io_cfg *cfg)
{
if (cfg)
{
free(cfg);
cfg = NULL;
}
}
static void pcap_io_cfg_print(const struct pcap_io_cfg *cfg)
{
if (cfg)
{
PCAP_IO_LOG_INFO("packet_io.mode : %s", cfg->mode);
PCAP_IO_LOG_INFO("packet_io.pcap_path : %s", cfg->pcap_path);
PCAP_IO_LOG_INFO("packet_io.thread_num : %ld", cfg->thread_num);
PCAP_IO_LOG_INFO("packet_io.packet_pool.capacity : %lu", cfg->capacity);
PCAP_IO_LOG_INFO("packet_io.ip_reassembly.fail_action : %lu", cfg->fail_action);
PCAP_IO_LOG_INFO("packet_io.ip_reassembly.timeout_ms : %lu", cfg->timeout_ms);
PCAP_IO_LOG_INFO("packet_io.ip_reassembly.frag_queue_num : %lu", cfg->frag_queue_num);
PCAP_IO_LOG_INFO("packet_io.ip_reassembly.frag_queue_size : %lu", cfg->frag_queue_size);
}
}
/******************************************************************************
* Private API -- pcap
******************************************************************************/
static void pcap_pkt_handler(u_char *user, const struct pcap_pkthdr *h, const u_char *bytes)
{
struct pcap_io *handle = (struct pcap_io *)user;
struct pcap_io *pcap_io = (struct pcap_io *)user;
// copy packet data to new memory
struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)calloc(1, sizeof(struct pcap_pkt) + h->caplen);
if (pcap_pkt == NULL)
{
@@ -141,36 +216,34 @@ static void pcap_pkt_handler(u_char *user, const struct pcap_pkthdr *h, const u_
pcap_pkt->len = h->caplen;
pcap_pkt->ts = h->ts;
memcpy((char *)pcap_pkt->data, bytes, h->caplen);
ATOMIC_INC(&handle->read_pcap_pkts);
ATOMIC_INC(&pcap_io->read_pcap_pkts);
// calculate packet hash
struct packet pkt;
memset(&pkt, 0, sizeof(struct packet));
packet_parse(&pkt, pcap_pkt->data, pcap_pkt->len);
uint64_t hash = packet_ldbc_hash(&pkt, PKT_LDBC_METH_OUTERMOST_INT_EXT_IP, PACKET_DIRECTION_OUTGOING);
// push packet to queue
struct packet_queue *queue = handle->queue[hash % handle->cfg.nr_worker_thread];
while (packet_queue_push(queue, pcap_pkt) == -1)
struct ring_buffer *ring = pcap_io->ring[hash % pcap_io->cfg->thread_num];
while (ring_buffer_push(ring, pcap_pkt) == -1)
{
if (ATOMIC_READ(&handle->io_thread_need_exit))
if (ATOMIC_READ(&pcap_io->io_thread_need_exit))
{
free(pcap_pkt);
PCAP_IO_LOG_FATAL("pcap io thread need exit");
pcap_breakloop(handle->pcap);
pcap_breakloop(pcap_io->pcap);
break;
}
usleep(1000);
}
if (ATOMIC_READ(&handle->io_thread_need_exit))
if (ATOMIC_READ(&pcap_io->io_thread_need_exit))
{
PCAP_IO_LOG_FATAL("pcap io thread need exit");
pcap_breakloop(handle->pcap);
pcap_breakloop(pcap_io->pcap);
}
}
static int pcap_io_handler(struct pcap_io *handle, const char *pcap_file)
static int pcap_io_handler(struct pcap_io *pcap_io, const char *pcap_file)
{
char resolved_path[256];
char pcap_errbuf[PCAP_ERRBUF_SIZE];
@@ -178,28 +251,28 @@ static int pcap_io_handler(struct pcap_io *handle, const char *pcap_file)
realpath(pcap_file, resolved_path);
PCAP_IO_LOG_FATAL("pcap %s in-processing", resolved_path)
handle->pcap = pcap_open_offline(resolved_path, pcap_errbuf);
if (handle->pcap == NULL)
pcap_io->pcap = pcap_open_offline(resolved_path, pcap_errbuf);
if (pcap_io->pcap == NULL)
{
PCAP_IO_LOG_ERROR("unable to open pcap file: %s, %s", resolved_path, pcap_errbuf);
return -1;
}
handle->read_pcap_files++;
pcap_loop(handle->pcap, -1, pcap_pkt_handler, (u_char *)handle);
pcap_close(handle->pcap);
pcap_io->read_pcap_files++;
pcap_loop(pcap_io->pcap, -1, pcap_pkt_handler, (u_char *)pcap_io);
pcap_close(pcap_io->pcap);
PCAP_IO_LOG_FATAL("pcap %s processed", resolved_path)
return 0;
}
static int all_packet_consumed(struct pcap_io *handle)
static int all_packet_consumed(struct pcap_io *pcap_io)
{
uint64_t consumed_pkts = 0;
uint64_t read_pcap_pkts = ATOMIC_READ(&handle->read_pcap_pkts);
for (uint16_t i = 0; i < handle->cfg.nr_worker_thread; i++)
uint64_t read_pcap_pkts = ATOMIC_READ(&pcap_io->read_pcap_pkts);
for (uint16_t i = 0; i < pcap_io->cfg->thread_num; i++)
{
consumed_pkts += ATOMIC_READ(&handle->stat[i].pkts_rx);
consumed_pkts += ATOMIC_READ(&pcap_io->stat[i].pkts_rx);
}
if (consumed_pkts == read_pcap_pkts)
{
@@ -213,49 +286,47 @@ static int all_packet_consumed(struct pcap_io *handle)
static void *pcap_io_thread(void *arg)
{
struct pcap_io *handle = (struct pcap_io *)arg;
__thread_local_logger = handle->logger;
struct pcap_io *pcap_io = (struct pcap_io *)arg;
__thread_local_logger = pcap_io->logger;
ATOMIC_SET(&handle->io_thread_is_runing, 1);
ATOMIC_SET(&pcap_io->io_thread_is_runing, 1);
PCAP_IO_LOG_FATAL("pcap io thread is running");
if (handle->cfg.mode == PACKET_IO_PCAPFILE)
if (strcmp(pcap_io->cfg->mode, "pcapfile") == 0)
{
pcap_io_handler(handle, handle->cfg.pcap_path);
pcap_io_handler(pcap_io, pcap_io->cfg->pcap_path);
}
else // PACKET_IO_PCAPLIST
else
{
FILE *fp = NULL;
if (strcmp(handle->cfg.pcap_path, "-") == 0)
if (strcmp(pcap_io->cfg->pcap_path, "-") == 0)
{
PCAP_IO_LOG_ERROR("pcap path is empty, read from stdin");
fp = stdin;
}
else
{
fp = fopen(handle->cfg.pcap_path, "r");
fp = fopen(pcap_io->cfg->pcap_path, "r");
if (fp == NULL)
{
PCAP_IO_LOG_ERROR("unable to open pcap path: %s", handle->cfg.pcap_path);
PCAP_IO_LOG_ERROR("unable to open pcap path: %s", pcap_io->cfg->pcap_path);
goto erro_out;
}
}
char line[PATH_MAX];
while (ATOMIC_READ(&handle->io_thread_need_exit) == 0 && fgets(line, sizeof(line), fp))
while (ATOMIC_READ(&pcap_io->io_thread_need_exit) == 0 && fgets(line, sizeof(line), fp))
{
if (line[0] == '#')
{
continue;
}
char *pos = strchr(line, '\n');
if (pos)
{
*pos = '\0';
}
pcap_io_handler(handle, line);
pcap_io_handler(pcap_io, line);
}
if (fp != stdin)
{
@@ -265,58 +336,201 @@ static void *pcap_io_thread(void *arg)
PCAP_IO_LOG_FATAL("pcap io thread read all pcap files");
erro_out:
while (ATOMIC_READ(&handle->io_thread_need_exit) == 0)
while (ATOMIC_READ(&pcap_io->io_thread_need_exit) == 0)
{
if (all_packet_consumed(handle))
if (all_packet_consumed(pcap_io))
{
ATOMIC_SET(&handle->io_thread_wait_exit, 1);
ATOMIC_SET(&pcap_io->io_thread_wait_exit, 1);
}
usleep(1000); // 1ms
}
PCAP_IO_LOG_FATAL("pcap io thread exit (read_pcap_files: %lu, read_pcap_pkts: %lu)", handle->read_pcap_files, ATOMIC_READ(&handle->read_pcap_pkts));
ATOMIC_SET(&handle->io_thread_is_runing, 0);
PCAP_IO_LOG_FATAL("pcap io thread exit (read_pcap_files: %lu, read_pcap_pkts: %lu)", pcap_io->read_pcap_files, ATOMIC_READ(&pcap_io->read_pcap_pkts));
ATOMIC_SET(&pcap_io->io_thread_is_runing, 0);
return NULL;
}
static void origin_free_cb(struct packet *pkt, void *args)
{
struct pcap_io *pcap_io = (struct pcap_io *)args;
struct packet_origin *origin = packet_get_origin(pkt);
struct pcap_pkt *pcap_pkt = origin->ctx;
struct packet_io_stat *stat = &pcap_io->stat[origin->thr_idx];
stat->pkts_user_freed++;
stat->bytes_user_freed += packet_get_raw_len(pkt);
free(pcap_pkt);
}
static struct packet *recv_packet(struct pcap_io *pcap_io, struct pcap_pkt *pcap_pkt, uint16_t thr_idx)
{
struct packet_io_stat *stat = &pcap_io->stat[thr_idx];
struct ip_reassembly *ip_reass = pcap_io->ip_reass[thr_idx];
struct packet_pool *pool = pcap_io->pool[thr_idx];
if (pcap_pkt == NULL)
{
return NULL;
}
stat->pkts_rx++;
stat->bytes_rx += pcap_pkt->len;
struct packet *pkt = packet_pool_pop(pool);
assert(pkt != NULL);
struct packet_origin origin = {
.type = ORIGIN_TYPE_PCAP,
.ctx = pcap_pkt,
.cb = origin_free_cb,
.args = pcap_io,
.thr_idx = thr_idx,
};
packet_parse(pkt, pcap_pkt->data, pcap_pkt->len);
memset(&pkt->meta, 0, sizeof(pkt->meta));
packet_set_action(pkt, PACKET_ACTION_FORWARD);
packet_set_timeval(pkt, &pcap_pkt->ts);
packet_set_origin(pkt, &origin);
if (packet_is_fragment(pkt))
{
return ip_reassembly_defrag(ip_reass, pkt, clock_get_real_time_ms());
}
else
{
return pkt;
}
}
static void send_packet(struct pcap_io *pcap_io, struct packet *pkt, uint16_t thr_idx)
{
struct pcap_pkt *pcap_pkt = NULL;
struct packet_io_stat *stat = &pcap_io->stat[thr_idx];
int len = packet_get_raw_len(pkt);
struct packet_origin *origin = packet_get_origin(pkt);
if (origin->type == ORIGIN_TYPE_PCAP)
{
pcap_pkt = (struct pcap_pkt *)origin->ctx;
free(pcap_pkt);
packet_pool_push(pcap_io->pool[thr_idx], pkt);
}
else
{
stat->pkts_injected++;
stat->bytes_injected += len;
struct tuple6 tuple;
char file[PATH_MAX] = {0};
char src_addr[INET6_ADDRSTRLEN] = {0};
char dst_addr[INET6_ADDRSTRLEN] = {0};
memset(&tuple, 0, sizeof(struct tuple6));
packet_get_innermost_tuple6(pkt, &tuple);
if (tuple.addr_family == AF_INET)
{
inet_ntop(AF_INET, &tuple.src_addr.v4, src_addr, INET6_ADDRSTRLEN);
inet_ntop(AF_INET, &tuple.dst_addr.v4, dst_addr, INET6_ADDRSTRLEN);
}
else
{
inet_ntop(AF_INET6, &tuple.src_addr.v6, src_addr, INET6_ADDRSTRLEN);
inet_ntop(AF_INET6, &tuple.dst_addr.v6, dst_addr, INET6_ADDRSTRLEN);
}
snprintf(file, sizeof(file), "inject-%s:%u-%s:%u-%lu.pcap", src_addr, ntohs(tuple.src_port), dst_addr, ntohs(tuple.dst_port), stat->pkts_injected);
if (packet_dump_pcap(pkt, file) == -1)
{
PCAP_IO_LOG_ERROR("unable to dump pcap file: %s", file);
}
else
{
PCAP_IO_LOG_FATAL("dump inject packet: %s", file);
}
packet_free(pkt);
}
stat->pkts_tx++;
stat->bytes_tx += len;
}
static void drop_packet(struct pcap_io *pcap_io, struct packet *pkt, uint16_t thr_idx)
{
struct packet_io_stat *stat = &pcap_io->stat[thr_idx];
int len = packet_get_raw_len(pkt);
struct packet_origin *origin = packet_get_origin(pkt);
stat->pkts_dropped++;
stat->bytes_dropped += len;
if (origin->type == ORIGIN_TYPE_PCAP)
{
struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)origin->ctx;
free(pcap_pkt);
packet_pool_push(pcap_io->pool[thr_idx], pkt);
}
else
{
packet_free(pkt);
}
}
/******************************************************************************
* Public API
******************************************************************************/
void *pcap_io_new(const struct packet_io_config *cfg)
void *pcap_io_new(const char *toml_file)
{
pthread_t tid;
struct pcap_io *handle = (struct pcap_io *)calloc(1, sizeof(struct pcap_io));
if (handle == NULL)
struct pcap_io *pcap_io = (struct pcap_io *)calloc(1, sizeof(struct pcap_io));
if (pcap_io == NULL)
{
PCAP_IO_LOG_ERROR("unable to allocate memory for pcap_io");
return NULL;
}
handle->logger = __thread_local_logger;
memcpy(&handle->cfg, cfg, sizeof(struct packet_io_config));
for (uint16_t i = 0; i < handle->cfg.nr_worker_thread; i++)
pcap_io->cfg = pcap_io_cfg_new(toml_file);
if (pcap_io->cfg == NULL)
{
handle->queue[i] = packet_queue_new(MAX_PACKET_QUEUE_SIZE);
if (handle->queue[i] == NULL)
PCAP_IO_LOG_ERROR("unable to create pcap_io_cfg");
goto error_out;
}
pcap_io_cfg_print(pcap_io->cfg);
pcap_io->logger = __thread_local_logger;
for (uint16_t i = 0; i < pcap_io->cfg->thread_num; i++)
{
pcap_io->ring[i] = ring_buffer_new(RING_BUFFER_MAX_SIZE);
if (pcap_io->ring[i] == NULL)
{
PCAP_IO_LOG_ERROR("unable to create packet queue");
PCAP_IO_LOG_ERROR("unable to create ring buffer");
goto error_out;
}
pcap_io->pool[i] = packet_pool_new(pcap_io->cfg->capacity);
if (pcap_io->pool[i] == NULL)
{
PCAP_IO_LOG_ERROR("unable to create packet pool");
goto error_out;
}
pcap_io->ip_reass[i] = ip_reassembly_new(pcap_io->cfg->timeout_ms, pcap_io->cfg->frag_queue_num, pcap_io->cfg->frag_queue_size);
if (pcap_io->ip_reass[i] == NULL)
{
PCAP_IO_LOG_ERROR("unable to create ip reassembly");
goto error_out;
}
}
if (pthread_create(&tid, NULL, pcap_io_thread, (void *)handle) != 0)
if (pthread_create(&tid, NULL, pcap_io_thread, (void *)pcap_io) != 0)
{
PCAP_IO_LOG_ERROR("unable to create pcap io thread");
goto error_out;
}
return handle;
return pcap_io;
error_out:
pcap_io_free(handle);
pcap_io_free(pcap_io);
return NULL;
}
@@ -333,11 +547,11 @@ void pcap_io_free(void *handle)
}
struct pcap_pkt *pcap_pkt = NULL;
for (uint16_t i = 0; i < pcap_io->cfg.nr_worker_thread; i++)
for (uint16_t i = 0; i < pcap_io->cfg->thread_num; i++)
{
while (1)
{
packet_queue_pop(pcap_io->queue[i], (void **)&pcap_pkt);
ring_buffer_pop(pcap_io->ring[i], (void **)&pcap_pkt);
if (pcap_pkt)
{
free(pcap_pkt);
@@ -347,9 +561,11 @@ void pcap_io_free(void *handle)
break;
}
}
packet_queue_free(pcap_io->queue[i]);
ip_reassembly_free(pcap_io->ip_reass[i]);
packet_pool_free(pcap_io->pool[i]);
ring_buffer_free(pcap_io->ring[i]);
}
pcap_io_cfg_free(pcap_io->cfg);
free(pcap_io);
pcap_io = NULL;
}
@@ -367,128 +583,76 @@ int pcap_io_init(void *handle __attribute__((unused)), uint16_t thr_idx __attrib
return 0;
}
uint16_t pcap_io_ingress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
int pcap_io_recv(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
uint16_t nr_packet_parsed = 0;
struct packet *pkt = NULL;
struct pcap_pkt *pcap_pkt = NULL;
struct pcap_io *pcap_io = (struct pcap_io *)handle;
struct packet_queue *queue = pcap_io->queue[thr_idx];
struct packet_io_stat *stat = &pcap_io->stat[thr_idx];
struct ring_buffer *ring = pcap_io->ring[thr_idx];
for (uint16_t i = 0; i < nr_pkts; i++)
int ret = 0;
for (int i = 0; i < nr_pkts; i++)
{
packet_queue_pop(queue, (void **)&pcap_pkt);
if (pcap_pkt == NULL)
ring_buffer_pop(ring, (void **)&pcap_pkt);
pkt = recv_packet(pcap_io, pcap_pkt, thr_idx);
if (pkt)
{
break;
}
else
{
ATOMIC_INC(&stat->pkts_rx);
stat->bytes_rx += pcap_pkt->len;
stat->raw_pkts_rx++;
stat->raw_bytes_rx += pcap_pkt->len;
pkt = &pkts[nr_packet_parsed];
packet_parse(pkt, pcap_pkt->data, pcap_pkt->len);
memset(&pkt->meta, 0, sizeof(pkt->meta));
packet_set_origin_ctx(pkt, pcap_pkt);
packet_set_action(pkt, PACKET_ACTION_FORWARD);
packet_set_timeval(pkt, &pcap_pkt->ts);
nr_packet_parsed++;
pkts[ret++] = pkt;
}
}
return nr_packet_parsed;
return ret;
}
void pcap_io_egress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
void pcap_io_send(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
int len;
struct tuple6 tuple;
struct packet *frag = NULL;
struct packet *pkt = NULL;
struct pcap_io *pcap_io = (struct pcap_io *)handle;
struct packet_io_stat *stat = &pcap_io->stat[thr_idx];
char file[PATH_MAX] = {0};
char src_addr[INET6_ADDRSTRLEN] = {0};
char dst_addr[INET6_ADDRSTRLEN] = {0};
for (uint16_t i = 0; i < nr_pkts; i++)
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
len = packet_get_raw_len(pkt);
pkt = pkts[i];
stat->pkts_tx++;
stat->bytes_tx += len;
if (packet_is_ctrl(pkt))
if (packet_is_defraged(pkt))
{
stat->ctrl_pkts_tx++;
stat->ctrl_bytes_tx += len;
while ((frag = packet_pop_frag(pkt)))
{
send_packet(pcap_io, frag, thr_idx);
}
packet_free(pkt);
}
else
{
stat->raw_pkts_tx++;
stat->raw_bytes_tx += len;
}
struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_origin_ctx(pkt);
if (pcap_pkt)
{
free(pcap_pkt);
}
else
{
stat->pkts_injected++;
stat->bytes_injected += len;
memset(&tuple, 0, sizeof(struct tuple6));
packet_get_innermost_tuple6(pkt, &tuple);
if (tuple.addr_family == AF_INET)
{
inet_ntop(AF_INET, &tuple.src_addr.v4, src_addr, INET6_ADDRSTRLEN);
inet_ntop(AF_INET, &tuple.dst_addr.v4, dst_addr, INET6_ADDRSTRLEN);
}
else
{
inet_ntop(AF_INET6, &tuple.src_addr.v6, src_addr, INET6_ADDRSTRLEN);
inet_ntop(AF_INET6, &tuple.dst_addr.v6, dst_addr, INET6_ADDRSTRLEN);
}
snprintf(file, sizeof(file), "inject-%s:%u-%s:%u-%lu.pcap", src_addr, ntohs(tuple.src_port), dst_addr, ntohs(tuple.dst_port), stat->pkts_injected);
if (packet_dump_pcap(pkt, file) == -1)
{
PCAP_IO_LOG_ERROR("unable to dump pcap file: %s", file);
}
else
{
PCAP_IO_LOG_FATAL("dump inject packet: %s", file);
}
send_packet(pcap_io, pkt, thr_idx);
}
pkts[i] = NULL;
}
}
void pcap_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
void pcap_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts)
{
struct packet *pkt = NULL;
struct packet *frag = NULL;
struct pcap_io *pcap_io = (struct pcap_io *)handle;
struct packet_io_stat *stat = &pcap_io->stat[thr_idx];
for (uint16_t i = 0; i < nr_pkts; i++)
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_origin_ctx(pkt);
if (pcap_pkt)
pkt = pkts[i];
if (packet_is_defraged(pkt))
{
stat->pkts_dropped++;
stat->bytes_dropped += packet_get_raw_len(pkt);
free(pcap_pkt);
while ((frag = packet_pop_frag(pkt)))
{
drop_packet(pcap_io, frag, thr_idx);
}
packet_free(pkt);
}
packet_free(pkt);
else
{
drop_packet(pcap_io, pkt, thr_idx);
}
pkts[i] = NULL;
}
}
@@ -497,9 +661,31 @@ void pcap_io_yield(void *handle __attribute__((unused)), uint16_t thr_idx __attr
return;
}
void pcap_io_polling(void *handle, uint16_t thr_idx)
{
struct pcap_io *pcap_io = (struct pcap_io *)handle;
struct ip_reassembly *ip_reass = pcap_io->ip_reass[thr_idx];
struct packet *pkt = NULL;
uint64_t now_ms = clock_get_real_time_ms();
while ((pkt = ip_reassembly_clean(ip_reass, now_ms)))
{
if (pcap_io->cfg->fail_action == 0)
{
send_packet(pcap_io, pkt, thr_idx);
}
else
{
drop_packet(pcap_io, pkt, thr_idx);
}
}
// TODO
// output stat
}
struct packet_io_stat *pcap_io_stat(void *handle, uint16_t thr_idx)
{
struct pcap_io *pcap_io = (struct pcap_io *)handle;
return &pcap_io->stat[thr_idx];
}

9
infra/packet_io/pcap_io.h
View File

@@ -7,15 +7,16 @@ extern "C"
#include "packet_io.h"
void *pcap_io_new(const struct packet_io_config *cfg);
void *pcap_io_new(const char *toml_file);
void pcap_io_free(void *handle);
int pcap_io_isbreak(void *handle);
int pcap_io_init(void *handle, uint16_t thr_idx);
uint16_t pcap_io_ingress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void pcap_io_egress(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
void pcap_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
int pcap_io_recv(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void pcap_io_send(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void pcap_io_drop(void *handle, uint16_t thr_idx, struct packet *pkts[], int nr_pkts);
void pcap_io_yield(void *handle, uint16_t thr_idx);
void pcap_io_polling(void *handle, uint16_t thr_idx);
struct packet_io_stat *pcap_io_stat(void *handle, uint16_t thr_idx);
#ifdef __cplusplus