odp_sched_latency.c

/* SPDX-License-Identifier: BSD-3-Clause
 * Copyright (c) 2016-2018 Linaro Limited
 * Copyright (c) 2020-2025 Nokia
 */
 
#include <string.h>
#include <stdlib.h>
#include <inttypes.h>
 
/* ODP main header */
#include <odp_api.h>
 
/* ODP helper for Linux apps */
#include <odp/helper/odph_api.h>
 
/* Result export helpers */
#include <export_results.h>
 
/* GNU lib C */
#include <getopt.h>
 
#define MAX_QUEUES    4096      
#define MAX_GROUPS        64
#define EVENT_POOL_SIZE   (1024 * 1024) 
#define MAIN_THREAD   1 
/* Test priorities */
#define NUM_PRIOS 2 
#define HI_PRIO   0
#define LO_PRIO   1
 
/* Test event forwarding mode */
#define EVENT_FORWARD_RAND 0
#define EVENT_FORWARD_INC  1
#define EVENT_FORWARD_NONE 2
 
#define MAX_BURST_SIZE 64
 
typedef enum {
    WARM_UP,  
    COOL_DOWN,
    TRAFFIC,  
    SAMPLE    
} event_type_t;
 
typedef struct {
    odp_time_t time_stamp;  
    event_type_t type;  
    int src_idx[NUM_PRIOS]; 
    int prio;       
    int warm_up_rounds; 
} test_event_t;
 
typedef struct {
    odp_time_t enq_time; 
    odp_event_t sample_collector[MAX_BURST_SIZE]; 
    odp_atomic_u32_t collector_idx; 
    odp_atomic_u32_t ready; 
} test_track_t;
 
typedef struct {
    unsigned int cpu_count; 
    odp_schedule_sync_t sync_type;  
    int forward_mode;   
    int num_group;
    int isolate;
    int test_rounds;    
    int warm_up_rounds; 
    int burst_size;     
    struct {
        int queues; 
        int events; 
        int samples;
        odp_bool_t events_per_queue; 
    } prio[NUM_PRIOS];
    odp_bool_t sample_per_prio; 
    uint64_t wait_ns;   
} test_args_t;
 
typedef struct {
    uint64_t events;   
    uint64_t samples;  
    odp_time_t tot;    
    odp_time_t min;    
    odp_time_t max;    
    uint64_t max_idx;  
} test_stat_t;
 
typedef struct ODP_ALIGNED_CACHE {
    test_stat_t prio[NUM_PRIOS]; 
} core_stat_t;
 
typedef struct {
    core_stat_t  core_stat[ODP_THREAD_COUNT_MAX];
    odp_barrier_t    barrier; 
    odp_pool_t       pool;    
    test_args_t      args;    
    odp_queue_t      queue[NUM_PRIOS][MAX_QUEUES]; 
    test_common_options_t common_options;   
    test_track_t     track[NUM_PRIOS]; 
    odp_schedule_group_t group[NUM_PRIOS][MAX_GROUPS];
 
} test_globals_t;
 
static void clear_sched_queues(test_globals_t *globals)
{
    odp_event_t ev;
    odp_buffer_t buf;
    test_event_t *event;
    int i, j;
    odp_queue_t fromq;
 
    /* Allocate the cool_down event. */
    buf = odp_buffer_alloc(globals->pool);
    if (buf == ODP_BUFFER_INVALID)
        ODPH_ABORT("Buffer alloc failed.\n");
 
    event = odp_buffer_addr(buf);
    event->type = COOL_DOWN;
    ev = odp_buffer_to_event(buf);
 
    for (i = 0; i < NUM_PRIOS; i++) {
        for (j = 0; j < globals->args.prio[i].queues; j++) {
            /* Enqueue cool_down event on each queue. */
            if (odp_queue_enq(globals->queue[i][j], ev))
                ODPH_ABORT("Queue enqueue failed.\n");
 
            /* Invoke scheduler until cool_down event has been
             * received. */
            while (1) {
                ev = odp_schedule(NULL, ODP_SCHED_WAIT);
                buf = odp_buffer_from_event(ev);
                event = odp_buffer_addr(buf);
                if (event->type == COOL_DOWN)
                    break;
                odp_event_free(ev);
            }
        }
    }
 
    /* Free the cool_down event. */
    odp_event_free(ev);
 
    /* Call odp_schedule() to trigger a release of any scheduler context. */
    ev = odp_schedule(&fromq, ODP_SCHED_NO_WAIT);
    if (ev != ODP_EVENT_INVALID)
        ODPH_ABORT("Queue %" PRIu64 " not empty.\n",
               odp_queue_to_u64(fromq));
}
 
