odp_pktio_perf.c

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright (c) 2015-2018 Linaro Limited
 */
 
#include <odp_api.h>
 
#include <odp/helper/odph_api.h>
 
#include <getopt.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <inttypes.h>
 
#define TEST_SKIP 77
 
#define PKT_BUF_NUM       (32 * 1024)
#define MAX_NUM_IFACES    2
#define TEST_HDR_MAGIC    0x92749451
#define MAX_WORKERS       (ODP_THREAD_COUNT_MAX - 1)
#define BATCH_LEN_MAX     32
 
/* Packet rate at which to start when using binary search */
#define RATE_SEARCH_INITIAL_PPS 1000000
 
/* When using the binary search method to determine the maximum
 * achievable packet rate, this value specifies how close the pass
 * and fail measurements must be before the test is terminated. */
#define RATE_SEARCH_ACCURACY_PPS 100000
 
/* Amount of time to wait, in nanoseconds, between the transmitter(s)
 * completing and the receiver(s) being shutdown. Any packets not
 * received by this time will be assumed to have been lost. */
#define SHUTDOWN_DELAY_NS (ODP_TIME_MSEC_IN_NS * 100)
 
/* Number of duration units in a second. */
#define T_SCALE 10
 
/* Default test duration in T_SCALE units */
#define DEFAULT_DURATION 1
 
#define PKT_HDR_LEN (sizeof(pkt_head_t) + ODPH_UDPHDR_LEN + \
             ODPH_IPV4HDR_LEN + ODPH_ETHHDR_LEN)
 
typedef struct {
    unsigned int cpu_count; /* CPU count */
    int      num_tx_workers;/* Number of CPUs to use for transmit */
    int      duration;  /* Time to run each iteration of the test for */
    uint32_t tx_batch_len;  /* Number of packets to send in a single
                   batch */
    int      schedule;  /* 1: receive packets via scheduler
                   0: receive packets via direct deq */
    uint32_t rx_batch_len;  /* Number of packets to receive in a single
                   batch */
    uint64_t pps;       /* Attempted packet rate */
    int      verbose;   /* Print verbose information, such as per
                   thread statistics */
    unsigned pkt_len;   /* Packet payload length in bytes (not
                   including headers) */
    int      search;    /* Set implicitly when pps is not configured.
                   Perform a search at different packet rates
                   to determine the maximum rate at which no
                   packet loss occurs. */
 
    char     *if_str;
    const char *ifaces[MAX_NUM_IFACES];
    int      num_ifaces;
} test_args_t;
 
typedef struct ODP_ALIGNED_CACHE {
    uint64_t rx_cnt;    /* Valid packets received */
    uint64_t rx_ignore; /* Ignored packets */
} pkt_rx_stats_t;
 
typedef struct ODP_ALIGNED_CACHE {
    uint64_t tx_cnt;    /* Packets transmitted */
    uint64_t alloc_failures;/* Packet allocation failures */
    uint64_t enq_failures;  /* Enqueue failures */
    odp_time_t idle_ticks;  /* Idle ticks count in TX loop */
} pkt_tx_stats_t;
 
/* Test global variables */
typedef struct {
    odp_instance_t instance;
    test_args_t args;
    odp_barrier_t rx_barrier;
    odp_barrier_t tx_barrier;
    odp_pktio_t pktio_tx;
    odp_pktio_t pktio_rx;
    /* Pool from which transmitted packets are allocated */
    odp_pool_t transmit_pkt_pool;
    pkt_rx_stats_t *rx_stats;
    pkt_tx_stats_t *tx_stats;
    uint8_t src_mac[ODPH_ETHADDR_LEN];
    uint8_t dst_mac[ODPH_ETHADDR_LEN];
    uint32_t rx_stats_size;
    uint32_t tx_stats_size;
    /* Indicate to the receivers to shutdown */
    odp_atomic_u32_t shutdown;
    /* Sequence number of IP packets */
    odp_atomic_u32_t ip_seq ODP_ALIGNED_CACHE;
} test_globals_t;
 
/* Status of max rate search */
typedef struct {
    uint64_t pps_curr; /* Current attempted PPS */
    uint64_t pps_pass; /* Highest passing PPS */
    uint64_t pps_fail; /* Lowest failing PPS */
    int      warmup;   /* Warmup stage - ignore results */
} test_status_t;
 
/* Thread specific arguments */
typedef struct {
    int batch_len; /* Number of packets per transmit batch */
    int duration;  /* Run duration in scaled time units */
    uint64_t pps;  /* Packets per second for this thread */
} thread_args_t;
 
typedef struct {
    odp_u32be_t magic; /* Packet header magic number */
} pkt_head_t;
 
/* Application global data */
static test_globals_t *gbl_args;
 
/*
 * Generate a single test packet for transmission.
 */
