🦄 refactor(directory structure): restructure and rename src to infra

infra/packet_io/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_library(packet_io dumpfile_io.cpp marsio_io.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 marsio pcap packet)

infra/packet_io/dumpfile_io.cpp

@@ -0,0 +1,514 @@
+#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 "tuple.h"
+#include "utils.h"
+#include "log_private.h"
+#include "dumpfile_io.h"
+#include "packet_dump.h"
+#include "packet_parser.h"
+#include "packet_private.h"
+
+#define PACKET_IO_LOG_FATAL(format, ...) STELLAR_LOG_FATAL(__thread_local_logger, "dumpfile", format, ##__VA_ARGS__)
+#define PACKET_IO_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "dumpfile", format, ##__VA_ARGS__)
+
+#define MAX_PACKET_QUEUE_SIZE (4096 * 1000)
+
+struct dumpfile_io
+{
+    enum packet_io_mode mode;
+    uint16_t nr_threads;
+    char dumpfile_path[256];
+
+    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;
+};
+
+struct pcap_pkt
+{
+    char *data;
+    int len;
+    struct timeval ts;
+};
+
+/******************************************************************************
+ * Private API -- queue
+ ******************************************************************************/
+
+struct packet_queue
+{
+    uint64_t *queue;
+    uint32_t size;
+    uint32_t head;
+    uint32_t tail;
+};
+
+static struct packet_queue *packet_queue_new(uint32_t size)
+{
+    struct packet_queue *queue = (struct packet_queue *)calloc(1, sizeof(struct packet_queue));
+    if (queue == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to new packet queue");
+        return NULL;
+    }
+
+    queue->queue = (uint64_t *)calloc(size, sizeof(uint64_t));
+    if (queue->queue == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to new packet queue");
+        free(queue);
+        return NULL;
+    }
+
+    queue->size = size;
+    queue->head = 0;
+    queue->tail = 0;
+
+    return queue;
+}
+
+static void packet_queue_free(struct packet_queue *queue)
+{
+    if (queue == NULL)
+    {
+        return;
+    }
+
+    if (queue->queue)
+    {
+        free(queue->queue);
+        queue->queue = NULL;
+    }
+
+    free(queue);
+}
+
+static int packet_queue_push(struct packet_queue *queue, void *data)
+{
+    if (__sync_val_compare_and_swap(&queue->queue[queue->tail], 0, data) != 0)
+    {
+        PACKET_IO_LOG_ERROR("packet queue is full, retry later");
+        return -1;
+    }
+
+    queue->tail = (queue->tail + 1) % queue->size;
+    return 0;
+}
+
+static void packet_queue_pop(struct packet_queue *queue, void **data)
+{
+    uint64_t read = ATOMIC_READ(&queue->queue[queue->head]);
+    if (read == 0)
+    {
+        *data = NULL;
+        return;
+    }
+    __sync_val_compare_and_swap(&queue->queue[queue->head], read, 0);
+    *data = (void *)read;
+    queue->head = (queue->head + 1) % queue->size;
+}
+
+/******************************************************************************
+ * Private API -- utils
+ ******************************************************************************/
+
+static void pcap_pkt_handler(u_char *user, const struct pcap_pkthdr *h, const u_char *bytes)
+{
+    struct dumpfile_io *handle = (struct dumpfile_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)
+    {
+        PACKET_IO_LOG_ERROR("unable to alloc packet");
+        return;
+    }
+    pcap_pkt->data = (char *)pcap_pkt + sizeof(struct pcap_pkt);
+    pcap_pkt->len = h->caplen;
+    pcap_pkt->ts = h->ts;
+    memcpy((char *)pcap_pkt->data, bytes, h->caplen);
+    ATOMIC_INC(&handle->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->nr_threads];
+    while (packet_queue_push(queue, pcap_pkt) == -1)
+    {
+        if (ATOMIC_READ(&handle->io_thread_need_exit))
+        {
+            free(pcap_pkt);
+            PACKET_IO_LOG_FATAL("dumpfile io thread need exit");
+            pcap_breakloop(handle->pcap);
+            break;
+        }
+        usleep(1000);
+    }
+
+    if (ATOMIC_READ(&handle->io_thread_need_exit))
+    {
+        PACKET_IO_LOG_FATAL("dumpfile io thread need exit");
+        pcap_breakloop(handle->pcap);
+    }
+}
+
+static int dumpfile_handler(struct dumpfile_io *handle, const char *pcap_file)
+{
+    char resolved_path[256];
+    char pcap_errbuf[PCAP_ERRBUF_SIZE];
+
+    realpath(pcap_file, resolved_path);
+    PACKET_IO_LOG_FATAL("dumpfile %s in-processing", resolved_path)
+
+    handle->pcap = pcap_open_offline(resolved_path, pcap_errbuf);
+    if (handle->pcap == NULL)
+    {
+        PACKET_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);
+
+    PACKET_IO_LOG_FATAL("dumpfile %s processed", resolved_path)
+
+    return 0;
+}
+
+static int all_packet_consumed(struct dumpfile_io *handle)
+{
+    uint64_t consumed_pkts = 0;
+    uint64_t read_pcap_pkts = ATOMIC_READ(&handle->read_pcap_pkts);
+    for (uint16_t i = 0; i < handle->nr_threads; i++)
+    {
+        consumed_pkts += ATOMIC_READ(&handle->stat[i].pkts_rx);
+    }
+    if (consumed_pkts == read_pcap_pkts)
+    {
+        return 1;
+    }
+    else
+    {
+        return 0;
+    }
+}
+
+static void *dumpfile_thread(void *arg)
+{
+    struct dumpfile_io *handle = (struct dumpfile_io *)arg;
+    __thread_local_logger = handle->logger;
+
+    ATOMIC_SET(&handle->io_thread_is_runing, 1);
+    PACKET_IO_LOG_FATAL("dumpfile io thread is running");
+
+    if (handle->mode == PACKET_IO_DUMPFILE)
+    {
+        dumpfile_handler(handle, handle->dumpfile_path);
+    }
+    else // PACKET_IO_DUMPFILELIST
+    {
+        FILE *fp = NULL;
+        if (strcmp(handle->dumpfile_path, "-") == 0)
+        {
+            PACKET_IO_LOG_ERROR("dumpfile list is empty, read from stdin");
+            fp = stdin;
+        }
+        else
+        {
+            fp = fopen(handle->dumpfile_path, "r");
+            if (fp == NULL)
+            {
+                PACKET_IO_LOG_ERROR("unable to open dumpfile list: %s", handle->dumpfile_path);
+                goto erro_out;
+            }
+        }
+
+        char line[PATH_MAX];
+        while (ATOMIC_READ(&handle->io_thread_need_exit) == 0 && fgets(line, sizeof(line), fp))
+        {
+            if (line[0] == '#')
+            {
+                continue;
+            }
+
+            char *pos = strchr(line, '\n');
+            if (pos)
+            {
+                *pos = '\0';
+            }
+
+            dumpfile_handler(handle, line);
+        }
+        if (fp != stdin)
+        {
+            fclose(fp);
+        }
+    }
+    PACKET_IO_LOG_FATAL("dumpfile io thread read all pcap files");
+
+erro_out:
+    while (ATOMIC_READ(&handle->io_thread_need_exit) == 0)
+    {
+        if (all_packet_consumed(handle))
+        {
+            ATOMIC_SET(&handle->io_thread_wait_exit, 1);
+        }
+
+        usleep(1000); // 1ms
+    }
+
+    PACKET_IO_LOG_FATAL("dumpfile 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);
+
+    return NULL;
+}
+
+/******************************************************************************
+ * Public API
+ ******************************************************************************/
+
+struct dumpfile_io *dumpfile_io_new(const char *dumpfile_path, enum packet_io_mode mode, uint16_t nr_threads)
+{
+    pthread_t tid;
+    struct dumpfile_io *handle = (struct dumpfile_io *)calloc(1, sizeof(struct dumpfile_io));
+    if (handle == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to allocate memory for dumpfile_io");
+        return NULL;
+    }
+
+    handle->mode = mode;
+    handle->nr_threads = nr_threads;
+    handle->logger = __thread_local_logger;
+    strncpy(handle->dumpfile_path, dumpfile_path, MIN(strlen(dumpfile_path), sizeof(handle->dumpfile_path)));
+
+    for (uint16_t i = 0; i < handle->nr_threads; i++)
+    {
+        handle->queue[i] = packet_queue_new(MAX_PACKET_QUEUE_SIZE);
+        if (handle->queue[i] == NULL)
+        {
+            PACKET_IO_LOG_ERROR("unable to create packet queue");
+            goto error_out;
+        }
+    }
+    if (pthread_create(&tid, NULL, dumpfile_thread, (void *)handle) != 0)
+    {
+        PACKET_IO_LOG_ERROR("unable to create packet io thread");
+        goto error_out;
+    }
+
+    return handle;
+
+error_out:
+    dumpfile_io_free(handle);
+    return NULL;
+}
+
+void dumpfile_io_free(struct dumpfile_io *handle)
+{
+    if (handle)
+    {
+        ATOMIC_SET(&handle->io_thread_need_exit, 1);
+
+        while (ATOMIC_READ(&handle->io_thread_is_runing))
+        {
+            usleep(1000);
+        }
+
+        struct pcap_pkt *pcap_pkt = NULL;
+        for (uint16_t i = 0; i < handle->nr_threads; i++)
+        {
+            while (1)
+            {
+                packet_queue_pop(handle->queue[i], (void **)&pcap_pkt);
+                if (pcap_pkt)
+                {
+                    free(pcap_pkt);
+                }
+                else
+                {
+                    break;
+                }
+            }
+
+            packet_queue_free(handle->queue[i]);
+        }
+        free(handle);
+        handle = NULL;
+    }
+}
+
+int dumpfile_io_isbreak(struct dumpfile_io *handle)
+{
+    return ATOMIC_READ(&handle->io_thread_wait_exit);
+}
+
+int dumpfile_io_init(struct dumpfile_io *handle __attribute__((unused)), uint16_t thr_idx __attribute__((unused)))
+{
+    return 0;
+}
+
+uint16_t dumpfile_io_input(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    struct packet_queue *queue = handle->queue[thr_idx];
+    struct packet_io_stat *stat = &handle->stat[thr_idx];
+    struct pcap_pkt *pcap_pkt = NULL;
+    struct packet *pkt;
+    uint16_t nr_parsed = 0;
+
+    for (uint16_t i = 0; i < nr_pkts; i++)
+    {
+        packet_queue_pop(queue, (void **)&pcap_pkt);
+        if (pcap_pkt == NULL)
+        {
+            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_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_parsed++;
+        }
+    }
+
+    return nr_parsed;
+}
+
+void dumpfile_io_output(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    int len;
+    struct packet *pkt = NULL;
+    struct packet_io_stat *stat = &handle->stat[thr_idx];
+
+    for (uint16_t i = 0; i < nr_pkts; i++)
+    {
+        pkt = &pkts[i];
+        len = packet_get_raw_len(pkt);
+
+        stat->pkts_tx++;
+        stat->bytes_tx += len;
+
+        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);
+        }
+        packet_free(pkt);
+    }
+}
+
+void dumpfile_io_drop(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    struct packet *pkt = NULL;
+    struct packet_io_stat *stat = &handle->stat[thr_idx];
+
+    for (uint16_t 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)
+        {
+            stat->pkts_dropped++;
+            stat->bytes_dropped += packet_get_raw_len(pkt);
+            free(pcap_pkt);
+        }
+        packet_free(pkt);
+    }
+}
+
+uint16_t dumpfile_io_inject(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    uint16_t len;
+    struct packet *pkt = NULL;
+    struct packet_io_stat *stat = &handle->stat[thr_idx];
+    struct tuple6 tuple;
+
+    char file[1024] = {0};
+    char src_addr[INET6_ADDRSTRLEN] = {0};
+    char dst_addr[INET6_ADDRSTRLEN] = {0};
+
+    for (uint16_t i = 0; i < nr_pkts; i++)
+    {
+        pkt = &pkts[i];
+        len = packet_get_raw_len(pkt);
+
+        stat->pkts_injected++;
+        stat->bytes_injected += len;
+
+        stat->raw_pkts_tx++;
+        stat->raw_bytes_tx += len;
+
+        stat->pkts_tx++;
+        stat->bytes_tx += 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)
+        {
+            PACKET_IO_LOG_ERROR("unable to dump pcap file: %s", file);
+        }
+        else
+        {
+            PACKET_IO_LOG_FATAL("dump inject packet: %s", file);
+        }
+        packet_free(pkt);
+    }
+
+    return nr_pkts;
+}
+
+void dumpfile_io_yield(struct dumpfile_io *handle __attribute__((unused)), uint16_t thr_idx __attribute__((unused)), uint64_t timeout_ms __attribute__((unused)))
+{
+    return;
+}
+
+struct packet_io_stat *dumpfile_io_stat(struct dumpfile_io *handle, uint16_t thr_idx)
+{
+    return &handle->stat[thr_idx];
+}