static int enqueue_events(int prio, int num_queues, int num_events,
              int num_samples, odp_bool_t div_events,
              test_globals_t *globals)
{
    odp_buffer_t buf[num_events + num_samples];
    odp_event_t ev[num_events + num_samples];
    odp_queue_t queue;
    test_event_t *event;
    int i, j, ret;
    int enq_events;
    int events_per_queue;
    int tot_events;
    int rdy_events = 0;
 
    tot_events = num_events + num_samples;
 
    if (!num_queues || !tot_events)
        return 0;
 
    events_per_queue = tot_events;
    if (div_events)
        events_per_queue = (tot_events + num_queues - 1) / num_queues;
 
    for (i = 0; i < num_queues; i++) {
        queue = globals->queue[prio][i];
 
        ret = odp_buffer_alloc_multi(globals->pool, buf,
                         events_per_queue);
        if (ret != events_per_queue) {
            ODPH_ERR("Buffer alloc failed. Try increasing EVENT_POOL_SIZE.\n");
            ret = ret < 0 ? 0 : ret;
            odp_buffer_free_multi(buf, ret);
            return -1;
        }
        for (j = 0; j < events_per_queue; j++) {
            if (!odp_buffer_is_valid(buf[j])) {
                ODPH_ERR("Buffer alloc failed\n");
                odp_buffer_free_multi(buf, events_per_queue);
                return -1;
            }
 
            event = odp_buffer_addr(buf[j]);
            memset(event, 0, sizeof(test_event_t));
 
            /* Latency isn't measured from the first processing
             * rounds. */
            if (num_samples > 0) {
                event->type = WARM_UP;
                event->warm_up_rounds = 0;
                num_samples--;
            } else {
                event->type = TRAFFIC;
            }
            event->src_idx[prio] = i;
            event->prio = prio;
            ev[j] = odp_buffer_to_event(buf[j]);
        }
 
        enq_events = 0;
        do {
            ret = odp_queue_enq_multi(queue, &ev[enq_events],
                          events_per_queue -
                          enq_events);
            if (ret < 0) {
                ODPH_ERR("Queue enqueue failed.\n");
                return -1;
            }
            enq_events += ret;
        } while (enq_events < events_per_queue);
 
        rdy_events += events_per_queue;
        if (div_events && rdy_events >= tot_events)
            return 0;
    }
    return 0;
}
 
static int output_results(test_globals_t *globals)
{
    test_stat_t *lat;
    odp_schedule_sync_t stype;
    test_stat_t total;
    test_args_t *args;
    uint64_t avg;
    unsigned int i, j;
 
    args = &globals->args;
    stype = globals->args.sync_type;
 
    printf("\n%s queue scheduling latency\n",
           (stype == ODP_SCHED_SYNC_ATOMIC) ? "ATOMIC" :
           ((stype == ODP_SCHED_SYNC_ORDERED) ? "ORDERED" : "PARALLEL"));
 
    printf("  Forwarding mode: %s\n",
           (args->forward_mode == EVENT_FORWARD_RAND) ? "random" :
           ((args->forward_mode == EVENT_FORWARD_INC) ? "incremental" :
        "none"));
 
    printf("  Scheduling mode: %s\n", args->burst_size ? "burst" : "single");
    if (args->burst_size)
        printf("  Burst size: %d\n", args->burst_size);
 
    printf("  LO_PRIO queues: %i\n", args->prio[LO_PRIO].queues);
    if (args->prio[LO_PRIO].events_per_queue)
        printf("  LO_PRIO event per queue: %i\n",
               args->prio[LO_PRIO].events);
    else
        printf("  LO_PRIO events: %i\n", args->prio[LO_PRIO].events);
 
    printf("  LO_PRIO sample events: %i\n", args->prio[LO_PRIO].samples);
 
    printf("  HI_PRIO queues: %i\n", args->prio[HI_PRIO].queues);
    if (args->prio[HI_PRIO].events_per_queue)
        printf("  HI_PRIO event per queue: %i\n\n",
               args->prio[HI_PRIO].events);
    else
        printf("  HI_PRIO events: %i\n", args->prio[HI_PRIO].events);
 
    printf("  HI_PRIO sample events: %i\n\n", args->prio[HI_PRIO].samples);
 
    if (globals->common_options.is_export) {
        if (test_common_write("high priority avg (ns),high priority min (ns),"
                      "high priority max (ns),low priority avg (ns),"
                      "low priority min (ns),low priority max (ns)\n")) {
            ODPH_ERR("Export failed\n");
            test_common_write_term();
            return -1;
        }
    }
 
    for (i = 0; i < NUM_PRIOS; i++) {
        memset(&total, 0, sizeof(test_stat_t));
        total.min = odp_time_global_from_ns(ODP_TIME_HOUR_IN_NS);
        total.max = ODP_TIME_NULL;
        total.tot = ODP_TIME_NULL;
 
        printf("%s priority\n"
               "Thread   Avg[ns]    Min[ns]    Max[ns]    Samples    Total      Max idx\n"
               "-----------------------------------------------------------------------\n",
               i == HI_PRIO ? "HIGH" : "LOW");
        for (j = 1; j <= args->cpu_count; j++) {
            lat = &globals->core_stat[j].prio[i];
 
            if (lat->samples == 0) {
                printf("%-8d N/A\n", j);
                continue;
            }
 
            if (odp_time_cmp(lat->max, total.max) > 0)
                total.max = lat->max;
            if (odp_time_cmp(lat->min, total.min) < 0)
                total.min = lat->min;
            total.tot = odp_time_sum(total.tot, lat->tot);
            total.samples += lat->samples;
            total.events += lat->events;
 
            avg = lat->samples ? odp_time_to_ns(lat->tot) / lat->samples : 0;
            printf("%-8d %-10" PRIu64 " %-10" PRIu64 " "
                   "%-10" PRIu64 " %-10" PRIu64 " %-10" PRIu64 " %-10" PRIu64 "\n",
                   j, avg, odp_time_to_ns(lat->min), odp_time_to_ns(lat->max),
                   lat->samples, lat->events, lat->max_idx);
        }
        printf("-----------------------------------------------------------------------\n");
        if (total.samples == 0) {
            printf("Total    N/A\n\n");
            continue;
        }
 
        avg = total.samples ? odp_time_to_ns(total.tot) / total.samples : 0;
        printf("Total    %-10" PRIu64 " %-10" PRIu64 " %-10" PRIu64 " "
               "%-10" PRIu64 " %-10" PRIu64 "\n\n", avg, odp_time_to_ns(total.min),
               odp_time_to_ns(total.max), total.samples, total.events);
 
        if (globals->common_options.is_export) {
            if (test_common_write("%" PRIu64 ",%" PRIu64 ",%" PRIu64 "%s",
                          avg, odp_time_to_ns(total.min),
                          odp_time_to_ns(total.max), i == 0 ? "," : "")) {
                ODPH_ERR("Export failed\n");
                test_common_write_term();
                return -1;
            }
        }
    }
 
    if (globals->common_options.is_export)
        test_common_write_term();
 
    return 0;
}
 
