odp_dmafwd.c

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright (c) 2023 Nokia
 */
 
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
 
#include <inttypes.h>
#include <stdlib.h>
#include <signal.h>
#include <stdio.h>
#include <unistd.h>
#include <time.h>
 
#include <odp_api.h>
#include <odp/helper/odph_api.h>
 
#define EXIT_NOT_SUP 2
#define PROG_NAME "odp_dmafwd"
#define DELIMITER ","
 
enum {
    SW_COPY = 0U,
    DMA_COPY_EV,
    DMA_COPY_POLL
};
 
#define DEF_CPY_TYPE SW_COPY
#define DEF_CNT 32768U
#define DEF_LEN 1024U
#define DEF_WORKERS 1U
#define DEF_TIME 0U
 
#define MAX_IFS 2U
#define MAX_OUT_QS 32U
#define MAX_BURST 32U
#define MAX_WORKERS (ODP_THREAD_COUNT_MAX - 1)
 
#define DIV_IF(a, b) ((b) > 0U ? ((a) / (b)) : 0U)
 
ODP_STATIC_ASSERT(MAX_IFS < UINT8_MAX, "Too large maximum interface count");
ODP_STATIC_ASSERT(MAX_OUT_QS < UINT8_MAX, "Too large maximum output queue count");
 
typedef struct {
    uint32_t burst_size;
    uint32_t num_pkts;
    uint32_t pkt_len;
    uint32_t cache_size;
} dynamic_defs_t;
 
typedef enum {
    PRS_OK,
    PRS_NOK,
    PRS_TERM,
    PRS_NOT_SUP
} parse_result_t;
 
typedef struct prog_config_s prog_config_t;
 
typedef struct {
    uint64_t copy_errs;
    uint64_t trs;
    uint64_t start_errs;
    uint64_t trs_errs;
    uint64_t buf_alloc_errs;
    uint64_t compl_alloc_errs;
    uint64_t pkt_alloc_errs;
    uint64_t trs_poll_errs;
    uint64_t trs_polled;
    uint64_t fwd_pkts;
    uint64_t discards;
    uint64_t sched_cc;
    uint64_t tot_cc;
    uint64_t sched_rounds;
    uint64_t pkt_cleanup;
} stats_t;
 
typedef struct ODP_ALIGNED_CACHE {
    prog_config_t *prog_config;
    odp_dma_t dma_handle;
    odp_pool_t compl_pool;
    odp_pool_t copy_pool;
    odp_pool_t trs_pool;
    odp_queue_t compl_q;
    odp_stash_t inflight_stash;
    stats_t stats;
    int thr_idx;
} thread_config_t;
 
typedef struct pktio_s {
    odp_pktout_queue_t out_qs[MAX_OUT_QS];
    char *name;
    odp_pktio_t handle;
    uint8_t num_out_qs;
} pktio_t;
 
typedef struct {
    odp_packet_t src_pkts[MAX_BURST];
    odp_packet_t dst_pkts[MAX_BURST];
    pktio_t *pktio;
    int num;
} transfer_t;
 
/* Function for initializing transfer structures */
typedef transfer_t *(*init_fn_t)(odp_dma_transfer_param_t *trs_param,
                 odp_dma_compl_param_t *compl_param, odp_dma_seg_t *src_segs,
                 odp_dma_seg_t *dst_segs, pktio_t *pktio, thread_config_t *config);
/* Function for starting transfers */
typedef odp_bool_t (*start_fn_t)(odp_dma_transfer_param_t *trs_param,
                 odp_dma_compl_param_t *compl_param, thread_config_t *config);
/* Function for setting up packets for copy */
typedef void (*pkt_fn_t)(odp_packet_t pkts[], int num, pktio_t *pktio, init_fn_t init_fn,
             start_fn_t start_fn, thread_config_t *config);
/* Function for draining and tearing down inflight operations */
typedef void (*drain_fn_t)(thread_config_t *config);
 
typedef struct prog_config_s {
    uint8_t pktio_idx_map[ODP_PKTIO_MAX_INDEX + 1];
    odph_thread_t thread_tbl[MAX_WORKERS];
    thread_config_t thread_config[MAX_WORKERS];
    pktio_t pktios[MAX_IFS];
    dynamic_defs_t dyn_defs;
    odp_instance_t odp_instance;
    odp_barrier_t init_barrier;
    odp_barrier_t term_barrier;
    odp_atomic_u32_t is_running;
    odp_pool_t pktio_pool;
    odp_pool_t copy_pool;
    odp_pool_t trs_pool;
 
    struct {
        init_fn_t init_fn;
        start_fn_t start_fn;
        pkt_fn_t pkt_fn;
        drain_fn_t drain_fn;
    };
 
    uint64_t inflight_obj_size;
    uint32_t burst_size;
    uint32_t num_pkts;
    uint32_t pkt_len;
    uint32_t cache_size;
    uint32_t num_inflight;
    uint32_t trs_cache_size;
    uint32_t compl_cache_size;
    uint32_t stash_cache_size;
    double time_sec;
    odp_stash_type_t stash_type;
    int num_thrs;
    uint8_t num_ifs;
    uint8_t copy_type;
} prog_config_t;
 
typedef struct {
    odp_packet_t pkts[MAX_BURST * 2U];
    pktio_t *pktio;
    int num;
} pkt_vec_t;
 
static prog_config_t *prog_conf;
 
static void terminate(int signal ODP_UNUSED)
{
    odp_atomic_store_u32(&prog_conf->is_running, 0U);
}
 
static void init_config(prog_config_t *config)
{
    odp_dma_capability_t dma_capa;
    uint32_t burst_size;
    odp_pool_capability_t pool_capa;
    odp_pool_param_t pool_param;
    thread_config_t *thr;
 
    memset(config, 0, sizeof(*config));
 
    if (odp_dma_capability(&dma_capa) == 0) {
        burst_size = ODPH_MIN(dma_capa.max_src_segs, dma_capa.max_dst_segs);
        burst_size = ODPH_MIN(burst_size, MAX_BURST);
        config->dyn_defs.burst_size = burst_size;
    }
 
    if (odp_pool_capability(&pool_capa) == 0) {
        config->dyn_defs.num_pkts = pool_capa.pkt.max_num > 0U ?
                        ODPH_MIN(pool_capa.pkt.max_num, DEF_CNT) : DEF_CNT;
        config->dyn_defs.pkt_len = pool_capa.pkt.max_len > 0U ?
                        ODPH_MIN(pool_capa.pkt.max_len, DEF_LEN) : DEF_LEN;
        odp_pool_param_init(&pool_param);
        config->dyn_defs.cache_size = pool_param.pkt.cache_size;
    }
 
    config->pktio_pool = ODP_POOL_INVALID;
    config->copy_pool = ODP_POOL_INVALID;
    config->trs_pool = ODP_POOL_INVALID;
    config->burst_size = config->dyn_defs.burst_size;
    config->num_pkts = config->dyn_defs.num_pkts;
    config->pkt_len = config->dyn_defs.pkt_len;
    config->cache_size = config->dyn_defs.cache_size;
    config->time_sec = DEF_TIME;
    config->num_thrs = DEF_WORKERS;
    config->copy_type = DEF_CPY_TYPE;
 
    for (int i = 0; i < MAX_WORKERS; ++i) {
        thr = &config->thread_config[i];
        thr->dma_handle = ODP_DMA_INVALID;
        thr->compl_pool = ODP_POOL_INVALID;
        thr->compl_q = ODP_QUEUE_INVALID;
        thr->inflight_stash = ODP_STASH_INVALID;
    }
 
    for (uint32_t i = 0U; i < MAX_IFS; ++i)
        config->pktios[i].handle = ODP_PKTIO_INVALID;
}
 
