Add packet IO module

* support marsio
    * support dumpfile ( 1 thread read dumpfile & N thread handle packet)

src/packet_io/CMakeLists.txt

@@ -0,0 +1,8 @@
+###############################################################################
+# packet_io
+###############################################################################
+
+add_library(packet_io packet_queue.cpp packet_io.cpp packet_io_dumpfile.cpp packet_io_marsio.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)

src/packet_io/packet_io.cpp

@@ -0,0 +1,120 @@
+#include <stdlib.h>
+#include <string.h>
+
+#include "packet_io.h"
+#include "packet_io_marsio.h"
+#include "packet_io_dumpfile.h"
+
+typedef void *on_create(void *config);
+typedef void on_destroy(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 **pkt);
+typedef void on_send(void *handle, uint16_t thread_id, struct packet *pkt);
+
+struct packet_io
+{
+    void *handle;
+    on_create *create;
+    on_destroy *destroy;
+    on_stat *stat;
+    on_init *init;
+    on_recv *recv;
+    on_send *send;
+};
+
+struct packet_io *packet_io_create(struct packet_io_config *config)
+{
+    struct packet_io *handle = (struct packet_io *)calloc(1, sizeof(struct packet_io));
+    if (handle == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to alloc packet io");
+        return NULL;
+    }
+
+    struct packet_io_marsio_confg marsio_config;
+    strncpy(marsio_config.app_symbol, config->app_symbol, sizeof(marsio_config.app_symbol));
+    strncpy(marsio_config.dev_symbol, config->dev_symbol, sizeof(marsio_config.dev_symbol));
+    memcpy(marsio_config.cpu_mask, config->cpu_mask, sizeof(marsio_config.cpu_mask));
+    marsio_config.nr_threads = config->nr_threads;
+
+    struct packet_io_dumpfile_confg dumpfile_config;
+    strncpy(dumpfile_config.dumpfile_dir, config->dumpfile_dir, sizeof(dumpfile_config.dumpfile_dir));
+    dumpfile_config.nr_threads = config->nr_threads;
+
+    void *_config = NULL;
+
+    if (config->mode == PACKET_IO_MARSIO)
+    {
+        _config = &marsio_config;
+        handle->create = (on_create *)packet_io_marsio_create;
+        handle->destroy = (on_destroy *)packet_io_marsio_destory;
+        handle->stat = (on_stat *)packet_io_marsio_stat;
+        handle->init = (on_init *)packet_io_marsio_init;
+        handle->recv = (on_recv *)packet_io_marsio_recv;
+        handle->send = (on_send *)packet_io_marsio_send;
+    }
+    else
+    {
+        _config = &dumpfile_config;
+        handle->create = (on_create *)packet_io_dumpfile_create;
+        handle->destroy = (on_destroy *)packet_io_dumpfile_destory;
+        handle->stat = (on_stat *)packet_io_dumpfile_stat;
+        handle->init = (on_init *)packet_io_dumpfile_init;
+        handle->recv = (on_recv *)packet_io_dumpfile_recv;
+        handle->send = (on_send *)packet_io_dumpfile_send;
+    }
+
+    handle->handle = handle->create(_config);
+    if (handle->handle == NULL)
+    {
+        goto error_out;
+    }
+
+    return handle;
+
+error_out:
+    packet_io_destroy(handle);
+    return NULL;
+}
+
+void packet_io_destroy(struct packet_io *handle)
+{
+    if (handle)
+    {
+        handle->destroy(handle->handle);
+        free(handle);
+        handle = NULL;
+    }
+}
+
+struct packet_io_stat *packet_io_get_stat(struct packet_io *handle)
+{
+    return (struct packet_io_stat *)handle->stat(handle->handle);
+}
+
+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);
+}
+
+int packet_io_init(struct packet_io *handle, uint16_t thread_id)
+{
+    return handle->init(handle->handle, thread_id);
+}
+
+int packet_io_recv(struct packet_io *handle, uint16_t thread_id, struct packet **pkt)
+{
+    return handle->recv(handle->handle, thread_id, pkt);
+}
+
+void packet_io_send(struct packet_io *handle, uint16_t thread_id, struct packet *pkt)
+{
+    handle->send(handle->handle, thread_id, pkt);
+}

src/packet_io/packet_io.h

