path: root/app/src/mrb.c

#include <assert.h>
#include <marsio.h>
#include <metadata_define.h>
#include <mrapp.h>
#include <mrb_define.h>
#include <mrdp_trace.h>
#include <protect.h>
#include <dp_trace.h>

#include <rte_common.h>
#include <rte_malloc.h>
#include <rte_mbuf.h>

void * marsio_buff_ctrlzone(marsio_buff_t * mr_buff, uint8_t id)
{
    return NULL;
}

void * marsio_buff_ctrlzone_data(marsio_buff_t * mr_buff, uint8_t id, uint8_t * size)
{
    return NULL;
}

void marsio_buff_ctrlzone_set(marsio_buff_t * mr_buff, uint8_t id, void * ptr_data, uint8_t size)
{
    return;
}

void marsio_buff_ctrlzone_reset(marsio_buff_t * mr_buff)
{
    struct mrb_metadata * mrb_meta = mrbuf_cz_data(mr_buff, MR_NODE_CTRLZONE_ID);
    struct dp_trace_buffer * dp_trace_buffer_ptr = mrb_meta->dp_trace_buffer;
    int8_t measurement_type = mrb_meta->measurement_type;

    /* reset the ctrlzone but not dp_trace_buffer */
    memset(mrbuf_cz_data(mr_buff, MR_NODE_CTRLZONE_ID), 0, sizeof(struct mrb_metadata));
    mrb_meta->dp_trace_buffer = dp_trace_buffer_ptr;
    mrb_meta->measurement_type = measurement_type;
}

void * mr_buffer_ctrlzone(struct rte_mbuf * mr_buff, uint8_t id)
{
    return NULL;
}

static struct rte_mempool_cache * mempool_cache_get(struct mr_instance * instance, struct rte_mempool * mp)
{
    if (!thread_info.is_dataplane_thread)
    {
        /* not a data plane thread, do not use any mp cache */
        return NULL;
    }

    thread_id_t tid = thread_info.thread_id;
    for (unsigned int i = 0; i < instance->nr_mp_cache_map; i++)
    {
        struct mp_cache_map * mp_cache_map_ptr = instance->mp_cache_map[i];
        if (mp_cache_map_ptr->mp != mp)
            continue;
        return mp_cache_map_ptr->mp_cache[tid];
    }

    return NULL;
}

int marsio_buff_alloc_v2(struct mr_instance * instance, marsio_buff_t * buff[], unsigned int nr_buffs)
{
    /* TODO: 要根据NUMA节点ID获取最近的pool */
    struct mr_vdev * vdev_desc = &instance->vdevs[thread_info.thread_id % instance->nr_vdevs];
    struct rte_mempool * mp = vdev_desc->vdi->direct_pool;
    struct rte_mempool_cache * mp_cache = mempool_cache_get(instance, mp);

    /* use the mp APIs with external mp */
    int ret = rte_mempool_generic_get(mp, (void **)buff, nr_buffs, mp_cache);
    if (unlikely(ret < 0))
    {
        return ret;
    }

    /* for each mbuf, reset the mbuf */
    for (unsigned int i = 0; i < nr_buffs; i++)
    {
        struct rte_mbuf * m = (struct rte_mbuf *)buff[i];
        __rte_mbuf_raw_sanity_check(m);
        rte_pktmbuf_reset(m);
        memset(mrbuf_cz_data(m, MR_NODE_CTRLZONE_ID), 0, sizeof(struct mrb_metadata));
    }

#ifndef NDEBUG
    /* 多线程锁，多线程同时失败时，只有一个线程能竞争到这个锁 */
    static pthread_mutex_t _mempool_list_dump_lock = PTHREAD_MUTEX_INITIALIZER;

    /* 申请内存失败，打印内存池数量分布情况统计 */
    if (unlikely(ret < 0))
    {
        pthread_mutex_lock(&_mempool_list_dump_lock);
        rte_mempool_list_dump(stderr);
        rte_panic("Cannot alloc mbufs from pool %p", mp);
    }

#endif

    if (unlikely(ret < 0))
        return ret;

    for (int i = 0; i < nr_buffs; i++)
    {
        struct rte_mbuf * m = (struct rte_mbuf *)buff[i];
        struct mrb_metadata * mrb_metadata = mrbuf_cz_data(m, MR_NODE_CTRLZONE_ID);
        mrb_metadata->packet_create_from_nf = 1;
    }

    instance->stat[thread_info.thread_id].mbuf_alloc_count += nr_buffs;
    return 0;
}

