path: root/service/src/node_shmdev.c

#include <rte_common.h>
#include <rte_debug.h>
#include <rte_ethdev.h>
#include <rte_ether.h>
#include <rte_graph.h>
#include <rte_graph_worker.h>
#include <rte_mbuf.h>

#include <MESA_prof_load.h>
#include <cJSON.h>
#include <mrb_define.h>
#include <sc_node.h>
#include <sc_node_common.h>
#include <sc_trace.h>
#include <sc_vdev.h>
#include <sys/ucontext.h>

#define MR_SHMDEV_STAT_ADD(st, graph_id, counter, value)                                                               \
    do                                                                                                                 \
    {                                                                                                                  \
        st[graph_id].counter += value;                                                                                 \
    } while (0)

enum
{
    SHMDEV_RX_NEXT_NODE_FORWARDER,
    SHMDEV_RX_NEXT_NODE_HEALTH_CHECK,
    SHMDEV_RX_NEXT_NODE_PKT_DROP,
    SHMDEV_RX_NEXT_NODE_MAX
};

struct shmdev_stat_per_lcore
{
    volatile uint64_t total_tx_pkts;
    volatile uint64_t total_rx_pkts;
    volatile uint64_t original_pkt;
    volatile uint64_t nf_create_pkt;
    volatile uint64_t to_eth_ingress_pkt;
} __rte_cache_aligned;

struct shmdev_stat_per_lcore shmdev_graph_stat[RTE_MAX_LCORE];

enum packet_direction
{
    PACKET_DIRECTION_TX,
    PACKET_DIRECTION_RX
};

/* Generate and store the trace information */
static __rte_always_inline void gen_store_trace_info(struct rte_node * node, struct rte_mbuf * mbuf,
                                                     struct vdev * shm_dev_desc, enum packet_direction direction)
{
    struct dp_trace_record_meta meta = {.appsym = MR_TRACE_APPSYM, .module = node->name};

    char str_record[MR_STRING_MAX];
    int len = snprintf(str_record, sizeof(str_record), "port:%u,%s", shm_dev_desc->port_id, shm_dev_desc->symbol);

    if (direction == PACKET_DIRECTION_RX)
    {
        len += snprintf(str_record + len, sizeof(str_record) - len, ",next:%s", node->nodes[0]->name);
    }

    /* Emit the trace record */
    dp_trace_record_emit_str(sc_main_get()->trace, mbuf, rte_lcore_id(), &meta, str_record);
}

uint16_t shmdev_rx_node_process(struct rte_graph * graph, struct rte_node * node, void ** objs, uint16_t cnt)
{
    struct dev_node_ctx * ctx = (struct dev_node_ctx *)node->ctx;

    struct mr_dev_desc * dev_desc = ctx->dev_desc;
    struct vdev * shm_dev_desc = dev_desc->shm_dev_desc;

    RTE_SET_USED(objs);
    RTE_SET_USED(cnt);

    struct mr_dev_desc_qid_map * qid_map = dev_desc->rx_qid_map;
    unsigned int core_id = rte_lcore_id();

    /* check the core can do recv for this device or not */
    if (qid_map->qid_enabled[core_id] == 0)
    {
        return 0;
    }

    unsigned int qid = qid_map->qid_map[core_id];
    unsigned int nr_mbufs =
        vdev_collect(shm_dev_desc, qid, (struct rte_mbuf **)node->objs, RTE_GRAPH_BURST_SIZE, VDEV_COLLECT_DATA);

    if (unlikely(nr_mbufs == 0))
        return 0;

    /* Save number of objects used */
    node->idx = nr_mbufs;
    unsigned int mbuf_alloc_at_nf = 0;
    unsigned int mbuf_alloc_at_serv = 0;

    /* Deal all pkt */
    for (unsigned int i = 0; i < node->idx; i++)
    {
        /* prefetch next mbuf */
        if (i < node->idx - 1)
        {
            rte_prefetch0(node->objs[i + 1]);
            rte_prefetch0(mrbuf_cz_data(node->objs[i + 1], MR_NODE_CTRLZONE_ID));
        }

        struct rte_mbuf * mbuf0 = (struct rte_mbuf *)node->objs[i];
        struct mrb_metadata * mrb_meta = mrbuf_cz_data(mbuf0, MR_NODE_CTRLZONE_ID);

        assert(mbuf0 != NULL);
        assert(mrb_meta != NULL);
        __rte_mbuf_sanity_check(mbuf0, 1);

        /* Check pkt */
        if (likely(mrb_meta->packet_create_from_nf == 0))
        {
            mbuf_alloc_at_serv++;
        }
        else
        {
            /* prepare to parse the ingress pkt */
            struct pkt_parser pkt_parser;
            pkt_parser_init(&pkt_parser, &mrb_meta->pkt_parser_result, LAYER_TYPE_ALL, MR_PKT_PARSER_LAYERS_MAX);
            pkt_parser_exec(&pkt_parser, mbuf0);
            mbuf_alloc_at_nf++;
        }

        /* Fill port ingress  */
        mrb_meta->port_ingress = dev_desc->port_id;

        /* Check if tracing is enabled for the current Mbuf */
        if (unlikely(dp_trace_record_can_emit(mbuf0)))
        {
            gen_store_trace_info_rx(node, mbuf0, dev_desc, qid);

            if (unlikely(mrb_meta->packet_create_from_nf == 1))
            {
                gen_store_trace_info_sid_list(node, mbuf0);
                gen_store_trace_info_pkt_parser(node, mbuf0);
            }
        }
    }

    /* Update total rx pkts */
    MR_SHMDEV_STAT_ADD(shmdev_graph_stat, graph->id, original_pkt, mbuf_alloc_at_serv);
    MR_SHMDEV_STAT_ADD(shmdev_graph_stat, graph->id, nf_create_pkt, mbuf_alloc_at_nf);
    MR_SHMDEV_STAT_ADD(shmdev_graph_stat, graph->id, total_rx_pkts, nr_mbufs);
    MR_SHMDEV_STAT_ADD(shmdev_graph_stat, graph->id, to_eth_ingress_pkt, nr_mbufs);

    /* move to next node */
    rte_node_next_stream_move(graph, node, 0);
    return nr_mbufs;
}