static odp_packet_t pktio_create_packet(uint32_t seq)
{
    odp_packet_t pkt;
    odph_ethhdr_t *eth;
    odph_ipv4hdr_t *ip;
    odph_udphdr_t *udp;
    char *buf;
    uint32_t offset;
    pkt_head_t pkt_hdr;
    size_t payload_len;
 
    payload_len = sizeof(pkt_hdr) + gbl_args->args.pkt_len;
 
    pkt = odp_packet_alloc(gbl_args->transmit_pkt_pool,
                   payload_len + ODPH_UDPHDR_LEN +
                   ODPH_IPV4HDR_LEN + ODPH_ETHHDR_LEN);
 
    if (pkt == ODP_PACKET_INVALID)
        return ODP_PACKET_INVALID;
 
    buf = odp_packet_data(pkt);
 
    /* Ethernet */
    offset = 0;
    odp_packet_l2_offset_set(pkt, offset);
    eth = (odph_ethhdr_t *)buf;
    memcpy(eth->src.addr, gbl_args->src_mac, ODPH_ETHADDR_LEN);
    memcpy(eth->dst.addr, gbl_args->dst_mac, ODPH_ETHADDR_LEN);
    eth->type = odp_cpu_to_be_16(ODPH_ETHTYPE_IPV4);
 
    /* IP */
    offset += ODPH_ETHHDR_LEN;
    odp_packet_l3_offset_set(pkt, ODPH_ETHHDR_LEN);
    ip = (odph_ipv4hdr_t *)(buf + ODPH_ETHHDR_LEN);
    ip->dst_addr = odp_cpu_to_be_32(0);
    ip->src_addr = odp_cpu_to_be_32(0);
    ip->ver_ihl = ODPH_IPV4 << 4 | ODPH_IPV4HDR_IHL_MIN;
    ip->tot_len = odp_cpu_to_be_16(payload_len + ODPH_UDPHDR_LEN +
                       ODPH_IPV4HDR_LEN);
    ip->ttl = 128;
    ip->proto = ODPH_IPPROTO_UDP;
    ip->id = odp_cpu_to_be_16(seq);
    ip->chksum = 0;
    odph_ipv4_csum_update(pkt);
 
    /* UDP */
    offset += ODPH_IPV4HDR_LEN;
    odp_packet_l4_offset_set(pkt, offset);
    udp = (odph_udphdr_t *)(buf + offset);
    udp->src_port = odp_cpu_to_be_16(0);
    udp->dst_port = odp_cpu_to_be_16(0);
    udp->length = odp_cpu_to_be_16(payload_len + ODPH_UDPHDR_LEN);
    udp->chksum = 0;
 
    /* payload */
    offset += ODPH_UDPHDR_LEN;
    pkt_hdr.magic = TEST_HDR_MAGIC;
    if (odp_packet_copy_from_mem(pkt, offset, sizeof(pkt_hdr),
                     &pkt_hdr) != 0)
        ODPH_ABORT("Failed to generate test packet.\n");
 
    return pkt;
}
 
/*
 * Check if a packet payload contains test payload magic number.
 */
static int pktio_pkt_has_magic(odp_packet_t pkt)
{
    size_t l4_off;
    pkt_head_t pkt_hdr;
 
    l4_off = ODPH_ETHHDR_LEN + ODPH_IPV4HDR_LEN;
    int ret = odp_packet_copy_to_mem(pkt,
                     l4_off + ODPH_UDPHDR_LEN,
                     sizeof(pkt_hdr), &pkt_hdr);
 
    if (ret != 0)
        return 0;
 
    if (pkt_hdr.magic == TEST_HDR_MAGIC)
        return 1;
 
    return 0;
}
 
/*
 * Allocate packets for transmission.
 */
static int alloc_packets(odp_packet_t *pkt_tbl, int num_pkts)
{
    int n;
    uint16_t seq;
 
    seq = odp_atomic_fetch_add_u32(&gbl_args->ip_seq, num_pkts);
    for (n = 0; n < num_pkts; ++n) {
        pkt_tbl[n] = pktio_create_packet(seq + n);
        if (pkt_tbl[n] == ODP_PACKET_INVALID)
            break;
    }
 
    return n;
}
 
static int send_packets(odp_pktout_queue_t pktout,
            odp_packet_t *pkt_tbl, unsigned pkts)
{
    unsigned tx_drops;
    unsigned sent = 0;
 
    if (pkts == 0)
        return 0;
 
    while (sent < pkts) {
        int ret;
 
        ret = odp_pktout_send(pktout, &pkt_tbl[sent], pkts - sent);
 
        if (odp_likely(ret > 0))
            sent += ret;
        else
            break;
    }
 
    tx_drops = pkts - sent;
 
    if (odp_unlikely(tx_drops))
        odp_packet_free_multi(&pkt_tbl[sent], tx_drops);
 
    return sent;
}
 
/*
 * Main packet transmit routine. Transmit packets at a fixed rate for
 * specified length of time.
 */