@@ -0,0 +1,70 @@
+#ifndef _PACKET_IO_H
+#define _PACKET_IO_H
+
+#ifdef __cpluscplus
+extern "C"
+{
+#endif
+
+#include "packet.h"
+#include "stellar.h"
+
+#define PACKET_IO_LOG_STATE(format, ...) LOG_STATE("packet_io", format, ##__VA_ARGS__)
+#define PACKET_IO_LOG_ERROR(format, ...) LOG_ERROR("packet_io", format, ##__VA_ARGS__)
+#define PACKET_IO_LOG_DEBUG(format, ...) LOG_DEBUG("packet_io", format, ##__VA_ARGS__)
+
+struct packet_io_stat
+{
+    uint64_t rx_pkts;
+    uint64_t tx_pkts;
+
+    uint64_t rx_bytes;
+    uint64_t tx_bytes;
+
+    uint64_t keepalive_pkts;
+    uint64_t keepalive_bytes;
+
+    uint64_t drop_pkts;
+    uint64_t drop_bytes;
+
+    uint64_t inject_pkts;
+    uint64_t inject_bytes;
+};
+
+enum packet_io_mode
+{
+    PACKET_IO_DUMPFILE = 0,
+    PACKET_IO_MARSIO = 1,
+};
+
+struct packet_io_config
+{
+    enum packet_io_mode mode;
+
+    // for dumpfile
+    char dumpfile_dir[256];
+
+    // for marsio
+    char app_symbol[64];
+    char dev_symbol[64];
+
+    uint8_t nr_threads;
+    uint16_t cpu_mask[MAX_THREAD_NUM];
+};
+
+struct packet_io;
+struct packet_io *packet_io_create(struct packet_io_config *config);
+void packet_io_destroy(struct packet_io *handle);
+struct packet_io_stat *packet_io_get_stat(struct packet_io *handle);
+void packet_io_print_stat(struct packet_io *handle);
+
+// return 0 if success, -1 if failed
+int packet_io_init(struct packet_io *handle, uint16_t thread_id);
+int packet_io_recv(struct packet_io *handle, uint16_t thread_id, struct packet **pkt);
+void packet_io_send(struct packet_io *handle, uint16_t thread_id, struct packet *pkt);
+
+#ifdef __cpluscplus
+}
+#endif
+
+#endif

src/packet_io/packet_io_dumpfile.cpp

