path: root/examples/l2fwd-nf.c


#include "mrapp.h"
#include <assert.h>
#include <marsio.h>
#include <pthread.h>
#include <rte_ether.h>
#include <sched.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

static char appsym[64] = "mrfwd";
static char dev_1_symbol[64] = "meth0";

cpu_set_t cpu_set_io;
unsigned int nr_thread;

struct mr_instance * mr_instance = NULL;
struct mr_vdev * dev_1_handler = NULL;
struct mr_sendpath * to_dev_1_sendpath = NULL;

#define BURST_MAX 64
unsigned int nr_burst = 32;

/* options */
unsigned int opt_dump_packet_metadata = 1;
unsigned int opt_inject_by_deep_copy = 0;

int str_uint_range_parse(const char * str, unsigned int * out, const size_t size);
static int mr_parse_corelist(const char * corelist, int * cores);

void dump_packet_metadata(marsio_buff_t * buff)
{
#if 0
    sid_t sids[MR_SID_LIST_MAXLEN];
    int nr_sids = marsio_buff_get_sid_list(buff, sids, MR_SID_LIST_MAXLEN);

    fprintf(stderr, "packet buf = %p, is_vlan_tci_set = %u, vlan_tci = %u\n", buff, is_vlan_tci_set, vlan_tci);
#endif
}

void * l2fwd_loop(void * arg)
{
    uintptr_t sid = (uintptr_t)arg;
    marsio_buff_t * rx_buff[BURST_MAX];
    marsio_buff_t * tx_buff[BURST_MAX];
    int ret = 0;

    marsio_thread_init(mr_instance);

    for (;;)
    {
        ret = marsio_recv_burst(dev_1_handler, sid, rx_buff, (int)nr_burst);
        if (ret == 0)
        {
            marsio_poll_wait(mr_instance, &dev_1_handler, 1, sid, -1);
        }

        for (int i = 0; i < ret; i++)
        {
            if (marsio_dp_trace_measurements_can_emit(mr_instance, rx_buff[i], DP_TRACE_MEASUREMENT_TYPE_TELEMETRY))
            {
                marsio_dp_trace_measurement_emit_str(mr_instance, rx_buff[i], DP_TRACE_MEASUREMENT_TYPE_TELEMETRY,
                                                     "I2fwd-nf", "test data path trace telemetry");
                marsio_dp_trace_measurement_emit_str(mr_instance, rx_buff[i], DP_TRACE_MEASUREMENT_TYPE_TELEMETRY,
                                                     "I2fwd-nf", "test data path trace telemetry 2");
            }

            if (marsio_dp_trace_measurements_can_emit(mr_instance, rx_buff[i], DP_TRACE_MEASUREMENT_TYPE_TRACE))
            {
                marsio_dp_trace_measurement_emit_fmt(mr_instance, rx_buff[i], DP_TRACE_MEASUREMENT_TYPE_TRACE,
                                                     "I2fwd-nf", "test format recod %s", "hello");
                marsio_dp_trace_measurement_emit_fmt(mr_instance, rx_buff[i], DP_TRACE_MEASUREMENT_TYPE_TRACE,
                                                     "I2fwd-nf", "test format recod %s", "hello 2");
            }

            if (opt_dump_packet_metadata)
            {
                dump_packet_metadata(rx_buff[i]);
            }

            if (opt_inject_by_deep_copy)
            {
                marsio_buff_t * orin_buff = rx_buff[i];
                marsio_buff_t * deep_copy_buff = NULL;
                marsio_buff_malloc_global(mr_instance, &deep_copy_buff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY);

                /* copy the packet body */
                unsigned int orin_pkt_len = marsio_buff_datalen(orin_buff);
                const char * orin_pkt_ptr = marsio_buff_mtod(orin_buff);

                char * deep_copy_ptr = marsio_buff_append(deep_copy_buff, orin_pkt_len);
                memcpy(deep_copy_ptr, orin_pkt_ptr, orin_pkt_len);

                /* get the route ctx */
                char route_ctx[64];
                marsio_buff_get_metadata(orin_buff, MR_BUFF_ROUTE_CTX, route_ctx, sizeof(route_ctx));

                /* get the sids */
                sid_t sids[MR_SID_LIST_MAXLEN];
                int nr_sids = marsio_buff_get_sid_list(orin_buff, sids, MR_SID_LIST_MAXLEN);

#if 0
                struct rte_ether_hdr * ether_hdr = (struct rte_ether_hdr *)orin_pkt_ptr;
                if(ether_hdr->ether_type != 0xAAAA)
                {
                    assert(nr_sids != 0);
                }
#endif

                /* set route ctx and sids for deep copy buff */
                marsio_buff_set_metadata(deep_copy_buff, MR_BUFF_ROUTE_CTX, route_ctx, sizeof(route_ctx));
                marsio_buff_set_sid_list(deep_copy_buff, sids, nr_sids);

                if (marsio_dp_trace_measurements_can_emit(mr_instance, deep_copy_buff, DP_TRACE_MEASUREMENT_TYPE_TRACE))
                {
                    marsio_dp_trace_measurement_emit_str(mr_instance, deep_copy_buff, DP_TRACE_MEASUREMENT_TYPE_TRACE,
                                                         "I2fwd-nf", "test for sending message directly");
                }

                tx_buff[i] = deep_copy_buff;
            }
        }

        if (opt_inject_by_deep_copy)
        {
            marsio_buff_free(mr_instance, rx_buff, ret, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY);
            marsio_send_burst(to_dev_1_sendpath, sid, tx_buff, (int)ret);
        }
        else
        {
            marsio_send_burst(to_dev_1_sendpath, sid, rx_buff, (int)ret);
        }
    }
}