static int join_groups(test_globals_t *globals, int thr)
{
    odp_thrmask_t thrmask;
    odp_schedule_group_t group;
    int i, num;
    int num_group = globals->args.num_group;
 
    if (num_group <= 0)
        return 0;
 
    num = num_group;
    if (globals->args.isolate)
        num = 2 * num_group;
 
    odp_thrmask_zero(&thrmask);
    odp_thrmask_set(&thrmask, thr);
 
    for (i = 0; i < num; i++) {
        if (globals->args.isolate)
            group = globals->group[i % 2][i / 2];
        else
            group = globals->group[0][i];
 
        if (odp_schedule_group_join(group, &thrmask)) {
            ODPH_ERR("Group join failed %i (thr %i)\n", i, thr);
            return -1;
        }
    }
 
    return 0;
}
 
static inline void record_stats(test_stat_t *stats, odp_time_t sched, odp_time_t enq)
{
    /* Skip measurement if enq has not been set yet */
    if (odp_unlikely(odp_time_cmp(enq, ODP_TIME_NULL) == 0))
        return;
 
    odp_time_t latency = odp_time_diff(sched, enq);
 
    if (odp_time_cmp(latency, stats->max) > 0) {
        stats->max = latency;
        stats->max_idx = stats->samples;
    }
 
    if (odp_time_cmp(latency, stats->min) < 0)
        stats->min = latency;
 
    stats->samples++;
    stats->tot = odp_time_sum(stats->tot, latency);
}
 
static inline int get_dst_idx(const test_args_t *args, const test_event_t *event)
{
    int dst_idx;
 
    if (args->forward_mode != EVENT_FORWARD_NONE)
        dst_idx = event->src_idx[event->prio] + 1;
    else
        dst_idx = event->src_idx[event->prio];
 
    if (dst_idx >= args->prio[event->prio].queues)
        dst_idx = 0;
 
    return dst_idx;
}
 
static int test_schedule_single(int thr, test_globals_t *globals)
{
    const int warm_up_rounds = globals->args.warm_up_rounds;
    const uint64_t test_rounds = globals->args.test_rounds * (uint64_t)1000000;
    const uint64_t wait_ns = globals->args.wait_ns;
    odp_event_t ev;
    odp_buffer_t buf;
    odp_queue_t dst_queue;
    uint64_t i;
    test_event_t *event;
    test_stat_t *stats;
    odp_time_t time;
    int dst_idx;
 
    for (i = 0; i < test_rounds; i++) {
        ev = odp_schedule(NULL, ODP_SCHED_WAIT);
        time = odp_time_global_strict();
        buf = odp_buffer_from_event(ev);
        event = odp_buffer_addr(buf);
 
        if (odp_unlikely(wait_ns))
            odp_time_wait_ns(wait_ns);
 
        stats = &globals->core_stat[thr].prio[event->prio];
 
        if (event->type == SAMPLE) {
            record_stats(stats, time, event->time_stamp);
 
            /* Move sample event to a different priority */
            if (!globals->args.sample_per_prio &&
                globals->args.prio[!event->prio].queues)
                event->prio = !event->prio;
        }
 
        if (odp_unlikely(event->type == WARM_UP)) {
            event->warm_up_rounds++;
            if (event->warm_up_rounds >= warm_up_rounds)
                event->type = SAMPLE;
        } else {
            stats->events++;
        }
 
        /* Move event to next queue if forwarding is enabled */
        dst_idx = get_dst_idx(&globals->args, event);
        event->src_idx[event->prio] = dst_idx;
        dst_queue = globals->queue[event->prio][dst_idx];
 
        if (event->type == SAMPLE)
            event->time_stamp = odp_time_global_strict();
 
        if (odp_queue_enq(dst_queue, ev)) {
            ODPH_ERR("[%i] Queue enqueue failed.\n", thr);
            odp_event_free(ev);
            return -1;
        }
    }
 
    return 0;
}
 
static inline int enq_burst(int thr, test_globals_t *globals, odp_event_t burst_array[],
                uint32_t burst_size)
{
    const test_args_t *args = &globals->args;
    test_track_t *track = globals->track;
    test_event_t *event;
    odp_queue_t dst_queue;
    uint32_t i;
    int num_enq, new_prio;
    int enq = 0;
    int dst_idx = 0;
 
    event = odp_buffer_addr(odp_buffer_from_event(burst_array[0]));
    if (event->type == SAMPLE) {
        /* Move sample event to different priority if possible */
        if (!args->sample_per_prio && args->prio[!event->prio].queues) {
            new_prio = !event->prio;
            event->prio = new_prio;
        } else {
            new_prio = event->prio;
        }
 
        dst_idx = get_dst_idx(&globals->args, event);
        dst_queue = globals->queue[event->prio][dst_idx];
 
        for (i = 0; i < burst_size; i++) {
            event = odp_buffer_addr(odp_buffer_from_event(burst_array[i]));
            event->prio = new_prio;
            event->src_idx[event->prio] = dst_idx;
        }
 
        track[event->prio].enq_time = odp_time_global_strict();
    } else {
        dst_idx = get_dst_idx(&globals->args, event);
        dst_queue = globals->queue[event->prio][dst_idx];
 
        for (i = 0; i < burst_size; i++) {
            event = odp_buffer_addr(odp_buffer_from_event(burst_array[i]));
            event->src_idx[event->prio] = dst_idx;
        }
    }
 
    do {
        num_enq = odp_queue_enq_multi(dst_queue, &burst_array[enq], burst_size - enq);
        if (odp_unlikely(num_enq < 0)) {
            ODPH_ERR("[%i] Queue enqueue failed.\n", thr);
            odp_event_free_multi(&burst_array[enq], burst_size - enq);
            return -1;
        }
        enq += num_enq;
    } while (enq < (int)burst_size);
 
    return 0;
}
 