static int run_thread_tx(void *arg)
{
    test_globals_t *globals;
    int thr_id;
    odp_pktout_queue_t pktout;
    pkt_tx_stats_t *stats;
    odp_time_t cur_time, send_time_end, send_duration;
    odp_time_t burst_gap_end, burst_gap;
    uint32_t batch_len;
    int unsent_pkts = 0;
    odp_packet_t tx_packet[BATCH_LEN_MAX];
    odp_time_t idle_start = ODP_TIME_NULL;
 
    thread_args_t *targs = arg;
 
    batch_len = targs->batch_len;
 
    if (batch_len > BATCH_LEN_MAX)
        batch_len = BATCH_LEN_MAX;
 
    thr_id = odp_thread_id();
 
    globals = odp_shm_addr(odp_shm_lookup("test_globals"));
    stats = &globals->tx_stats[thr_id];
 
    if (odp_pktout_queue(globals->pktio_tx, &pktout, 1) != 1)
        ODPH_ABORT("Failed to get output queue for thread %d\n",
               thr_id);
 
    burst_gap = odp_time_local_from_ns(
            ODP_TIME_SEC_IN_NS / (targs->pps / targs->batch_len));
    send_duration =
        odp_time_local_from_ns(targs->duration * ODP_TIME_SEC_IN_NS / T_SCALE);
 
    odp_barrier_wait(&globals->tx_barrier);
 
    cur_time     = odp_time_local();
    send_time_end = odp_time_sum(cur_time, send_duration);
    burst_gap_end = cur_time;
    while (odp_time_cmp(send_time_end, cur_time) > 0) {
        unsigned alloc_cnt = 0, tx_cnt;
 
        if (odp_time_cmp(burst_gap_end, cur_time) > 0) {
            cur_time = odp_time_local();
            if (!odp_time_cmp(idle_start, ODP_TIME_NULL))
                idle_start = cur_time;
            continue;
        }
 
        if (odp_time_cmp(idle_start, ODP_TIME_NULL) > 0) {
            odp_time_t diff = odp_time_diff(cur_time, idle_start);
 
            stats->idle_ticks =
                odp_time_sum(diff, stats->idle_ticks);
 
            idle_start = ODP_TIME_NULL;
        }
 
        burst_gap_end = odp_time_sum(burst_gap_end, burst_gap);
 
        alloc_cnt = alloc_packets(tx_packet, batch_len - unsent_pkts);
        if (alloc_cnt != batch_len)
            stats->alloc_failures++;
 
        tx_cnt = send_packets(pktout, tx_packet, alloc_cnt);
        unsent_pkts = alloc_cnt - tx_cnt;
        stats->enq_failures += unsent_pkts;
        stats->tx_cnt += tx_cnt;
 
        cur_time = odp_time_local();
    }
 
    if (gbl_args->args.verbose)
        printf(" %02d: TxPkts %-8" PRIu64 " EnqFail %-6" PRIu64
               " AllocFail %-6" PRIu64 " Idle %" PRIu64 "ms\n",
               thr_id, stats->tx_cnt, stats->enq_failures,
               stats->alloc_failures,
               odp_time_to_ns(stats->idle_ticks) /
               (uint64_t)ODP_TIME_MSEC_IN_NS);
 
    return 0;
}
 
static int receive_packets(odp_queue_t queue,
               odp_event_t *event_tbl, unsigned num_pkts)
{
    int n_ev = 0;
 
    if (num_pkts == 0)
        return 0;
 
    if (queue != ODP_QUEUE_INVALID) {
        if (num_pkts == 1) {
            event_tbl[0] = odp_queue_deq(queue);
            n_ev = event_tbl[0] != ODP_EVENT_INVALID;
        } else {
            n_ev = odp_queue_deq_multi(queue, event_tbl, num_pkts);
        }
    } else {
        if (num_pkts == 1) {
            event_tbl[0] = odp_schedule(NULL, ODP_SCHED_NO_WAIT);
            n_ev = event_tbl[0] != ODP_EVENT_INVALID;
        } else {
            n_ev = odp_schedule_multi(NULL, ODP_SCHED_NO_WAIT,
                          event_tbl, num_pkts);
        }
    }
    return n_ev;
}
 
static int run_thread_rx(void *arg)
{
    test_globals_t *globals;
    int thr_id, batch_len;
    odp_queue_t queue = ODP_QUEUE_INVALID;
    odp_packet_t pkt;
 
    thread_args_t *targs = arg;
 
    batch_len = targs->batch_len;
 
    if (batch_len > BATCH_LEN_MAX)
        batch_len = BATCH_LEN_MAX;
 
    thr_id = odp_thread_id();
 
    globals = odp_shm_addr(odp_shm_lookup("test_globals"));
 
    pkt_rx_stats_t *stats = &globals->rx_stats[thr_id];
 
    if (gbl_args->args.schedule == 0) {
        if (odp_pktin_event_queue(globals->pktio_rx, &queue, 1) != 1)
            ODPH_ABORT("No input queue.\n");
    }
 
    odp_barrier_wait(&globals->rx_barrier);
    while (1) {
        odp_event_t ev[BATCH_LEN_MAX];
        int i, n_ev;
 
        n_ev = receive_packets(queue, ev, batch_len);
 
        for (i = 0; i < n_ev; ++i) {
            if (odp_event_type(ev[i]) == ODP_EVENT_PACKET) {
                pkt = odp_packet_from_event(ev[i]);
                if (pktio_pkt_has_magic(pkt))
                    stats->rx_cnt++;
                else
                    stats->rx_ignore++;
            }
            odp_event_free(ev[i]);
        }
        if (n_ev == 0 && odp_atomic_load_u32(&gbl_args->shutdown))
            break;
    }
 
    return 0;
}
 