static void print_usage(dynamic_defs_t *dyn_defs)
{
    printf("\n"
           "DMA performance tester with packet I/O. Receive and forward packets after\n"
           "software copy or DMA offload copy.\n"
           "\n"
           "Usage: " PROG_NAME " OPTIONS\n"
           "\n"
           "  E.g. " PROG_NAME " -i eth0\n"
           "       " PROG_NAME " -i eth0 -t 0\n"
           "       " PROG_NAME " -i eth0 -t 1 -b 15 -l 4096 -c 5\n"
           "\n"
           "Mandatory OPTIONS:\n"
           "\n"
           "  -i, --interfaces   Ethernet interfaces for packet I/O, comma-separated, no\n"
           "                     spaces.\n"
           "\n"
           "Optional OPTIONS:\n"
           "\n"
           "  -t, --copy_type    Type of copy. %u by default.\n"
           "                         0: SW\n"
           "                         1: DMA with event completion\n"
           "                         2: DMA with poll completion\n"
           "  -b, --burst_size   Copy burst size. This many packets are accumulated before\n"
           "                     copy. %u by default.\n"
           "  -n, --num_pkts     Number of packet buffers allocated for packet I/O pool.\n"
           "                     %u by default.\n"
           "  -l, --pkt_len      Maximum size of packet buffers in packet I/O pool. %u by\n"
           "                     default.\n"
           "  -c, --worker_count Amount of workers. %u by default.\n"
           "  -C, --cache_size   Maximum cache size for pools. %u by default.\n"
           "  -T, --time_sec     Time in seconds to run. 0 means infinite. %u by default.\n"
           "  -h, --help         This help.\n"
           "\n", DEF_CPY_TYPE, dyn_defs->burst_size, dyn_defs->num_pkts, dyn_defs->pkt_len,
           DEF_WORKERS, dyn_defs->cache_size, DEF_TIME);
}
 
static void parse_interfaces(prog_config_t *config, const char *optarg)
{
    char *tmp_str = strdup(optarg), *tmp;
 
    if (tmp_str == NULL)
        return;
 
    tmp = strtok(tmp_str, DELIMITER);
 
    while (tmp && config->num_ifs < MAX_IFS) {
        config->pktios[config->num_ifs].name = strdup(tmp);
 
        if (config->pktios[config->num_ifs].name != NULL)
            ++config->num_ifs;
 
        tmp = strtok(NULL, DELIMITER);
    }
 
    free(tmp_str);
}
 
static odp_bool_t get_stash_capa(odp_stash_capability_t *stash_capa, odp_stash_type_t *stash_type)
{
    if (odp_stash_capability(stash_capa, ODP_STASH_TYPE_FIFO) == 0) {
        *stash_type = ODP_STASH_TYPE_FIFO;
        return true;
    }
 
    if (odp_stash_capability(stash_capa, ODP_STASH_TYPE_DEFAULT) == 0) {
        *stash_type = ODP_STASH_TYPE_DEFAULT;
        return true;
    }
 
    return false;
}
 