@@ -0,0 +1,191 @@
+#include <pcap/pcap.h>
+#include <pthread.h>
+#include <unistd.h>
+
+#include "stellar.h"
+#include "file_scan.h"
+#include "packet_io.h"
+#include "packet_utils.h"
+#include "packet_queue.h"
+#include "packet_io_dumpfile.h"
+
+#define MAX_PACKET_QUEUE_SIZE (4096 * 1000)
+
+struct packet_io_dumpfile
+{
+    uint8_t nr_threads;
+    char dumpfile_dir[256];
+
+    pcap_t *pcap;
+    struct packet_queue *queue[MAX_THREAD_NUM];
+    struct packet_io_stat stat;
+    uint64_t io_thread_need_exit;
+    uint64_t io_thread_is_runing;
+};
+
+/******************************************************************************
+ * Private API
+ ******************************************************************************/
+
+static void pcap_handle(u_char *user, const struct pcap_pkthdr *h, const u_char *bytes)
+{
+    struct packet_io_dumpfile *handle = (struct packet_io_dumpfile *)user;
+
+    struct packet *pkt = packet_new(h->caplen);
+    if (pkt == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to alloc packet");
+        return;
+    }
+
+    memcpy((char *)pkt->data_ptr, bytes, h->caplen);
+    packet_parse(pkt, pkt->data_ptr, h->caplen);
+
+    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);
+
+    if (ATOMIC_READ(&handle->io_thread_need_exit))
+    {
+        PACKET_IO_LOG_STATE("dumpfile io thread need exit");
+        pcap_breakloop(handle->pcap);
+    }
+}
+
+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)
+
+    handle->pcap = pcap_open_offline(file, NULL);
+    if (handle->pcap == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to open pcap file: %s", file);
+        return -1;
+    }
+    pcap_loop(handle->pcap, -1, pcap_handle, (u_char *)handle);
+    pcap_close(handle->pcap);
+
+    PACKET_IO_LOG_STATE("dumpfile %s processed", file)
+
+    return 0;
+}
+
+static void *dumpfile_thread_cycle(void *arg)
+{
+    struct packet_io_dumpfile *handle = (struct packet_io_dumpfile *)arg;
+
+    ATOMIC_SET(&handle->io_thread_is_runing, 1);
+    PACKET_IO_LOG_STATE("dumpfile io thread is running");
+
+    file_scan(handle->dumpfile_dir, dumpfile_handle, arg);
+
+    PACKET_IO_LOG_STATE("dumpfile io thread is exiting");
+    ATOMIC_SET(&handle->io_thread_is_runing, 0);
+
+    return NULL;
+}
+
+/******************************************************************************
+ * Public API
+ ******************************************************************************/
+
+struct packet_io_dumpfile *packet_io_dumpfile_create(struct packet_io_dumpfile_confg *config)
+{
+    pthread_t tid;
+    struct packet_io_dumpfile *handle = (struct packet_io_dumpfile *)calloc(1, sizeof(struct packet_io_dumpfile));
+    if (handle == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to allocate memory for packet_io_dumpfile");
+        return NULL;
+    }
+
+    handle->nr_threads = config->nr_threads;
+    strncpy(handle->dumpfile_dir, config->dumpfile_dir, sizeof(handle->dumpfile_dir));
+
+    for (uint16_t i = 0; i < handle->nr_threads; i++)
+    {
+        handle->queue[i] = packet_queue_create(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_cycle, (void *)handle) != 0)
+    {
+        PACKET_IO_LOG_ERROR("unable to create packet io thread");
+        goto error_out;
+    }
+
+    return handle;
+
+error_out:
+    packet_io_dumpfile_destory(handle);
+    return NULL;
+}
+
+void packet_io_dumpfile_destory(struct packet_io_dumpfile *handle)
+{
+    if (handle)
+    {
+        ATOMIC_SET(&handle->io_thread_need_exit, 1);
+
+        while (ATOMIC_READ(&handle->io_thread_is_runing))
+        {
+            usleep(1000);
+        }
+
+        for (uint16_t i = 0; i < handle->nr_threads; i++)
+        {
+            packet_queue_destory(handle->queue[i]);
+        }
+        free(handle);
+        handle = NULL;
+    }
+}
+
+struct packet_io_stat *packet_io_dumpfile_stat(struct packet_io_dumpfile *handle)
+{
+    return &handle->stat;
+}
+
+int packet_io_dumpfile_init(struct packet_io_dumpfile *handle, uint16_t thread_id)
+{
+    return 0;
+}
+
+int packet_io_dumpfile_recv(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet **pkt)
+{
+    struct packet_queue *queue = handle->queue[thread_id];
+
+    packet_queue_pop(queue, pkt);
+    if (*pkt == NULL)
+    {
+        return -1;
+    }
+    else
+    {
+        ATOMIC_ADD(&handle->stat.rx_pkts, 1);
+        ATOMIC_ADD(&handle->stat.rx_bytes, packet_get_len(*pkt));
+        return 0;
+    }
+}
+
+void packet_io_dumpfile_send(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkt)
+{
+    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));
+    }
+
+    packet_free(pkt);
+}

src/packet_io/packet_io_dumpfile.h

@@ -0,0 +1,31 @@
+#ifndef _PACKET_IO_DUMPFILE_H
+#define _PACKET_IO_DUMPFILE_H
+
+#ifdef __cpluscplus
+extern "C"
+{
+#endif
+
+#include "packet.h"
+
+struct packet_io_dumpfile_confg
+{
+    char dumpfile_dir[256];
+    uint8_t nr_threads;
+};
+
+struct packet_io_dumpfile;
+
+struct packet_io_dumpfile *packet_io_dumpfile_create(struct packet_io_dumpfile_confg *config);
+void packet_io_dumpfile_destory(struct packet_io_dumpfile *handle);
+struct packet_io_stat *packet_io_dumpfile_stat(struct packet_io_dumpfile *handle);
+
+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 **pkt);
+void packet_io_dumpfile_send(struct packet_io_dumpfile *handle, uint16_t thread_id, struct packet *pkt);
+
+#ifdef __cpluscplus
+}
+#endif
+
+#endif

src/packet_io/packet_io_marsio.cpp