static inline int collect_enq_sample(int thr, test_globals_t *globals, odp_event_t ev,
                     odp_time_t time, int prio)
{
    const uint32_t burst_size = globals->args.burst_size;
    test_stat_t *stats = &globals->core_stat[thr].prio[prio];
    test_track_t *track = &globals->track[prio];
    odp_atomic_u32_t *collector_idx = &track->collector_idx;
    odp_atomic_u32_t *ready = &track->ready;
    odp_event_t *sample_collector = track->sample_collector;
    odp_event_t local_collector[MAX_BURST_SIZE];
    uint32_t idx, i;
 
    idx = odp_atomic_fetch_inc_u32(collector_idx);
    sample_collector[idx] = ev;
    odp_atomic_add_rel_u32(ready, 1);
 
    if (idx == burst_size - 1) {
        while (odp_atomic_load_acq_u32(ready) != burst_size)
            odp_cpu_pause();
 
        record_stats(stats, time, track->enq_time);
 
        /* Snapshot the burst to avoid data race with scheduling threads */
        for (i = 0; i < burst_size; i++)
            local_collector[i] = sample_collector[i];
 
        odp_atomic_store_u32(collector_idx, 0);
        odp_atomic_store_u32(ready, 0);
 
        if (enq_burst(thr, globals, local_collector, burst_size))
            return -1;
    }
 
    return 0;
}
 
static int test_schedule_burst(int thr, test_globals_t *globals)
{
    const int warm_up_rounds = globals->args.warm_up_rounds;
    const uint64_t test_rounds = globals->args.test_rounds * (uint64_t)1000000;
    const uint32_t burst_size = globals->args.burst_size;
    const uint64_t wait_ns = globals->args.wait_ns;
    uint32_t num_sched_ev;
    odp_event_t ev_burst[MAX_BURST_SIZE];
    odp_event_t traffic_collector[MAX_BURST_SIZE];
    odp_event_t ev;
    test_event_t *event;
    test_stat_t *stats;
    odp_time_t time;
    uint32_t traffic_idx = 0;
    uint64_t i;
    uint32_t j;
 
    for (i = 0; i < test_rounds; i++) {
        num_sched_ev = odp_schedule_multi(NULL, ODP_SCHED_WAIT,
                          ev_burst, burst_size);
        time = odp_time_global_strict();
 
        if (odp_unlikely(wait_ns))
            odp_time_wait_ns(wait_ns);
 
        for (j = 0; j < num_sched_ev; j++) {
            ev = ev_burst[j];
            event = odp_buffer_addr(odp_buffer_from_event(ev));
            stats = &globals->core_stat[thr].prio[event->prio];
 
            if (odp_unlikely(event->type == WARM_UP)) {
                event->warm_up_rounds++;
                if (event->warm_up_rounds >= warm_up_rounds)
                    event->type = SAMPLE;
            } else {
                stats->events++;
            }
 
            /* Collect events for burst enqueueing */
            if (event->type == SAMPLE) {
                if (odp_unlikely(collect_enq_sample(thr, globals, ev, time,
                                    event->prio)))
                    return -1;
            } else {
                traffic_collector[traffic_idx++] = ev;
            }
        }
 
        if (traffic_idx) {
            if (odp_unlikely(enq_burst(thr, globals, traffic_collector,
                           traffic_idx) == -1))
                return -1;
            traffic_idx = 0;
        }
    }
    return 0;
}
 
static int test_schedule(int thr, test_globals_t *globals)
{
    odp_event_t ev;
    test_track_t *track;
    uint32_t cnt;
 
    globals->core_stat[thr].prio[HI_PRIO].min = odp_time_global_from_ns(ODP_TIME_HOUR_IN_NS);
    globals->core_stat[thr].prio[HI_PRIO].max = ODP_TIME_NULL;
    globals->core_stat[thr].prio[HI_PRIO].tot = ODP_TIME_NULL;
    globals->core_stat[thr].prio[LO_PRIO].min = odp_time_global_from_ns(ODP_TIME_HOUR_IN_NS);
    globals->core_stat[thr].prio[LO_PRIO].max = ODP_TIME_NULL;
    globals->core_stat[thr].prio[LO_PRIO].tot = ODP_TIME_NULL;
 
    odp_barrier_wait(&globals->barrier);
 
    if (globals->args.burst_size) {
        if (test_schedule_burst(thr, globals))
            return -1;
    } else {
        if (test_schedule_single(thr, globals))
            return -1;
    }
 
    /* Clear possible locally stored buffers */
    odp_schedule_pause();
 
    while (1) {
        odp_queue_t src_queue;
 
        ev = odp_schedule(&src_queue, ODP_SCHED_NO_WAIT);
 
        if (ev == ODP_EVENT_INVALID)
            break;
 
        if (odp_queue_enq(src_queue, ev)) {
            ODPH_ERR("[%i] Queue enqueue failed.\n", thr);
            odp_event_free(ev);
            return -1;
        }
    }
 
    odp_barrier_wait(&globals->barrier);
 
    if (thr == MAIN_THREAD) {
        if (globals->args.burst_size) {
            for (int i = 0; i < NUM_PRIOS; i++) {
                track = &globals->track[i];
                cnt = odp_atomic_load_u32(&track->collector_idx);
 
                if (cnt) {
                    odp_event_free_multi(track->sample_collector, cnt);
                    odp_atomic_store_u32(&track->collector_idx, 0);
                    odp_atomic_store_u32(&track->ready, 0);
                }
            }
        }
        odp_schedule_resume();
        clear_sched_queues(globals);
        if (output_results(globals))
            return -1;
    }
 
    return 0;
}
 
