path: root/infra/src/vnode_mirror.c

/*	Virtual Data Deliver Node - Packet Deliver Class
    Author	:	Lu Qiuwen<[email protected]>
                Zheng Chao<[email protected]>
    Date	:	2017-01-09
*/

#include <assert.h>
#include <protect.h>
#include <rte_malloc.h>
#include <rte_mbuf.h>
#include <rte_ring.h>
#include <rte_spinlock.h>
#include <rte_version.h>
#include <sys/eventfd.h>
#include <unistd.h>

#include "vnode_common.h"

static inline void dist_tunnel_flush(struct vnode_prod * prod, struct vnode_cons * cons, unsigned int prodq,
                                     unsigned int consq, struct tunnel_desc * desc)
{
    struct vnode_cons_notify * cons_notify_ctx;

    /* nothing to send */
    if (desc->sz_en_buffer_used == 0)
    {
        return;
    }

    unsigned int nr_ring_count = desc->shared_credict_counter != NULL ? rte_ring_count(desc->tunnel_object) : 0;
    unsigned int nr_ring_to_use = nr_ring_count + desc->sz_en_buffer_used;
    unsigned int nr_shared_credict = 0;

    if (nr_ring_to_use > desc->tunnel_exclusive_size)
    {
        /* need to apply shared credict */
        assert(nr_ring_to_use >= (desc->tunnel_exclusive_size + desc->shared_credict_used));

        nr_shared_credict = nr_ring_to_use - (desc->tunnel_exclusive_size + desc->shared_credict_used);
        while (1)
        {
            uint32_t cur_value = rte_atomic32_read(desc->shared_credict_counter);
            uint32_t new_value = cur_value > nr_shared_credict ? cur_value - nr_shared_credict : 0;

            if (rte_atomic32_cmpset((volatile uint32_t *)desc->shared_credict_counter, cur_value, new_value))
            {
                nr_shared_credict = cur_value > new_value ? cur_value - new_value : 0;
                break;
            }
        }

        desc->shared_credict_used += nr_shared_credict;
        assert(desc->shared_credict_used <= (desc->tunnel_size - desc->tunnel_exclusive_size));
    }

    assert((desc->tunnel_exclusive_size + desc->shared_credict_used) >= nr_ring_count);

    unsigned int n_can_send = (desc->tunnel_exclusive_size + desc->shared_credict_used) - nr_ring_count;
    unsigned int n_to_send = RTE_MIN(desc->sz_en_buffer_used, n_can_send);

    size_t n_send_len = 0;
    for (unsigned int k = 0; k < n_to_send; k++)
    {
        n_send_len += rte_pktmbuf_data_len(desc->en_buffer[k]);
    }

    for (unsigned int k = 0; k < n_to_send; k++)
    {
        struct rte_mbuf * mbuf = desc->en_buffer[k];

        /* flush all cache lines in mbuf to release l1d */
        rte_cldemote(rte_mbuf_data_addr_default(mbuf));

        /* flush the mbuf's data first, and then flush the metadata of mbufs */
        /* because to know where is mbuf's data, we need to access mbuf's first cache line */
        rte_cldemote(mbuf);
        rte_cldemote(RTE_PTR_ADD(mbuf, RTE_CACHE_LINE_SIZE));
        rte_cldemote(RTE_PTR_ADD(mbuf, RTE_CACHE_LINE_SIZE * 2));
    }

    unsigned int n_send = rte_ring_sp_enqueue_burst(desc->tunnel_object, (void **)desc->en_buffer, n_to_send, NULL);
    unsigned int n_send_missed = desc->sz_en_buffer_used - n_send;

    /* packet is missed */
    if (unlikely(n_send_missed != 0))
    {
        for (unsigned int k = n_send; k < desc->sz_en_buffer_used; k++)
        {
            struct rte_mbuf * object_to_be_free = desc->en_buffer[k];
            n_send_len -= rte_pktmbuf_data_len(object_to_be_free);
            rte_pktmbuf_free(object_to_be_free);
        }

        /* return the shared credict */
        unsigned int nr_shared_credict_to_release = RTE_MIN(n_send_missed, nr_shared_credict);
        if (nr_shared_credict_to_release > 0)
        {
            rte_atomic32_add(desc->shared_credict_counter, (int32_t)nr_shared_credict_to_release);
            desc->shared_credict_used -= nr_shared_credict_to_release;
        }

        assert(desc->shared_credict_used <= (desc->tunnel_size - desc->tunnel_exclusive_size));
    }

    cons_notify_ctx = &cons->notify[consq];
    if (cons_notify_ctx->enable)
    {
        /* wakeup the cons when it is waiting */
        if (cons_notify_ctx->cons_running_status == CONS_STATUS_WAITING)
        {
            eventfd_write(cons_notify_ctx->cons_notify_eventfd, 1);
        }
    }

    // 更新生产者统计计数
    VNODE_STAT_UPDATE(prod, prodq, on_line, desc->sz_en_buffer_used);
    VNODE_STAT_UPDATE(prod, prodq, deliver, n_send);
    VNODE_STAT_UPDATE(prod, prodq, missed, n_send_missed);
    VNODE_STAT_UPDATE(prod, prodq, total_len, n_send_len);

    // 更新消费者统计计数
    VNODE_STAT_UPDATE(cons, consq, on_line, desc->sz_en_buffer_used);
    VNODE_STAT_UPDATE(cons, consq, deliver, n_send);
    VNODE_STAT_UPDATE(cons, consq, missed, n_send_missed);
    VNODE_STAT_UPDATE(cons, consq, total_len, n_send_len);

    // Batch size
#if 0
    VNODE_STAT_UPDATE(prod, prodq, batch_size_total, desc->sz_en_buffer_used);
    VNODE_STAT_UPDATE(prod, prodq, batch_size_count, 1);

    // Ring -- Prod
    VNODE_STAT_UPDATE_AVG(prod, prodq, ring_elem_count_avg, nr_ring_count);
    VNODE_STAT_UPDATE_AVG(prod, prodq, ring_shared_credict_avg, nr_shared_credict);
#endif

    // 清空缓冲区
    desc->sz_en_buffer_used = 0;
}