@@ -0,0 +1,242 @@
+#include <sched.h>
+#include <netinet/ether.h>
+
+#include "stellar.h"
+#include "marsio.h"
+#include "packet_io.h"
+#include "packet_utils.h"
+#include "packet_io_marsio.h"
+
+struct packet_io_marsio
+{
+    struct mr_instance *mr_ins;
+    struct mr_vdev *mr_dev;
+    struct mr_sendpath *mr_path;
+
+    struct packet_io_stat stat;
+};
+
+/******************************************************************************
+ * Private API
+ ******************************************************************************/
+
+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));
+    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));
+    if (packet_get_type(pkt) == PACKET_TYPE_CTRL)
+    {
+        marsio_buff_set_ctrlbuf(mbuff);
+    }
+}
+
+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;
+    }
+}
+
+/******************************************************************************
+ * Public API
+ ******************************************************************************/
+
+struct packet_io_marsio *packet_io_marsio_create(struct packet_io_marsio_confg *config)
+{
+    int opt = 1;
+    cpu_set_t coremask;
+    CPU_ZERO(&coremask);
+    for (uint8_t i = 0; i < config->nr_threads; i++)
+    {
+        CPU_SET(config->cpu_mask[i], &coremask);
+    }
+
+    struct packet_io_marsio *handle = (struct packet_io_marsio *)calloc(1, sizeof(struct packet_io_marsio));
+    if (handle == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to allocate memory for packet_io_marsio");
+        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(coremask));
+    marsio_option_set(handle->mr_ins, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt));
+
+    if (marsio_init(handle->mr_ins, config->app_symbol) != 0)
+    {
+        PACKET_IO_LOG_ERROR("unable to init marsio instance");
+        goto error_out;
+    }
+
+    handle->mr_dev = marsio_open_device(handle->mr_ins, config->dev_symbol, config->nr_threads, config->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:
+    packet_io_marsio_destory(handle);
+    return NULL;
+}
+
+void packet_io_marsio_destory(struct packet_io_marsio *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;
+    }
+}
+
+struct packet_io_stat *packet_io_marsio_stat(struct packet_io_marsio *handle)
+{
+    return &handle->stat;
+}
+
+int packet_io_marsio_init(struct packet_io_marsio *handle, uint16_t thread_id)
+{
+    if (marsio_thread_init(handle->mr_ins) != 0)
+    {
+        PACKET_IO_LOG_ERROR("unable to init marsio thread");
+        return -1;
+    }
+
+    return 0;
+}
+
+int packet_io_marsio_recv(struct packet_io_marsio *handle, uint16_t thread_id, struct packet **pkt)
+{
+    marsio_buff_t *rx_buff;
+    marsio_buff_t *rx_buffs[1];
+    thread_local struct packet thd_pkt;
+
+retry:
+    if (marsio_recv_burst(handle->mr_dev, thread_id, rx_buffs, 1) <= 0)
+    {
+        *pkt = NULL;
+        return -1;
+    }
+
+    rx_buff = rx_buffs[0];
+    char *data = marsio_buff_mtod(rx_buff);
+    int len = marsio_buff_datalen(rx_buff);
+
+    ATOMIC_ADD(&handle->stat.rx_pkts, 1);
+    ATOMIC_ADD(&handle->stat.rx_bytes, len);
+
+    if (is_keepalive_packet(data, len))
+    {
+        ATOMIC_ADD(&handle->stat.keepalive_pkts, 1);
+        ATOMIC_ADD(&handle->stat.keepalive_bytes, len);
+
+        ATOMIC_ADD(&handle->stat.tx_pkts, 1);
+        ATOMIC_ADD(&handle->stat.tx_bytes, len);
+        marsio_send_burst(handle->mr_path, thread_id, rx_buffs, 1);
+        goto retry;
+    }
+
+    metadata_to_packet(rx_buff, &thd_pkt);
+    packet_parse(&thd_pkt, data, len);
+    *pkt = &thd_pkt;
+
+    return 0;
+}
+
+void packet_io_marsio_send(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkt)
+{
+    marsio_buff_t *tx_buff = (marsio_buff_t *)packet_get_io_ctx(pkt);
+    if (packet_get_action(pkt) == PACKET_ACTION_DROP)
+    {
+        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
+        }
+    }
+    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");
+                return;
+            }
+            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));
+        }
+
+        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);
+    }
+
+    packet_free(pkt);
+}