static parse_result_t check_options(prog_config_t *config)
{
    odp_dma_capability_t dma_capa;
    uint32_t burst_size;
    odp_stash_capability_t stash_capa;
    const uint64_t obj_size = sizeof(odp_dma_transfer_id_t);
    uint64_t max_num;
    odp_pool_capability_t pool_capa;
 
    if (config->num_ifs == 0U) {
        ODPH_ERR("Invalid number of interfaces: %u (min: 1, max: %u)\n", config->num_ifs,
             MAX_IFS);
        return PRS_NOK;
    }
 
    if (config->copy_type != SW_COPY && config->copy_type != DMA_COPY_EV &&
        config->copy_type != DMA_COPY_POLL) {
        ODPH_ERR("Invalid copy type: %u\n", config->copy_type);
        return PRS_NOK;
    }
 
    if (config->num_thrs <= 0 || config->num_thrs > MAX_WORKERS) {
        ODPH_ERR("Invalid worker count: %d (min: 1, max: %d)\n", config->num_thrs,
             MAX_WORKERS);
        return PRS_NOK;
    }
 
    if (odp_dma_capability(&dma_capa) < 0) {
        ODPH_ERR("Error querying DMA capabilities\n");
        return PRS_NOK;
    }
 
    if ((uint32_t)config->num_thrs > dma_capa.max_sessions) {
        ODPH_ERR("Unsupported DMA session count: %d (max: %u)\n", config->num_thrs,
             dma_capa.max_sessions);
        return PRS_NOT_SUP;
    }
 
    burst_size = ODPH_MIN(dma_capa.max_src_segs, dma_capa.max_dst_segs);
    burst_size = ODPH_MIN(burst_size, MAX_BURST);
 
    if (config->burst_size == 0U || config->burst_size > burst_size) {
        ODPH_ERR("Invalid segment count for DMA: %u (min: 1, max: %u)\n",
             config->burst_size, burst_size);
        return PRS_NOK;
    }
 
    if (config->pkt_len > dma_capa.max_seg_len) {
        ODPH_ERR("Invalid packet length for DMA: %u (max: %u)\n", config->pkt_len,
             dma_capa.max_seg_len);
        return PRS_NOK;
    }
 
    config->num_inflight = dma_capa.max_transfers;
 
    if (odp_pool_capability(&pool_capa) < 0) {
        ODPH_ERR("Error querying pool capabilities\n");
        return PRS_NOK;
    }
 
    if (config->cache_size < pool_capa.pkt.min_cache_size ||
        config->cache_size > pool_capa.pkt.max_cache_size) {
        ODPH_ERR("Invalid pool cache size: %u (min: %u, max: %u)\n", config->cache_size,
             pool_capa.pkt.min_cache_size, pool_capa.pkt.max_cache_size);
        return PRS_NOK;
    }
 
    if (config->copy_type != SW_COPY)
        config->trs_cache_size = ODPH_MIN(ODPH_MAX(config->cache_size,
                               pool_capa.buf.min_cache_size),
                          pool_capa.buf.max_cache_size);
 
    if (config->copy_type == DMA_COPY_EV) {
        if ((dma_capa.compl_mode_mask & ODP_DMA_COMPL_EVENT) == 0U ||
            !dma_capa.queue_type_sched) {
            ODPH_ERR("Unsupported DMA completion mode: event (mode support: %x, "
                 "scheduled queue support: %u)\n", dma_capa.compl_mode_mask,
                 dma_capa.queue_type_sched);
            return PRS_NOT_SUP;
        }
 
        if ((uint32_t)config->num_thrs > dma_capa.pool.max_pools) {
            ODPH_ERR("Invalid amount of DMA completion pools: %d (max: %u)\n",
                 config->num_thrs, dma_capa.pool.max_pools);
            return PRS_NOK;
        }
 
        if (config->num_inflight > dma_capa.pool.max_num) {
            ODPH_ERR("Invalid amount of DMA completion events: %u (max: %u)\n",
                 config->num_inflight, dma_capa.pool.max_num);
            return PRS_NOK;
        }
 
        config->compl_cache_size = ODPH_MIN(ODPH_MAX(config->cache_size,
                                 dma_capa.pool.min_cache_size),
                            dma_capa.pool.max_cache_size);
    } else if (config->copy_type == DMA_COPY_POLL) {
        if ((dma_capa.compl_mode_mask & ODP_DMA_COMPL_POLL) == 0U) {
            ODPH_ERR("Unsupported DMA completion mode: poll (mode support: %x)\n",
                 dma_capa.compl_mode_mask);
            return PRS_NOT_SUP;
        }
 
        if (!get_stash_capa(&stash_capa, &config->stash_type)) {
            ODPH_ERR("Error querying stash capabilities\n");
            return PRS_NOK;
        }
 
        if ((uint32_t)config->num_thrs > stash_capa.max_stashes) {
            ODPH_ERR("Invalid amount of stashes: %d (max: %u)\n", config->num_thrs,
                 stash_capa.max_stashes);
            return PRS_NOK;
        }
 
        if (obj_size == sizeof(uint8_t)) {
            max_num = stash_capa.max_num.u8;
        } else if (obj_size == sizeof(uint16_t)) {
            max_num = stash_capa.max_num.u16;
        } else if (obj_size <= sizeof(uint32_t)) {
            max_num = stash_capa.max_num.u32;
        } else if (obj_size <= sizeof(uint64_t)) {
            max_num = stash_capa.max_num.u64;
        } else if (obj_size <= sizeof(odp_u128_t)) {
            max_num = stash_capa.max_num.u128;
        } else {
            ODPH_ERR("Invalid stash object size: %" PRIu64 "\n", obj_size);
            return PRS_NOK;
        }
 
        if (config->num_inflight > max_num) {
            ODPH_ERR("Invalid stash size: %u (max: %" PRIu64 ")\n",
                 config->num_inflight, max_num);
            return PRS_NOK;
        }
 
        config->inflight_obj_size = obj_size;
        config->stash_cache_size = ODPH_MIN(config->cache_size, stash_capa.max_cache_size);
    }
 
    if (config->num_pkts == 0U ||
        (pool_capa.pkt.max_num > 0U && config->num_pkts > pool_capa.pkt.max_num)) {
        ODPH_ERR("Invalid pool packet count: %u (min: 1, max: %u)\n", config->num_pkts,
             pool_capa.pkt.max_num);
        return PRS_NOK;
    }
 
    if (config->pkt_len == 0U ||
        (pool_capa.pkt.max_len > 0U && config->pkt_len > pool_capa.pkt.max_len)) {
        ODPH_ERR("Invalid pool packet length: %u (min: 1, max: %u)\n", config->pkt_len,
             pool_capa.pkt.max_len);
        return PRS_NOK;
    }
 
    if (config->num_inflight > pool_capa.buf.max_num) {
        ODPH_ERR("Invalid pool buffer count: %u (max: %u)\n", config->num_inflight,
             pool_capa.buf.max_num);
        return PRS_NOK;
    }
 
    return PRS_OK;
}
 
static parse_result_t parse_options(int argc, char **argv, prog_config_t *config)
{
    int opt;
 
    static const struct option longopts[] = {
        { "interfaces", required_argument, NULL, 'i' },
        { "copy_type", required_argument, NULL, 't' },
        { "burst_size", required_argument, NULL, 'b' },
        { "num_pkts", required_argument, NULL, 'n' },
        { "pkt_len", required_argument, NULL, 'l' },
        { "worker_count", required_argument, NULL, 'c' },
        { "cache_size", required_argument, NULL, 'C' },
        { "time_sec", required_argument, NULL, 'T' },
        { "help", no_argument, NULL, 'h' },
        { NULL, 0, NULL, 0 }
    };
 
    static const char *shortopts = "i:t:b:n:l:c:C:T:h";
 
    init_config(config);
 
    while (1) {
        opt = getopt_long(argc, argv, shortopts, longopts, NULL);
 
        if (opt == -1)
            break;
 
        switch (opt) {
        case 'i':
            parse_interfaces(config, optarg);
            break;
        case 't':
            config->copy_type = atoi(optarg);
            break;
        case 'b':
            config->burst_size = atoi(optarg);
            break;
        case 'n':
            config->num_pkts = atoi(optarg);
            break;
        case 'l':
            config->pkt_len = atoi(optarg);
            break;
        case 'c':
            config->num_thrs = atoi(optarg);
            break;
        case 'C':
            config->cache_size = atoi(optarg);
            break;
        case 'T':
            config->time_sec = atof(optarg);
            break;
        case 'h':
            print_usage(&config->dyn_defs);
            return PRS_TERM;
        case '?':
        default:
            print_usage(&config->dyn_defs);
            return PRS_NOK;
        }
    }
 
    return check_options(config);
}
 
static parse_result_t setup_program(int argc, char **argv, prog_config_t *config)
{
    struct sigaction action = { .sa_handler = terminate };
 
    if (sigemptyset(&action.sa_mask) == -1 || sigaddset(&action.sa_mask, SIGINT) == -1 ||
        sigaddset(&action.sa_mask, SIGTERM) == -1 ||
        sigaddset(&action.sa_mask, SIGHUP) == -1 || sigaction(SIGINT, &action, NULL) == -1 ||
        sigaction(SIGTERM, &action, NULL) == -1 || sigaction(SIGHUP, &action, NULL) == -1) {
        ODPH_ERR("Error installing signal handler\n");
        return PRS_NOK;
    }
 
    return parse_options(argc, argv, config);
}
 
static inline int send_packets(odp_pktout_queue_t queue, odp_packet_t pkts[], int num)
{
    int ret = odp_pktout_send(queue, pkts, num);
 
    if (odp_unlikely(ret < num)) {
        ret = ret < 0 ? 0 : ret;
        odp_packet_free_multi(&pkts[ret], num - ret);
    }
 
    return ret;
}
 
