feature: packet IO support IP reassembly

9 files changed, 1078 insertions, 938 deletions

diff --git a/infra/packet_io/CMakeLists.txt b/infra/packet_io/CMakeLists.txt
index 8edf21d..036318d 100644
--- a/infra/packet_io/CMakeLists.txt
+++ b/infra/packet_io/CMakeLists.txt
@@ -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)
\ No newline at end of file
+target_link_libraries(packet_io marsio pcap packet_manager ip_reassembly)
\ No newline at end of file
diff --git a/infra/packet_io/mars_io.c b/infra/packet_io/mars_io.c
new file mode 100644
index 0000000..dfd1f49
--- /dev/null
+++ b/infra/packet_io/mars_io.c
@@ -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];
+}
diff --git a/infra/packet_io/mars_io.h b/infra/packet_io/mars_io.h
new file mode 100644
index 0000000..66e41ee
--- /dev/null
+++ b/infra/packet_io/mars_io.h
@@ -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
diff --git a/infra/packet_io/marsio_io.c b/infra/packet_io/marsio_io.c
deleted file mode 100644
index cf0b6d8..0000000
--- a/infra/packet_io/marsio_io.c
+++ /dev/null
@@ -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];
-}
diff --git a/infra/packet_io/marsio_io.h b/infra/packet_io/marsio_io.h
deleted file mode 100644
index ab2f37c..0000000
--- a/infra/packet_io/marsio_io.h
+++ /dev/null
@@ -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
diff --git a/infra/packet_io/packet_io.c b/infra/packet_io/packet_io.c
index 68b4f12..d0efc32 100644
--- a/infra/packet_io/packet_io.c
+++ b/infra/packet_io/packet_io.c
@@ -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);
diff --git a/infra/packet_io/packet_io.h b/infra/packet_io/packet_io.h
index efb9453..4d9d457 100644
--- a/infra/packet_io/packet_io.h
+++ b/infra/packet_io/packet_io.h
@@ -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
diff --git a/infra/packet_io/pcap_io.c b/infra/packet_io/pcap_io.c
index 00220c9..139dc70 100644
--- a/infra/packet_io/pcap_io.c
+++ b/infra/packet_io/pcap_io.c
@@ -1,35 +1,58 @@
-#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)
+#define RING_BUFFER_MAX_SIZE (4096 * 1000)
+
+struct pcap_pkt
+{
+    char *data;
+    int len;
+    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 packet_io_config cfg;
+    struct pcap_io_cfg *cfg;
 
     pcap_t *pcap;
     struct logger *logger;
-    struct packet_queue *queue[MAX_THREAD_NUM];
+    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;
@@ -38,99 +61,151 @@ struct pcap_io
     uint64_t read_pcap_pkts;
 };
 
-struct pcap_pkt
-{
-    char *data;
-    int len;
-    struct timeval ts;
-};
-
 /******************************************************************************
- * 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)
-    {
-        return;
-    }
-
-    if (queue->queue)
+    if (ring)
     {
-        free(queue->queue);
-        queue->queue = NULL;
+        if (ring->buff)
+        {
+            free(ring->buff);
+            ring->buff = NULL;
+        }
+        free(ring);
+        ring = 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 -- 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 -- utils
+ * 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));
+    pcap_io->cfg = pcap_io_cfg_new(toml_file);
+    if (pcap_io->cfg == NULL)
+    {
+        PCAP_IO_LOG_ERROR("unable to create pcap_io_cfg");
+        goto error_out;
+    }
 
-    for (uint16_t i = 0; i < handle->cfg.nr_worker_thread; i++)
+    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++)
     {
-        handle->queue[i] = packet_queue_new(MAX_PACKET_QUEUE_SIZE);
-        if (handle->queue[i] == NULL)
+        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,139 +583,109 @@ 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)
-        {
-            break;
-        }
-        else
+        ring_buffer_pop(ring, (void **)&pcap_pkt);
+        pkt = recv_packet(pcap_io, pcap_pkt, thr_idx);
+        if (pkt)
         {
-            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;
+            send_packet(pcap_io, pkt, thr_idx);
         }
+        pkts[i] = NULL;
+    }
+}
 
-        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);
+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;
 
-            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);
+    for (int i = 0; i < nr_pkts; i++)
+    {
+        pkt = pkts[i];
 
-            if (packet_dump_pcap(pkt, file) == -1)
-            {
-                PCAP_IO_LOG_ERROR("unable to dump pcap file: %s", file);
-            }
-            else
+        if (packet_is_defraged(pkt))
+        {
+            while ((frag = packet_pop_frag(pkt)))
             {
-                PCAP_IO_LOG_FATAL("dump inject packet: %s", file);
+                drop_packet(pcap_io, frag, thr_idx);
             }
+            packet_free(pkt);
+        }
+        else
+        {
+            drop_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_yield(void *handle __attribute__((unused)), uint16_t thr_idx __attribute__((unused)))
+{
+    return;
+}
+
+void pcap_io_polling(void *handle, uint16_t thr_idx)
 {
-    struct packet *pkt = NULL;
     struct pcap_io *pcap_io = (struct pcap_io *)handle;
-    struct packet_io_stat *stat = &pcap_io->stat[thr_idx];
+    struct ip_reassembly *ip_reass = pcap_io->ip_reass[thr_idx];
+    struct packet *pkt = NULL;
+    uint64_t now_ms = clock_get_real_time_ms();
 
-    for (uint16_t i = 0; i < nr_pkts; i++)
+    while ((pkt = ip_reassembly_clean(ip_reass, now_ms)))
     {
-        pkt = &pkts[i];
-        struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_origin_ctx(pkt);
-        if (pcap_pkt)
+        if (pcap_io->cfg->fail_action == 0)
         {
-            stat->pkts_dropped++;
-            stat->bytes_dropped += packet_get_raw_len(pkt);
-            free(pcap_pkt);
+            send_packet(pcap_io, pkt, thr_idx);
+        }
+        else
+        {
+            drop_packet(pcap_io, pkt, thr_idx);
         }
-        packet_free(pkt);
     }
-}
 
-void pcap_io_yield(void *handle __attribute__((unused)), uint16_t thr_idx __attribute__((unused)))
-{
-    return;
+    // 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];
 }
\ No newline at end of file
diff --git a/infra/packet_io/pcap_io.h b/infra/packet_io/pcap_io.h
index 6619f46..9556eda 100644
--- a/infra/packet_io/pcap_io.h
+++ b/infra/packet_io/pcap_io.h
@@ -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