void marsio_buff_free_v2(struct mr_instance * instance, marsio_buff_t * buff[], unsigned int nr_buffs)
{
    for (unsigned int i = 0; i < nr_buffs; i++)
    {
        struct rte_mbuf * m = (struct rte_mbuf *)buff[i];
        struct rte_mbuf * m_next = NULL;

        if (marsio_dp_trace_measurements_can_emit(instance, m, DP_TRACE_MEASUREMENT_TYPE_TELEMETRY))
        {
            marsio_dp_trace_measurement_emit_str(instance, m, DP_TRACE_MEASUREMENT_TYPE_TELEMETRY, NULL,
                                                 "packet dropped");
        }

        if (marsio_dp_trace_measurements_can_emit(instance, m, DP_TRACE_MEASUREMENT_TYPE_TRACE))
        {
            marsio_dp_trace_measurement_emit_str(instance, m, DP_TRACE_MEASUREMENT_TYPE_TRACE, NULL, "packet dropped");
        }

        struct mrb_metadata * mrb_meta = rte_mbuf_to_priv(m);
        mrb_meta->egress_action = MR_EGRESS_ACTION_DROP;
        marsio_dp_trace_record_write(instance, (struct rte_mbuf *)m);

        __rte_mbuf_sanity_check(m, 1);
        while (m != NULL)
        {
            m_next = m->next;
            m = rte_pktmbuf_prefree_seg(m);
            if (likely(m != NULL))
            {
                RTE_ASSERT(!RTE_MBUF_CLONED(m) && (!RTE_MBUF_HAS_EXTBUF(m) || RTE_MBUF_HAS_PINNED_EXTBUF(m)));
                __rte_mbuf_raw_sanity_check(m);

                /* get the local cache by mp */
                struct rte_mempool_cache * mp_cache = mempool_cache_get(instance, m->pool);
                rte_mempool_generic_put(m->pool, (void **)&m, 1, mp_cache);
            }

            m = m_next;
        }
    }

    instance->stat[thread_info.thread_id].mbuf_free_count += nr_buffs;
}

int marsio_buff_malloc_device(struct mr_vdev * vdev, marsio_buff_t * marsio_buff[], unsigned int nr_mbufs,
                              int socket_id, int thread_id)
{
    RTE_SET_USED(socket_id);
    RTE_SET_USED(thread_id);
    return marsio_buff_alloc_v2(vdev->instance, marsio_buff, nr_mbufs);
}

int marsio_buff_malloc_global(struct mr_instance * instance, marsio_buff_t * marsio_buff[], unsigned int nr_mbufs,
                              int socket_id, int thread_id)
{
    RTE_SET_USED(socket_id);
    RTE_SET_USED(thread_id);
    return marsio_buff_alloc_v2(instance, marsio_buff, nr_mbufs);
}

void marsio_buff_free(struct mr_instance * instance, marsio_buff_t * marsio_buff[], unsigned int nr_mbufs,
                      int socket_id, int thread_id)
{
    RTE_SET_USED(socket_id);
    RTE_SET_USED(thread_id);
    marsio_buff_free_v2(instance, marsio_buff, nr_mbufs);
}

void marsio_buff_do_rehash(struct mr_instance * mr_instance, marsio_buff_t * m)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);

    /* parse the packet again */
    struct pkt_parser pkt_parser;
    pkt_parser_init(&pkt_parser, &mrb_metadata->pkt_parser_result, LAYER_TYPE_ALL, MR_PKT_PARSER_LAYERS_MAX);
    pkt_parser_exec(&pkt_parser, mbuf);

    struct rte_mbuf * mbufs[1] = {mbuf};
    struct pkt_parser_result * parser_results[1] = {&mrb_metadata->pkt_parser_result};

    /* clear the txq hash */
    // mbuf->hash.txadapter.txq = 0;

    /* do the rehash */
    distributer_calculate_from_parser_results(mr_instance->dist_object, mbufs, parser_results, 1);
}