static void sw_copy_and_send_packets(odp_packet_t pkts[], int num, pktio_t *pktio,
                     init_fn_t init_fn ODP_UNUSED, start_fn_t start_fn ODP_UNUSED,
                     thread_config_t *config)
{
    odp_packet_t old_pkt, new_pkt;
    odp_pool_t copy_pool = config->copy_pool;
    odp_packet_t out_pkts[num];
    int num_out_pkts = 0, num_sent;
    stats_t *stats = &config->stats;
 
    for (int i = 0; i < num; ++i) {
        old_pkt = pkts[i];
        new_pkt = odp_packet_copy(old_pkt, copy_pool);
 
        if (new_pkt != ODP_PACKET_INVALID)
            out_pkts[num_out_pkts++] = new_pkt;
        else
            ++stats->copy_errs;
 
        odp_packet_free(old_pkt);
    }
 
    if (num_out_pkts > 0) {
        num_sent = send_packets(pktio->out_qs[config->thr_idx % pktio->num_out_qs],
                    out_pkts, num_out_pkts);
        stats->fwd_pkts += num_sent;
        stats->discards += num_out_pkts - num_sent;
    }
}
 
static transfer_t *init_dma_ev_trs(odp_dma_transfer_param_t *trs_param,
                   odp_dma_compl_param_t *compl_param, odp_dma_seg_t *src_segs,
                   odp_dma_seg_t *dst_segs, pktio_t *pktio,
                   thread_config_t *config)
{
    odp_buffer_t buf;
    stats_t *stats = &config->stats;
    transfer_t *trs;
    odp_dma_compl_t c_ev;
 
    buf = odp_buffer_alloc(config->trs_pool);
 
    if (odp_unlikely(buf == ODP_BUFFER_INVALID)) {
        ++stats->buf_alloc_errs;
        return NULL;
    }
 
    trs = (transfer_t *)odp_buffer_addr(buf);
    trs->num = 0;
    trs->pktio = pktio;
    trs_param->src_format = ODP_DMA_FORMAT_PACKET;
    trs_param->dst_format = ODP_DMA_FORMAT_PACKET;
    trs_param->num_src = 0U;
    trs_param->num_dst = 0U;
    trs_param->src_seg = src_segs;
    trs_param->dst_seg = dst_segs;
    compl_param->compl_mode = ODP_DMA_COMPL_EVENT;
    c_ev = odp_dma_compl_alloc(config->compl_pool);
 
    if (odp_unlikely(c_ev == ODP_DMA_COMPL_INVALID)) {
        odp_buffer_free(buf);
        ++stats->compl_alloc_errs;
        return NULL;
    }
 
    compl_param->event = odp_dma_compl_to_event(c_ev);
    compl_param->queue = config->compl_q;
    compl_param->user_ptr = buf;
    memset(src_segs, 0, sizeof(*src_segs) * MAX_BURST);
    memset(dst_segs, 0, sizeof(*dst_segs) * MAX_BURST);
 
    return trs;
}
 
static transfer_t *init_dma_poll_trs(odp_dma_transfer_param_t *trs_param,
                     odp_dma_compl_param_t *compl_param, odp_dma_seg_t *src_segs,
                     odp_dma_seg_t *dst_segs, pktio_t *pktio,
                     thread_config_t *config)
{
    odp_buffer_t buf;
    stats_t *stats = &config->stats;
    transfer_t *trs;
 
    buf = odp_buffer_alloc(config->trs_pool);
 
    if (odp_unlikely(buf == ODP_BUFFER_INVALID)) {
        ++stats->buf_alloc_errs;
        return NULL;
    }
 
    trs = (transfer_t *)odp_buffer_addr(buf);
    trs->num = 0;
    trs->pktio = pktio;
    trs_param->src_format = ODP_DMA_FORMAT_PACKET;
    trs_param->dst_format = ODP_DMA_FORMAT_PACKET;
    trs_param->num_src = 0U;
    trs_param->num_dst = 0U;
    trs_param->src_seg = src_segs;
    trs_param->dst_seg = dst_segs;
    compl_param->compl_mode = ODP_DMA_COMPL_POLL;
    compl_param->transfer_id = odp_dma_transfer_id_alloc(config->dma_handle);
 
    if (odp_unlikely(compl_param->transfer_id == ODP_DMA_TRANSFER_ID_INVALID)) {
        odp_buffer_free(buf);
        ++stats->compl_alloc_errs;
        return NULL;
    }
 
    compl_param->user_ptr = buf;
    memset(src_segs, 0, sizeof(*src_segs) * MAX_BURST);
    memset(dst_segs, 0, sizeof(*dst_segs) * MAX_BURST);
 
    return trs;
}
 
static odp_bool_t start_dma_ev_trs(odp_dma_transfer_param_t *trs_param,
                   odp_dma_compl_param_t *compl_param, thread_config_t *config)
{
    const int ret = odp_dma_transfer_start(config->dma_handle, trs_param, compl_param);
 
    if (odp_unlikely(ret <= 0)) {
        odp_buffer_free(compl_param->user_ptr);
        odp_event_free(compl_param->event);
        return false;
    }
 
    return true;
}
 
static odp_bool_t start_dma_poll_trs(odp_dma_transfer_param_t *trs_param,
                     odp_dma_compl_param_t *compl_param, thread_config_t *config)
{
    const int ret = odp_dma_transfer_start(config->dma_handle, trs_param, compl_param);
 
    if (odp_unlikely(ret <= 0)) {
        odp_buffer_free(compl_param->user_ptr);
        odp_dma_transfer_id_free(config->dma_handle, compl_param->transfer_id);
        return false;
    }
 
    if (odp_unlikely(odp_stash_put(config->inflight_stash, &compl_param->transfer_id, 1) != 1))
        /* Should not happen, but make it visible if it somehow does */
        ODPH_ABORT("DMA inflight transfer stash overflow, aborting");
 
    return true;
}
 
static void dma_copy(odp_packet_t pkts[], int num, pktio_t *pktio, init_fn_t init_fn,
             start_fn_t start_fn, thread_config_t *config)
{
    odp_dma_transfer_param_t trs_param;
    odp_dma_compl_param_t compl_param;
    odp_packet_t pkt;
    transfer_t *trs = NULL;
    odp_dma_seg_t src_segs[MAX_BURST], dst_segs[MAX_BURST];
    uint32_t num_segs = 0U, pkt_len;
    odp_pool_t copy_pool = config->copy_pool;
    stats_t *stats = &config->stats;
 
    odp_dma_transfer_param_init(&trs_param);
    odp_dma_compl_param_init(&compl_param);
 
    for (int i = 0; i < num; ++i) {
        pkt = pkts[i];
 
        if (odp_unlikely(trs == NULL)) {
            trs = init_fn(&trs_param, &compl_param, src_segs, dst_segs, pktio, config);
 
            if (trs == NULL) {
                odp_packet_free(pkt);
                continue;
            }
        }
 
        pkt_len = odp_packet_len(pkt);
        src_segs[num_segs].packet = pkt;
        src_segs[num_segs].len = pkt_len;
        dst_segs[num_segs].packet = odp_packet_alloc(copy_pool, pkt_len);
 
        if (odp_unlikely(dst_segs[num_segs].packet == ODP_PACKET_INVALID)) {
            odp_packet_free(pkt);
            ++stats->pkt_alloc_errs;
            continue;
        }
 
        dst_segs[num_segs].len = pkt_len;
        trs->src_pkts[num_segs] = src_segs[num_segs].packet;
        trs->dst_pkts[num_segs] = dst_segs[num_segs].packet;
        ++trs->num;
        ++trs_param.num_src;
        ++trs_param.num_dst;
        ++num_segs;
    }
 
    if (num_segs > 0U)
        if (odp_unlikely(!start_fn(&trs_param, &compl_param, config))) {
            odp_packet_free_multi(trs->src_pkts, trs->num);
            odp_packet_free_multi(trs->dst_pkts, trs->num);
            ++stats->start_errs;
        }
}
 
