path: root/service/src/node_vwire.c

#include "sc_trace.h"
#include <cJSON.h>
#include <rte_branch_prediction.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 <link_db.h>
#include <sc_node.h>
#include <sc_node_common.h>

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

enum
{
    VWIRE_INGRESS_NEXT_NODE_CLASSIFIER,
    VWIRE_INGRESS_NEXT_NODE_PKT_DROP,
    VWIRE_INGRESS_NEXT_NODE_MAX
};

enum
{
    VWIRE_EGRESS_NEXT_NODE_ETH_EGRESS,
    VWIRE_EGRESS_NEXT_NODE_PKT_DROP,
    VWIRE_EGRESS_NEXT_NODE_MAX
};

struct vwire_forward_rule
{
    struct mr_dev_desc * int_dev;
    struct mr_dev_desc * ext_dev;
    port_id_t int_dev_port_id;
    port_id_t ext_dev_port_id;
    unsigned int vwire_id;
};

struct vwire_map_result
{
    struct vwire_forward_rule * rule_object;
    port_id_t egress_port_id;
    unsigned int is_ext_to_int_dir;
};

/* Vwire ingress stats struct */
struct vwire_ingress_stats
{
    volatile uint64_t total_pkts;
    volatile uint64_t pkts_per_batch;
    volatile uint64_t drop_pkts;
    volatile uint64_t to_classifier;
} __rte_cache_aligned;

/* Vwire egress stats struct */
struct vwire_egress_stats
{
    volatile uint64_t total_pkts;
    volatile uint64_t pkts_per_batch;
    volatile uint64_t drop_pkts;
} __rte_cache_aligned;

struct vwire_node_main
{
    struct vwire_map_result vwire_map[MR_DEVICE_MAX];
    struct vwire_forward_rule ** forward_rules;
    struct link_db_ctx * link_db_ctx;
    unsigned int nr_forward_rules;
    uint32_t nr_sid;
    uint32_t sid_start;
    uint32_t sid_end;
};

static struct vwire_node_main * p_vwire_node_main = NULL;
static unsigned int nr_max_vwires = 256;
static struct vwire_ingress_stats ingress_stats_per_graph[RTE_MAX_LCORE] = {};
static struct vwire_egress_stats egress_stats_per_graph[RTE_MAX_LCORE] = {};

int olp_set_used(char * dev_sym, uint32_t channel_id);

/* Get max vwires num */
unsigned int nr_max_vwires_get()
{
    return nr_max_vwires;
}

/* Get max vwires start sid */
uint32_t vwire_sid_start_get()
{
    return p_vwire_node_main->sid_start;
}

/* Vwire service id check */
int vwire_id_check(uint32_t vwire_id)
{
    if (vwire_id >= nr_max_vwires)
        return RT_ERR;

    struct vwire_forward_rule * forward_rule = p_vwire_node_main->forward_rules[vwire_id];

    if (forward_rule == NULL)
        return RT_ERR;

    if (forward_rule->vwire_id != vwire_id)
        return RT_ERR;

    return RT_SUCCESS;
}

/* Generate and store the trace information for ingress */
static __rte_always_inline void gen_store_trace_info_ingress(struct rte_node * node, struct rte_mbuf * mbuf,
                                                             uint16_t next_node_index, uint16_t prepend_sid)
{
    /* Populate the next node infomation */
    char str_record[MR_STRING_MAX];
    struct mrb_metadata * mrb_meta = mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    int len = snprintf(str_record, sizeof(str_record), "next node:%s", node->nodes[next_node_index]->name);

    /* Populate the reason for next node */
    if (next_node_index == VWIRE_INGRESS_NEXT_NODE_PKT_DROP)
    {
        len += snprintf(str_record + len, sizeof(str_record) - len, ", rsn:match failed");
    }
    else
    {
        /* Populate the vwire id */
        len += snprintf(str_record + len, sizeof(str_record) - len, ", vwire id:%u, trf lk id:%u, prep sid:%u",
                        mrb_meta->adapter_id, mrb_meta->traffic_link_id, prepend_sid);

        /* Populate the sids information */
        len += embed_sid_info(mbuf, str_record + len, sizeof(str_record) - len);
    }

    /* Emit the trace record */
    struct dp_trace_record_meta meta = {
        .measurement_type = DP_TRACE_MEASUREMENT_TYPE_TRACE, .appsym = MR_TRACE_APPSYM, .module = node->name};
    dp_trace_record_emit_str(sc_main_get()->trace, mbuf, rte_lcore_id(), &meta, str_record);
}