static inline void dist_tunnel_enqueue(struct vnode_prod * prod, struct vnode_cons * cons, unsigned int prodq,
                                       unsigned int consq, struct tunnel_desc * desc, struct rte_mbuf * obj)
{
#if VNODE_CHECK_THREAD_SAFE
    assert(rte_spinlock_trylock(&desc->lock_thread_safe_check));
#endif

    desc->en_buffer[desc->sz_en_buffer_used++] = obj;
    assert(desc->sz_en_buffer_used <= desc->sz_en_buffer);

#if 0
    // append the object at the tail of enqueue buffer.
    unsigned int pos;
    pos = desc->sz_en_buffer_used;

    assert(pos < desc->sz_en_buffer);
    desc->en_buffer[pos++] = obj;

    // the enqueue buffer is not full, return
    if (likely(pos < desc->sz_en_buffer))
    {
        desc->sz_en_buffer_used = pos;
        goto out;
    }

    desc->sz_en_buffer_used = desc->sz_en_buffer;
#endif

#if 0
    dist_tunnel_flush(prod, cons, prodq, consq, desc);
#endif

#if VNODE_CHECK_THREAD_SAFE

out:
    rte_spinlock_unlock(&desc->lock_thread_safe_check);
#endif

    return;
}

static inline int dist_tunnel_dequeue(struct tunnel_desc * desc, void * obj, int nr_max_obj)
{
    unsigned int nr_deq = rte_ring_sc_dequeue_burst(desc->tunnel_object, obj, nr_max_obj, NULL);
    unsigned int shared_credict_to_release = RTE_MIN(nr_deq, desc->shared_credict_used);

    if (shared_credict_to_release > 0)
    {
        rte_atomic32_add(desc->shared_credict_counter, (int32_t)shared_credict_to_release);
        desc->shared_credict_used -= shared_credict_to_release;
    }

    assert(desc->shared_credict_used <= (desc->tunnel_size - desc->tunnel_exclusive_size));
    return (int)nr_deq;
}

// Tunnel Block Data Operation, send objects according for their hash value,
// only use by prod.
// TODO: rewrite in SSE/SSE2/AVX/AVX2 intrinsics