void marsio_buff_append_pkt(marsio_buff_t * head, marsio_buff_t * next)
{
    return;
}

void marsio_buff_append_seg(marsio_buff_t * head, marsio_buff_t * next)
{
    PROTECT_rte_mbuf_unpoison_meta((struct rte_mbuf *)head);
    PROTECT_rte_mbuf_unpoison_meta((struct rte_mbuf *)next);

    rte_pktmbuf_chain((struct rte_mbuf *)head, (struct rte_mbuf *)next);

    PROTECT_rte_mbuf_poison_meta((struct rte_mbuf *)head);
    PROTECT_rte_mbuf_poison_meta((struct rte_mbuf *)next);
}

marsio_buff_t * marsio_buff_getnext_seg(marsio_buff_t * m)
{
    PROTECT_rte_mbuf_unpoison_meta((struct rte_mbuf *)m);
    return (marsio_buff_t *)(((struct rte_mbuf *)m)->next);
    PROTECT_rte_mbuf_poison_meta((struct rte_mbuf *)m);
}

marsio_buff_t * marsio_buff_getnext_pkt(marsio_buff_t * m)
{
    return NULL;
}

void marsio_buff_chain_pkt(marsio_buff_t * pkt, marsio_buff_t * next)
{
}

void marsio_buff_reset(marsio_buff_t * m)
{
    PROTECT_rte_mbuf_unpoison_meta(m);
    struct mrb_zone_idx *cz_first = NULL, *cz_last = NULL;

    /* 引用计数检查，修改报文时，不应该被其他的地方引用 */
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;

    /* 断言，调试模式直接抛出异常，非调试模式返回 */
    assert(rte_mbuf_refcnt_read(__mbuf) == 1);
    if (unlikely(rte_mbuf_refcnt_read(__mbuf) != 1))
        return;

    rte_pktmbuf_reset((struct rte_mbuf *)m);
    return;

    if (mrbuf_cz_num(m) > 0)
    {
        cz_first = mrbuf_cz(m, 0);
        cz_last = mrbuf_cz(m, mrbuf_cz_num(m) - 1);
        memset(mrbuf_cz_data(m, 0), 0, cz_last->offset + cz_last->size - cz_first->offset);
    }

    PROTECT_rte_mbuf_poison_meta(m);
}

char * marsio_buff_mtod(marsio_buff_t * m)
{
    PROTECT_rte_mbuf_unpoison_meta((struct rte_mbuf *)m);
    PROTECT_rte_mbuf_unpoison_data((struct rte_mbuf *)m);
    return rte_pktmbuf_mtod((struct rte_mbuf *)m, char *);
}

uint32_t marsio_buff_buflen(marsio_buff_t * m)
{
    return PROTECT_rte_pktmbuf_pkt_len((struct rte_mbuf *)m);
}

uint32_t marsio_buff_datalen(marsio_buff_t * m)
{
    return PROTECT_rte_pktmbuf_data_len((struct rte_mbuf *)m);
}

struct rte_mbuf * __pktmbuf_alloc(struct mr_instance * instance, int socket_id, int thread_id)
{
    struct rte_mbuf * __mbuf = NULL;
    marsio_buff_malloc_global(instance, (marsio_buff_t **)&__mbuf, 1, socket_id, thread_id);
    return __mbuf;
}

marsio_buff_t * marsio_buff_clone_deep(struct mr_instance * instance, marsio_buff_t * md, int socket_id, int thread_id)
{
    struct rte_mbuf *__mc, *__mi, **__prev;
    struct rte_mbuf * __md = (struct rte_mbuf *)md;

    uint32_t pktlen;
    uint8_t nseg;

    if (unlikely((__mc = __pktmbuf_alloc(instance, socket_id, thread_id)) == NULL))
        return NULL;

    __mi = __mc;
    __prev = &__mi->next;
    pktlen = __md->pkt_len;
    nseg = 0;

    do
    {
        nseg++;

        uint16_t __md_datalen = rte_pktmbuf_data_len(__md);
        void * __md_data = rte_pktmbuf_mtod(__md, void *);
        void * __mi_data = rte_pktmbuf_append(__mi, __md_datalen);
        rte_memcpy(__mi_data, __md_data, __md_datalen);

        *__prev = __mi;
        __prev = &__mi->next;
    } while ((__md = __md->next) != NULL && (__mi = __pktmbuf_alloc(instance, socket_id, thread_id)) != NULL);

    *__prev = NULL;
    __mc->nb_segs = nseg;
    __mc->pkt_len = pktlen;

    /* Allocation of new indirect segment failed */
    if (unlikely(__mi == NULL))
    {
        marsio_rte_pktmbuf_free(__mc);
        return NULL;
    }

    __rte_mbuf_sanity_check(__mc, 1);
    return (marsio_buff_t *)__mc;
}