static void drain_events(thread_config_t *config ODP_UNUSED)
{
    odp_event_t ev;
    odp_event_type_t type;
    odp_dma_result_t res;
    odp_buffer_t buf;
    transfer_t *trs;
 
    while (true) {
        ev = odp_schedule(NULL, odp_schedule_wait_time(100 * ODP_TIME_MSEC_IN_NS));
 
        if (ev == ODP_EVENT_INVALID)
            break;
 
        type = odp_event_type(ev);
 
        if (type == ODP_EVENT_DMA_COMPL) {
            memset(&res, 0, sizeof(res));
            odp_dma_compl_result(odp_dma_compl_from_event(ev), &res);
            buf = (odp_buffer_t)res.user_ptr;
            trs = (transfer_t *)odp_buffer_addr(buf);
            odp_packet_free_multi(trs->src_pkts, trs->num);
            odp_packet_free_multi(trs->dst_pkts, trs->num);
            odp_buffer_free(buf);
        }
 
        odp_event_free(ev);
    }
}
 
static void drain_polled(thread_config_t *config)
{
    odp_dma_transfer_id_t id;
    odp_dma_result_t res;
    int ret;
    odp_buffer_t buf;
    transfer_t *trs;
 
    while (true) {
        if (odp_stash_get(config->inflight_stash, &id, 1) != 1)
            break;
 
        memset(&res, 0, sizeof(res));
 
        do {
            ret = odp_dma_transfer_done(config->dma_handle, id, &res);
        } while (ret == 0);
 
        odp_dma_transfer_id_free(config->dma_handle, id);
 
        if (ret < 0)
            continue;
 
        buf = (odp_buffer_t)res.user_ptr;
        trs = (transfer_t *)odp_buffer_addr(buf);
        odp_packet_free_multi(trs->src_pkts, trs->num);
        odp_packet_free_multi(trs->dst_pkts, trs->num);
        odp_buffer_free(buf);
    }
}
 
static odp_bool_t setup_copy(prog_config_t *config)
{
    odp_pool_param_t pool_param;
    thread_config_t *thr;
    const odp_dma_param_t dma_param = {
        .direction = ODP_DMA_MAIN_TO_MAIN,
        .type = ODP_DMA_TYPE_COPY,
        .compl_mode_mask = ODP_DMA_COMPL_EVENT | ODP_DMA_COMPL_POLL,
        .mt_mode = ODP_DMA_MT_SERIAL,
        .order = ODP_DMA_ORDER_NONE };
    odp_dma_pool_param_t compl_pool_param;
    odp_queue_param_t queue_param;
    odp_stash_param_t stash_param;
 
    odp_pool_param_init(&pool_param);
    pool_param.pkt.seg_len = config->pkt_len;
    pool_param.pkt.len = config->pkt_len;
    pool_param.pkt.num = config->num_pkts;
    pool_param.pkt.cache_size = config->cache_size;
    pool_param.type = ODP_POOL_PACKET;
    config->copy_pool = odp_pool_create(PROG_NAME "_copy", &pool_param);
 
    if (config->copy_pool == ODP_POOL_INVALID) {
        ODPH_ERR("Error creating packet copy pool\n");
        return false;
    }
 
    if (config->copy_type == SW_COPY) {
        config->pkt_fn = sw_copy_and_send_packets;
 
        for (int i = 0; i < config->num_thrs; ++i)
            config->thread_config[i].copy_pool = config->copy_pool;
 
        return true;
    }
 
    pool_param.buf.num = config->num_inflight;
    pool_param.buf.size = sizeof(transfer_t);
    pool_param.buf.cache_size = config->trs_cache_size;
    pool_param.type = ODP_POOL_BUFFER;
    config->trs_pool = odp_pool_create(PROG_NAME "_dma_trs", &pool_param);
 
    if (config->trs_pool == ODP_POOL_INVALID) {
        ODPH_ERR("Error creating DMA transfer tracking pool\n");
        return false;
    }
 
    for (int i = 0; i < config->num_thrs; ++i) {
        thr = &config->thread_config[i];
        thr->copy_pool = config->copy_pool;
        thr->trs_pool = config->trs_pool;
        thr->dma_handle = odp_dma_create(PROG_NAME "_dma", &dma_param);
 
        if (thr->dma_handle == ODP_DMA_INVALID) {
            ODPH_ERR("Error creating DMA session\n");
            return false;
        }
 
        if (config->copy_type == DMA_COPY_EV) {
            odp_dma_pool_param_init(&compl_pool_param);
            compl_pool_param.num = config->num_inflight;
            compl_pool_param.cache_size = config->compl_cache_size;
            thr->compl_pool = odp_dma_pool_create(PROG_NAME "_dma_compl",
                                  &compl_pool_param);
 
            if (thr->compl_pool == ODP_POOL_INVALID) {
                ODPH_ERR("Error creating DMA event completion pool\n");
                return false;
            }
 
            odp_queue_param_init(&queue_param);
            queue_param.type = ODP_QUEUE_TYPE_SCHED;
            queue_param.sched.sync = ODP_SCHED_SYNC_PARALLEL;
            queue_param.sched.prio = odp_schedule_max_prio();
            thr->compl_q = odp_queue_create(PROG_NAME "_dma_compl", &queue_param);
 
            if (thr->compl_q == ODP_QUEUE_INVALID) {
                ODPH_ERR("Error creating DMA completion queue\n");
                return false;
            }
 
            config->init_fn = init_dma_ev_trs;
            config->start_fn = start_dma_ev_trs;
            config->drain_fn = drain_events;
        } else {
            odp_stash_param_init(&stash_param);
            stash_param.type = config->stash_type;
            stash_param.put_mode = ODP_STASH_OP_LOCAL;
            stash_param.get_mode = ODP_STASH_OP_LOCAL;
            stash_param.num_obj = config->num_inflight;
            stash_param.obj_size = config->inflight_obj_size;
            stash_param.cache_size = config->stash_cache_size;
            thr->inflight_stash = odp_stash_create("_dma_inflight", &stash_param);
 
            if (thr->inflight_stash == ODP_STASH_INVALID) {
                ODPH_ERR("Error creating DMA inflight transfer stash\n");
                return false;
            }
 
            config->init_fn = init_dma_poll_trs;
            config->start_fn = start_dma_poll_trs;
            config->drain_fn = drain_polled;
        }
    }
 
    config->pkt_fn = dma_copy;
 
    return true;
}
 
