path: root/service/src/node_bfd.c

#include <bits/stdint-uintn.h>
#include <cJSON.h>
#include <common.h>
#include <ldbc.h>
#include <mrb_define.h>
#include <netinet/in.h>
#include <rte_graph.h>
#include <rte_graph_worker.h>
#include <sc_common.h>
#include <sc_node.h>
#include <sc_node_common.h>
#include <sc_vdev.h>
#include <sys/socket.h>

/* Global Config */
#define MR_BFD_MAX_SESSION_NUM 128
#define MR_BFD_START_DISCRIMINATOR 1000
#define MR_BFD_MAX_SESSION_BUF_NUM MR_BFD_MAX_SESSION_NUM + 1
#define MR_BFD_MAX_SESSION_ID 1023
#define MR_BFD_INVALID_SESSION_ID MR_BFD_MAX_SESSION_NUM
#define TIMEOUT_CHECK_US 500000

#define MR_BFD_STAT_ADD_FOR_TOTAL(st, gid, counter, value)                                                             \
    do                                                                                                                 \
    {                                                                                                                  \
        st->graph_stat[gid].counter += value;                                                                          \
    } while (0)

#define MR_BFD_STAT_ADD_FOR_SESSION_ID(st, gid, sessionid, counter, value)                                             \
    do                                                                                                                 \
    {                                                                                                                  \
        st->graph_stat[gid].stat_for_session[sessionid].counter += value;                                              \
    } while (0)

#define MR_BFD_STAT_UPDATE_TSC_FOR_SESSION_ID(st, gid, sessionid, counter, value)                                      \
    do                                                                                                                 \
    {                                                                                                                  \
        st->graph_stat[gid].stat_for_session[sessionid].counter = value;                                               \
    } while (0)

/* Bfd Next Node */
enum
{
    BFD_NEXT_PKT_DROP = 0,
    BFD_NEXT_ETH_EGRESS,
    BFD_NEXT_MAX
};

/* Bfd State */
enum
{
    BFD_STATE_ADMINDOWN = 0,
    BFD_STATE_DOWN,
    BFD_STATE_INIT,
    BFD_STATE_UP
};

struct bfd_header_t
{
    uint8_t version_diag;
#if __BYTE_ORDER == __LITTLE_ENDIAN
    uint8_t flags : 6;
    uint8_t state : 2;
#elif __BYTE_ORDER == __BIG_ENDIAN
    uint8_t state : 2;
    uint8_t flags : 6;
#endif

    uint8_t detect_time_multiplier;
    uint8_t length;
    uint8_t my_discriminator[4];
    uint8_t your_discriminator[4];
    uint8_t desired_min_tx_interval[4];
    uint8_t required_min_rx_interval[4];
    uint8_t required_min_echo_interval[4];
};

/* Bfd Stat Field */
struct bfd_stat_for_session
{
    volatile uint64_t rx_pkts;
    volatile uint64_t prev_tsc;
    uint32_t etherfabric_ip;
} __rte_cache_aligned;

/* Bfd Stat Struct */
struct bfd_stat
{
    volatile uint64_t deal_pkts;
    volatile uint64_t drop_for_unexpected_pkts;
    volatile uint64_t drop_for_no_free_session_id;
    struct bfd_stat_for_session stat_for_session[MR_BFD_MAX_SESSION_NUM];
} __rte_cache_aligned;

/* Bfd Main Struct */
struct node_bfd_main
{
    struct bfd_stat graph_stat[RTE_MAX_LCORE];
};

/* Global Bfd Main */
static struct node_bfd_main * global_bfd_main = NULL;

/*************************************   Bfd Config    **************************************/
/* Init Bfd Servicd */
int bfd_init(struct sc_main * sc)
{
    struct node_bfd_main * bfd_main = NULL;

    bfd_main = ZMALLOC(sizeof(struct node_bfd_main));
    MR_VERIFY_MALLOC(bfd_main);
    sc->bfd_node_main = bfd_main;
    global_bfd_main = bfd_main;
    return RT_SUCCESS;
}