static int run_thread(void *arg ODP_UNUSED)
{
    odp_shm_t shm;
    test_globals_t *globals;
    test_args_t *args;
    int thr;
 
    thr = odp_thread_id();
 
    shm     = odp_shm_lookup("test_globals");
    globals = odp_shm_addr(shm);
 
    if (globals == NULL) {
        ODPH_ERR("Shared mem lookup failed\n");
        return -1;
    }
 
    if (join_groups(globals, thr))
        return -1;
 
    if (thr == MAIN_THREAD) {
        args = &globals->args;
 
        if (enqueue_events(HI_PRIO, args->prio[HI_PRIO].queues,
                   args->prio[HI_PRIO].events, args->prio[HI_PRIO].samples,
                   !args->prio[HI_PRIO].events_per_queue,
                   globals))
            return -1;
 
        if (enqueue_events(LO_PRIO, args->prio[LO_PRIO].queues,
                   args->prio[LO_PRIO].events, args->prio[LO_PRIO].samples,
                   !args->prio[LO_PRIO].events_per_queue,
                   globals))
            return -1;
    }
 
    if (test_schedule(thr, globals))
        return -1;
 
    return 0;
}
 
static void usage(void)
{
    printf("\n"
           "OpenDataPlane scheduler latency benchmark application.\n"
           "\n"
           "Usage: ./odp_sched_latency [options]\n"
           "Optional OPTIONS:\n"
           "  -c, --count <number> CPU count, 0=all available, default=1\n"
           "  -b, --burst <num> Test burst size. Latency is measured over whole bursts.\n"
           "  -d, --duration <number> Test duration in scheduling rounds (millions), default=10, min=1\n"
           "  -f, --forward-mode <mode> Selection of target queue\n"
           "               0: Random (default)\n"
           "               1: Incremental\n"
           "               2: Use source queue\n"
           "  -g, --num_group <num>  Number of schedule groups. Round robins queues into groups.\n"
           "                         -1: SCHED_GROUP_WORKER\n"
           "                          0: SCHED_GROUP_ALL (default)\n"
           "  -i, --isolate <mode> Select if shared or isolated groups are used. Ignored when num_group <= 0.\n"
           "                       0: All queues share groups (default)\n"
           "                       1: Separate groups for high and low priority queues. Creates 2xnum_group groups.\n"
           "  -l, --lo-prio-queues <number> Number of low priority scheduled queues (default=64)\n"
           "  -t, --hi-prio-queues <number> Number of high priority scheduled queues (default=16)\n"
           "  -m, --lo-prio-events-per-queue <number> Number of events per low priority queue (default=32).\n"
           "                                          Does not include sample event.\n"
           "  -n, --hi-prio-events-per-queue <number> Number of events per high priority queues (default=0)\n"
           "                                          Does not include sample event.\n"
           "  -o, --lo-prio-events <number> Total number of low priority events. Overrides the\n"
           "                number of events per queue, does not include sample event.\n"
           "  -p, --hi-prio-events <number> Total number of high priority events. Overrides the\n"
           "                number of events per queue, does not include sample event.\n"
           "  -r  --sample-per-prio Allocate a separate sample event for each priority. By default\n"
           "            a single sample event is used and its priority is changed after\n"
           "            each processing round.\n"
           "  -s, --sync  Scheduled queues' sync type\n"
           "               0: ODP_SCHED_SYNC_PARALLEL (default)\n"
           "               1: ODP_SCHED_SYNC_ATOMIC\n"
           "               2: ODP_SCHED_SYNC_ORDERED\n"
           "  -w, --warm-up <number> Number of warm-up rounds, default=100, min=1\n"
           "  -W, --wait-ns <number> Number of nsec to wait per schedule round to simulate work (default=0)\n"
           "  -h, --help   Display help and exit.\n\n");
}
 