static odp_bool_t setup_pktios(prog_config_t *config)
{
    odp_pool_param_t pool_param;
    pktio_t *pktio;
    odp_pktio_param_t pktio_param;
    odp_pktio_capability_t capa;
    uint32_t num_input_qs, num_output_qs;
    odp_pktin_queue_param_t pktin_param;
    odp_pktout_queue_param_t pktout_param;
 
    odp_pool_param_init(&pool_param);
    pool_param.pkt.seg_len = config->pkt_len;
    pool_param.pkt.len = config->pkt_len;
    pool_param.pkt.num = config->num_pkts;
    pool_param.pkt.cache_size = config->cache_size;
    pool_param.type = ODP_POOL_PACKET;
    config->pktio_pool = odp_pool_create(PROG_NAME, &pool_param);
 
    if (config->pktio_pool == ODP_POOL_INVALID) {
        ODPH_ERR("Error creating packet I/O pool\n");
        return false;
    }
 
    for (uint32_t i = 0U; i < config->num_ifs; ++i) {
        pktio = &config->pktios[i];
        odp_pktio_param_init(&pktio_param);
        pktio_param.in_mode = ODP_PKTIN_MODE_SCHED;
        pktio_param.out_mode = ODP_PKTOUT_MODE_DIRECT;
        pktio->handle = odp_pktio_open(pktio->name, config->pktio_pool, &pktio_param);
 
        if (pktio->handle == ODP_PKTIO_INVALID) {
            ODPH_ERR("Error opening packet I/O (%s)\n", pktio->name);
            return false;
        }
 
        config->pktio_idx_map[odp_pktio_index(pktio->handle)] = i;
 
        if (odp_pktio_capability(pktio->handle, &capa) < 0) {
            ODPH_ERR("Error querying packet I/O capabilities (%s)\n", pktio->name);
            return false;
        }
 
        num_input_qs = ODPH_MIN((uint32_t)config->num_thrs, capa.max_input_queues);
        num_output_qs = ODPH_MIN((uint32_t)config->num_thrs, capa.max_output_queues);
        num_output_qs = ODPH_MIN(num_output_qs, MAX_OUT_QS);
        odp_pktin_queue_param_init(&pktin_param);
 
        if (num_input_qs > 1) {
            pktin_param.hash_enable = true;
            pktin_param.hash_proto.proto.ipv4 = 1U;
        }
 
        pktin_param.num_queues = num_input_qs;
        pktin_param.queue_param.sched.prio = odp_schedule_default_prio();
 
        if (odp_pktin_queue_config(pktio->handle, &pktin_param) < 0) {
            ODPH_ERR("Error configuring packet I/O input queues (%s)\n", pktio->name);
            return false;
        }
 
        odp_pktout_queue_param_init(&pktout_param);
 
        if (num_output_qs == (uint32_t)config->num_thrs)
            pktout_param.op_mode = ODP_PKTIO_OP_MT_UNSAFE;
 
        pktout_param.num_queues = num_output_qs;
        pktio->num_out_qs = num_output_qs;
 
        if (odp_pktout_queue_config(pktio->handle, &pktout_param) < 0) {
            ODPH_ERR("Error configuring packet I/O output queues (%s)\n", pktio->name);
            return false;
        }
 
        if (odp_pktout_queue(pktio->handle, pktio->out_qs, num_output_qs) !=
            (int)num_output_qs) {
            ODPH_ERR("Error querying packet I/O output queues (%s)\n", pktio->name);
            return false;
        }
 
        if (odp_pktio_start(pktio->handle) < 0) {
            ODPH_ERR("Error starting packet I/O (%s)\n", pktio->name);
            return false;
        }
    }
 
    return true;
}
 
static inline void send_dma_poll_trs_pkts(int burst_size, thread_config_t *config)
{
    odp_stash_t stash_handle = config->inflight_stash;
    odp_dma_transfer_id_t ids[burst_size], id;
    int32_t num;
    odp_dma_t dma_handle = config->dma_handle;
    odp_dma_result_t res;
    int ret;
    odp_buffer_t buf;
    transfer_t *trs;
    pktio_t *pktio;
    int num_sent;
    stats_t *stats = &config->stats;
 
    while (true) {
        num = odp_stash_get(stash_handle, &ids, burst_size);
 
        if (num <= 0)
            break;
 
        for (int32_t i = 0; i < num; ++i) {
            id = ids[i];
            ret = odp_dma_transfer_done(dma_handle, id, &res);
 
            if (ret == 0) {
                if (odp_unlikely(odp_stash_put(stash_handle, &id, 1) != 1))
                    /* Should not happen, but make it visible if it somehow
                     * does */
                    ODPH_ABORT("DMA inflight transfer stash overflow,"
                           " aborting");
 
                ++stats->trs_polled;
                continue;
            }
 
            odp_dma_transfer_id_free(dma_handle, id);
 
            if (ret < 0) {
                ++stats->trs_poll_errs;
                continue;
            }
 
            buf = (odp_buffer_t)res.user_ptr;
            trs = (transfer_t *)odp_buffer_addr(buf);
 
            if (res.success) {
                pktio = trs->pktio;
                num_sent = send_packets(pktio->out_qs[config->thr_idx %
                                    pktio->num_out_qs],
                            trs->dst_pkts, trs->num);
                ++stats->trs;
                stats->fwd_pkts += num_sent;
                stats->discards += trs->num - num_sent;
            } else {
                odp_packet_free_multi(trs->dst_pkts, trs->num);
                ++stats->trs_errs;
            }
 
            odp_packet_free_multi(trs->src_pkts, trs->num);
            odp_buffer_free(buf);
        }
    }
}
 
static inline void send_dma_ev_trs_pkts(odp_dma_compl_t compl_ev, thread_config_t *config)
{
    odp_dma_result_t res;
    odp_buffer_t buf;
    transfer_t *trs;
    pktio_t *pktio;
    int num_sent;
    stats_t *stats = &config->stats;
 
    memset(&res, 0, sizeof(res));
    odp_dma_compl_result(compl_ev, &res);
    buf = (odp_buffer_t)res.user_ptr;
    trs = (transfer_t *)odp_buffer_addr(buf);
 
    if (res.success) {
        pktio = trs->pktio;
        num_sent = send_packets(pktio->out_qs[config->thr_idx % pktio->num_out_qs],
                    trs->dst_pkts, trs->num);
        ++stats->trs;
        stats->fwd_pkts += num_sent;
        stats->discards += trs->num - num_sent;
    } else {
        odp_packet_free_multi(trs->dst_pkts, trs->num);
        ++stats->trs_errs;
    }
 
    odp_packet_free_multi(trs->src_pkts, trs->num);
    odp_buffer_free(buf);
    odp_dma_compl_free(compl_ev);
}
 