static __rte_always_inline uint16_t vwire_ingress_node_process(struct rte_graph * graph, struct rte_node * node,
                                                               void ** objs, uint16_t cnt)
{
    /* Get stream for the speculated next node */
    unsigned int next_index = VWIRE_INGRESS_NEXT_NODE_CLASSIFIER;
    void ** to_next = rte_node_next_stream_get(graph, node, next_index, cnt);
    struct vwire_ingress_stats stats = {};

    struct rte_mbuf ** mbufs = (struct rte_mbuf **)objs;
    void ** from = (void **)objs;

    unsigned int n_left_from = cnt;
    unsigned int last_spec = 0;
    unsigned int held = 0;

    while (n_left_from > 0)
    {
        struct rte_mbuf * mbuf0 = mbufs[0];
        struct mrb_metadata * priv_data = mrbuf_cz_data(mbuf0, MR_NODE_CTRLZONE_ID);
        unsigned int next0 = 0;
        uint16_t prepend_sid = 0;

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

        mbufs += 1;
        n_left_from -= 1;

        struct vwire_map_result * map_result = &p_vwire_node_main->vwire_map[priv_data->port_ingress];
        if (likely(map_result->rule_object != NULL))
        {
            /* this packet will go to classifier node */
            next0 = VWIRE_INGRESS_NEXT_NODE_CLASSIFIER;
            stats.to_classifier++;

            /* Insert sid*/
            prepend_sid = p_vwire_node_main->sid_start + map_result->rule_object->vwire_id;
            sid_list_prepend(&priv_data->sid_list, &prepend_sid, 1);

            /* Traffic link id lookup */
            uint16_t result;
            struct link_db_match_field match_field;
            match_field.vwire_id = map_result->rule_object->vwire_id;
            link_db_match(p_vwire_node_main->link_db_ctx, &match_field, 1, &result);

            /* Save traffic link id */
            priv_data->traffic_link_id = result;

            /* Save vwire id */
            priv_data->adapter_type = ADAPTER_TYPE_VWIRE;
            priv_data->adapter_id = map_result->rule_object->vwire_id;

            /* Save dir */
            priv_data->dir = map_result->is_ext_to_int_dir;
        }
        else
        {
            /* nothing to map, drop this packet */
            next0 = VWIRE_INGRESS_NEXT_NODE_PKT_DROP;
            stats.drop_pkts++;
        }

        /* Check if tracing is enabled for the current Mbuf */
        if (unlikely(dp_trace_record_can_emit(mbuf0, DP_TRACE_MEASUREMENT_TYPE_TRACE)))
        {
            gen_store_trace_info_ingress(node, mbuf0, next0, prepend_sid);
            // gen_store_trace_info_sid_list(node, mbuf0);
        }

        if (unlikely(dp_trace_record_can_emit(mbuf0, DP_TRACE_MEASUREMENT_TYPE_TELEMETRY)))
        {
            gen_store_telemetry_info_adapter(mbuf0);
        }

        if (unlikely(next_index ^ next0))
        {
            /* Copy things successfully speculated till now */
            rte_memcpy(to_next, from, last_spec * sizeof(from[0]));
            from += last_spec;
            to_next += last_spec;
            held += last_spec;
            last_spec = 0;

            rte_node_enqueue_x1(graph, node, next0, from[0]);
            from += 1;
        }
        else
        {
            last_spec += 1;
        }
    }

    /* Update graph stats */
    struct vwire_ingress_stats * graph_stats = &ingress_stats_per_graph[graph->id];
    graph_stats->total_pkts += cnt;
    graph_stats->pkts_per_batch = cnt;
    graph_stats->drop_pkts += stats.drop_pkts;
    graph_stats->to_classifier += stats.to_classifier;

    /* home run */
    if (likely(last_spec == cnt))
    {
        rte_node_next_stream_move(graph, node, next_index);
        return cnt;
    }

    held += last_spec;
    rte_memcpy(to_next, from, last_spec * sizeof(from[0]));
    rte_node_next_stream_put(graph, node, next_index, held);
    return 0;
}