/************************************** Bfd Node **************************************/
/* Get Free Session Id */
uint16_t get_free_session_id(struct bfd_stat * graph_stat, struct rte_ipv4_hdr * ipv4_hdr)
{
    for (int i = 0; i < MR_BFD_MAX_SESSION_NUM; i++)
    {
        if (graph_stat->stat_for_session[i].etherfabric_ip != 0)
            continue;
        graph_stat->stat_for_session[i].etherfabric_ip = ipv4_hdr->src_addr;
        return i;
    }
    return MR_BFD_INVALID_SESSION_ID;
}

/* Get Bfd Session Id */
uint16_t get_bfd_session_id(struct bfd_stat * graph_stat, struct rte_ipv4_hdr * ipv4_hdr)
{
    for (int i = 0; i < MR_BFD_MAX_SESSION_NUM; i++)
    {
        if (graph_stat->stat_for_session[i].etherfabric_ip == ipv4_hdr->src_addr)
            return i;
    }

    return get_free_session_id(graph_stat, ipv4_hdr);
}

/* Reply Bfd Request */
uint8_t bfd_reply(struct rte_ether_hdr * ether_hdr, struct rte_ipv4_hdr * ipv4_hdr, struct rte_udp_hdr * udp_hdr,
                  uint16_t bfd_session_id)
{
    /* Swap ether_hdr */
    struct rte_ether_addr swap_addr;
    rte_ether_addr_copy(&ether_hdr->dst_addr, &swap_addr);
    rte_ether_addr_copy(&ether_hdr->src_addr, &ether_hdr->dst_addr);
    rte_ether_addr_copy(&swap_addr, &ether_hdr->src_addr);

    /* Swap ipv4_hdr */
    uint32_t swap_ip = ipv4_hdr->dst_addr;
    ipv4_hdr->dst_addr = ipv4_hdr->src_addr;
    ipv4_hdr->src_addr = swap_ip;

    /* Swap discriminator */
    struct bfd_header_t * bfd_hdr = (struct bfd_header_t *)(udp_hdr + 1);
    uint8_t swap_discriminator[4];
    uint32_t discriminator = htonl(MR_BFD_START_DISCRIMINATOR + bfd_session_id);
    memcpy(swap_discriminator, bfd_hdr->my_discriminator, sizeof(swap_discriminator));
    memcpy(bfd_hdr->my_discriminator, &discriminator, sizeof(swap_discriminator));
    memcpy(bfd_hdr->your_discriminator, swap_discriminator, sizeof(swap_discriminator));

    /* Set udp check sum */
    udp_hdr->dgram_cksum = 0;

    /* Set Bfd State */
    uint8_t bfd_state = bfd_hdr->state;
    if (bfd_state == BFD_STATE_DOWN)
    {
        bfd_hdr->state = BFD_STATE_INIT;
        return bfd_state;
    }
    else if (bfd_state == BFD_STATE_INIT)
    {
        bfd_hdr->state = BFD_STATE_UP;
        return bfd_state;
    }
    bfd_hdr->state = BFD_STATE_UP;
    return bfd_state;
}

/* Bfd Node Init Function */
static int bfd_node_init(const struct rte_graph * graph, struct rte_node * node)
{
    return 0;
}