uint16_t shmdev_tx_node_process(struct rte_graph * graph, struct rte_node * node, void ** objs, uint16_t cnt)
{
    struct dev_node_ctx * ctx = (struct dev_node_ctx *)node->ctx;
    struct rte_mbuf ** mbufs = (struct rte_mbuf **)node->objs;

    struct mr_dev_desc * dev_desc = ctx->dev_desc;
    struct vdev * shm_dev_desc = dev_desc->shm_dev_desc;
    unsigned int nr_pkts = cnt;

    MR_SHMDEV_STAT_ADD(shmdev_graph_stat, graph->id, total_tx_pkts, cnt);

    for (unsigned int i = 0; i < nr_pkts; i++)
    {
        /* Check if tracing is enabled for the current Mbuf */
        if (unlikely(dp_trace_record_can_emit(mbufs[i])))
        {
            gen_store_trace_info_tx(node, mbufs[i], dev_desc, graph->id);
            // gen_store_trace_info_rte_mbuf(node, mbufs[i]);
        }
    }

    while (nr_pkts)
    {
        unsigned int nr_mbufs_this_batch = (nr_pkts > RTE_GRAPH_BURST_SIZE) ? RTE_GRAPH_BURST_SIZE : nr_pkts;
        vdev_dispatch(shm_dev_desc, graph->id, mbufs, nr_mbufs_this_batch, 0);

        nr_pkts -= nr_mbufs_this_batch;
        mbufs += nr_mbufs_this_batch;

        /* retrieve the backpressure packets */
        struct rte_mbuf * rt_mbufs[RTE_GRAPH_BURST_SIZE];
        int ret = vdev_rt_pkts_retrieve(shm_dev_desc, graph->id, rt_mbufs, RTE_GRAPH_BURST_SIZE);

        /* these packet to pkt drop node */
        if (unlikely(ret > 0))
        {
            rte_node_enqueue(graph, node, 0, (void **)rt_mbufs, ret);
        }
    }

    return cnt;
}

/************************************** Shmdev Statistics **************************************/
cJSON * shmdev_tx_node_monit_loop(struct sc_main * sc)
{
    unsigned int nr_graph_total = sc->nr_io_thread;
    cJSON * json_root = cJSON_CreateObject();

    for (uint32_t graph_id = 0; graph_id < nr_graph_total; graph_id++)
    {
        struct shmdev_stat_per_lcore * stat_item = &shmdev_graph_stat[graph_id];
        if (stat_item->total_tx_pkts == 0)
            continue;

        cJSON * graph_obj = cJSON_CreateObject();
        cJSON_AddNumberToObject(graph_obj, "total_tx_pkts", stat_item->total_tx_pkts);

        char graph_index[MR_STRING_MAX] = {};
        sprintf(graph_index, "graph-%u", graph_id);
        cJSON_AddItemToObject(json_root, graph_index, graph_obj);
    }

    cJSON_AddNumberToObject(json_root, "total_graph_num", nr_graph_total);
    return json_root;
}

cJSON * shmdev_rx_node_monit_loop(struct sc_main * sc)
{
    unsigned int nr_graph_total = sc->nr_io_thread;
    cJSON * json_root = cJSON_CreateObject();

    for (uint32_t graph_id = 0; graph_id < nr_graph_total; graph_id++)
    {
        struct shmdev_stat_per_lcore * stat_item = &shmdev_graph_stat[graph_id];
        if (stat_item->total_rx_pkts == 0)
            continue;

        cJSON * graph_obj = cJSON_CreateObject();
        cJSON_AddNumberToObject(graph_obj, "total_rx_pkts", stat_item->total_rx_pkts);
        cJSON_AddNumberToObject(graph_obj, "original_pkt", stat_item->original_pkt);
        cJSON_AddNumberToObject(graph_obj, "nf_create_pkt", stat_item->nf_create_pkt);
        cJSON_AddNumberToObject(graph_obj, "to_eth_ingress_pkt", stat_item->to_eth_ingress_pkt);

        char graph_index[MR_STRING_MAX] = {};
        sprintf(graph_index, "graph-%u", graph_id);
        cJSON_AddItemToObject(json_root, graph_index, graph_obj);
    }

    cJSON_AddNumberToObject(json_root, "total_graph_num", nr_graph_total);
    return json_root;
}