/*
 * Process the results from a single fixed rate test run to determine whether
 * it passed or failed. Pass criteria are that the requested transmit packet
 * rate was achieved and that all of the transmitted packets were received.
 */
static int process_results(uint64_t expected_tx_cnt,
               test_status_t *status)
{
    int fail = 0;
    uint64_t drops = 0;
    uint64_t rx_pkts = 0;
    uint64_t tx_pkts = 0;
    uint64_t attempted_pps;
    int i;
    char str[512];
    int len = 0;
 
    for (i = 0; i < odp_thread_count_max(); ++i) {
        rx_pkts += gbl_args->rx_stats[i].rx_cnt;
        tx_pkts += gbl_args->tx_stats[i].tx_cnt;
    }
 
    if (rx_pkts == 0) {
        ODPH_ERR("no packets received\n");
        return -1;
    }
 
    if (tx_pkts < (expected_tx_cnt - (expected_tx_cnt / 100))) {
        /* failed to transmit packets at (99% of) requested rate */
        fail = 1;
    } else if (tx_pkts > rx_pkts) {
        /* failed to receive all of the transmitted packets */
        fail = 1;
        drops = tx_pkts - rx_pkts;
    }
 
    attempted_pps = status->pps_curr;
 
    len += snprintf(&str[len], sizeof(str) - 1 - len,
            "PPS: %-8" PRIu64 " ", attempted_pps);
    len += snprintf(&str[len], sizeof(str) - 1 - len,
            "Succeeded: %-4s ", fail ? "No" : "Yes");
    len += snprintf(&str[len], sizeof(str) - 1 - len,
            "TxPkts: %-8" PRIu64 " ", tx_pkts);
    len += snprintf(&str[len], sizeof(str) - 1 - len,
            "RxPkts: %-8" PRIu64 " ", rx_pkts);
    len += snprintf(&str[len], sizeof(str) - 1 - len,
            "DropPkts: %-8" PRIu64 " ", drops);
    printf("%s\n", str);
 
    if (gbl_args->args.search == 0) {
        printf("Result: %s\n", fail ? "FAILED" : "PASSED");
        return fail ? -1 : 0;
    }
 
    if (fail && (status->pps_fail == 0 ||
             attempted_pps < status->pps_fail)) {
        status->pps_fail = attempted_pps;
    } else if (attempted_pps > status->pps_pass) {
        status->pps_pass = attempted_pps;
    }
 
    if (status->pps_fail == 0) {
        /* ramping up, double the previously attempted pps */
        status->pps_curr *= 2;
    } else {
        /* set the new target to half way between the upper and lower
         * limits */
        status->pps_curr = status->pps_pass +
                   ((status->pps_fail - status->pps_pass) / 2);
    }
 
    /* stop once the pass and fail measurements are within range */
    if ((status->pps_fail - status->pps_pass) < RATE_SEARCH_ACCURACY_PPS) {
        unsigned pkt_len = gbl_args->args.pkt_len + PKT_HDR_LEN;
        int mbps = (pkt_len * status->pps_pass * 8) / 1024 / 1024;
 
        printf("Maximum packet rate: %" PRIu64 " PPS (%d Mbps)\n",
               status->pps_pass, mbps);
 
        return 0;
    }
 
    return 1;
}
 
static int setup_txrx_masks(odp_cpumask_t *thd_mask_tx,
                odp_cpumask_t *thd_mask_rx)
{
    odp_cpumask_t cpumask;
    int num_workers, num_tx_workers, num_rx_workers;
    int i, cpu;
 
    num_workers =
        odp_cpumask_default_worker(&cpumask,
                       gbl_args->args.cpu_count);
    if (num_workers < 2) {
        ODPH_ERR("Need at least two cores\n");
        return TEST_SKIP;
    }
 
    if (num_workers > MAX_WORKERS) {
        ODPH_DBG("Worker count limited to MAX_WORKERS define (=%d)\n",
             MAX_WORKERS);
        num_workers = MAX_WORKERS;
    }
 
    if (gbl_args->args.num_tx_workers) {
        if (gbl_args->args.num_tx_workers > (num_workers - 1)) {
            ODPH_ERR("Invalid TX worker count\n");
            return -1;
        }
        num_tx_workers = gbl_args->args.num_tx_workers;
    } else {
        /* default is to split the available cores evenly into TX and
         * RX workers, favour TX for odd core count */
        num_tx_workers = (num_workers + 1) / 2;
    }
 
    odp_cpumask_zero(thd_mask_tx);
    odp_cpumask_zero(thd_mask_rx);
 
    cpu = odp_cpumask_first(&cpumask);
    for (i = 0; i < num_workers; ++i) {
        if (i < num_tx_workers)
            odp_cpumask_set(thd_mask_tx, cpu);
        else
            odp_cpumask_set(thd_mask_rx, cpu);
        cpu = odp_cpumask_next(&cpumask, cpu);
    }
 
    num_rx_workers = odp_cpumask_count(thd_mask_rx);
 
    odp_barrier_init(&gbl_args->rx_barrier, num_rx_workers + 1);
    odp_barrier_init(&gbl_args->tx_barrier, num_tx_workers + 1);
 
    return 0;
}
 