/* Bfd Node Process Function */
static __rte_always_inline uint16_t bfd_node_process(struct rte_graph * graph, struct rte_node * node, void ** objs,
                                                     uint16_t cnt)
{
    uint16_t last_spec = 0, next_node_index = 0;
    uint16_t n_left_from = cnt;
    struct rte_mbuf * mbuf0;
    struct rte_mbuf ** pkts = (struct rte_mbuf **)objs;
    void ** batch_pkts = objs;
    uint16_t batch_next_node_index = BFD_NEXT_ETH_EGRESS;
    struct node_bfd_main * bfd_main = global_bfd_main;
    rte_graph_t gid = graph->id;
    uint64_t drop_for_unexpected_pkts = 0;
    uint64_t drop_for_no_free_session_id = 0;

    /* Single Packet Processing */
    while (n_left_from > 0)
    {
        mbuf0 = pkts[0];
        pkts += 1;
        n_left_from -= 1;

        /* Get Private Data */
        struct mrb_metadata * private_ctrlzone = mrbuf_cz_data(mbuf0, MR_NODE_CTRLZONE_ID);
        struct pkt_parser_result * pkt_parser_result = &private_ctrlzone->pkt_parser_result;
        uint16_t ingress_port_id = private_ctrlzone->port_ingress;

        static const uint16_t expect_layers[] = {
            LAYER_TYPE_ID_ETHER,
            LAYER_TYPE_ID_IPV4,
            LAYER_TYPE_ID_UDP,
        };

        int expect_result = complex_layer_type_expect(pkt_parser_result, expect_layers, RTE_DIM(expect_layers));
        if (unlikely(expect_result < 0))
        {
            drop_for_unexpected_pkts++;
            next_node_index = BFD_NEXT_PKT_DROP;
            goto node_enqueue;
        }

        struct rte_ether_hdr * ether_hdr = rte_pktmbuf_mtod(mbuf0, struct rte_ether_hdr *);
        struct rte_ipv4_hdr * ipv4_hdr =
            complex_layer_jump_to_outermost(pkt_parser_result, rte_pktmbuf_mtod(mbuf0, void *), LAYER_TYPE_ID_IPV4);
        struct rte_udp_hdr * udp_hdr =
            complex_layer_jump_to_outermost(pkt_parser_result, rte_pktmbuf_mtod(mbuf0, void *), LAYER_TYPE_ID_UDP);

        assert(ipv4_hdr != NULL);
        assert(udp_hdr != NULL);

        /* Get Session Id */
        uint16_t bfd_session_id = get_bfd_session_id(&bfd_main->graph_stat[gid], ipv4_hdr);
        if (unlikely(bfd_session_id == MR_BFD_INVALID_SESSION_ID))
        {
            drop_for_no_free_session_id++;
            next_node_index = BFD_NEXT_PKT_DROP;
            goto node_enqueue;
        }

        /* Reply Bfd Request */
        bfd_reply(ether_hdr, ipv4_hdr, udp_hdr, bfd_session_id);
        MR_BFD_STAT_ADD_FOR_SESSION_ID(bfd_main, gid, bfd_session_id, rx_pkts, 1);

        /* Update prev_tsc */
        MR_BFD_STAT_UPDATE_TSC_FOR_SESSION_ID(bfd_main, gid, bfd_session_id, prev_tsc, rte_rdtsc());

        /* From Ingress Port Id Get Next Node Index (TX) */
        private_ctrlzone->port_egress = ingress_port_id;
        next_node_index = BFD_NEXT_ETH_EGRESS;

    node_enqueue:
        /* Judge The Next Index Whether To Change */
        if (unlikely(batch_next_node_index != next_node_index))
        {
            /* If The Next Index Has Been Changed,Enqueue Last Pkts */
            rte_node_enqueue(graph, node, batch_next_node_index, batch_pkts, last_spec);
            batch_pkts += last_spec;
            last_spec = 1;
            batch_next_node_index = next_node_index;
        }
        else
        {
            /* If The Next Index Not Change, Update The Lasts */
            last_spec++;
        }
    }

    /* Update graph stat */
    struct bfd_stat * graph_stat = &bfd_main->graph_stat[gid];
    graph_stat->deal_pkts += cnt;
    graph_stat->drop_for_unexpected_pkts += drop_for_unexpected_pkts;
    graph_stat->drop_for_no_free_session_id += drop_for_no_free_session_id;

    /* Process The Remaining Packets */
    if (likely(last_spec > 0))
        rte_node_enqueue(graph, node, batch_next_node_index, batch_pkts, last_spec);
    return cnt;
}