static void parse_args(int argc, char *argv[], test_args_t *args)
{
    int opt;
    int i;
 
    static const struct option longopts[] = {
        {"count", required_argument, NULL, 'c'},
        {"burst", required_argument, NULL, 'b'},
        {"duration", required_argument, NULL, 'd'},
        {"forward-mode", required_argument, NULL, 'f'},
        {"num_group", required_argument, NULL, 'g'},
        {"isolate", required_argument, NULL, 'i'},
        {"lo-prio-queues", required_argument, NULL, 'l'},
        {"hi-prio-queues", required_argument, NULL, 't'},
        {"lo-prio-events-per-queue", required_argument, NULL, 'm'},
        {"hi-prio-events-per-queue", required_argument, NULL, 'n'},
        {"lo-prio-events", required_argument, NULL, 'o'},
        {"hi-prio-events", required_argument, NULL, 'p'},
        {"sync", required_argument, NULL, 's'},
        {"warm-up", required_argument, NULL, 'w'},
        {"wait-ns", required_argument, NULL, 'W'},
        {"sample-per-prio", no_argument, NULL, 'r'},
        {"help", no_argument, NULL, 'h'},
        {NULL, 0, NULL, 0}
    };
 
    static const char *shortopts = "+c:b:d:f:g:i:l:t:m:n:o:p:s:w:W:rh";
 
    args->cpu_count = 1;
    args->forward_mode = EVENT_FORWARD_RAND;
    args->num_group = 0;
    args->isolate = 0;
    args->test_rounds = 10;
    args->warm_up_rounds = 100;
    args->sync_type = ODP_SCHED_SYNC_PARALLEL;
    args->sample_per_prio = 0;
    args->prio[LO_PRIO].queues = 64;
    args->prio[HI_PRIO].queues = 16;
    args->prio[LO_PRIO].events = 32;
    args->prio[HI_PRIO].events = 0;
    args->prio[LO_PRIO].events_per_queue = 1;
    args->prio[HI_PRIO].events_per_queue = 0;
    args->prio[LO_PRIO].samples = 0;
    args->prio[HI_PRIO].samples = 1;
 
    while (1) {
        opt = getopt_long(argc, argv, shortopts, longopts, NULL);
 
        if (opt == -1)
            break;  /* No more options */
 
        switch (opt) {
        case 'c':
            args->cpu_count = atoi(optarg);
            break;
        case 'd':
            args->test_rounds = atoi(optarg);
            break;
        case 'f':
            args->forward_mode = atoi(optarg);
            break;
        case 'g':
            args->num_group = atoi(optarg);
            break;
        case 'i':
            args->isolate = atoi(optarg);
            break;
        case 'l':
            args->prio[LO_PRIO].queues = atoi(optarg);
            break;
        case 't':
            args->prio[HI_PRIO].queues = atoi(optarg);
            break;
        case 'm':
            args->prio[LO_PRIO].events = atoi(optarg);
            args->prio[LO_PRIO].events_per_queue = 1;
            break;
        case 'n':
            args->prio[HI_PRIO].events = atoi(optarg);
            args->prio[HI_PRIO].events_per_queue = 1;
            break;
        case 'o':
            args->prio[LO_PRIO].events = atoi(optarg);
            args->prio[LO_PRIO].events_per_queue = 0;
            break;
        case 'p':
            args->prio[HI_PRIO].events = atoi(optarg);
            args->prio[HI_PRIO].events_per_queue = 0;
            break;
        case 's':
            i = atoi(optarg);
            if (i == 1)
                args->sync_type = ODP_SCHED_SYNC_ATOMIC;
            else if (i == 2)
                args->sync_type = ODP_SCHED_SYNC_ORDERED;
            else
                args->sync_type = ODP_SCHED_SYNC_PARALLEL;
            break;
        case 'r':
            args->sample_per_prio = 1;
            break;
        case 'w':
            args->warm_up_rounds = atoi(optarg);
            break;
        case 'W':
            args->wait_ns = atoll(optarg);
            break;
        case 'b':
            args->burst_size = atoi(optarg);
            break;
        case 'h':
            usage();
            exit(EXIT_SUCCESS);
            break;
 
        default:
            break;
        }
    }
 
    /* Make sure arguments are valid */
    /* -1 for main thread */
    if (args->cpu_count > ODP_THREAD_COUNT_MAX - 1)
        args->cpu_count = ODP_THREAD_COUNT_MAX - 1;
    if (args->prio[LO_PRIO].queues > MAX_QUEUES)
        args->prio[LO_PRIO].queues = MAX_QUEUES;
    if (args->prio[HI_PRIO].queues > MAX_QUEUES)
        args->prio[HI_PRIO].queues = MAX_QUEUES;
    if (args->test_rounds < 1)
        args->test_rounds = 1;
    if (!args->prio[HI_PRIO].queues && !args->prio[LO_PRIO].queues) {
        printf("No queues configured\n");
        usage();
        exit(EXIT_FAILURE);
    }
    if (args->forward_mode > EVENT_FORWARD_NONE ||
        args->forward_mode < EVENT_FORWARD_RAND) {
        printf("Invalid forwarding mode\n");
        usage();
        exit(EXIT_FAILURE);
    }
 
    if (args->num_group > MAX_GROUPS) {
        ODPH_ERR("Too many groups. Max supported %i.\n", MAX_GROUPS);
        exit(EXIT_FAILURE);
    }
 
    if (args->prio[HI_PRIO].queues == 0 || args->sample_per_prio)
        args->prio[LO_PRIO].samples = 1;
 
    if (args->burst_size < 0 || args->burst_size > MAX_BURST_SIZE) {
        ODPH_ERR("Invalid burst size %i, min 0, max %i\n",
             args->burst_size, MAX_BURST_SIZE);
        exit(EXIT_FAILURE);
    }
 
    if (args->burst_size) {
        args->prio[HI_PRIO].samples  = args->burst_size;
        if (args->sample_per_prio)
            args->prio[LO_PRIO].samples = args->burst_size;
    }
}
 
static void randomize_queues(odp_queue_t queues[], uint32_t num, uint64_t *seed)
{
    uint32_t i;
 
    for (i = 0; i < num; i++) {
        uint32_t new_index;
        odp_queue_t swap_queue;
        odp_queue_t cur_queue = queues[i];
 
        odp_random_test_data((uint8_t *)&new_index, sizeof(new_index),
                     seed);
        new_index = new_index % num;
        swap_queue = queues[new_index];
 
        queues[new_index] = cur_queue;
        queues[i] = swap_queue;
    }
}
 
static int create_groups(test_globals_t *globals, odp_schedule_group_t group[], int num)
{
    odp_schedule_capability_t sched_capa;
    odp_thrmask_t zeromask;
    int i, j, max;
 
    if (num <= 0)
        return 0;
 
    if (odp_schedule_capability(&sched_capa)) {
        ODPH_ERR("Schedule capability failed\n");
        return 0;
    }
 
    max = sched_capa.max_groups - 3;
    if (num > max) {
        printf("Too many schedule groups %i (max %u)\n", num, max);
        return 0;
    }
 
    for (i = 0; i < NUM_PRIOS; i++)
        for (j = 0; j < MAX_GROUPS; j++)
            globals->group[i][j] = ODP_SCHED_GROUP_INVALID;
 
    odp_thrmask_zero(&zeromask);
 
    for (i = 0; i < num; i++) {
        group[i] = odp_schedule_group_create("test_group", &zeromask);
 
        if (group[i] == ODP_SCHED_GROUP_INVALID) {
            ODPH_ERR("Group create failed %i\n", i);
            break;
        }
 
        if (globals->args.isolate) {
            globals->group[i % 2][i / 2] = group[i];
        } else {
            globals->group[0][i] = group[i];
            globals->group[1][i] = group[i];
        }
    }
 
    return i;
}
 