static inline void dist_tunnel_block_flush(struct tunnel_block * block, int prodq)
{
    for (unsigned int consq = 0; consq < block->nr_consq; consq++)
    {
        struct tunnel_desc * tunnel = *tunnel_block_locate(block, prodq, consq);
        dist_tunnel_flush(block->prod, block->cons, prodq, consq, tunnel);
    }
}

static inline void dist_tunnel_block_enqueue_with_hash(struct tunnel_block * block, int prodq, struct rte_mbuf * obj[],
                                                       uint32_t hash[], int nr_obj)
{
    assert(nr_obj <= MR_LIBVNODE_MAX_SZ_BURST);
    for (unsigned int i = 0; i < nr_obj; i++)
    {
        assert(obj[i] != NULL);
        unsigned int consq = hash[i] % block->nr_consq;
        struct tunnel_desc * tunnel = *tunnel_block_locate(block, prodq, consq);
        dist_tunnel_enqueue(block->prod, block->cons, prodq, consq, tunnel, obj[i]);
    }

    dist_tunnel_block_flush(block, prodq);
}

#if 0

#define __FWDSTEP 4

static inline void dist_tunnel_block_enqueue_with_hash(struct tunnel_block * block,
    int prodq, struct rte_mbuf * obj[], uint32_t hash[], int nr_obj)
{
    struct tunnel_desc * tunnel;
    unsigned int idx = 0;
    unsigned int consq;

    switch (nr_obj % 4)
    {
    case 0:
        while (idx != nr_obj) {
            consq = hash[idx] % block->nr_consq;
            tunnel = *tunnel_block_locate(block, prodq, consq);
            dist_tunnel_enqueue(block->prod, block->cons, prodq, consq, tunnel, obj[idx]);
            idx++;
    case 3:
        consq = hash[idx] % block->nr_consq;
        tunnel = *tunnel_block_locate(block, prodq, consq);
        dist_tunnel_enqueue(block->prod, block->cons, prodq, consq, tunnel, obj[idx]);
        idx++;
    case 2:
        consq = hash[idx] % block->nr_consq;
        tunnel = *tunnel_block_locate(block, prodq, consq);
        dist_tunnel_enqueue(block->prod, block->cons, prodq, consq, tunnel, obj[idx]);
        idx++;
    case 1:
        consq = hash[idx] % block->nr_consq;
        tunnel = *tunnel_block_locate(block, prodq, consq);
        dist_tunnel_enqueue(block->prod, block->cons, prodq, consq, tunnel, obj[idx]);
        idx++;
        }
    }
}

#endif

// Tunnel Block Dequeue, dequeue from block, only used by cons.
// TODO: rewrite in SSE/SSE2/AVX/AVX2 intrinsics

static inline int dist_tunnel_block_dequeue(struct tunnel_block * block, int consq, struct rte_mbuf * obj[],
                                            int nr_max_obj)
{
    unsigned int nr_obj = 0, nr_obj_recv = 0;
    unsigned int nr_obj_left = nr_max_obj;

    for (int prodq = 0; prodq < block->nr_prodq; prodq++)
    {
        struct tunnel_desc * tunnel = *tunnel_block_locate(block, prodq, consq);
        nr_obj_recv = dist_tunnel_dequeue(tunnel, &obj[nr_obj], nr_obj_left);
        nr_obj += nr_obj_recv;
        nr_obj_left -= nr_obj_recv;
    }

    assert(nr_obj <= nr_max_obj);
    return nr_obj;
}



/*	VNode Data Operation Functions
    VNode Enqueue/Dequeue Data
*/

#if VNODE_MIRROR_CLONE_BY_REFCNT
static inline int __mirror_clone_objects(struct rte_mempool * clone_pool, struct rte_mbuf * source_objects[],
                                         struct rte_mbuf * cloned_objects[], unsigned int nr_source_objects)
{
    for (int i = 0; i < nr_source_objects; i++)
    {
        PROTECT_rte_mbuf_refcnt_update(source_objects[i], 1);
    }

    rte_memcpy(cloned_objects, source_objects, sizeof(struct rte_mbuf *) * nr_source_objects);
    return 0;
}

#else