int help()
{
    fprintf(stderr, "parameter error.\n");
    exit(EXIT_SUCCESS);
}

void signalHandler(int signum)
{
    printf("Received CTRL+C signal (SIGINT)\n");
    exit(0);
}

int main(int argc, char * argv[])
{
    if (signal(SIGINT, signalHandler) == SIG_ERR)
    {
        perror("Error setting up signal handler");
        return 1;
    }

    int opt = 0;
    while ((opt = getopt(argc, argv, "s:d:a:c:l:b:h?")) != -1)
    {
        char * endptr = NULL;
        switch (opt)
        {
        case '?':
        case 'h': {
            help();
            break;
        }
        case 's': {
            snprintf(dev_1_symbol, sizeof(dev_1_symbol), "%s", optarg);
            break;
        }
        case 'a': {
            snprintf(appsym, sizeof(appsym), "%s", optarg);
            break;
        }
        case 'c': {
            uint64_t cpu_mask = strtoull(optarg, &endptr, 0);
            if (cpu_mask == 0 && endptr == optarg)
                help();
            CPU_ZERO(&cpu_set_io);
            for (unsigned long bit_iter = 0; bit_iter < sizeof(cpu_mask) * 8; bit_iter++)
            {
                if ((cpu_mask & (1ULL << bit_iter)))
                {
                    CPU_SET(bit_iter, &cpu_set_io);
                }
            }
            break;
        }
        case 'l': {
            int lcore_indexes[RTE_MAX_LCORE];
            if (mr_parse_corelist(optarg, lcore_indexes) < 0)
            {
                fprintf(stderr, "%s\n", "invalid core list syntax");
                help();
            }
            CPU_ZERO(&cpu_set_io);
            for (unsigned int i = 0; i < RTE_MAX_LCORE; i++)
            {
                if (lcore_indexes[i] == -1)
                {
                    continue;
                }
                CPU_SET(i, &cpu_set_io);
            }
            break;
        }
        case 'b': {
            nr_burst = strtoull(optarg, &endptr, 0);
            if (nr_burst == 0 && endptr == optarg)
                help();
            break;
        }

        default:
            help();
            break;
        }
    }

    if (CPU_COUNT(&cpu_set_io) == 0)
    {
        fprintf(stderr, "please set the io cores parameter.\n");
        help();
    }

    mr_instance = marsio_create();
    if (mr_instance == NULL)
    {
        fprintf(stderr, "Marsio instance create failed. ");
        abort();
    }

    unsigned int opt_value = 1;
    marsio_option_set(mr_instance, MARSIO_OPT_EXIT_WHEN_ERR, &opt_value, sizeof(opt_value));
    marsio_option_set(mr_instance, MARSIO_OPT_THREAD_MASK_IN_CPUSET, &cpu_set_io, sizeof(cpu_set_io));
    marsio_init(mr_instance, appsym);

    nr_thread = CPU_COUNT(&cpu_set_io);
    dev_1_handler = marsio_open_device(mr_instance, dev_1_symbol, nr_thread, nr_thread);
    to_dev_1_sendpath = marsio_sendpath_create_by_vdev(dev_1_handler);

    fprintf(stdout, "Thread Count = %d\n", nr_thread);
    pthread_t tmp_pid[nr_thread];
    for (int i = 0; i < nr_thread; i++)
    {
        pthread_create(&tmp_pid[i], NULL, l2fwd_loop, (void *)(uintptr_t)i);
    }

    for (int i = 0; i < nr_thread; i++)
    {
        pthread_join(tmp_pid[i], NULL);
    }

    marsio_destory(mr_instance);
    fprintf(stdout, "L2FWD is terminated. ");
    return 0;
}