/*
 * Run a single instance of the throughput test. When attempting to determine
 * the maximum packet rate this will be invoked multiple times with the only
 * difference between runs being the target PPS rate.
 */
static int run_test_single(odp_cpumask_t *thd_mask_tx,
               odp_cpumask_t *thd_mask_rx,
               test_status_t *status)
{
    odph_thread_t thread_tbl[MAX_WORKERS];
    odph_thread_common_param_t thr_common;
    odph_thread_param_t thr_param;
    thread_args_t args_tx, args_rx;
    uint64_t expected_tx_cnt;
    int num_tx_workers, num_rx_workers;
 
    odp_atomic_store_u32(&gbl_args->shutdown, 0);
 
    memset(thread_tbl, 0, sizeof(thread_tbl));
    memset(gbl_args->rx_stats, 0, gbl_args->rx_stats_size);
    memset(gbl_args->tx_stats, 0, gbl_args->tx_stats_size);
 
    expected_tx_cnt = status->pps_curr * gbl_args->args.duration / T_SCALE;
 
    /* start receiver threads first */
 
    num_rx_workers = odp_cpumask_count(thd_mask_rx);
    args_rx.batch_len = gbl_args->args.rx_batch_len;
 
    odph_thread_common_param_init(&thr_common);
    thr_common.instance = gbl_args->instance;
    thr_common.cpumask = thd_mask_rx;
    thr_common.share_param = 1;
 
    odph_thread_param_init(&thr_param);
    thr_param.start = run_thread_rx;
    thr_param.arg = &args_rx;
    thr_param.thr_type = ODP_THREAD_WORKER;
 
    odph_thread_create(thread_tbl, &thr_common, &thr_param, num_rx_workers);
    odp_barrier_wait(&gbl_args->rx_barrier);
 
    /* then start transmitters */
 
    num_tx_workers    = odp_cpumask_count(thd_mask_tx);
    args_tx.pps       = status->pps_curr / num_tx_workers;
    args_tx.duration  = gbl_args->args.duration;
    args_tx.batch_len = gbl_args->args.tx_batch_len;
 
    odph_thread_common_param_init(&thr_common);
    thr_common.instance = gbl_args->instance;
    thr_common.cpumask = thd_mask_tx;
    thr_common.share_param = 1;
 
    odph_thread_param_init(&thr_param);
    thr_param.start = run_thread_tx;
    thr_param.arg = &args_tx;
    thr_param.thr_type = ODP_THREAD_WORKER;
 
    odph_thread_create(&thread_tbl[num_rx_workers], &thr_common, &thr_param, num_tx_workers);
    odp_barrier_wait(&gbl_args->tx_barrier);
 
    /* wait for transmitter threads to terminate */
    odph_thread_join(&thread_tbl[num_rx_workers], num_tx_workers);
 
    /* delay to allow transmitted packets to reach the receivers */
    odp_time_wait_ns(SHUTDOWN_DELAY_NS);
 
    /* indicate to the receivers to exit */
    odp_atomic_store_u32(&gbl_args->shutdown, 1);
 
    /* wait for receivers */
    odph_thread_join(thread_tbl, num_rx_workers);
 
    if (!status->warmup)
        return process_results(expected_tx_cnt, status);
 
    return 1;
}
 
static int run_test(void)
{
    int ret;
    int i;
    odp_cpumask_t txmask, rxmask;
    test_status_t status = {
        .pps_curr = gbl_args->args.pps,
        .pps_pass = 0,
        .pps_fail = 0,
        .warmup = 1,
    };
 
    ret = setup_txrx_masks(&txmask, &rxmask);
    if (ret)
        return ret;
 
    printf("Starting test with params:\n");
    printf("\tTransmit workers:     \t%d\n", odp_cpumask_count(&txmask));
    printf("\tReceive workers:      \t%d\n", odp_cpumask_count(&rxmask));
    printf("\tDuration (seconds):   \t%d\n", gbl_args->args.duration);
    printf("\tTransmit batch length:\t%" PRIu32 "\n",
           gbl_args->args.tx_batch_len);
    printf("\tReceive batch length: \t%" PRIu32 "\n",
           gbl_args->args.rx_batch_len);
    printf("\tPacket receive method:\t%s\n",
           gbl_args->args.schedule ? "schedule" : "plain");
    printf("\tInterface(s):         \t");
    for (i = 0; i < gbl_args->args.num_ifaces; ++i)
        printf("%s ", gbl_args->args.ifaces[i]);
    printf("\n");
 
    /* first time just run the test but throw away the results */
    run_test_single(&txmask, &rxmask, &status);
    status.warmup = 0;
 
    while (1) {
        ret = run_test_single(&txmask, &rxmask, &status);
        if (ret <= 0)
            break;
    }
 
    return ret;
}
 