infra/packet_io/dumpfile_io.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "packet_io.h"
+
+struct dumpfile_io;
+struct dumpfile_io *dumpfile_io_new(const char *dumpfile_path, enum packet_io_mode mode, uint16_t nr_threads);
+void dumpfile_io_free(struct dumpfile_io *handle);
+int dumpfile_io_isbreak(struct dumpfile_io *handle);
+
+int dumpfile_io_init(struct dumpfile_io *handle, uint16_t thr_idx);
+uint16_t dumpfile_io_input(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void dumpfile_io_output(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void dumpfile_io_drop(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+uint16_t dumpfile_io_inject(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void dumpfile_io_yield(struct dumpfile_io *handle, uint16_t thr_idx, uint64_t timeout_ms);
+struct packet_io_stat *dumpfile_io_stat(struct dumpfile_io *handle, uint16_t thr_idx);
+
+#ifdef __cplusplus
+}
+#endif

infra/packet_io/marsio_io.cpp

@@ -0,0 +1,428 @@
+#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_private.h"
+#include "packet_parser.h"
+#include "packet_private.h"
+
+#define PACKET_IO_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "marsio", format, ##__VA_ARGS__)
+
+struct marsio_io
+{
+    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
+    {
+        PACKET_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
+    {
+        PACKET_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
+    {
+        PACKET_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
+    {
+        PACKET_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
+    {
+        PACKET_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
+    {
+        PACKET_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)
+    {
+        PACKET_IO_LOG_ERROR("failed to set route ctx");
+    }
+
+    if (marsio_buff_set_sid_list(mbuff, (sid_t *)sids->sid, sids->used) != 0)
+    {
+        PACKET_IO_LOG_ERROR("failed to set sids");
+    }
+
+    if (marsio_buff_set_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) != 0)
+    {
+        PACKET_IO_LOG_ERROR("failed to set session id");
+    }
+
+// TODO
+#if 0
+    if (marsio_buff_set_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) != 0)
+    {
+        PACKET_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)
+    {
+        PACKET_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)
+    {
+        PACKET_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
+ ******************************************************************************/
+
+struct marsio_io *marsio_io_new(const char *app_symbol, const char *dev_symbol, uint16_t *cpu_mask, uint16_t nr_threads)
+{
+    int opt = 1;
+    cpu_set_t coremask;
+    CPU_ZERO(&coremask);
+    for (uint16_t i = 0; i < nr_threads; i++)
+    {
+        CPU_SET(cpu_mask[i], &coremask);
+    }
+
+    struct marsio_io *handle = (struct marsio_io *)calloc(1, sizeof(struct marsio_io));
+    if (handle == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to allocate memory for marsio_io");
+        return NULL;
+    }
+
+    handle->mr_ins = marsio_create();
+    if (handle->mr_ins == NULL)
+    {
+        PACKET_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, app_symbol) != 0)
+    {
+        PACKET_IO_LOG_ERROR("unable to init marsio instance");
+        goto error_out;
+    }
+    handle->mr_dev = marsio_open_device(handle->mr_ins, dev_symbol, nr_threads, nr_threads);
+    if (handle->mr_dev == NULL)
+    {
+        PACKET_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)
+    {
+        PACKET_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(struct marsio_io *handle)
+{
+    if (handle)
+    {
+        if (handle->mr_path)
+        {
+            marsio_sendpath_destory(handle->mr_path);
+            handle->mr_path = NULL;
+        }
+        if (handle->mr_dev)
+        {
+            marsio_close_device(handle->mr_dev);
+            handle->mr_dev = NULL;
+        }
+        if (handle->mr_ins)
+        {
+            marsio_destory(handle->mr_ins);
+            handle->mr_ins = NULL;
+        }
+        free(handle);
+        handle = NULL;
+    }
+}
+
+int marsio_io_init(struct marsio_io *handle, uint16_t thr_idx __attribute__((unused)))
+{
+    if (marsio_thread_init(handle->mr_ins) != 0)
+    {
+        PACKET_IO_LOG_ERROR("unable to init marsio thread");
+        return -1;
+    }
+
+    return 0;
+}
+
+uint16_t marsio_io_input(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    struct packet *pkt;
+    marsio_buff_t *mbuff;
+    marsio_buff_t *rx_buffs[RX_BURST_MAX];
+    struct packet_io_stat *stat = &handle->stat[thr_idx];
+    uint16_t nr_parsed = 0;
+    int len;
+    char *data;
+
+    int nr_recv = marsio_recv_burst(handle->mr_dev, thr_idx, rx_buffs, MIN(RX_BURST_MAX, nr_pkts));
+    if (nr_recv <= 0)
+    {
+        return nr_parsed;
+    }
+
+    for (int i = 0; i < nr_recv; 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(handle->mr_path, thr_idx, &mbuff, 1);
+            continue;
+        }
+
+        pkt = &pkts[nr_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_parsed;
+}
+
+void marsio_io_output(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    struct packet *pkt;
+    marsio_buff_t *mbuff;
+    struct packet_io_stat *stat = &handle->stat[thr_idx];
+    int len;
+
+    for (uint16_t i = 0; i < nr_pkts; i++)
+    {
+        pkt = &pkts[i];
+        len = packet_get_raw_len(pkt);
+
+        stat->pkts_tx++;
+        stat->bytes_tx += len;
+
+        mbuff = (marsio_buff_t *)packet_get_origin_ctx(pkt);
+        assert(mbuff != NULL);
+        metadata_from_packet_to_mbuff(pkt, mbuff);
+
+        if (marsio_buff_is_ctrlbuf(mbuff))
+        {
+            stat->ctrl_pkts_tx++;
+            stat->ctrl_bytes_tx += len;
+        }
+        else
+        {
+            stat->raw_pkts_tx++;
+            stat->raw_bytes_tx += len;
+        }
+
+        marsio_send_burst(handle->mr_path, thr_idx, &mbuff, 1);
+        packet_free(pkt);
+    }
+}
+
+void marsio_io_drop(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    struct packet *pkt;
+    marsio_buff_t *mbuff;
+    struct packet_io_stat *stat = &handle->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(handle->mr_ins, &mbuff, 1, 0, thr_idx);
+        }
+        packet_free(pkt);
+    }
+}
+
+uint16_t marsio_io_inject(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    int len;
+    char *ptr;
+    uint16_t nr_inject = 0;
+    struct packet *pkt;
+    marsio_buff_t *mbuff;
+    struct packet_io_stat *stat = &handle->stat[thr_idx];
+
+    for (uint16_t i = 0; i < nr_pkts; i++)
+    {
+        pkt = &pkts[i];
+        len = packet_get_raw_len(pkt);
+
+        if (marsio_buff_malloc_global(handle->mr_ins, &mbuff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY) < 0)
+        {
+            PACKET_IO_LOG_ERROR("unable to allocate marsio buffer for inject packet");
+            continue;
+        }
+
+        stat->pkts_injected++;
+        stat->bytes_injected += len;
+
+        stat->raw_pkts_tx++;
+        stat->raw_bytes_tx += len;
+
+        stat->pkts_tx++;
+        stat->bytes_tx += len;
+
+        nr_inject++;
+
+        ptr = marsio_buff_append(mbuff, len);
+        memcpy(ptr, packet_get_raw_data(pkt), len);
+        metadata_from_packet_to_mbuff(pkt, mbuff);
+
+        marsio_send_burst_with_options(handle->mr_path, thr_idx, &mbuff, 1, MARSIO_SEND_OPT_REHASH);
+        packet_free(pkt);
+    }
+
+    return nr_inject;
+}
+
+void marsio_io_yield(struct marsio_io *handle, uint16_t thr_idx, uint64_t timeout_ms)
+{
+    struct mr_vdev *vdevs[1] = {
+        handle->mr_dev,
+    };
+
+    marsio_poll_wait(handle->mr_ins, vdevs, 1, thr_idx, timeout_ms);
+}
+
+struct packet_io_stat *marsio_io_stat(struct marsio_io *handle, uint16_t thr_idx)
+{
+    return &handle->stat[thr_idx];
+}

infra/packet_io/marsio_io.h

@@ -0,0 +1,24 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "packet_io.h"
+
+struct marsio_io;
+struct marsio_io *marsio_io_new(const char *app_symbol, const char *dev_symbol, uint16_t *cpu_mask, uint16_t nr_threads);
+void marsio_io_free(struct marsio_io *handle);
+
+int marsio_io_init(struct marsio_io *handle, uint16_t thr_idx);
+uint16_t marsio_io_input(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void marsio_io_output(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void marsio_io_drop(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+uint16_t marsio_io_inject(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void marsio_io_yield(struct marsio_io *handle, uint16_t thr_idx, uint64_t timeout_ms);
+struct packet_io_stat *marsio_io_stat(struct marsio_io *handle, uint16_t thr_idx);
+
+#ifdef __cplusplus
+}
+#endif

infra/packet_io/packet_io.cpp

@@ -0,0 +1,154 @@
+#include <stdlib.h>
+#include <string.h>
+
+#include "marsio_io.h"
+#include "dumpfile_io.h"
+
+struct packet_io
+{
+    enum packet_io_mode mode;
+    struct marsio_io *marsio;
+    struct dumpfile_io *dumpfile;
+};
+
+struct packet_io *packet_io_new(struct packet_io_options *opts)
+{
+    struct packet_io *packet_io = (struct packet_io *)calloc(1, sizeof(struct packet_io));
+    if (packet_io == NULL)
+    {
+        return NULL;
+    }
+
+    packet_io->mode = opts->mode;
+    if (opts->mode == PACKET_IO_MARSIO)
+    {
+        packet_io->marsio = marsio_io_new(opts->app_symbol, opts->dev_symbol, opts->cpu_mask, opts->nr_threads);
+    }
+    else
+    {
+        packet_io->dumpfile = dumpfile_io_new(opts->dumpfile_path, packet_io->mode, opts->nr_threads);
+    }
+    if (packet_io->marsio == NULL && packet_io->dumpfile == NULL)
+    {
+        goto error_out;
+    }
+
+    return packet_io;
+
+error_out:
+    packet_io_free(packet_io);
+    return NULL;
+}
+
+void packet_io_free(struct packet_io *packet_io)
+{
+    if (packet_io)
+    {
+        if (likely(packet_io->mode == PACKET_IO_MARSIO))
+        {
+            marsio_io_free(packet_io->marsio);
+        }
+        else
+        {
+            dumpfile_io_free(packet_io->dumpfile);
+        }
+        free(packet_io);
+        packet_io = NULL;
+    }
+}
+
+int packet_io_isbreak(struct packet_io *packet_io) // used for dumpfile mode
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        return 0;
+    }
+    else
+    {
+        return dumpfile_io_isbreak(packet_io->dumpfile);
+    }
+}
+
+int packet_io_init(struct packet_io *packet_io, uint16_t thr_idx)
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        return marsio_io_init(packet_io->marsio, thr_idx);
+    }
+    else
+    {
+        return dumpfile_io_init(packet_io->dumpfile, thr_idx);
+    }
+}
+
+uint16_t packet_io_input(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        return marsio_io_input(packet_io->marsio, thr_idx, pkts, nr_pkts);
+    }
+    else
+    {
+        return dumpfile_io_input(packet_io->dumpfile, thr_idx, pkts, nr_pkts);
+    }
+}
+
+void packet_io_output(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        marsio_io_output(packet_io->marsio, thr_idx, pkts, nr_pkts);
+    }
+    else
+    {
+        dumpfile_io_output(packet_io->dumpfile, thr_idx, pkts, nr_pkts);
+    }
+}
+
+void packet_io_drop(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        marsio_io_drop(packet_io->marsio, thr_idx, pkts, nr_pkts);
+    }
+    else
+    {
+        dumpfile_io_drop(packet_io->dumpfile, thr_idx, pkts, nr_pkts);
+    }
+}
+
+uint16_t packet_io_inject(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts)
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        return marsio_io_inject(packet_io->marsio, thr_idx, pkts, nr_pkts);
+    }
+    else
+    {
+        return dumpfile_io_inject(packet_io->dumpfile, thr_idx, pkts, nr_pkts);
+    }
+}
+
+void packet_io_yield(struct packet_io *packet_io, uint16_t thr_idx, uint64_t timeout_ms)
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        marsio_io_yield(packet_io->marsio, thr_idx, timeout_ms);
+    }
+    else
+    {
+        dumpfile_io_yield(packet_io->dumpfile, thr_idx, timeout_ms);
+    }
+}
+
+struct packet_io_stat *packet_io_stat(struct packet_io *packet_io, uint16_t thr_idx)
+{
+    if (likely(packet_io->mode == PACKET_IO_MARSIO))
+    {
+        return marsio_io_stat(packet_io->marsio, thr_idx);
+    }
+    else
+    {
+        return dumpfile_io_stat(packet_io->dumpfile, thr_idx);
+    }
+}