static void __buff_clone_ctrlzone(marsio_buff_t * mc, marsio_buff_t * md)
{
    struct rte_mbuf * __mc = (struct rte_mbuf *)mc;
    struct rte_mbuf * __md = (struct rte_mbuf *)md;

    /* 复制Priv部分的数据 */
    while (__mc != NULL && __md != NULL)
    {
        struct mrb_priv_zone * __mc_priv_zone = mrbuf_priv(__mc);
        struct mrb_priv_zone * __mi_priv_zone = mrbuf_priv(__md);

        /* 断言，MBUF的私有区域一定大于控制域的大小 */
        assert(rte_pktmbuf_priv_size(__mc->pool) >= sizeof(struct mrb_priv_zone));
        rte_memcpy(__mc_priv_zone, __mi_priv_zone, rte_pktmbuf_priv_size(__mc->pool));
        ((struct mrb_metadata *)__mc_priv_zone)->measurement_type = DP_TRACE_MEASUREMENT_TYPE_UNKNOWN;
        ((struct mrb_metadata *)__mc_priv_zone)->dp_trace_buffer = NULL;

        /* 下一个链表项 */
        __mc = __mc->next;
        __md = __md->next;
    }

    /* 断言，__mc和__mi同时到达链表的终点，否则说明链表克隆时出现了问题。 */
    assert(__mc == NULL && __md == NULL);
}

static void __memcpy_operator_buff(struct rte_mbuf * __mi, struct rte_mbuf * __md)
{
    void * __md_buff = rte_mbuf_to_baddr(__md);
    void * __mi_buff = rte_mbuf_to_baddr(__mi);

    assert(__md->buf_len == __mi->buf_len);
    size_t bufflen = RTE_MIN(__md->buf_len, __mi->buf_len);

    rte_memcpy(__mi_buff, __md_buff, bufflen);
    __mi->data_len = __md->data_len;
    __mi->data_off = __md->data_off;
    return;
}

static void __memcpy_operator_data(struct rte_mbuf * __mi, struct rte_mbuf * __md)
{
    uint16_t __md_datalen = rte_pktmbuf_data_len(__md);
    void * __md_data = rte_pktmbuf_mtod(__md, void *);
    void * __mi_data = rte_pktmbuf_append(__mi, __md_datalen);
    rte_memcpy(__mi_data, __md_data, __md_datalen);
    return;
}

static marsio_buff_t * __buff_clone_memcpy(struct mr_instance * instance, marsio_buff_t * md, int socket_id,
                                           int thread_id,
                                           void (*memcpy_operator)(struct rte_mbuf * __mi, struct rte_mbuf * __md))
{
    struct rte_mbuf *__mc, *__mi, **__prev;
    struct rte_mbuf * __md = (struct rte_mbuf *)md;

    uint32_t pktlen;
    uint8_t nseg;

    if (unlikely((__mc = __pktmbuf_alloc(instance, socket_id, thread_id)) == NULL))
        return NULL;

    __mi = __mc;
    __prev = &__mi->next;
    pktlen = __md->pkt_len;
    nseg = 0;

    do
    {
        nseg++;
        memcpy_operator(__mi, __md);

        *__prev = __mi;
        __prev = &__mi->next;
    } while ((__md = __md->next) != NULL && (__mi = __pktmbuf_alloc(instance, socket_id, thread_id)) != NULL);

    *__prev = NULL;
    __mc->nb_segs = nseg;
    __mc->pkt_len = pktlen;

    /* Allocation of new indirect segment failed */
    if (unlikely(__mi == NULL))
    {
        marsio_rte_pktmbuf_free(__mc);
        return NULL;
    }

    __rte_mbuf_sanity_check(__mc, 1);
    return (marsio_buff_t *)__mc;
}