static odp_pktio_t create_pktio(const char *iface, int schedule)
{
    odp_pool_t pool;
    odp_pktio_t pktio;
    char pool_name[ODP_POOL_NAME_LEN];
    odp_pool_param_t params;
    odp_pktio_param_t pktio_param;
 
    odp_pool_param_init(&params);
    params.pkt.len     = PKT_HDR_LEN + gbl_args->args.pkt_len;
    params.pkt.seg_len = params.pkt.len;
    params.pkt.num     = PKT_BUF_NUM;
    params.type        = ODP_POOL_PACKET;
 
    snprintf(pool_name, sizeof(pool_name), "pkt_pool_%s", iface);
    pool = odp_pool_create(pool_name, &params);
    if (pool == ODP_POOL_INVALID)
        return ODP_PKTIO_INVALID;
 
    odp_pktio_param_init(&pktio_param);
 
    if (schedule)
        pktio_param.in_mode = ODP_PKTIN_MODE_SCHED;
    else
        pktio_param.in_mode = ODP_PKTIN_MODE_QUEUE;
 
    pktio = odp_pktio_open(iface, pool, &pktio_param);
 
    return pktio;
}
 
static int test_init(void)
{
    odp_pool_param_t params;
    const char *iface;
    int schedule;
    odp_pktio_config_t cfg;
 
    odp_pool_param_init(&params);
    params.pkt.len     = PKT_HDR_LEN + gbl_args->args.pkt_len;
    params.pkt.seg_len = params.pkt.len;
    params.pkt.num     = PKT_BUF_NUM;
    params.type        = ODP_POOL_PACKET;
 
    gbl_args->transmit_pkt_pool = odp_pool_create("pkt_pool_transmit",
                              &params);
    if (gbl_args->transmit_pkt_pool == ODP_POOL_INVALID)
        ODPH_ABORT("Failed to create transmit pool\n");
 
    odp_atomic_init_u32(&gbl_args->ip_seq, 0);
    odp_atomic_init_u32(&gbl_args->shutdown, 0);
 
    iface    = gbl_args->args.ifaces[0];
    schedule = gbl_args->args.schedule;
 
    if (schedule)
        odp_schedule_config(NULL);
 
    /* create pktios and associate input/output queues */
    gbl_args->pktio_tx = create_pktio(iface, schedule);
    if (gbl_args->args.num_ifaces > 1) {
        iface = gbl_args->args.ifaces[1];
        gbl_args->pktio_rx = create_pktio(iface, schedule);
    } else {
        gbl_args->pktio_rx = gbl_args->pktio_tx;
    }
 
    odp_pktio_mac_addr(gbl_args->pktio_tx, gbl_args->src_mac,
               ODPH_ETHADDR_LEN);
    odp_pktio_mac_addr(gbl_args->pktio_rx, gbl_args->dst_mac,
               ODPH_ETHADDR_LEN);
 
    if (gbl_args->pktio_rx == ODP_PKTIO_INVALID ||
        gbl_args->pktio_tx == ODP_PKTIO_INVALID) {
        ODPH_ERR("failed to open pktio\n");
        return -1;
    }
 
    /* Create single queue with default parameters */
    if (odp_pktout_queue_config(gbl_args->pktio_tx, NULL)) {
        ODPH_ERR("failed to configure pktio_tx queue\n");
        return -1;
    }
 
    /* Configure also input side (with defaults) */
    if (odp_pktin_queue_config(gbl_args->pktio_tx, NULL)) {
        ODPH_ERR("failed to configure pktio_tx queue\n");
        return -1;
    }
 
    /* Disable packet parsing as this is done in the driver where it
     * affects scalability.
     */
    odp_pktio_config_init(&cfg);
    cfg.parser.layer = ODP_PROTO_LAYER_NONE;
    odp_pktio_config(gbl_args->pktio_rx, &cfg);
 
    if (gbl_args->args.num_ifaces > 1) {
        if (odp_pktout_queue_config(gbl_args->pktio_rx, NULL)) {
            ODPH_ERR("failed to configure pktio_rx queue\n");
            return -1;
        }
 
        if (odp_pktin_queue_config(gbl_args->pktio_rx, NULL)) {
            ODPH_ERR("failed to configure pktio_rx queue\n");
            return -1;
        }
    }
 
    if (odp_pktio_start(gbl_args->pktio_tx) != 0)
        return -1;
    if (gbl_args->args.num_ifaces > 1 &&
        odp_pktio_start(gbl_args->pktio_rx))
        return -1;
 
    return 0;
}
 