static inline void push_packet(odp_packet_t pkt, pkt_vec_t pkt_vecs[], uint8_t *pktio_idx_map)
{
    uint8_t idx = pktio_idx_map[odp_packet_input_index(pkt)];
    pkt_vec_t *pkt_vec = &pkt_vecs[idx];
 
    pkt_vec->pkts[pkt_vec->num++] = pkt;
}
 
static inline void pop_packets(pkt_vec_t *pkt_vec, int num_procd)
{
    pkt_vec->num -= num_procd;
 
    for (int i = 0, j = num_procd; i < pkt_vec->num; ++i, ++j)
        pkt_vec->pkts[i] = pkt_vec->pkts[j];
}
 
static void free_pending_packets(pkt_vec_t pkt_vecs[], uint32_t num_ifs, stats_t *stats)
{
    for (uint32_t i = 0U; i < num_ifs; ++i) {
        stats->pkt_cleanup += pkt_vecs[i].num;
        odp_packet_free_multi(pkt_vecs[i].pkts, pkt_vecs[i].num);
    }
}
 
static int process_packets(void *args)
{
    thread_config_t *config = args;
    const uint8_t num_ifs = config->prog_config->num_ifs;
    pkt_vec_t pkt_vecs[num_ifs], *pkt_vec;
    odp_atomic_u32_t *is_running = &config->prog_config->is_running;
    uint64_t c1, c2, c3, c4, cdiff = 0U, rounds = 0U;
    const uint8_t copy_type = config->prog_config->copy_type;
    const int burst_size = config->prog_config->burst_size;
    odp_event_t evs[burst_size];
    int num_evs;
    odp_event_t ev;
    odp_event_type_t type;
    uint8_t *pktio_map = config->prog_config->pktio_idx_map;
    stats_t *stats = &config->stats;
    init_fn_t init_fn = config->prog_config->init_fn;
    start_fn_t start_fn = config->prog_config->start_fn;
    pkt_fn_t pkt_fn = config->prog_config->pkt_fn;
 
    for (uint32_t i = 0U; i < num_ifs; ++i) {
        pkt_vecs[i].pktio = &config->prog_config->pktios[i];
        pkt_vecs[i].num = 0;
    }
 
    config->thr_idx = odp_thread_id();
    odp_barrier_wait(&config->prog_config->init_barrier);
    c1 = odp_cpu_cycles();
 
    while (odp_atomic_load_u32(is_running)) {
        c3 = odp_cpu_cycles();
        num_evs = odp_schedule_multi_no_wait(NULL, evs, burst_size);
        c4 = odp_cpu_cycles();
        cdiff += odp_cpu_cycles_diff(c4, c3);
        ++rounds;
 
        if (copy_type == DMA_COPY_POLL)
            send_dma_poll_trs_pkts(burst_size, config);
 
        if (num_evs == 0)
            continue;
 
        for (int i = 0; i < num_evs; ++i) {
            ev = evs[i];
            type = odp_event_type(ev);
 
            if (type == ODP_EVENT_DMA_COMPL) {
                send_dma_ev_trs_pkts(odp_dma_compl_from_event(ev), config);
            } else if (type == ODP_EVENT_PACKET) {
                push_packet(odp_packet_from_event(ev), pkt_vecs, pktio_map);
            } else {
                odp_event_free(ev);
                ++stats->discards;
            }
        }
 
        for (uint32_t i = 0U; i < num_ifs; ++i) {
            pkt_vec = &pkt_vecs[i];
 
            if (pkt_vec->num >= burst_size) {
                pkt_fn(pkt_vec->pkts, burst_size, pkt_vec->pktio, init_fn,
                       start_fn, config);
                pop_packets(pkt_vec, burst_size);
            }
        }
    }
 
    c2 = odp_cpu_cycles();
    stats->sched_cc = cdiff;
    stats->tot_cc = odp_cpu_cycles_diff(c2, c1);
    stats->sched_rounds = rounds;
    free_pending_packets(pkt_vecs, num_ifs, stats);
    odp_barrier_wait(&config->prog_config->term_barrier);
 
    if (config->prog_config->drain_fn)
        config->prog_config->drain_fn(config);
 
    return 0;
}
 
static odp_bool_t setup_workers(prog_config_t *config)
{
    odp_cpumask_t cpumask;
    int num_workers;
    odph_thread_common_param_t thr_common;
    odph_thread_param_t thr_param[config->num_thrs];
 
    num_workers = odp_cpumask_default_worker(&cpumask, config->num_thrs);
    odph_thread_common_param_init(&thr_common);
    thr_common.instance = config->odp_instance;
    thr_common.cpumask = &cpumask;
 
    for (int i = 0; i < config->num_thrs; ++i) {
        odph_thread_param_init(&thr_param[i]);
        thr_param[i].start = process_packets;
        thr_param[i].thr_type = ODP_THREAD_WORKER;
        config->thread_config[i].prog_config = config;
        thr_param[i].arg = &config->thread_config[i];
    }
 
    num_workers = odph_thread_create(config->thread_tbl, &thr_common, thr_param, num_workers);
 
    if (num_workers != config->num_thrs) {
        ODPH_ERR("Error configuring worker threads\n");
        return false;
    }
 
    return true;
}
 
static odp_bool_t setup_test(prog_config_t *config)
{
    odp_barrier_init(&config->init_barrier, config->num_thrs + 1);
    odp_barrier_init(&config->term_barrier, config->num_thrs + 1);
 
    if (!setup_copy(config))
        return false;
 
    if (!setup_pktios(config))
        return false;
 
    if (!setup_workers(config))
        return false;
 
    odp_barrier_wait(&config->init_barrier);
 
    return true;
}
 
static void stop_test(prog_config_t *config)
{
    for (uint32_t i = 0U; i < config->num_ifs; ++i)
        if (config->pktios[i].handle != ODP_PKTIO_INVALID)
            (void)odp_pktio_stop(config->pktios[i].handle);
 
    odp_barrier_wait(&config->term_barrier);
    (void)odph_thread_join(config->thread_tbl, config->num_thrs);
}
 
static void teardown(prog_config_t *config)
{
    thread_config_t *thr;
 
    for (uint32_t i = 0U; i < config->num_ifs; ++i) {
        free(config->pktios[i].name);
 
        if (config->pktios[i].handle != ODP_PKTIO_INVALID)
            (void)odp_pktio_close(config->pktios[i].handle);
    }
 
    if (config->pktio_pool != ODP_POOL_INVALID)
        (void)odp_pool_destroy(config->pktio_pool);
 
    for (int i = 0; i < config->num_thrs; ++i) {
        thr = &config->thread_config[i];
 
        if (thr->inflight_stash != ODP_STASH_INVALID)
            (void)odp_stash_destroy(thr->inflight_stash);
 
        if (thr->compl_q != ODP_QUEUE_INVALID)
            (void)odp_queue_destroy(thr->compl_q);
 
        if (thr->compl_pool != ODP_POOL_INVALID)
            (void)odp_pool_destroy(thr->compl_pool);
 
        if (thr->dma_handle != ODP_DMA_INVALID)
            (void)odp_dma_destroy(thr->dma_handle);
    }
 
    if (config->copy_pool != ODP_POOL_INVALID)
        (void)odp_pool_destroy(config->copy_pool);
 
    if (config->trs_pool != ODP_POOL_INVALID)
        (void)odp_pool_destroy(config->trs_pool);
}
 