static inline int __mirror_clone_objects(struct rte_mempool * clone_pool, struct rte_mbuf * source_objects[],
                                         struct rte_mbuf * cloned_objects[], unsigned int nr_source_objects)
{
    // 全部复制成功标志位，避免在循环体内判断，减少分支判断造成的流水线中断
    uintptr_t tag_clone_objects = 1;

    for (int i = 0; i < nr_source_objects; i++)
    {
        cloned_objects[i] = rte_pktmbuf_clone(source_objects[i], clone_pool);
        tag_clone_objects = (tag_clone_objects && cloned_objects[i]);
    }

    if (likely(tag_clone_objects))
        return 0;

    /* 克隆失败，释放所有已经克隆的报文 */
    for (int i = 0; i < nr_source_objects; i++)
    {
        PROTECT_rte_pktmbuf_free(cloned_objects[i]);
    }

    return -1;
}

#endif

int vnode_mirror_enqueue_bulk(struct vnode_prod * prod, unsigned int prodq, struct rte_mbuf * objects[],
                              uint32_t hash[], int nr_objects)
{
    assert(nr_objects <= MR_LIBVNODE_MAX_SZ_BURST);

    struct tunnel_block * block = ACCESS_ONCE(prod->block);
    if (unlikely(block == NULL))
    {
        goto failure;
    }

    dist_tunnel_block_enqueue_with_hash(block, (int)prodq, objects, hash, nr_objects);
    return 0;

failure:
    for (int i = 0; i < nr_objects; i++)
    {
        rte_pktmbuf_free(objects[i]);
    }

    VNODE_STAT_UPDATE(prod, prodq, on_line, nr_objects);
    VNODE_STAT_UPDATE(prod, prodq, missed, nr_objects);
    return 0;
}

int vnode_mirror_dequeue_burst(struct vnode_cons * cons, unsigned int consq, struct rte_mbuf * objects[],
                               int nr_max_objects)
{
    struct tunnel_block * block = ACCESS_ONCE(cons->block);
    if (likely(block != NULL))
    {
        return dist_tunnel_block_dequeue(block, consq, objects, nr_max_objects);
    }
    else
    {
        return 0;
    }
}

void vnode_mirror_flush(struct vnode_prod * prod, unsigned int prodq)
{
    struct tunnel_block * block = prod->block;
    if (unlikely(block != NULL))
    {
        dist_tunnel_block_flush(block, prodq);
    }
}

struct vnode * vnode_mirror_create(const char * sym, unsigned int sz_exclusive, unsigned int sz_shared,
                                   unsigned int sz_buffer, unsigned int notify_cons_when_rx,
                                   unsigned int batch_interval_us)
{
    struct vnode * vnode_common = __vnode_common_create(sym, sz_exclusive, sz_buffer, notify_cons_when_rx);

    if (vnode_common == NULL)
    {
        MR_ERROR("Mirror vnode %s create failed. ", sym);
        return NULL;
    }

    vnode_common->batch_interval_tsc = batch_interval_us * rte_get_timer_cycles() / US_PER_S;
    vnode_common->sz_shared = sz_shared;

    rte_atomic32_init(&vnode_common->shared_credict_counter);
    return vnode_common;
}

int vnode_mirror_delete(struct vnode * vnode)
{
    return __vnode_common_delete(vnode);
}

void vnode_mirror_common_unpoison(struct vnode * vnode)
{
    __vnode_common_unpoison(vnode);
}

__USE_COMMON_VNODE_CREATE_PROD(mirror)
__USE_COMMON_VNODE_CREATE_CONS(mirror)
__USE_COMMON_VNODE_DELETE_PROD(mirror)
__USE_COMMON_VNODE_DELETE_CONS(mirror)
__USE_COMMON_VNODE_CONS_LOOKUP(mirror)
__USE_COMMON_VNODE_PROD_LOOKUP(mirror)
__USE_COMMON_VNODE_PROD_STAT_GET(mirror)
__USE_COMMON_VNODE_CONS_STAT_GET(mirror)
__USE_COMMON_VNODE_PROD_ATTACH(mirror)
__USE_COMMON_VNODE_CONS_ATTACH(mirror)
__USE_COMMON_VNODE_UNPOISON_PROD(mirror)
__USE_COMMON_VNODE_UNPOISON_CONS(mirror)
__USE_COMMON_VNODE_NOTIFY_CTX_CONS(mirror)