marsio_buff_t * marsio_buff_clone_with_options(struct mr_instance * instance, marsio_buff_t * md, int socket_id,
                                               int thread_id, uint16_t options)
{
    marsio_buff_t * mi = NULL;

    if (options & MR_BUFF_CLONE_BUFF)
    {
        mi = __buff_clone_memcpy(instance, md, socket_id, thread_id, __memcpy_operator_buff);
    }
    else if (options & MR_BUFF_CLONE_DATA)
    {
        mi = __buff_clone_memcpy(instance, md, socket_id, thread_id, __memcpy_operator_data);
    }

    if (unlikely(mi == NULL))
        return mi;
    if (options & MR_BUFF_CLONE_CTRLZONE)
    {
        __buff_clone_ctrlzone(mi, md);
    }

    return mi;
}

char * marsio_buff_prepend(marsio_buff_t * m, uint16_t len)
{
    /* 引用计数检查，修改报文时，不应该被其他的地方引用 */
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;
    if (unlikely(PROTECT_rte_mbuf_refcnt_read(__mbuf) != 1))
        return NULL;

    return PROTECT_rte_pktmbuf_prepend((struct rte_mbuf *)m, len);
}

char * marsio_buff_append(marsio_buff_t * m, uint16_t len)
{
    /* 引用计数检查，修改报文时，不应该被其他的地方引用 */
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;
    if (unlikely(PROTECT_rte_mbuf_refcnt_read(__mbuf) != 1))
        return NULL;

    return PROTECT_rte_pktmbuf_append((struct rte_mbuf *)m, len);
}

char * marsio_buff_adj(marsio_buff_t * m, uint16_t len)
{
    /* 引用计数检查，修改报文时，不应该被其他的地方引用 */
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;
    if (unlikely(PROTECT_rte_mbuf_refcnt_read(__mbuf) != 1))
        return NULL;

    return PROTECT_rte_pktmbuf_adj((struct rte_mbuf *)m, len);
}

int marsio_buff_trim(marsio_buff_t * m, uint16_t len)
{
    /* 引用计数检查，修改报文时，不应该被其他的地方引用 */
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;
    if (unlikely(PROTECT_rte_mbuf_refcnt_read(__mbuf) != 1))
        return -1;

    return PROTECT_rte_pktmbuf_trim((struct rte_mbuf *)m, len);
}

char * marsio_buff_offset_set(marsio_buff_t * m, off_t offset, int is_relative)
{
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;
    if (unlikely(rte_mbuf_refcnt_read(__mbuf) != 1))
        goto errout;

    int __offset = is_relative ? __mbuf->data_off + offset : offset;
    if (unlikely(__offset < 0 || __offset > __mbuf->buf_len))
        goto errout;

    __mbuf->data_off = (unsigned int)__offset;
    return rte_pktmbuf_mtod(__mbuf, char *);

errout:
    return NULL;
}

off_t marsio_buff_offset_get(marsio_buff_t * m)
{
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;
    return __mbuf->data_off;
}

static inline struct rte_mbuf * __copy_on_write_common(struct mr_instance * instance, struct rte_mbuf * __mbuf)
{
    if (likely(rte_mbuf_refcnt_read(__mbuf) == 1))
        return __mbuf;

    /* 深拷贝，此时报文有其他人引用 */
    /* TODO:  根据当前的上下文确定SOCKET-ID和LCORE-ID */
    struct rte_mbuf * __cloned_mbuf = marsio_buff_clone_with_options(
        instance, __mbuf, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY, MR_BUFF_CLONE_BUFF | MR_BUFF_CLONE_CTRLZONE);

    assert(__cloned_mbuf != NULL);

    /* 对于传入的报文，释放所有权，外面用户不应该再使用传入的MBUF */
    unsigned int thread_id = thread_info.thread_id;
    instance->stat[thread_id].mbuf_free_count += 1;

    marsio_rte_pktmbuf_free(__mbuf);
    return __cloned_mbuf;
}