static int empty_inq(odp_pktio_t pktio)
{
    odp_queue_t queue = ODP_QUEUE_INVALID;
    odp_event_t ev;
    odp_queue_type_t q_type;
 
    if (odp_pktin_event_queue(pktio, &queue, 1) != 1)
        return -1;
 
    q_type = odp_queue_type(queue);
 
    /* flush any pending events */
    while (1) {
        if (q_type == ODP_QUEUE_TYPE_PLAIN)
            ev = odp_queue_deq(queue);
        else
            ev = odp_schedule(NULL, ODP_SCHED_NO_WAIT);
 
        if (ev != ODP_EVENT_INVALID)
            odp_event_free(ev);
        else
            break;
    }
 
    return 0;
}
 
static int test_term(void)
{
    char pool_name[ODP_POOL_NAME_LEN];
    odp_pool_t pool;
    int i;
    int ret = 0;
 
    if (gbl_args->pktio_tx != gbl_args->pktio_rx) {
        if (odp_pktio_stop(gbl_args->pktio_tx)) {
            ODPH_ERR("Failed to stop pktio_tx\n");
            return -1;
        }
 
        if (odp_pktio_close(gbl_args->pktio_tx)) {
            ODPH_ERR("Failed to close pktio_tx\n");
            ret = -1;
        }
    }
 
    empty_inq(gbl_args->pktio_rx);
 
    if (odp_pktio_stop(gbl_args->pktio_rx)) {
        ODPH_ERR("Failed to stop pktio_rx\n");
        return -1;
    }
 
    if (odp_pktio_close(gbl_args->pktio_rx) != 0) {
        ODPH_ERR("Failed to close pktio_rx\n");
        ret = -1;
    }
 
    for (i = 0; i < gbl_args->args.num_ifaces; ++i) {
        snprintf(pool_name, sizeof(pool_name),
             "pkt_pool_%s", gbl_args->args.ifaces[i]);
        pool = odp_pool_lookup(pool_name);
        if (pool == ODP_POOL_INVALID)
            continue;
 
        if (odp_pool_destroy(pool) != 0) {
            ODPH_ERR("Failed to destroy pool %s\n", pool_name);
            ret = -1;
        }
    }
 
    if (odp_pool_destroy(gbl_args->transmit_pkt_pool) != 0) {
        ODPH_ERR("Failed to destroy transmit pool\n");
        ret = -1;
    }
 
    free(gbl_args->args.if_str);
 
    if (odp_shm_free(odp_shm_lookup("test_globals")) != 0) {
        ODPH_ERR("Failed to free test_globals\n");
        ret = -1;
    }
    if (odp_shm_free(odp_shm_lookup("test_globals.rx_stats")) != 0) {
        ODPH_ERR("Failed to free test_globals.rx_stats\n");
        ret = -1;
    }
    if (odp_shm_free(odp_shm_lookup("test_globals.tx_stats")) != 0) {
        ODPH_ERR("Failed to free test_globals.tx_stats\n");
        ret = -1;
    }
 
    return ret;
}
 
static void usage(void)
{
    printf("\nUsage: odp_pktio_perf [options]\n\n");
    printf("  -c, --count <number>   CPU count, 0=all available, default=2\n");
    printf("  -t, --txcount <number> Number of CPUs to use for TX\n");
    printf("                         default: cpu_count+1/2\n");
    printf("  -b, --txbatch <length> Number of packets per TX batch\n");
    printf("                         default: %d\n", BATCH_LEN_MAX);
    printf("  -p, --plain            Plain input queue for packet RX\n");
    printf("                         default: disabled (use scheduler)\n");
    printf("  -R, --rxbatch <length> Number of packets per RX batch\n");
    printf("                         default: %d\n", BATCH_LEN_MAX);
    printf("  -l, --length <length>  Additional payload length in bytes\n");
    printf("                         default: 0\n");
    printf("  -r, --rate <number>    Attempted packet rate in PPS\n");
    printf("  -i, --interface <list> List of interface names to use\n");
    printf("  -d, --duration <secs>  Duration of each test iteration\n");
    printf("  -v, --verbose          Print verbose information\n");
    printf("  -h, --help             This help\n");
    printf("\n");
}
 