static void print_stats(const prog_config_t *config)
{
    const stats_t *stats;
    const char *align1 = config->copy_type == DMA_COPY_EV ? " " : "";
    const char *align2 = config->copy_type == SW_COPY ? "  " :
                config->copy_type == DMA_COPY_EV ? "                  " :
                    "                 ";
 
    printf("\n==================\n\n"
           "DMA forwarder done\n\n"
           "    copy mode:       %s\n"
           "    burst size:      %u\n"
           "    packet length:   %u\n"
           "    max cache size:  %u\n", config->copy_type == SW_COPY ? "SW" :
           config->copy_type == DMA_COPY_EV ? "DMA-event" : "DMA-poll",
           config->burst_size, config->pkt_len, config->cache_size);
 
    for (int i = 0; i < config->num_thrs; ++i) {
        stats = &config->thread_config[i].stats;
 
        printf("\n    worker %d:\n", i);
 
        if (config->copy_type == SW_COPY) {
            printf("        packet copy errors: %" PRIu64 "\n",
                   stats->copy_errs);
        } else {
            printf("        successful DMA transfers:          %s%" PRIu64 "\n"
                   "        DMA transfer start errors:         %s%" PRIu64 "\n"
                   "        DMA transfer errors:               %s%" PRIu64 "\n"
                   "        transfer buffer allocation errors: %s%" PRIu64 "\n"
                   "        copy packet allocation errors:     %s%" PRIu64 "\n",
                   align1, stats->trs, align1, stats->start_errs, align1,
                   stats->trs_errs, align1, stats->buf_alloc_errs, align1,
                   stats->pkt_alloc_errs);
 
            if (config->copy_type == DMA_COPY_EV)
                printf("        completion event allocation errors: %" PRIu64 "\n",
                       stats->compl_alloc_errs);
            else
                printf("        transfer ID allocation errors:     %" PRIu64 "\n"
                       "        transfer poll errors:              %" PRIu64 "\n"
                       "        transfers polled:                  %" PRIu64 "\n",
                       stats->compl_alloc_errs, stats->trs_poll_errs,
                       stats->trs_polled);
        }
 
        printf("        packets forwarded:%s%" PRIu64 "\n"
               "        packets dropped:  %s%" PRIu64 "\n"
               "        packets cleanup:  %s%" PRIu64 "\n"
               "        call cycles per schedule round:\n"
               "            total:    %" PRIu64 "\n"
               "            schedule: %" PRIu64 "\n"
               "            rounds:   %" PRIu64 "\n", align2, stats->fwd_pkts, align2,
               stats->discards, align2, stats->pkt_cleanup, DIV_IF(stats->tot_cc,
               stats->sched_rounds), DIV_IF(stats->sched_cc, stats->sched_rounds),
               stats->sched_rounds);
    }
 
    printf("\n==================\n");
}
 
int main(int argc, char **argv)
{
    odph_helper_options_t odph_opts;
    odp_init_t init_param;
    odp_instance_t odp_instance;
    odp_shm_t shm_cfg = ODP_SHM_INVALID;
    int ret = EXIT_SUCCESS;
    parse_result_t parse_res;
 
    argc = odph_parse_options(argc, argv);
 
    if (odph_options(&odph_opts) == -1) {
        ODPH_ERR("Error while reading ODP helper options, exiting\n");
        exit(EXIT_FAILURE);
    }
 
    odp_init_param_init(&init_param);
    init_param.mem_model = odph_opts.mem_model;
 
    if (odp_init_global(&odp_instance, &init_param, NULL)) {
        ODPH_ERR("ODP global init failed, exiting\n");
        exit(EXIT_FAILURE);
    }
 
    if (odp_init_local(odp_instance, ODP_THREAD_CONTROL)) {
        ODPH_ERR("ODP local init failed, exiting\n");
        exit(EXIT_FAILURE);
    }
 
    shm_cfg = odp_shm_reserve(PROG_NAME "_cfg", sizeof(prog_config_t), ODP_CACHE_LINE_SIZE,
                  0U);
 
    if (shm_cfg == ODP_SHM_INVALID) {
        ODPH_ERR("Error reserving shared memory\n");
        ret = EXIT_FAILURE;
        goto out;
    }
 
    prog_conf = odp_shm_addr(shm_cfg);
 
    if (prog_conf == NULL) {
        ODPH_ERR("Error resolving shared memory address\n");
        ret = EXIT_FAILURE;
        goto out;
    }
 
    parse_res = setup_program(argc, argv, prog_conf);
 
    if (parse_res == PRS_NOK) {
        ret = EXIT_FAILURE;
        goto out_test;
    }
 
    if (parse_res == PRS_TERM) {
        ret = EXIT_SUCCESS;
        goto out_test;
    }
 
    if (parse_res == PRS_NOT_SUP) {
        ret = EXIT_NOT_SUP;
        goto out_test;
    }
 
    if (odp_schedule_config(NULL) < 0) {
        ODPH_ERR("Error configuring scheduler\n");
        ret = EXIT_FAILURE;
        goto out_test;
    }
 
    prog_conf->odp_instance = odp_instance;
    odp_atomic_init_u32(&prog_conf->is_running, 1U);
 
    if (!setup_test(prog_conf)) {
        ret = EXIT_FAILURE;
        goto out_test;
    }
 
    if (prog_conf->time_sec > 0.001) {
        struct timespec ts;
 
        ts.tv_sec = prog_conf->time_sec;
        ts.tv_nsec = (prog_conf->time_sec - ts.tv_sec) * ODP_TIME_SEC_IN_NS;
        nanosleep(&ts, NULL);
        odp_atomic_store_u32(&prog_conf->is_running, 0U);
    } else {
        while (odp_atomic_load_u32(&prog_conf->is_running))
            sleep(1U);
    }
 
    stop_test(prog_conf);
    print_stats(prog_conf);
 
out_test:
    teardown(prog_conf);
 
out:
    if (shm_cfg != ODP_SHM_INVALID)
        (void)odp_shm_free(shm_cfg);
 
    if (odp_term_local()) {
        ODPH_ERR("ODP local terminate failed, exiting\n");
        exit(EXIT_FAILURE);
    }
 
    if (odp_term_global(odp_instance)) {
        ODPH_ERR("ODP global terminate failed, exiting\n");
        exit(EXIT_FAILURE);
    }
 
    return ret;
}