/* 支持写时复制的报文修改裁剪函数 */
marsio_buff_t * marsio_buff_prepend_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out)
{
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;

    /* 写时复制检测 */
    struct rte_mbuf * __op_mbuf = __copy_on_write_common(instance, __mbuf);
    if (unlikely(__op_mbuf == NULL))
        return NULL;

    /* 执行操作 */
    void * __ptr_out = rte_pktmbuf_prepend(__op_mbuf, len);

    /* 检测，如果等于空，操作失败 */
    assert(__ptr_out != NULL);
    if (ptr_out != NULL)
        *ptr_out = __ptr_out;
    return __op_mbuf;
}

marsio_buff_t * marsio_buff_append_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out)
{
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;

    /* 写时复制检测 */
    struct rte_mbuf * __op_mbuf = __copy_on_write_common(instance, __mbuf);
    if (unlikely(__op_mbuf == NULL))
        return NULL;

    /* 执行操作 */
    void * __ptr_out = rte_pktmbuf_append(__op_mbuf, len);

    /* 检测，如果等于空，操作失败 */
    assert(__ptr_out != NULL);
    if (ptr_out != NULL)
        *ptr_out = __ptr_out;
    return __op_mbuf;
}

marsio_buff_t * marsio_buff_adj_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out)
{
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;

    /* 写时复制检测 */
    struct rte_mbuf * __op_mbuf = __copy_on_write_common(instance, __mbuf);
    if (unlikely(__op_mbuf == NULL))
        return NULL;

    /* 执行操作 */
    void * __ptr_out = rte_pktmbuf_adj(__op_mbuf, len);

    /* 检测，如果等于空，操作失败 */
    assert(__ptr_out != NULL);
    if (ptr_out != NULL)
        *ptr_out = __ptr_out;
    return __op_mbuf;
}

marsio_buff_t * marsio_buff_trim_cw(struct mr_instance * instance, marsio_buff_t * m, uint16_t len, void ** ptr_out)
{
    struct rte_mbuf * __mbuf = (struct rte_mbuf *)m;

    /* 写时复制检测 */
    struct rte_mbuf * __op_mbuf = __copy_on_write_common(instance, __mbuf);
    if (unlikely(__op_mbuf == NULL))
        return NULL;

    /* 执行操作 */
    int ret = rte_pktmbuf_trim(__op_mbuf, len);
    assert(ret == 0);

    /* 成功，返回报文头部，否则ptr_out返回的是空 */
    if (ptr_out != NULL && ret == 0)
    {
        *ptr_out = rte_pktmbuf_mtod(__op_mbuf, void *);
    }
    else if (ptr_out != NULL && ret != 0)
    {
        *ptr_out = NULL;
    }

    return __op_mbuf;
}

uint16_t marsio_buff_headroom(const marsio_buff_t * m)
{
    return rte_pktmbuf_headroom((const struct rte_mbuf *)m);
}

uint16_t marsio_buff_tailroom(const marsio_buff_t * m)
{
    return rte_pktmbuf_tailroom((const struct rte_mbuf *)m);
}

uint32_t marsio_get_pkt_type(marsio_buff_t * m)
{
    return ((struct rte_mbuf *)m)->packet_type;
}

int marsio_buff_is_ctrlbuf(marsio_buff_t * m)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    return mrb_metadata->is_ctrlbuf;
}

void marsio_buff_set_ctrlbuf(marsio_buff_t * m)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    mrb_metadata->is_ctrlbuf = 1;
}

void marsio_pktmbuf_dump(FILE * f, const marsio_buff_t * m, unsigned dump_len)
{
    rte_pktmbuf_dump(f, (const struct rte_mbuf *)m, dump_len);
}

uint64_t marsio_buff_get_timestamp(marsio_buff_t * m)
{
    return 0;
}

void marsio_buff_set_timestamp(marsio_buff_t * m, uint64_t timestamp)
{
}

int marsio_buff_get_timestamp_ex(marsio_buff_t * m, enum mr_timestamp_type ts_type, struct timespec * ts)
{
    return RT_SUCCESS;
}

