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

reactor packet I/O & duplicated packet filter & evicted session filter

Browse Source
This commit is contained in:
luwenpeng
2024-03-09 19:28:14 +08:00
parent ee35a26a9d
commit 512dfddd03
79 changed files with 1974 additions and 2093 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/packet_io/CMakeLists.txt
View File

@@ -1,8 +1,4 @@
###############################################################################
# packet_io
###############################################################################
add_library(packet_io packet_queue.cpp packet_io.cpp packet_io_dumpfile.cpp packet_io_marsio.cpp)
add_library(packet_io packet_io_dumpfile.cpp packet_io_marsio.cpp lock_free_queue.cpp packet_io.cpp)
target_include_directories(packet_io PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_include_directories(packet_io PUBLIC ${CMAKE_SOURCE_DIR}/src/stellar)
target_link_libraries(packet_io mrzcpd packet file pcap)
target_link_libraries(packet_io mrzcpd pcap packet file)

40
src/packet_io/packet_queue.cpp → src/packet_io/lock_free_queue.cpp
View File

@@ -2,11 +2,10 @@
#include <stdlib.h>
#include <unistd.h>
#include "packet_io.h"
#include "packet_utils.h"
#include "packet_queue.h"
#include "stellar.h"
#include "lock_free_queue.h"
struct packet_queue
struct lock_free_queue
{
uint64_t *queue;
uint32_t size;
@@ -14,19 +13,19 @@ struct packet_queue
uint32_t tail;
};
struct packet_queue *packet_queue_new(uint32_t size)
struct lock_free_queue *lock_free_queue_new(uint32_t size)
{
struct packet_queue *queue = (struct packet_queue *)calloc(1, sizeof(struct packet_queue));
struct lock_free_queue *queue = (struct lock_free_queue *)calloc(1, sizeof(struct lock_free_queue));
if (queue == NULL)
{
PACKET_IO_LOG_ERROR("unable to alloc packet queue");
LOCK_FREE_QUEUE_LOG_ERROR("unable to new lock free queue");
return NULL;
}
queue->queue = (uint64_t *)calloc(size, sizeof(uint64_t));
if (queue->queue == NULL)
{
PACKET_IO_LOG_ERROR("unable to alloc packet queue buffer");
LOCK_FREE_QUEUE_LOG_ERROR("unable to new lock free queue");
free(queue);
return NULL;
}
@@ -38,22 +37,17 @@ struct packet_queue *packet_queue_new(uint32_t size)
return queue;
}
void packet_queue_free(struct packet_queue *queue)
void lock_free_queue_free(struct lock_free_queue *queue)
{
if (queue == NULL)
{
return;
}
struct packet *pkt = NULL;
while (1)
// wait queue is empty
while (queue->head != queue->tail)
{
packet_queue_pop(queue, &pkt);
if (pkt == NULL)
{
break;
}
packet_free(pkt);
usleep(1000);
}
if (queue->queue)
@@ -65,13 +59,13 @@ void packet_queue_free(struct packet_queue *queue)
free(queue);
}
void packet_queue_push(struct packet_queue *queue, struct packet *pkt)
void lock_free_queue_push(struct lock_free_queue *queue, void *data)
{
uint64_t wait = 1000;
retry:
if (__sync_val_compare_and_swap(&queue->queue[queue->tail], 0, pkt) != 0)
if (__sync_val_compare_and_swap(&queue->queue[queue->tail], 0, data) != 0)
{
PACKET_IO_LOG_ERROR("packet queue is full, retry later");
LOCK_FREE_QUEUE_LOG_ERROR("lock free queue is full, retry later");
usleep(wait);
wait *= 2;
goto retry;
@@ -80,15 +74,15 @@ retry:
queue->tail = (queue->tail + 1) % queue->size;
}
void packet_queue_pop(struct packet_queue *queue, struct packet **pkt)
void lock_free_queue_pop(struct lock_free_queue *queue, void **data)
{
uint64_t read = ATOMIC_READ(&queue->queue[queue->head]);
if (read == 0)
{
*pkt = NULL;
*data = NULL;
return;
}
__sync_val_compare_and_swap(&queue->queue[queue->head], read, 0);
*pkt = (struct packet *)read;
*data = (void *)read;
queue->head = (queue->head + 1) % queue->size;
}

26
src/packet_io/lock_free_queue.h Normal file
View File

@@ -0,0 +1,26 @@
#ifndef _LOCK_FREE_QUEUE_H
#define _LOCK_FREE_QUEUE_H
#ifdef __cpluscplus
extern "C"
{
#endif
#include "log.h"
#define LOCK_FREE_QUEUE_LOG_ERROR(format, ...) LOG_ERROR("lock free queue", format, ##__VA_ARGS__)
#define LOCK_FREE_QUEUE_LOG_DEBUG(format, ...) LOG_DEBUG("lock free queue", format, ##__VA_ARGS__)
struct lock_free_queue;
struct lock_free_queue *lock_free_queue_new(uint32_t size);
void lock_free_queue_free(struct lock_free_queue *queue);
void lock_free_queue_push(struct lock_free_queue *queue, void *data);
void lock_free_queue_pop(struct lock_free_queue *queue, void **data);
#ifdef __cpluscplus
}
#endif
#endif

117
src/packet_io/packet_io.cpp
View File

@@ -5,22 +5,26 @@
#include "packet_io_marsio.h"
#include "packet_io_dumpfile.h"
typedef void *new_cb(void *options);
typedef void free_cb(void *handle);
typedef void *stat_cb(void *handle);
typedef int init_cb(void *handle, uint16_t thread_id);
typedef int recv_cb(void *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
typedef void send_cb(void *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
typedef void *on_new(void *options);
typedef void on_free(void *handle);
typedef void *on_stat(void *handle);
typedef int on_init(void *handle, uint16_t thread_id);
typedef int on_recv(void *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
typedef void on_send(void *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
typedef void on_drop(void *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
typedef void on_inject(void *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
struct packet_io
{
void *handle;
new_cb *on_new;
free_cb *on_free;
stat_cb *on_stat;
init_cb *on_init;
recv_cb *on_recv;
send_cb *on_send;
on_new *new_func;
on_free *free_func;
on_stat *stat_func;
on_init *init_func;
on_recv *recv_func;
on_send *send_func;
on_drop *drop_func;
on_inject *inject_func;
};
struct packet_io *packet_io_new(struct packet_io_options *opts)
@@ -32,40 +36,44 @@ struct packet_io *packet_io_new(struct packet_io_options *opts)
return NULL;
}
struct packet_io_marsio_opts marsio_opts;
strncpy(marsio_opts.app_symbol, opts->app_symbol, sizeof(marsio_opts.app_symbol));
strncpy(marsio_opts.dev_symbol, opts->dev_symbol, sizeof(marsio_opts.dev_symbol));
memcpy(marsio_opts.cpu_mask, opts->cpu_mask, sizeof(marsio_opts.cpu_mask));
marsio_opts.nr_threads = opts->nr_threads;
struct packet_io_dumpfile_opts dumpfile_opts;
strncpy(dumpfile_opts.dumpfile_dir, opts->dumpfile_dir, sizeof(dumpfile_opts.dumpfile_dir));
dumpfile_opts.nr_threads = opts->nr_threads;
void *_opts = NULL;
struct packet_io_marsio_options opts_marsio;
struct packet_io_dumpfile_options opts_dumpfile;
void *opts_ptr = NULL;
if (opts->mode == PACKET_IO_MARSIO)
{
_opts = &marsio_opts;
handle->on_new = (new_cb *)packet_io_marsio_new;
handle->on_free = (free_cb *)packet_io_marsio_free;
handle->on_stat = (stat_cb *)packet_io_marsio_stat;
handle->on_init = (init_cb *)packet_io_marsio_init;
handle->on_recv = (recv_cb *)packet_io_marsio_recv;
handle->on_send = (send_cb *)packet_io_marsio_send;
strncpy(opts_marsio.app_symbol, opts->app_symbol, sizeof(opts_marsio.app_symbol));
strncpy(opts_marsio.dev_symbol, opts->dev_symbol, sizeof(opts_marsio.dev_symbol));
memcpy(opts_marsio.cpu_mask, opts->cpu_mask, sizeof(opts_marsio.cpu_mask));
opts_marsio.nr_threads = opts->nr_threads;
opts_ptr = &opts_marsio;
handle->new_func = (on_new *)packet_io_marsio_new;
handle->free_func = (on_free *)packet_io_marsio_free;
handle->stat_func = (on_stat *)packet_io_marsio_get_stat;
handle->init_func = (on_init *)packet_io_marsio_init;
handle->recv_func = (on_recv *)packet_io_marsio_ingress;
handle->send_func = (on_send *)packet_io_marsio_egress;
handle->drop_func = (on_drop *)packet_io_marsio_drop;
handle->inject_func = (on_inject *)packet_io_marsio_inject;
}
else
{
_opts = &dumpfile_opts;
handle->on_new = (new_cb *)packet_io_dumpfile_new;
handle->on_free = (free_cb *)packet_io_dumpfile_free;
handle->on_stat = (stat_cb *)packet_io_dumpfile_stat;
handle->on_init = (init_cb *)packet_io_dumpfile_init;
handle->on_recv = (recv_cb *)packet_io_dumpfile_recv;
handle->on_send = (send_cb *)packet_io_dumpfile_send;
strncpy(opts_dumpfile.dumpfile_dir, opts->dumpfile_dir, sizeof(opts_dumpfile.dumpfile_dir));
opts_dumpfile.nr_threads = opts->nr_threads;
opts_ptr = &opts_dumpfile;
handle->new_func = (on_new *)packet_io_dumpfile_new;
handle->free_func = (on_free *)packet_io_dumpfile_free;
handle->stat_func = (on_stat *)packet_io_dumpfile_get_stat;
handle->init_func = (on_init *)packet_io_dumpfile_init;
handle->recv_func = (on_recv *)packet_io_dumpfile_ingress;
handle->send_func = (on_send *)packet_io_dumpfile_egress;
handle->drop_func = (on_drop *)packet_io_dumpfile_drop;
handle->inject_func = (on_inject *)packet_io_dumpfile_inject;
}
handle->handle = handle->on_new(_opts);
handle->handle = handle->new_func(opts_ptr);
if (handle->handle == NULL)
{
goto error_out;
@@ -82,39 +90,40 @@ void packet_io_free(struct packet_io *handle)
{
if (handle)
{
handle->on_free(handle->handle);
handle->free_func(handle->handle);
free(handle);
handle = NULL;
}
}
void packet_io_print_stat(struct packet_io *handle)
{
struct packet_io_stat *stat = packet_io_get_stat(handle);
PACKET_IO_LOG_DEBUG("rx_pkts : %lu, rx_bytes : %lu", stat->rx_pkts, stat->rx_bytes);
PACKET_IO_LOG_DEBUG("tx_pkts : %lu, tx_bytes : %lu", stat->tx_pkts, stat->tx_bytes);
PACKET_IO_LOG_DEBUG("drop_pkts : %lu, drop_bytes : %lu", stat->drop_pkts, stat->drop_bytes);
PACKET_IO_LOG_DEBUG("inject_pkts : %lu, inject_bytes : %lu", stat->inject_pkts, stat->inject_bytes);
PACKET_IO_LOG_DEBUG("keepalive_pkts : %lu, keepalive_bytes : %lu", stat->keepalive_pkts, stat->keepalive_bytes);
}
struct packet_io_stat *packet_io_get_stat(struct packet_io *handle)
{
return (struct packet_io_stat *)handle->on_stat(handle->handle);
return (struct packet_io_stat *)handle->stat_func(handle->handle);
}
// return 0: success
// return -1: failed
int packet_io_init(struct packet_io *handle, uint16_t thread_id)
{
return handle->on_init(handle->handle, thread_id);
return handle->init_func(handle->handle, thread_id);
}
// return number of packets received
int packet_io_ingress(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
return handle->on_recv(handle->handle, thread_id, pkts, nr_pkts);
return handle->recv_func(handle->handle, thread_id, pkts, nr_pkts);
}
void packet_io_egress(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
handle->on_send(handle->handle, thread_id, pkts, nr_pkts);
handle->send_func(handle->handle, thread_id, pkts, nr_pkts);
}
void packet_io_drop(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
handle->drop_func(handle->handle, thread_id, pkts, nr_pkts);
}
void packet_io_inject(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
handle->inject_func(handle->handle, thread_id, pkts, nr_pkts);
}

9
src/packet_io/packet_io.h
View File

@@ -55,13 +55,18 @@ struct packet_io_options
struct packet_io;
struct packet_io *packet_io_new(struct packet_io_options *opts);
void packet_io_free(struct packet_io *handle);
void packet_io_print_stat(struct packet_io *handle);
struct packet_io_stat *packet_io_get_stat(struct packet_io *handle);
// return 0 if success, -1 if failed
// TODO performance optimization egress for multi-packet
// return 0: success
// return -1: failed
int packet_io_init(struct packet_io *handle, uint16_t thread_id);
// return number of packets received
int packet_io_ingress(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_egress(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_drop(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_inject(struct packet_io *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
#ifdef __cpluscplus
}

121
src/packet_io/packet_io_dumpfile.cpp
View File

@@ -1,12 +1,14 @@
#include <pcap/pcap.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include "stellar.h"
#include "file_scan.h"
#include "packet_io.h"
#include "packet_utils.h"
#include "packet_queue.h"
#include "lock_free_queue.h"
#include "packet_io_dumpfile.h"
#define MAX_PACKET_QUEUE_SIZE (4096 * 1000)
@@ -17,12 +19,18 @@ struct packet_io_dumpfile
char dumpfile_dir[256];
pcap_t *pcap;
struct packet_queue *queue[MAX_THREAD_NUM];
struct lock_free_queue *queue[MAX_THREAD_NUM];
struct packet_io_stat stat;
uint64_t io_thread_need_exit;
uint64_t io_thread_is_runing;
};
struct pcap_pkt
{
char *data;
int len;
};
/******************************************************************************
* Private API
******************************************************************************/
@@ -31,19 +39,25 @@ static void pcap_handle(u_char *user, const struct pcap_pkthdr *h, const u_char
{
struct packet_io_dumpfile *handle = (struct packet_io_dumpfile *)user;
struct packet *pkt = packet_new(h->caplen);
if (pkt == NULL)
// 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)
{
PACKET_IO_LOG_ERROR("unable to alloc packet");
return;
}
pcap_pkt->data = (char *)pcap_pkt + sizeof(struct pcap_pkt);
pcap_pkt->len = h->caplen;
memcpy((char *)pcap_pkt->data, bytes, h->caplen);
memcpy((char *)pkt->data_ptr, bytes, h->caplen);
packet_parse(pkt, pkt->data_ptr, h->caplen);
// calculate packet hash
struct packet pkt;
packet_parse(&pkt, pcap_pkt->data, pcap_pkt->len);
uint64_t hash = packet_get_hash(&pkt, LDBC_METHOD_HASH_INT_IP_AND_EXT_IP, 0);
uint64_t hash = packet_get_hash(pkt, LDBC_METHOD_HASH_INT_IP_AND_EXT_IP, 0);
struct packet_queue *queue = handle->queue[hash % handle->nr_threads];
packet_queue_push(queue, pkt);
// push packet to queue
struct lock_free_queue *queue = handle->queue[hash % handle->nr_threads];
lock_free_queue_push(queue, pcap_pkt);
if (ATOMIC_READ(&handle->io_thread_need_exit))
{
@@ -56,7 +70,7 @@ static int dumpfile_handle(const char *file, void *arg)
{
struct packet_io_dumpfile *handle = (struct packet_io_dumpfile *)arg;
PACKET_IO_LOG_STATE("dumpfile %s inprocessing", file)
PACKET_IO_LOG_STATE("dumpfile %s in-processing", file)
handle->pcap = pcap_open_offline(file, NULL);
if (handle->pcap == NULL)
@@ -91,7 +105,7 @@ static void *dumpfile_thread_cycle(void *arg)
* Public API
******************************************************************************/
struct packet_io_dumpfile *packet_io_dumpfile_new(struct packet_io_dumpfile_opts *opts)
struct packet_io_dumpfile *packet_io_dumpfile_new(struct packet_io_dumpfile_options *opts)
{
pthread_t tid;
struct packet_io_dumpfile *handle = (struct packet_io_dumpfile *)calloc(1, sizeof(struct packet_io_dumpfile));
@@ -106,7 +120,7 @@ struct packet_io_dumpfile *packet_io_dumpfile_new(struct packet_io_dumpfile_opts
for (uint16_t i = 0; i < handle->nr_threads; i++)
{
handle->queue[i] = packet_queue_new(MAX_PACKET_QUEUE_SIZE);
handle->queue[i] = lock_free_queue_new(MAX_PACKET_QUEUE_SIZE);
if (handle->queue[i] == NULL)
{
PACKET_IO_LOG_ERROR("unable to create packet queue");
@@ -140,14 +154,14 @@ void packet_io_dumpfile_free(struct packet_io_dumpfile *handle)
for (uint16_t i = 0; i < handle->nr_threads; i++)
{
packet_queue_free(handle->queue[i]);
lock_free_queue_free(handle->queue[i]);
}
free(handle);
handle = NULL;
}
}
struct packet_io_stat *packet_io_dumpfile_stat(struct packet_io_dumpfile *handle)
struct packet_io_stat *packet_io_dumpfile_get_stat(struct packet_io_dumpfile *handle)
{
return &handle->stat;
}
@@ -157,55 +171,84 @@ int packet_io_dumpfile_init(struct packet_io_dumpfile *handle, uint16_t thread_i
return 0;
}
int packet_io_dumpfile_recv(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
int packet_io_dumpfile_ingress(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
struct packet_queue *queue = handle->queue[thread_id];
struct packet *pkt = NULL;
struct lock_free_queue *queue = handle->queue[thread_id];
struct pcap_pkt *pcap_pkt = NULL;
int nr_parsed = 0;
for (int i = 0; i < nr_pkts; i++)
{
packet_queue_pop(queue, &pkt);
if (pkt == NULL)
lock_free_queue_pop(queue, (void **)&pcap_pkt);
if (pcap_pkt == NULL)
{
break;
}
else
{
ATOMIC_ADD(&handle->stat.rx_pkts, 1);
ATOMIC_ADD(&handle->stat.rx_bytes, packet_get_len(pkt));
ATOMIC_ADD(&handle->stat.rx_bytes, pcap_pkt->len);
struct packet *temp = &pkts[nr_parsed++];
memset(temp, 0, sizeof(struct packet));
packet_parse(temp, pkt->data_ptr, pkt->data_len);
packet_set_io_ctx(temp, pkt);
packet_set_type(temp, PACKET_TYPE_DATA);
packet_set_action(temp, PACKET_ACTION_FORWARD);
struct packet *pkt = &pkts[nr_parsed++];
memset(pkt, 0, sizeof(struct packet));
packet_parse(pkt, pcap_pkt->data, pcap_pkt->len);
packet_set_user_data(pkt, pcap_pkt);
packet_set_type(pkt, PACKET_TYPE_DATA);
packet_set_action(pkt, PACKET_ACTION_FORWARD);
}
}
return nr_parsed;
}
void packet_io_dumpfile_send(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
// pkts from packet_io_dumpfile_ingress
void packet_io_dumpfile_egress(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
struct packet *pkt = NULL;
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
if (packet_get_action(pkt) == PACKET_ACTION_DROP)
{
ATOMIC_ADD(&handle->stat.drop_pkts, 1);
ATOMIC_ADD(&handle->stat.drop_bytes, packet_get_len(pkt));
}
else
{
ATOMIC_ADD(&handle->stat.tx_pkts, 1);
ATOMIC_ADD(&handle->stat.tx_bytes, packet_get_len(pkt));
}
ATOMIC_ADD(&handle->stat.tx_pkts, 1);
ATOMIC_ADD(&handle->stat.tx_bytes, packet_get_len(pkt));
struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_user_data(pkt);
assert(pcap_pkt != NULL);
free(pcap_pkt);
}
}
// pkts from packet_io_dumpfile_ingress
void packet_io_dumpfile_drop(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
struct packet *pkt = NULL;
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
ATOMIC_ADD(&handle->stat.drop_pkts, 1);
ATOMIC_ADD(&handle->stat.drop_bytes, packet_get_len(pkt));
struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_user_data(pkt);
assert(pcap_pkt != NULL);
free(pcap_pkt);
}
}
// pkts build by packet_new
void packet_io_dumpfile_inject(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
struct packet *pkt = NULL;
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
ATOMIC_ADD(&handle->stat.inject_pkts, 1);
ATOMIC_ADD(&handle->stat.inject_bytes, packet_get_len(pkt));
struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_user_data(pkt);
assert(pcap_pkt == NULL);
packet_free((struct packet *)packet_get_io_ctx(pkt));
packet_free(pkt);
}
}

15
src/packet_io/packet_io_dumpfile.h
View File

@@ -8,7 +8,7 @@ extern "C"
#include "packet.h"
struct packet_io_dumpfile_opts
struct packet_io_dumpfile_options
{
char dumpfile_dir[256];
uint8_t nr_threads;
@@ -16,13 +16,18 @@ struct packet_io_dumpfile_opts
struct packet_io_dumpfile;
struct packet_io_dumpfile *packet_io_dumpfile_new(struct packet_io_dumpfile_opts *opts);
struct packet_io_dumpfile *packet_io_dumpfile_new(struct packet_io_dumpfile_options *opts);
void packet_io_dumpfile_free(struct packet_io_dumpfile *handle);
struct packet_io_stat *packet_io_dumpfile_stat(struct packet_io_dumpfile *handle);
struct packet_io_stat *packet_io_dumpfile_get_stat(struct packet_io_dumpfile *handle);
// return 0: success
// return -1: failed
int packet_io_dumpfile_init(struct packet_io_dumpfile *handle, uint16_t thread_id);
int packet_io_dumpfile_recv(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_dumpfile_send(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
// return number of packets received
int packet_io_dumpfile_ingress(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_dumpfile_egress(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_dumpfile_drop(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_dumpfile_inject(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
#ifdef __cpluscplus
}

123
src/packet_io/packet_io_marsio.cpp
View File

@@ -1,10 +1,13 @@
#include <sched.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <netinet/ether.h>
#include "stellar.h"
#include "marsio.h"
#include "packet.h"
#include "packet_io.h"
#include "packet_utils.h"
#include "packet_io_marsio.h"
struct packet_io_marsio
@@ -22,21 +25,25 @@ struct packet_io_marsio
static void metadata_to_packet(marsio_buff_t *mbuff, struct packet *pkt)
{
packet_set_io_ctx(pkt, mbuff);
pkt->sid_used = marsio_buff_get_sid_list(mbuff, pkt->sid_list, MAX_SID_NUM);
pkt->route_len = marsio_buff_get_metadata(mbuff, MR_BUFF_ROUTE_CTX, pkt->route_ctx, MAX_ROUTE_LEN);
marsio_buff_get_metadata(mbuff, MR_BUFF_SESSION_ID, &(pkt->session_id), sizeof(pkt->session_id));
marsio_buff_get_metadata(mbuff, MR_BUFF_DIR, &(pkt->direction), sizeof(pkt->direction));
struct metadata *meta = &pkt->meta;
packet_set_user_data(pkt, mbuff);
meta->sid.used = marsio_buff_get_sid_list(mbuff, meta->sid.list, MAX_SID_NUM);
meta->route.len = marsio_buff_get_metadata(mbuff, MR_BUFF_ROUTE_CTX, meta->route.data, MAX_ROUTE_LEN);
marsio_buff_get_metadata(mbuff, MR_BUFF_SESSION_ID, &(meta->session_id), sizeof(meta->session_id));
marsio_buff_get_metadata(mbuff, MR_BUFF_DIR, &(meta->direction), sizeof(meta->direction));
packet_set_type(pkt, marsio_buff_is_ctrlbuf(mbuff) ? PACKET_TYPE_CTRL : PACKET_TYPE_DATA);
packet_set_action(pkt, PACKET_ACTION_FORWARD);
}
static void metadata_to_mbuff(marsio_buff_t *mbuff, struct packet *pkt)
{
marsio_buff_set_sid_list(mbuff, pkt->sid_list, pkt->sid_used);
marsio_buff_set_metadata(mbuff, MR_BUFF_ROUTE_CTX, pkt->route_ctx, pkt->route_len);
marsio_buff_set_metadata(mbuff, MR_BUFF_SESSION_ID, &(pkt->session_id), sizeof(pkt->session_id));
marsio_buff_set_metadata(mbuff, MR_BUFF_DIR, &(pkt->direction), sizeof(pkt->direction));
struct metadata *meta = &pkt->meta;
marsio_buff_set_sid_list(mbuff, meta->sid.list, meta->sid.used);
marsio_buff_set_metadata(mbuff, MR_BUFF_ROUTE_CTX, meta->route.data, meta->route.len);
marsio_buff_set_metadata(mbuff, MR_BUFF_SESSION_ID, &(meta->session_id), sizeof(meta->session_id));
marsio_buff_set_metadata(mbuff, MR_BUFF_DIR, &(meta->direction), sizeof(meta->direction));
if (packet_get_type(pkt) == PACKET_TYPE_CTRL)
{
marsio_buff_set_ctrlbuf(mbuff);
@@ -65,7 +72,7 @@ static int is_keepalive_packet(const char *data, int len)
* Public API
******************************************************************************/
struct packet_io_marsio *packet_io_marsio_new(struct packet_io_marsio_opts *opts)
struct packet_io_marsio *packet_io_marsio_new(struct packet_io_marsio_options *opts)
{
int opt = 1;
cpu_set_t coremask;
@@ -146,7 +153,7 @@ void packet_io_marsio_free(struct packet_io_marsio *handle)
}
}
struct packet_io_stat *packet_io_marsio_stat(struct packet_io_marsio *handle)
struct packet_io_stat *packet_io_marsio_get_stat(struct packet_io_marsio *handle)
{
return &handle->stat;
}
@@ -162,7 +169,7 @@ int packet_io_marsio_init(struct packet_io_marsio *handle, uint16_t thread_id)
return 0;
}
int packet_io_marsio_recv(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
int packet_io_marsio_ingress(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
marsio_buff_t *rx_buff;
marsio_buff_t *rx_buffs[RX_BURST_MAX];
@@ -197,13 +204,38 @@ int packet_io_marsio_recv(struct packet_io_marsio *handle, uint16_t thread_id, s
continue;
}
packet_parse(&pkts[nr_parsed], raw_data, raw_len);
metadata_to_packet(rx_buff, &pkts[nr_parsed++]);
}
return nr_parsed;
}
void packet_io_marsio_send(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
// pkts from packet_io_marsio_ingress
void packet_io_marsio_egress(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
struct packet *pkt;
struct metadata *meta;
marsio_buff_t *tx_buff;
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
ATOMIC_ADD(&handle->stat.tx_pkts, 1);
ATOMIC_ADD(&handle->stat.tx_bytes, packet_get_len(pkt));
// update sid list
tx_buff = (marsio_buff_t *)packet_get_user_data(pkt);
assert(tx_buff != NULL);
meta = &pkt->meta;
marsio_buff_set_sid_list(tx_buff, meta->sid.list, meta->sid.used);
marsio_send_burst(handle->mr_path, thread_id, &tx_buff, 1);
}
}
// pkts from packet_io_marsio_ingress
void packet_io_marsio_drop(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
struct packet *pkt;
marsio_buff_t *tx_buff;
@@ -211,43 +243,42 @@ void packet_io_marsio_send(struct packet_io_marsio *handle, uint16_t thread_id,
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
tx_buff = (marsio_buff_t *)packet_get_io_ctx(pkt);
if (packet_get_action(pkt) == PACKET_ACTION_DROP)
tx_buff = (marsio_buff_t *)packet_get_user_data(pkt);
assert(tx_buff != NULL);
ATOMIC_ADD(&handle->stat.drop_pkts, 1);
ATOMIC_ADD(&handle->stat.drop_bytes, packet_get_len(pkt));
marsio_buff_free(handle->mr_ins, &tx_buff, 1, 0, thread_id);
}
}
// pkts from packet_new
void packet_io_marsio_inject(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts)
{
struct packet *pkt;
marsio_buff_t *tx_buff;
for (int i = 0; i < nr_pkts; i++)
{
pkt = &pkts[i];
tx_buff = (marsio_buff_t *)packet_get_user_data(pkt);
assert(tx_buff == NULL);
if (marsio_buff_malloc_global(handle->mr_ins, &tx_buff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY) < 0)
{
if (tx_buff)
{
ATOMIC_ADD(&handle->stat.drop_pkts, 1);
ATOMIC_ADD(&handle->stat.drop_bytes, packet_get_len(pkt));
marsio_buff_free(handle->mr_ins, &tx_buff, 1, 0, thread_id);
}
else
{
// do nothing
}
PACKET_IO_LOG_ERROR("unable to alloc tx buffer");
continue;
}
else
{
if (tx_buff == NULL)
{
if (marsio_buff_malloc_global(handle->mr_ins, &tx_buff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY) < 0)
{
PACKET_IO_LOG_ERROR("unable to alloc tx buffer");
goto fast_end;
}
ATOMIC_ADD(&handle->stat.inject_pkts, 1);
ATOMIC_ADD(&handle->stat.inject_bytes, packet_get_len(pkt));
ATOMIC_ADD(&handle->stat.inject_pkts, 1);
ATOMIC_ADD(&handle->stat.inject_bytes, packet_get_len(pkt));
char *dst = marsio_buff_append(tx_buff, packet_get_len(pkt));
memcpy(dst, packet_get_data(pkt), packet_get_len(pkt));
}
char *dst = marsio_buff_append(tx_buff, packet_get_len(pkt));
memcpy(dst, packet_get_data(pkt), packet_get_len(pkt));
ATOMIC_ADD(&handle->stat.tx_pkts, 1);
ATOMIC_ADD(&handle->stat.tx_bytes, packet_get_len(pkt));
metadata_to_mbuff(tx_buff, pkt);
marsio_send_burst(handle->mr_path, thread_id, &tx_buff, 1);
}
ATOMIC_ADD(&handle->stat.tx_pkts, 1);
ATOMIC_ADD(&handle->stat.tx_bytes, packet_get_len(pkt));
metadata_to_mbuff(tx_buff, pkt);
marsio_send_burst(handle->mr_path, thread_id, &tx_buff, 1);
fast_end:
packet_free(pkt);
}
}

15
src/packet_io/packet_io_marsio.h
View File

@@ -8,7 +8,7 @@ extern "C"
#include "packet.h"
struct packet_io_marsio_opts
struct packet_io_marsio_options
{
char app_symbol[64];
char dev_symbol[64];
@@ -18,13 +18,18 @@ struct packet_io_marsio_opts
struct packet_io_marsio;
struct packet_io_marsio *packet_io_marsio_new(struct packet_io_marsio_opts *opts);
struct packet_io_marsio *packet_io_marsio_new(struct packet_io_marsio_options *opts);
void packet_io_marsio_free(struct packet_io_marsio *handle);
struct packet_io_stat *packet_io_marsio_stat(struct packet_io_marsio *handle);
struct packet_io_stat *packet_io_marsio_get_stat(struct packet_io_marsio *handle);
// return 0: success
// return -1: failed
int packet_io_marsio_init(struct packet_io_marsio *handle, uint16_t thread_id);
int packet_io_marsio_recv(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_marsio_send(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
// return number of packets received
int packet_io_marsio_ingress(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_marsio_egress(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_marsio_drop(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
void packet_io_marsio_inject(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkts, int nr_pkts);
#ifdef __cpluscplus
}

26
src/packet_io/packet_queue.h
View File

@@ -1,26 +0,0 @@
#ifndef _PACKET_QUEUE_H
#define _PACKET_QUEUE_H
#ifdef __cpluscplus
extern "C"
{
#endif
#include <stdint.h>
struct packet_queue;
struct packet_queue *packet_queue_new(uint32_t size);
void packet_queue_free(struct packet_queue *queue);
int packet_queue_is_full(struct packet_queue *queue);
int packet_queue_is_empty(struct packet_queue *queue);
void packet_queue_push(struct packet_queue *queue, struct packet *pkt);
void packet_queue_pop(struct packet_queue *queue, struct packet **pkt);
#ifdef __cpluscplus
}
#endif
#endif