/* Generate and store the trace information for egress */
static __rte_always_inline void gen_store_trace_info_egress(struct rte_node * node, struct rte_mbuf * mbuf,
                                                            uint16_t next_node_index)
{
    /* Populate the next node infomation */
    char str_record[MR_STRING_MAX];
    struct mrb_metadata * mrb_meta = mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    int len = snprintf(str_record, sizeof(str_record), "next node:%s", node->nodes[next_node_index]->name);

    /* Populate the drop reason */
    if (next_node_index == VWIRE_EGRESS_NEXT_NODE_PKT_DROP)
    {
        len += snprintf(str_record + len, sizeof(str_record) - len, ", rsn:null device");
    }
    else
    {
        /* Populate the vwire id */
        len += snprintf(str_record + len, sizeof(str_record) - len, ", vwire id:%u, tx:%u", mrb_meta->adapter_id,
                        mrb_meta->port_egress);
    }

    /* Emit the trace record */
    struct dp_trace_record_meta meta = {
        .measurement_type = DP_TRACE_MEASUREMENT_TYPE_TRACE, .appsym = MR_TRACE_APPSYM, .module = node->name};
    dp_trace_record_emit_str(sc_main_get()->trace, mbuf, rte_lcore_id(), &meta, str_record);
}

static __rte_always_inline uint16_t vwire_egress_node_process(struct rte_graph * graph, struct rte_node * node,
                                                              void ** objs, uint16_t cnt)
{
    /* Get stream for the speculated next node */
    unsigned int next_index = VWIRE_EGRESS_NEXT_NODE_PKT_DROP;
    void ** to_next = rte_node_next_stream_get(graph, node, next_index, cnt);
    struct vwire_egress_stats stats = {};

    struct rte_mbuf ** mbufs = (struct rte_mbuf **)objs;
    void ** from = (void **)objs;

    unsigned int n_left_from = cnt;
    unsigned int last_spec = 0;
    unsigned int held = 0;

    while (n_left_from > 0)
    {
        struct rte_mbuf * mbuf0 = mbufs[0];
        struct mrb_metadata * priv_data = mrbuf_cz_data(mbuf0, MR_NODE_CTRLZONE_ID);
        unsigned int next0 = 0;

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

        mbufs += 1;
        n_left_from -= 1;

        /* Get vwire item */
        unsigned int vwire_id = priv_data->cur_sid - p_vwire_node_main->sid_start;
        assert(vwire_id < nr_max_vwires);

        struct vwire_forward_rule * vwire_fwd_rule = p_vwire_node_main->forward_rules[vwire_id];
        assert(vwire_fwd_rule != NULL);

        if (priv_data->dir == 0)
        {
            /* Internal to External */
            priv_data->port_egress = vwire_fwd_rule->ext_dev_port_id;
        }
        else
        {
            /* External to Internal */
            priv_data->port_egress = vwire_fwd_rule->int_dev_port_id;
        }

        /* null device */
        if (vwire_fwd_rule->int_dev == NULL || vwire_fwd_rule->ext_dev == NULL)
        {
            /* nothing to map, drop this packet */
            next0 = VWIRE_EGRESS_NEXT_NODE_PKT_DROP;
            stats.drop_pkts++;
        }
        else
        {
            /* goto eth_egress node */
            next0 = VWIRE_EGRESS_NEXT_NODE_ETH_EGRESS;
        }

        /* Check if tracing is enabled for the current Mbuf */
        if (unlikely(dp_trace_record_can_emit(mbuf0, DP_TRACE_MEASUREMENT_TYPE_TRACE)))
        {
            gen_store_trace_info_egress(node, mbuf0, next0);
        }

        if (unlikely(next_index ^ next0))
        {
            /* Copy things successfully speculated till now */
            rte_memcpy(to_next, from, last_spec * sizeof(from[0]));
            from += last_spec;
            to_next += last_spec;
            held += last_spec;
            last_spec = 0;

            rte_node_enqueue_x1(graph, node, next0, from[0]);
            from += 1;
        }
        else
        {
            last_spec += 1;
        }
    }

    /* Update graph stats */
    struct vwire_egress_stats * graph_stats = &egress_stats_per_graph[graph->id];
    graph_stats->total_pkts += cnt;
    graph_stats->pkts_per_batch = cnt;
    graph_stats->drop_pkts += stats.drop_pkts;

    /* home run */
    if (likely(last_spec == cnt))
    {
        rte_node_next_stream_move(graph, node, next_index);
        return cnt;
    }

    held += last_spec;
    rte_memcpy(to_next, from, last_spec * sizeof(from[0]));
    rte_node_next_stream_put(graph, node, next_index, held);

    return cnt;
}