src/packet_io/packet_io_marsio.h

@@ -0,0 +1,33 @@
+#ifndef _PACKET_IO_MARSIO_H
+#define _PACKET_IO_MARSIO_H
+
+#ifdef __cpluscplus
+extern "C"
+{
+#endif
+
+#include "packet.h"
+
+struct packet_io_marsio_confg
+{
+    char app_symbol[64];
+    char dev_symbol[64];
+    uint8_t nr_threads;
+    uint16_t cpu_mask[MAX_THREAD_NUM];
+};
+
+struct packet_io_marsio;
+
+struct packet_io_marsio *packet_io_marsio_create(struct packet_io_marsio_confg *config);
+void packet_io_marsio_destory(struct packet_io_marsio *handle);
+struct packet_io_stat *packet_io_marsio_stat(struct packet_io_marsio *handle);
+
+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 **pkt);
+void packet_io_marsio_send(struct packet_io_marsio *handle, uint16_t thread_id, struct packet *pkt);
+
+#ifdef __cpluscplus
+}
+#endif
+
+#endif

src/packet_io/packet_queue.cpp

@@ -0,0 +1,112 @@
+#include <pthread.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#include "packet_io.h"
+#include "packet_utils.h"
+#include "packet_queue.h"
+
+struct packet_queue
+{
+    pthread_mutex_t lock;
+    struct packet **queue;
+    uint32_t size;
+    uint32_t head;
+    uint32_t tail;
+};
+
+struct packet_queue *packet_queue_create(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 alloc packet queue");
+        return NULL;
+    }
+
+    queue->queue = (struct packet **)calloc(size, sizeof(struct packet *));
+    if (queue->queue == NULL)
+    {
+        PACKET_IO_LOG_ERROR("unable to alloc packet queue buffer");
+        free(queue);
+        return NULL;
+    }
+
+    queue->size = size;
+    queue->head = 0;
+    queue->tail = 0;
+    pthread_mutex_init(&queue->lock, NULL);
+
+    return queue;
+}
+
+void packet_queue_destory(struct packet_queue *queue)
+{
+    if (queue == NULL)
+    {
+        return;
+    }
+
+    struct packet *pkt = NULL;
+    while (1)
+    {
+        packet_queue_pop(queue, &pkt);
+        if (pkt == NULL)
+        {
+            break;
+        }
+        packet_free(pkt);
+    }
+
+    if (queue->queue)
+    {
+        free(queue->queue);
+        queue->queue = NULL;
+    }
+
+    pthread_mutex_destroy(&queue->lock);
+
+    free(queue);
+}
+
+int packet_queue_is_full(struct packet_queue *queue)
+{
+    return (queue->tail + 1) % queue->size == queue->head;
+}
+
+int packet_queue_is_empty(struct packet_queue *queue)
+{
+    return queue->head == queue->tail;
+}
+
+void packet_queue_push(struct packet_queue *queue, struct packet *pkt)
+{
+retry:
+    pthread_mutex_lock(&queue->lock);
+    if (packet_queue_is_full(queue))
+    {
+        PACKET_IO_LOG_ERROR("packet queue is full, retry later");
+        pthread_mutex_unlock(&queue->lock);
+        sleep(1);
+        goto retry;
+    }
+
+    queue->queue[queue->tail] = pkt;
+    queue->tail = (queue->tail + 1) % queue->size;
+    pthread_mutex_unlock(&queue->lock);
+}
+
+void packet_queue_pop(struct packet_queue *queue, struct packet **pkt)
+{
+    pthread_mutex_lock(&queue->lock);
+    if (packet_queue_is_empty(queue))
+    {
+        pthread_mutex_unlock(&queue->lock);
+        *pkt = NULL;
+        return;
+    }
+
+    *pkt = queue->queue[queue->head];
+    queue->head = (queue->head + 1) % queue->size;
+    pthread_mutex_unlock(&queue->lock);
+}

src/packet_io/packet_queue.h

@@ -0,0 +1,26 @@
+#ifndef _PACKET_QUEUE_H
+#define _PACKET_QUEUE_H
+
+#ifdef __cpluscplus
+extern "C"
+{
+#endif
+
+#include <stdint.h>
+
+struct packet_queue;
+
+struct packet_queue *packet_queue_create(uint32_t size);
+void packet_queue_destory(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