static int destroy_groups(odp_schedule_group_t group[], int num)
{
    int i;
 
    if (num <= 0)
        return 0;
 
    for (i = 0; i < num; i++) {
        if (odp_schedule_group_destroy(group[i])) {
            ODPH_ERR("Group destroy failed %i\n", i);
            return -1;
        }
    }
 
    return 0;
}
 
static int calc_queue_sizes(test_globals_t *globals, uint32_t queue_size[])
{
    odp_schedule_capability_t capa;
    test_args_t *args = &globals->args;
    const uint32_t min_queue_size = 256;
    uint32_t tot_queues = 0;
 
    if (odp_schedule_capability(&capa)) {
        ODPH_ERR("Schedule capability failed\n");
        return -1;
    }
 
    for (int i = 0; i < NUM_PRIOS; i++) {
        uint32_t events = args->prio[i].events;
        int queues = args->prio[i].queues;
 
        if (!args->prio[i].events_per_queue && queues)
            events = (events + queues - 1) / queues;
 
        /* Events may stack up if forwarding is enabled */
        if (args->forward_mode != EVENT_FORWARD_NONE)
            events *= queues;
 
        /* Reserve room for sample event */
        events++;
 
        queue_size[i] = ODPH_MAX(events, min_queue_size);
 
        if (capa.max_queue_size && queue_size[i] > capa.max_queue_size) {
            ODPH_ERR("Warn: queues may not be able to store all events (required size "
                "%" PRIu32 ", max supported %" PRIu32 ")\n", queue_size[i],
                capa.max_queue_size);
            queue_size[i] = capa.max_queue_size;
        }
        tot_queues += queues;
    }
 
    if (tot_queues > capa.max_queues) {
        ODPH_ERR("Requested %" PRIu32 " queues, max %" PRIu32 " supported\n",
             tot_queues, capa.max_queues);
        return -1;
    }
 
    return 0;
}
 
static int create_queues(test_globals_t *globals)
{
    odp_queue_param_t param;
    test_args_t *args = &globals->args;
    int num_group = args->num_group;
    uint32_t queue_size[NUM_PRIOS];
 
    if (calc_queue_sizes(globals, queue_size))
        return -1;
 
    odp_queue_param_init(&param);
    param.type = ODP_QUEUE_TYPE_SCHED;
    param.sched.sync = args->sync_type;
 
    for (int i = 0; i < NUM_PRIOS; i++) {
        char name[] = "sched_XX_YY";
        odp_queue_t queue;
        odp_schedule_group_t grp = num_group < 0 ? ODP_SCHED_GROUP_WORKER :
                               ODP_SCHED_GROUP_ALL;
        const int prio = i == HI_PRIO ? odp_schedule_max_prio() :
                        odp_schedule_min_prio();
 
        param.sched.prio = prio;
        param.size = queue_size[i];
 
        /* Replace XX and YY in name to differentiate queues */
        name[6] = '0' + (prio / 10);
        name[7] = '0' + prio - (10 * (prio / 10));
 
        for (int j = 0; j < args->prio[i].queues; j++) {
            name[9]  = '0' + j / 10;
            name[10] = '0' + j - 10 * (j / 10);
 
            /* Round robin queues into groups */
            if (num_group > 0)
                grp = globals->group[i][j % num_group];
 
            param.sched.group = grp;
 
            queue = odp_queue_create(name, &param);
 
            if (queue == ODP_QUEUE_INVALID) {
                ODPH_ERR("Scheduled queue create failed\n");
                return -1;
            }
 
            globals->queue[i][j] = queue;
        }
        if (args->forward_mode == EVENT_FORWARD_RAND) {
            uint64_t seed = i;
 
            randomize_queues(globals->queue[i], args->prio[i].queues, &seed);
        }
    }
    return 0;
}
 
static void init_global_data(test_globals_t *globals, test_args_t *args,
                 test_common_options_t common_options)
{
    memset(globals, 0, sizeof(test_globals_t));
    memcpy(&globals->args, args, sizeof(test_args_t));
 
    globals->common_options = common_options;
 
    odp_barrier_init(&globals->barrier, globals->args.cpu_count);
 
    globals->pool = ODP_POOL_INVALID;
    for (int i = 0; i < NUM_PRIOS; i++) {
        for (int j = 0; j < MAX_GROUPS; j++) {
            globals->group[i][j] = ODP_SCHED_GROUP_INVALID;
            globals->queue[i][j] = ODP_QUEUE_INVALID;
        }
 
        globals->track[i].enq_time = ODP_TIME_NULL;
        odp_atomic_init_u32(&globals->track[i].ready, 0);
        odp_atomic_init_u32(&globals->track[i].collector_idx, 0);
    }
}
 