static struct rte_node_register vwire_ingress_node_base = {
    .process = vwire_ingress_node_process,
    .name = "vwire_ingress",
    .init = NULL,
    .nb_edges = VWIRE_INGRESS_NEXT_NODE_MAX,
    .next_nodes =
        {
            [VWIRE_INGRESS_NEXT_NODE_CLASSIFIER] = "classifier",
            [VWIRE_INGRESS_NEXT_NODE_PKT_DROP] = "pkt_drop_trap",
        },
};

static struct rte_node_register vwire_egress_node_base = {
    .process = vwire_egress_node_process,
    .name = "vwire_egress",
    .init = NULL,
    .nb_edges = VWIRE_EGRESS_NEXT_NODE_MAX,
    .next_nodes =
        {
            [VWIRE_EGRESS_NEXT_NODE_ETH_EGRESS] = "eth_egress",
            [VWIRE_EGRESS_NEXT_NODE_PKT_DROP] = "pkt_drop_trap",
        },
};

RTE_NODE_REGISTER(vwire_ingress_node_base);
RTE_NODE_REGISTER(vwire_egress_node_base);

static int vwire_forward_rule_add__DEV_TYPE(struct vwire_node_main * vwire_main, struct sc_main * sc,
                                            const char * int_dev, const char * ext_dev, unsigned int vwire_id)
{
    struct vwire_forward_rule * fwd_rule = ZMALLOC(sizeof(struct vwire_forward_rule));
    MR_VERIFY_MALLOC(fwd_rule);

    /* internal device/tag */
    struct mr_dev_desc * int_dev_desc = NULL;
    struct mr_dev_desc * ext_dev_desc = NULL;

    if (strcmp(int_dev, "null") == 0)
    {
        int_dev_desc = NULL;
    }
    else
    {
        int_dev_desc = mr_dev_desc_lookup(sc->devmgr_main, int_dev);
        if (int_dev_desc != NULL)
        {
            fwd_rule->int_dev = int_dev_desc;
            fwd_rule->int_dev_port_id = int_dev_desc->port_id;
        }
        else
        {
            MR_ERROR("Internal Device %s is not existed, cannot create virtual-wire rule.", int_dev);
            goto errout;
        }
    }

    if (strcmp(ext_dev, "null") == 0)
    {
        ext_dev_desc = NULL;
    }
    else
    {
        ext_dev_desc = mr_dev_desc_lookup(sc->devmgr_main, ext_dev);
        if (ext_dev_desc != NULL)
        {
            fwd_rule->ext_dev = ext_dev_desc;
            fwd_rule->ext_dev_port_id = ext_dev_desc->port_id;
        }
        else
        {
            MR_ERROR("External Device %s is not existed, cannot create virtual-wire rule.", ext_dev);
            goto errout;
        }
    }

    /* construct the forward lookup map, internal->external */
    struct vwire_map_result * int_to_ext_map_result = NULL;
    struct vwire_map_result * ext_to_int_map_result = NULL;

    if (fwd_rule->int_dev != NULL)
    {
        int_to_ext_map_result = &vwire_main->vwire_map[fwd_rule->int_dev_port_id];
        if (int_to_ext_map_result->rule_object != NULL)
        {
            MR_ERROR("Internal Device %s has already be a member of virtual-wire group, rejected.", int_dev);
            goto errout;
        }
    }

    if (fwd_rule->ext_dev != NULL)
    {
        ext_to_int_map_result = &vwire_main->vwire_map[fwd_rule->ext_dev_port_id];
        if (ext_to_int_map_result->rule_object != NULL)
        {
            MR_ERROR("External Device %s has already be a member of virtual-wire group, rejected.", ext_dev);
            goto errout;
        }
    }

    if (int_to_ext_map_result != NULL)
    {
        int_to_ext_map_result->rule_object = fwd_rule;
        int_to_ext_map_result->is_ext_to_int_dir = 0;
        int_to_ext_map_result->egress_port_id = fwd_rule->ext_dev_port_id;
    }

    if (ext_to_int_map_result != NULL)
    {
        ext_to_int_map_result->rule_object = fwd_rule;
        ext_to_int_map_result->egress_port_id = fwd_rule->int_dev_port_id;
        ext_to_int_map_result->is_ext_to_int_dir = 1;
    }

    vwire_main->nr_forward_rules++;
    fwd_rule->vwire_id = vwire_id;
    vwire_main->forward_rules[vwire_id] = fwd_rule;

    MR_INFO("Virtual Wire: ID=%d, internal_dev=%s, external_dev=%s created.", fwd_rule->vwire_id, int_dev, ext_dev);
    return RT_SUCCESS;

errout:
    if (fwd_rule != NULL)
    {
        FREE(fwd_rule);
    }

    return RT_ERR;
}