// copy from lib/eal/common/eal_common_options.c
static int check_core_list(int * lcores, unsigned int count)
{
    char lcorestr[RTE_MAX_LCORE * 10];
    bool overflow = false;
    int len = 0, ret;
    unsigned int i;

    for (i = 0; i < count; i++)
    {
        if (lcores[i] < RTE_MAX_LCORE)
            continue;

        fprintf(stdout, "lcore %d >= RTE_MAX_LCORE (%d)", lcores[i], RTE_MAX_LCORE);
        overflow = true;
    }
    if (!overflow)
        return 0;

    /*
     * We've encountered a core that's greater than RTE_MAX_LCORE,
     * suggest using --lcores option to map lcores onto physical cores
     * greater than RTE_MAX_LCORE.
     */
    for (i = 0; i < count; i++)
    {
        ret = snprintf(&lcorestr[len], sizeof(lcorestr) - len, "%d@%d,", i, lcores[i]);
        if (ret > 0)
            len = len + ret;
    }
    if (len > 0)
        lcorestr[len - 1] = 0;
    fprintf(stdout,
            "To use high physical core ids, "
            "please use --lcores to map them to lcore ids below RTE_MAX_LCORE, "
            "e.g. --lcores %s",
            lcorestr);
    return -1;
}

static int mr_parse_corelist(const char * corelist, int * cores)
{
    unsigned int count = 0, i;
    int lcores[RTE_MAX_LCORE];
    char * end = NULL;
    int min, max;
    int idx;

    for (idx = 0; idx < RTE_MAX_LCORE; idx++)
        cores[idx] = -1;

    /* Remove all blank characters ahead */
    while (isblank(*corelist))
        corelist++;

    /* Get list of cores */
    min = -1;
    do
    {
        while (isblank(*corelist))
            corelist++;
        if (*corelist == '\0')
            return -1;
        errno = 0;
        idx = strtol(corelist, &end, 10);
        if (errno || end == NULL)
            return -1;
        if (idx < 0)
            return -1;
        while (isblank(*end))
            end++;
        if (*end == '-')
        {
            min = idx;
        }
        else if ((*end == ',') || (*end == '\0'))
        {
            max = idx;
            if (min == -1)
                min = idx;
            for (idx = min; idx <= max; idx++)
            {
                bool dup = false;

                /* Check if this idx is already present */
                for (i = 0; i < count; i++)
                {
                    if (lcores[i] == idx)
                        dup = true;
                }
                if (dup)
                    continue;
                if (count >= RTE_MAX_LCORE)
                {
                    fprintf(stdout, "Too many lcores provided. Cannot exceed RTE_MAX_LCORE (%d)", RTE_MAX_LCORE);
                    return -1;
                }
                lcores[count++] = idx;
            }
            min = -1;
        }
        else
            return -1;
        corelist = end + 1;
    } while (*end != '\0');

    if (count == 0)
        return -1;

    if (check_core_list(lcores, count))
        return -1;

    /*
     * Now that we've got a list of cores no longer than RTE_MAX_LCORE,
     * and no lcore in that list is greater than RTE_MAX_LCORE, populate
     * the cores array.
     */
    do
    {
        count--;
        cores[lcores[count]] = count;
    } while (count != 0);

    return 0;
}

#if 0
// Remove the following code later
int core_list_parse(const char * str)
{
    unsigned int io_cores[128] = {};
    int nr_io_cores = 0;

    nr_io_cores = str_uint_range_parse(str, io_cores, sizeof(io_cores) / sizeof(io_cores[0]));
    if (nr_io_cores < 0)
    {
        return -1;
    }

    CPU_ZERO(&cpu_set_io);
    for (unsigned int i = 0; i < nr_io_cores; i++)
    {
        CPU_SET(io_cores[i], &cpu_set_io);
    }
    return 0;
}

int str_uint_range_parse(const char * str, unsigned int * out, const size_t size)
{
    // The strtok() is not reentrant function. It is not recommended.
    int ret = 0;
    char * line = strdup(str);
    char * saveptr;

    int num_cnt = 0;
    char * currnum = strtok_r(line, " ,", &saveptr);
    while (currnum != NULL)
    {
        if (num_cnt >= size)
        {
            ret = -1;
            goto end;
        }
        out[num_cnt++] = atoi(currnum);
        currnum = strtok_r(NULL, " ,", &saveptr);
    }
    ret = num_cnt;

end:
    free(line);
    return ret;
}
#endif