/* Bfd Node Base */
static struct rte_node_register bfd_node_base = {
    .process = bfd_node_process,
    .name = "bfd",
    .init = bfd_node_init,
    .nb_edges = BFD_NEXT_MAX,
    .next_nodes =
        {
            [BFD_NEXT_PKT_DROP] = "pkt_drop",
            [BFD_NEXT_ETH_EGRESS] = "eth_egress",
        },
};
RTE_NODE_REGISTER(bfd_node_base);

/************************************** Bfd Statistics **************************************/
cJSON * bfd_node_monit_loop(struct sc_main * sc)
{
    uint32_t session_num = 0;
    struct node_bfd_main * bfd_main = sc->bfd_node_main;
    unsigned int nr_graph_total = sc->nr_io_thread;
    const uint64_t drain_tsc = (rte_get_tsc_hz() + US_PER_S - 1) / US_PER_S * TIMEOUT_CHECK_US;
    cJSON * json_root = cJSON_CreateObject();

    for (uint32_t graph_id = 0; graph_id < nr_graph_total; graph_id++)
    {
        struct bfd_stat * stat_item = &bfd_main->graph_stat[graph_id];
        if (stat_item->deal_pkts == 0)
            continue;

        cJSON * graph_obj = cJSON_CreateObject();
        cJSON_AddNumberToObject(graph_obj, "deal_pkts", stat_item->deal_pkts);
        cJSON_AddNumberToObject(graph_obj, "drop_for_unexpected_pkts", stat_item->drop_for_unexpected_pkts);
        cJSON_AddNumberToObject(graph_obj, "drop_for_no_free_session_id", stat_item->drop_for_no_free_session_id);

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

    for (int session_index = 0; session_index < MR_BFD_MAX_SESSION_NUM; session_index++)
    {
        uint64_t max_tsc = 0;
        uint32_t latest_graph_id = 0;
        for (uint32_t graph_id = 0; graph_id < nr_graph_total; graph_id++)
        {
            struct bfd_stat * stat_item = &bfd_main->graph_stat[graph_id];
            if (stat_item->deal_pkts == 0)
                continue;

            struct bfd_stat_for_session * session_stat_item = &stat_item->stat_for_session[session_index];
            if (session_stat_item->rx_pkts == 0)
                continue;

            if (session_stat_item->prev_tsc > max_tsc)
            {
                max_tsc = session_stat_item->prev_tsc;
                latest_graph_id = graph_id;
            }
        }

        if (max_tsc != 0)
        {
            struct in_addr ip_addr;
            struct bfd_stat * stat_item = &bfd_main->graph_stat[latest_graph_id];
            struct bfd_stat_for_session * session_stat_item = &stat_item->stat_for_session[session_index];
            cJSON * session_obj = cJSON_CreateObject();

            cJSON_AddNumberToObject(session_obj, "bfd_session_id", session_index);
            ip_addr.s_addr = session_stat_item->etherfabric_ip;
            cJSON_AddStringToObject(session_obj, "device_ip", inet_ntoa(ip_addr));

            uint64_t prev_tsc = session_stat_item->prev_tsc;
            uint64_t cur_tsc = rte_rdtsc();
            uint64_t diff_tsc = cur_tsc - prev_tsc;
            if (unlikely(diff_tsc > drain_tsc))
                cJSON_AddStringToObject(session_obj, "device_status", "Down");
            else
                cJSON_AddStringToObject(session_obj, "device_status", "Up");

            char bfd_session_id[MR_STRING_MAX] = {};
            snprintf(bfd_session_id, sizeof(bfd_session_id), "bfd-session-%u", session_index);
            cJSON_AddItemToObject(json_root, bfd_session_id, session_obj);
            session_num++;
        }
    }

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