int vwire_init(struct sc_main * sc)
{
    struct vwire_node_main * vwire_node_main = ZMALLOC(sizeof(struct vwire_node_main));
    MR_VERIFY_MALLOC(vwire_node_main);

    /* Get ef max entry,default is 256 */
    MESA_load_profile_uint_def(sc->local_cfgfile, "limits", "nr_max_vwires", &nr_max_vwires, 256);

    vwire_node_main->forward_rules = ZMALLOC(sizeof(struct vwire_forward_rule *) * nr_max_vwires);
    MR_VERIFY_MALLOC(vwire_node_main->forward_rules);

    p_vwire_node_main = vwire_node_main;

    /* Get sid range start */
    uint32_t sid_start;
    MESA_load_profile_uint_def(sc->local_cfgfile, "vwires", "sid_start", &sid_start, 300);

    /* Get sid range end */
    uint32_t sid_end;
    MESA_load_profile_uint_def(sc->local_cfgfile, "vwires", "sid_end", &sid_end, 400);

    /* Check sid range */
    if (sid_start > sid_end)
    {
        MR_ERROR("Service Vwire 'sid_end' less than 'sid_start' .");
        return RT_ERR;
    }

    /* load vwire rules */
    int ret = 0;
    for (unsigned int i = 0; i < nr_max_vwires; i++)
    {
        char sec_symbol[MR_SYMBOL_MAX] = {};
        snprintf(sec_symbol, sizeof(sec_symbol) - 1, "vwire:%u", i);

        char str_int_dev[MR_SYMBOL_MAX] = {};
        ret = MESA_load_profile_string_nodef(sc->local_cfgfile, sec_symbol, "interface_int", str_int_dev,
                                             sizeof(str_int_dev) - 1);

        if (ret < 0)
        {
            continue;
        }

        char str_ext_dev[MR_SYMBOL_MAX] = {};
        ret = MESA_load_profile_string_nodef(sc->local_cfgfile, sec_symbol, "interface_ext", str_ext_dev,
                                             sizeof(str_ext_dev) - 1);

        if (ret < 0)
        {
            MR_DEBUG("interface_ext is not existed in section %s, ignore this section", sec_symbol);
            continue;
        }

        unsigned int vwire_id;
        ret = MESA_load_profile_uint_nodef(sc->local_cfgfile, sec_symbol, "vwire_id", &vwire_id);
        if (ret < 0)
        {
            MR_DEBUG("vwire_id is not existed in section %s, ignore this section", sec_symbol);
            continue;
        }

        if (vwire_id >= nr_max_vwires)
        {
            MR_ERROR("Service Vwire id:%u out of range:%u .", vwire_id, nr_max_vwires);
            return RT_ERR;
        }

        ret = vwire_forward_rule_add__DEV_TYPE(p_vwire_node_main, sc, str_int_dev, str_ext_dev, vwire_id);
        if (ret < 0)
        {
            return RT_ERR;
        }

        char str_obp_dev[MR_SYMBOL_MAX] = {};
        ret = MESA_load_profile_string_nodef(sc->local_cfgfile, sec_symbol, "obp_device", str_obp_dev,
                                             sizeof(str_obp_dev) - 1);
        if (ret > 0)
        {
            uint32_t channel_id = 0;
            ret = MESA_load_profile_uint_nodef(sc->local_cfgfile, sec_symbol, "obp_segment", &channel_id);
            if (ret >= 0)
            {
                olp_set_used(str_obp_dev, channel_id);
            }
        }
    }

    /* Check group num */
    uint32_t nr_sid = sid_end - sid_start + 1;
    if (p_vwire_node_main->nr_forward_rules > nr_sid)
    {
        MR_ERROR("Service Vwire rules num:%u out of sid num: %u .", p_vwire_node_main->nr_forward_rules, nr_sid);
        return RT_ERR;
    }

    /* inserter sid to forwarder table */
    for (int i = 0; i < nr_max_vwires; i++)
    {
        struct vwire_forward_rule * fwd_rule = p_vwire_node_main->forward_rules[i];
        if (fwd_rule == NULL)
            continue;

        uint16_t sid = sid_start + fwd_rule->vwire_id;
        forwarder_table_insert(sid, FORWARDER_TYPE_VWIRE);
    }

    /* Save sid info */
    p_vwire_node_main->nr_sid = nr_sid;
    p_vwire_node_main->sid_start = sid_start;
    p_vwire_node_main->sid_end = sid_end;

    /* Link db ctx create */
    uint32_t max_entries;
    MESA_load_profile_uint_def(sc->local_cfgfile, "limits", "nr_max_link_dbs", &max_entries, 64);
    p_vwire_node_main->link_db_ctx = link_db_create(LINK_DB_TYPE_VWIRE, max_entries);

    if (p_vwire_node_main->link_db_ctx == NULL)
    {
        return RT_ERR;
    }

    ret = link_db_config_parse(sc->local_cfgfile, p_vwire_node_main->link_db_ctx);
    if (ret < 0)
    {
        return RT_ERR;
    }

    return 0;
}

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

    uint64_t total_pkts[nr_graphs];
    uint64_t pkts_per_batch[nr_graphs];
    uint64_t drop_pkts[nr_graphs];
    uint64_t to_classifier[nr_graphs];

    for (uint32_t graph_id = 0; graph_id < nr_graphs; graph_id++)
    {
        struct vwire_ingress_stats * stats = &ingress_stats_per_graph[graph_id];
        if (stats->total_pkts == 0)
        {
            total_pkts[graph_id] = 0;
            pkts_per_batch[graph_id] = 0;
            drop_pkts[graph_id] = 0;
            to_classifier[graph_id] = 0;
            continue;
        }

        total_pkts[graph_id] = stats->total_pkts;
        pkts_per_batch[graph_id] = stats->pkts_per_batch;
        drop_pkts[graph_id] = stats->drop_pkts;
        to_classifier[graph_id] = stats->to_classifier;
    }

    cJSON * json_total_pkts = create_uint64_array(total_pkts, nr_graphs);
    cJSON_AddItemToObject(json_root, "vwire ingress, total_pkts", json_total_pkts);

    cJSON * json_pkts_per_batch = create_uint64_array(pkts_per_batch, nr_graphs);
    cJSON_AddItemToObject(json_root, "vwire ingress, pkts_per_batch", json_pkts_per_batch);

    cJSON * json_to_classifier = create_uint64_array(to_classifier, nr_graphs);
    cJSON_AddItemToObject(json_root, "vwire ingress, to_classifier", json_to_classifier);

    cJSON * json_drop_pkts = create_uint64_array(drop_pkts, nr_graphs);
    cJSON_AddItemToObject(json_root, "vwire ingress, drop_pkts", json_drop_pkts);

    return json_root;
}

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

    uint64_t total_pkts[nr_graphs];
    uint64_t pkts_per_batch[nr_graphs];
    uint64_t drop_pkts[nr_graphs];

    for (uint32_t graph_id = 0; graph_id < nr_graphs; graph_id++)
    {
        struct vwire_egress_stats * stats = &egress_stats_per_graph[graph_id];
        if (stats->total_pkts == 0)
        {
            total_pkts[graph_id] = 0;
            pkts_per_batch[graph_id] = 0;
            drop_pkts[graph_id] = 0;

            continue;
        }

        total_pkts[graph_id] = stats->total_pkts;
        pkts_per_batch[graph_id] = stats->pkts_per_batch;
        drop_pkts[graph_id] = stats->drop_pkts;
    }

    cJSON * json_total_pkts = create_uint64_array(total_pkts, nr_graphs);
    cJSON_AddItemToObject(json_root, "vwire egress, total_pkts", json_total_pkts);

    cJSON * json_pkts_per_batch = create_uint64_array(pkts_per_batch, nr_graphs);
    cJSON_AddItemToObject(json_root, "vwire egress, pkts_per_batch", json_pkts_per_batch);

    cJSON * json_drop_pkts = create_uint64_array(drop_pkts, nr_graphs);
    cJSON_AddItemToObject(json_root, "vwire egress, drop_pkts", json_drop_pkts);

    return json_root;
}