infra/packet_io/packet_io.h

@@ -0,0 +1,86 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include <stdint.h>
+#include <limits.h>
+
+#include "utils.h"
+
+struct __attribute__((aligned(64))) packet_io_stat
+{
+    // device packet
+    uint64_t pkts_rx;
+    uint64_t bytes_rx;
+
+    uint64_t pkts_tx;
+    uint64_t bytes_tx;
+
+    // keep-alive packet
+    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;
+
+    // inject packet
+    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;
+};
+
+enum packet_io_mode
+{
+    PACKET_IO_DUMPFILE = 0,
+    PACKET_IO_DUMPFILELIST = 1,
+    PACKET_IO_MARSIO = 2,
+};
+
+struct packet_io_options
+{
+    enum packet_io_mode mode;
+
+    // for dumpfile
+    char dumpfile_path[PATH_MAX];
+
+    // for marsio
+    char app_symbol[64];
+    char dev_symbol[64];
+
+    uint16_t nr_threads;
+    uint16_t cpu_mask[MAX_THREAD_NUM];
+};
+
+struct packet_io;
+struct packet_io *packet_io_new(struct packet_io_options *opts);
+void packet_io_free(struct packet_io *packet_io);
+int packet_io_isbreak(struct packet_io *packet_io); // used for dumpfile mode
+
+int packet_io_init(struct packet_io *packet_io, uint16_t thr_idx);
+uint16_t packet_io_input(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void packet_io_output(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void packet_io_drop(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+uint16_t packet_io_inject(struct packet_io *packet_io, uint16_t thr_idx, struct packet *pkts, uint16_t nr_pkts);
+void packet_io_yield(struct packet_io *packet_io, uint16_t thr_idx, uint64_t timeout_ms);
+struct packet_io_stat *packet_io_stat(struct packet_io *packet_io, uint16_t thr_idx);
+
+#ifdef __cplusplus
+}
+#endif