void marsio_buff_set_rehash_index(marsio_buff_t * m, uint32_t hash)
{
    struct rte_mbuf * __m = (struct rte_mbuf *)m;
    __m->hash.usr = hash;
}

uint32_t marsio_buff_get_rehash_index(marsio_buff_t * m)
{
    struct rte_mbuf * __m = (struct rte_mbuf *)m;
    return __m->hash.usr;
}

int marsio_buff_get_metadata(marsio_buff_t * m, enum mr_buff_metadata_type type, void * data, unsigned int sz_data)
{
    return buffer_metadata_get(m, type, data, sz_data);
}

int marsio_buff_set_metadata(marsio_buff_t * m, enum mr_buff_metadata_type type, void * data, unsigned int sz_data)
{
    return buffer_metadata_set(m, type, data, sz_data);
}

int marsio_buff_get_sid_list(marsio_buff_t * m, sid_t * out_slist, uint8_t sz_out_slist)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    return sid_list_get(&mrb_metadata->sid_list, out_slist, sz_out_slist);
}

int marsio_buff_set_sid_list(marsio_buff_t * m, sid_t * slist, uint8_t sz_slist)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);

    return sid_list_set(&mrb_metadata->sid_list, slist, sz_slist);
}

int marsio_buff_append_sid_list(marsio_buff_t * m, sid_t * slist, uint8_t sz_slist)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    return sid_list_append(&mrb_metadata->sid_list, slist, sz_slist);
}

int marsio_buff_prepend_sid_list(marsio_buff_t * m, sid_t * slist, uint8_t sz_slist)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    return sid_list_prepend(&mrb_metadata->sid_list, slist, sz_slist);
}

int marsio_buff_get_current_sid(marsio_buff_t * m, sid_t * sid)
{
    struct rte_mbuf * mbuf = (struct rte_mbuf *)m;
    struct mrb_metadata * mrb_metadata = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);

    *sid = mrb_metadata->cur_sid;
    return 0;
}

int marsio_buff_unset_metadata(marsio_buff_t * m, enum mr_buff_metadata_type type)
{
    return 0;
}

struct tlv_header_define
{
    uint16_t be_type;
    uint16_t be_length;
    char value[0];
};

enum
{
    SMARTOFFLOAD_REQ_TYPE_PKT_HEADER = 0,
    SMARTOFFLOAD_REQ_TYPE_AGE = 1,
};

marsio_buff_t * marsio_buff_malloc_smartoffload(struct mr_vdev * vdev, const char * pkt, unsigned int pkt_len)
{
#if 0
    marsio_buff_t * bufs[1];
    int ret = marsio_buff_malloc_device(vdev, bufs, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY);

    if (unlikely(ret < 0))
    {
        goto errout;
    }

    struct rte_mbuf * __mbuf = (struct rte_mbuf *)bufs[0];
    assert(__mbuf != NULL);

    static uint64_t __counter = 0;
    __mbuf->hash.usr = __counter++;

    char * cur_ptr = NULL;
    cur_ptr = rte_pktmbuf_append(__mbuf, sizeof(struct tlv_header_define) + sizeof(uint32_t));

    if (unlikely(cur_ptr == NULL))
    {
        goto errout;
    }

    /* flow age */
    struct tlv_header_define * flow_age_header = (struct tlv_header_define *)cur_ptr;
    flow_age_header->be_type = ntohs(SMARTOFFLOAD_REQ_TYPE_AGE);
    flow_age_header->be_length = ntohs(sizeof(uint32_t));
    *(uint32_t *)flow_age_header->value = ntohl(1);

    /* packet */
    cur_ptr = rte_pktmbuf_append(__mbuf, sizeof(struct tlv_header_define) + pkt_len);
    if (unlikely(cur_ptr == NULL))
    {
        goto errout;
    }

    struct tlv_header_define * pkt_header = (struct tlv_header_define *)cur_ptr;
    pkt_header->be_type = ntohs(SMARTOFFLOAD_REQ_TYPE_PKT_HEADER);
    pkt_header->be_length = ntohs(pkt_len);
    rte_memcpy(pkt_header->value, pkt, pkt_len);

    return bufs[0];

errout:
    marsio_buff_free(vdev->instance, bufs, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY);
    return NULL;

#endif

    return NULL;
}