int main(int argc, char *argv[])
{
    odp_instance_t instance;
    odp_init_t init_param;
    odph_helper_options_t helper_options;
    odph_thread_common_param_t thr_common;
    odph_thread_param_t thr_param;
    odp_cpumask_t cpumask;
    odp_pool_capability_t pool_capa;
    odp_pool_param_t params;
    test_globals_t *globals;
    test_args_t args;
    char cpumaskstr[ODP_CPUMASK_STR_SIZE];
    uint32_t pool_size;
    int i, j, ret;
    int num_group, tot_group;
    odp_schedule_group_t group[2 * MAX_GROUPS];
    odph_thread_t thread_tbl[ODP_THREAD_COUNT_MAX];
    int err = 0;
    int num_workers = 0;
    odp_shm_t shm = ODP_SHM_INVALID;
    odp_pool_t pool = ODP_POOL_INVALID;
    test_common_options_t common_options;
 
    printf("\nODP scheduling latency benchmark starts\n\n");
 
    /* 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);
    }
 
    argc = test_common_parse_options(argc, argv);
    if (test_common_options(&common_options)) {
        ODPH_ERR("Error: reading test options failed\n");
        exit(EXIT_FAILURE);
    }
 
    odp_init_param_init(&init_param);
    init_param.mem_model = helper_options.mem_model;
 
    memset(&args, 0, sizeof(args));
    parse_args(argc, argv, &args);
 
    /* ODP global init */
    if (odp_init_global(&instance, &init_param, NULL)) {
        ODPH_ERR("ODP global init failed.\n");
        exit(EXIT_FAILURE);
    }
 
    /*
     * Init this thread. It makes also ODP calls when
     * setting up resources for worker threads.
     */
    if (odp_init_local(instance, ODP_THREAD_CONTROL)) {
        ODPH_ERR("ODP global init failed.\n");
        exit(EXIT_FAILURE);
    }
 
    odp_sys_info_print();
 
    num_group = args.num_group;
 
    tot_group = 0;
    if (num_group > 0)
        tot_group = args.isolate ? 2 * num_group : num_group;
 
    /* Get default worker cpumask */
    if (args.cpu_count)
        num_workers = args.cpu_count;
 
    num_workers = odp_cpumask_default_worker(&cpumask, num_workers);
    args.cpu_count = num_workers;
 
    (void)odp_cpumask_to_str(&cpumask, cpumaskstr, sizeof(cpumaskstr));
 
    printf("Test options:\n");
    printf("  Worker threads:   %i\n", num_workers);
    printf("  First CPU:        %i\n", odp_cpumask_first(&cpumask));
    printf("  CPU mask:         %s\n", cpumaskstr);
    printf("  Test rounds:      %iM\n", args.test_rounds);
    printf("  Warm-up rounds:   %i\n", args.warm_up_rounds);
    printf("  Wait nsec:        %" PRIu64 "\n", args.wait_ns);
    printf("  Isolated groups:  %i\n", args.isolate);
    printf("  Number of groups: %i\n", num_group);
    printf("  Created groups:   %i\n", tot_group);
    printf("\n");
 
    shm = odp_shm_reserve("test_globals", sizeof(test_globals_t), ODP_CACHE_LINE_SIZE, 0);
    if (shm == ODP_SHM_INVALID) {
        ODPH_ERR("Shared memory reserve failed.\n");
        err = -1;
        goto error;
    }
 
    globals = odp_shm_addr(shm);
    init_global_data(globals, &args, common_options);
    odp_schedule_config(NULL);
 
    /*
     * Create event pool
     */
    if (odp_pool_capability(&pool_capa)) {
        ODPH_ERR("pool capa failed\n");
        err = -1;
        goto error;
    }
 
    pool_size = EVENT_POOL_SIZE;
    if (pool_capa.buf.max_num && pool_capa.buf.max_num < EVENT_POOL_SIZE)
        pool_size = pool_capa.buf.max_num;
 
    odp_pool_param_init(&params);
    params.buf.size  = sizeof(test_event_t);
    params.buf.align = 0;
    params.buf.num   = pool_size;
    params.type      = ODP_POOL_BUFFER;
 
    pool = odp_pool_create("event_pool", &params);
 
    if (pool == ODP_POOL_INVALID) {
        ODPH_ERR("Pool create failed.\n");
        err = -1;
        goto error;
    }
    globals->pool = pool;
 
    /* Create groups */
    ret = create_groups(globals, group, tot_group);
    if (ret != tot_group) {
        ODPH_ERR("Group create failed.\n");
        tot_group = ret;
        err = -1;
        goto error;
    }
 
    if (create_queues(globals)) {
        ODPH_ERR("Creating test queues failed.\n");
        err = -1;
        goto error;
    }
 
    /* Create and launch worker threads */
    memset(thread_tbl, 0, sizeof(thread_tbl));
 
    odph_thread_common_param_init(&thr_common);
    thr_common.instance = instance;
    thr_common.cpumask = &cpumask;
    thr_common.share_param = 1;
 
    odph_thread_param_init(&thr_param);
    thr_param.start = run_thread;
    thr_param.arg = NULL;
    thr_param.thr_type = ODP_THREAD_WORKER;
 
    odph_thread_create(thread_tbl, &thr_common, &thr_param, num_workers);
 
    /* Wait for worker threads to terminate */
    if (odph_thread_join(thread_tbl, num_workers) != num_workers)
        err = -1;
 
    printf("ODP scheduling latency test complete\n\n");
 
    for (i = 0; i < NUM_PRIOS; i++) {
        odp_queue_t queue;
        int num_queues;
 
        num_queues = args.prio[i].queues;
 
        for (j = 0; j < num_queues; j++) {
            queue = globals->queue[i][j];
            if (odp_queue_destroy(queue)) {
                ODPH_ERR("Queue destroy failed [%i][%i]\n", i, j);
                err = -1;
                break;
            }
        }
    }
 
error:
    if (destroy_groups(group, tot_group)) {
        ODPH_ERR("Group destroy failed\n");
        err = -1;
    }
 
    if (pool != ODP_POOL_INVALID) {
        if (odp_pool_destroy(pool)) {
            ODPH_ERR("Pool destroy failed\n");
            err = -1;
        }
    }
 
    if (shm != ODP_SHM_INVALID) {
        if (odp_shm_free(shm)) {
            ODPH_ERR("SHM destroy failed\n");
            err = -1;
        }
    }
 
    err += odp_term_local();
    err += odp_term_global(instance);
 
    return err;
}