static void parse_args(int argc, char *argv[], test_args_t *args)
{
    int opt;
 
    static const struct option longopts[] = {
        {"count",     required_argument, NULL, 'c'},
        {"txcount",   required_argument, NULL, 't'},
        {"txbatch",   required_argument, NULL, 'b'},
        {"plain",     no_argument,       NULL, 'p'},
        {"rxbatch",   required_argument, NULL, 'R'},
        {"length",    required_argument, NULL, 'l'},
        {"rate",      required_argument, NULL, 'r'},
        {"interface", required_argument, NULL, 'i'},
        {"duration",  required_argument, NULL, 'd'},
        {"verbose",   no_argument,       NULL, 'v'},
        {"help",      no_argument,       NULL, 'h'},
        {NULL, 0, NULL, 0}
    };
 
    static const char *shortopts = "+c:t:b:pR:l:r:i:d:vh";
 
    args->cpu_count      = 2;
    args->num_tx_workers = 0; /* defaults to cpu_count+1/2 */
    args->tx_batch_len   = BATCH_LEN_MAX;
    args->rx_batch_len   = BATCH_LEN_MAX;
    args->duration       = DEFAULT_DURATION;
    args->pps            = RATE_SEARCH_INITIAL_PPS;
    args->search         = 1;
    args->schedule       = 1;
    args->verbose        = 0;
 
    while (1) {
        opt = getopt_long(argc, argv, shortopts,
                  longopts, NULL);
 
        if (opt == -1)
            break;
 
        switch (opt) {
        case 'h':
            usage();
            exit(EXIT_SUCCESS);
        case 'c':
            args->cpu_count = atoi(optarg);
            break;
        case 't':
            args->num_tx_workers = atoi(optarg);
            break;
        case 'd':
            args->duration = atoi(optarg) * T_SCALE;
            break;
        case 'r':
            args->pps     = atoi(optarg);
            args->search  = 0;
            args->verbose = 1;
            break;
        case 'i':
        {
            char *token;
 
            args->if_str = malloc(strlen(optarg) + 1);
 
            if (!args->if_str)
                ODPH_ABORT("Failed to alloc iface storage\n");
 
            strcpy(args->if_str, optarg);
            for (token = strtok(args->if_str, ",");
                 token != NULL && args->num_ifaces < MAX_NUM_IFACES;
                 token = strtok(NULL, ","))
                args->ifaces[args->num_ifaces++] = token;
        }
            break;
        case 'p':
            args->schedule = 0;
            break;
        case 'b':
            args->tx_batch_len = atoi(optarg);
            break;
        case 'R':
            args->rx_batch_len = atoi(optarg);
            break;
        case 'v':
            args->verbose = 1;
            break;
        case 'l':
            args->pkt_len = atoi(optarg);
            break;
        }
    }
 
    if (args->num_ifaces == 0) {
        args->ifaces[0] = "loop";
        args->num_ifaces = 1;
    }
}
 
int main(int argc, char **argv)
{
    int ret;
    odp_shm_t shm;
    int max_thrs;
    odph_helper_options_t helper_options;
    odp_instance_t instance;
    odp_init_t init_param;
 
    /* Let helper collect its own arguments (e.g. --odph_proc) */
    argc = odph_parse_options(argc, argv);
    if (odph_options(&helper_options)) {
        ODPH_ERR("Error: reading ODP helper options failed.\n");
        exit(EXIT_FAILURE);
    }
 
    odp_init_param_init(&init_param);
    init_param.mem_model = helper_options.mem_model;
 
    if (odp_init_global(&instance, &init_param, NULL) != 0)
        ODPH_ABORT("Failed global init.\n");
 
    if (odp_init_local(instance, ODP_THREAD_CONTROL) != 0)
        ODPH_ABORT("Failed local init.\n");
 
    odp_sys_info_print();
 
    shm = odp_shm_reserve("test_globals",
                  sizeof(test_globals_t), ODP_CACHE_LINE_SIZE, 0);
    if (shm == ODP_SHM_INVALID)
        ODPH_ABORT("Shared memory reserve failed.\n");
 
    gbl_args = odp_shm_addr(shm);
    if (gbl_args == NULL)
        ODPH_ABORT("Shared memory reserve failed.\n");
    memset(gbl_args, 0, sizeof(test_globals_t));
 
    max_thrs = odp_thread_count_max();
 
    gbl_args->instance = instance;
    gbl_args->rx_stats_size = max_thrs * sizeof(pkt_rx_stats_t);
    gbl_args->tx_stats_size = max_thrs * sizeof(pkt_tx_stats_t);
 
    shm = odp_shm_reserve("test_globals.rx_stats",
                  gbl_args->rx_stats_size,
                  ODP_CACHE_LINE_SIZE, 0);
    if (shm == ODP_SHM_INVALID)
        ODPH_ABORT("Shared memory reserve failed.\n");
 
    gbl_args->rx_stats = odp_shm_addr(shm);
 
    if (gbl_args->rx_stats == NULL)
        ODPH_ABORT("Shared memory reserve failed.\n");
 
    memset(gbl_args->rx_stats, 0, gbl_args->rx_stats_size);
 
    shm = odp_shm_reserve("test_globals.tx_stats",
                  gbl_args->tx_stats_size,
                  ODP_CACHE_LINE_SIZE, 0);
    if (shm == ODP_SHM_INVALID)
        ODPH_ABORT("Shared memory reserve failed.\n");
 
    gbl_args->tx_stats = odp_shm_addr(shm);
 
    if (gbl_args->tx_stats == NULL)
        ODPH_ABORT("Shared memory reserve failed.\n");
 
    memset(gbl_args->tx_stats, 0, gbl_args->tx_stats_size);
 
    parse_args(argc, argv, &gbl_args->args);
 
    ret = test_init();
 
    if (ret == 0) {
        ret = run_test();
        test_term();
    }
 
    odp_term_local();
    odp_term_global(instance);
 
    return ret;
}