| Message ID | 1451562390-9196-1-git-send-email-ivan.khoronzhuk@linaro.org |
|---|---|
| State | Superseded |
| Headers | show |
You need to create a .gitignore file in example/time and add the output binary (odp_global_time_test) to it for this patch to be complete. Other than that: On Thu, Dec 31, 2015 at 5:46 AM, Ivan Khoronzhuk <ivan.khoronzhuk@linaro.org > wrote: > This test example uses linux helper functions to schedule worker > threads in order to check global time monotony between the threads. > Each thread receives event from it's own queue, checks global time on > correctness, writes in the buffer new current global time, > chooses randomly next thread queue, sends event and waits on next > event on it's own queue. Using for each thread it's own queue guarantees > that event is read by not the same thread that sends it. When global time > source is correct the time received with event by a thread must be less > than read on the moment of receiving. > > Signed-off-by: Ivan Khoronzhuk <ivan.khoronzhuk@linaro.org> > Reviewed-and-tested-by: Bill Fischofer <bill.fischofer@linaro.org> > --- > > Since v2: > - replaced many includes on odp.h > - corrected a little error path > - increased num of iterations > - rebased on ODPv1.6 > > configure.ac | 1 + > example/Makefile.am | 2 +- > example/time/Makefile.am | 10 ++ > example/time/time_global_test.c | 352 > ++++++++++++++++++++++++++++++++++++++++ > 4 files changed, 364 insertions(+), 1 deletion(-) > create mode 100644 example/time/Makefile.am > create mode 100644 example/time/time_global_test.c > > diff --git a/configure.ac b/configure.ac > index 2626fd7..b284384 100644 > --- a/configure.ac > +++ b/configure.ac > @@ -338,6 +338,7 @@ AC_CONFIG_FILES([Makefile > example/generator/Makefile > example/ipsec/Makefile > example/packet/Makefile > + example/time/Makefile > example/timer/Makefile > helper/Makefile > helper/test/Makefile > diff --git a/example/Makefile.am b/example/Makefile.am > index 353f397..39d9b01 100644 > --- a/example/Makefile.am > +++ b/example/Makefile.am > @@ -1 +1 @@ > -SUBDIRS = classifier generator ipsec packet timer > +SUBDIRS = classifier generator ipsec packet time timer > diff --git a/example/time/Makefile.am b/example/time/Makefile.am > new file mode 100644 > index 0000000..915593a > --- /dev/null > +++ b/example/time/Makefile.am > @@ -0,0 +1,10 @@ > +include $(top_srcdir)/example/Makefile.inc > + > +bin_PROGRAMS = odp_time_global_test$(EXEEXT) > +odp_time_global_test_LDFLAGS = $(AM_LDFLAGS) -static > +odp_time_global_test_CFLAGS = $(AM_CFLAGS) -I${top_srcdir}/example > + > +noinst_HEADERS = \ > + $(top_srcdir)/example/example_debug.h > + > +dist_odp_time_global_test_SOURCES = time_global_test.c > diff --git a/example/time/time_global_test.c > b/example/time/time_global_test.c > new file mode 100644 > index 0000000..de0ab47 > --- /dev/null > +++ b/example/time/time_global_test.c > @@ -0,0 +1,352 @@ > +/* Copyright (c) 2015, Linaro Limited > + * All rights reserved. > + * > + * SPDX-License-Identifier: BSD-3-Clause > + */ > + > +#include <odp.h> > +#include <example_debug.h> > +#include <odp/helper/linux.h> > + > +#define MAX_WORKERS 32 > +#define ITERATION_NUM 2048 > +#define LOG_LINE_SIZE 153 > +#define LOG_BASE 8 > + > +#define QUEUE_NAME_PREFIX "thread_queue_" > + > +typedef struct { > + odp_time_t timestamp; > + int id; > +} timestamp_event_t; > + > +typedef struct { > + uint8_t thr; > + uint8_t id; > + uint64_t ns; > +} log_entry_t; > + > +typedef struct { > + uint32_t iteration_num; > + odp_atomic_u32_t iteration_counter; > + odp_atomic_u32_t id_counter; > + odp_atomic_u32_t log_counter; > + odp_atomic_u32_t err_counter; > + odp_barrier_t start_barrier; > + odp_barrier_t end_barrier; > + int thread_num; > + log_entry_t *log; > + int log_enries_num; > +} test_globals_t; > + > +static void print_log(test_globals_t *gbls) > +{ > + uint32_t err_num; > + int i, j, k, padding; > + char line[LOG_LINE_SIZE]; > + > + memset(line, '-', LOG_LINE_SIZE - 1); > + line[LOG_LINE_SIZE - 1] = 0; > + for (i = 1; i <= gbls->thread_num; i++) { > + printf("\n==== history of %d buffer, time,ns (thread) > ====\n%s\n", > + i, line); > + > + /* print log for buffer */ > + k = 0; > + for (j = 0; j < gbls->log_enries_num; j++) > + if (gbls->log[j].id == i) { > + printf("%10" PRIu64 " (%-3d)", > gbls->log[j].ns, > + gbls->log[j].thr); > + > + if (!(++k % LOG_BASE)) > + printf(" |\n"); > + else > + printf(" =>"); > + } > + > + if ((k % 8)) { > + padding = (LOG_BASE - k % LOG_BASE) * 19 - 4; > + printf(" end%*c\n%s\n", padding, '|', line); > + } else { > + printf("%s\n", line); > + } > + } > + > + printf("\n\n"); > + > + err_num = odp_atomic_load_u32(&gbls->err_counter); > + if (err_num) > + printf("Number of errors: %u\n", err_num); > +} > + > +static void > +generate_next_queue(test_globals_t *gbls, odp_queue_t *queue, unsigned > int id) > +{ > + int thr; > + unsigned int rand_id; > + char queue_name[sizeof(QUEUE_NAME_PREFIX) + 2]; > + > + thr = odp_thread_id(); > + > + /* generate next random id */ > + do { > + odp_random_data((uint8_t *)&rand_id, sizeof(rand_id), 1); > + rand_id = rand_id % gbls->thread_num + 1; > + } while (rand_id == id); > + > + sprintf(queue_name, QUEUE_NAME_PREFIX "%d", rand_id); > + *queue = odp_queue_lookup(queue_name); > + > + if (ODP_QUEUE_INVALID == *queue) > + EXAMPLE_ABORT("Cannot lookup thread queue \"%s\", thread > %d\n", > + queue_name, thr); > +} > + > +static void test_global_timestamps(test_globals_t *gbls, > + odp_queue_t queue, unsigned int id) > +{ > + int thr; > + int log_entry; > + odp_event_t ev; > + odp_time_t time; > + odp_buffer_t buf; > + odp_queue_t queue_next; > + timestamp_event_t *timestamp_ev; > + > + thr = odp_thread_id(); > + while (odp_atomic_load_u32(&gbls->iteration_counter) < > + gbls->iteration_num) { > + ev = odp_queue_deq(queue); > + > + if (ev == ODP_EVENT_INVALID) > + continue; > + > + buf = odp_buffer_from_event(ev); > + timestamp_ev = (timestamp_event_t *)odp_buffer_addr(buf); > + > + time = odp_time_global(); > + if (odp_time_cmp(time, timestamp_ev->timestamp) < 0) { > + EXAMPLE_ERR("timestamp is less than previous > time_prev=%" PRIu64 "ns, time_next=%" PRIu64 "ns, thread %d\n", > + > odp_time_to_ns(timestamp_ev->timestamp), > + odp_time_to_ns(time), thr); > + odp_atomic_inc_u32(&gbls->err_counter); > + } > + > + /* update the log */ > + log_entry = odp_atomic_fetch_inc_u32(&gbls->log_counter); > + gbls->log[log_entry].ns = > + odp_time_to_ns(timestamp_ev->timestamp); > + gbls->log[log_entry].id = timestamp_ev->id; > + gbls->log[log_entry].thr = thr; > + > + /* assign new current time and send */ > + generate_next_queue(gbls, &queue_next, id); > + timestamp_ev->timestamp = time; > + if (odp_queue_enq(queue_next, ev)) > + EXAMPLE_ABORT("Cannot enqueue event %" PRIu64 " on > queue %" PRIu64 ", thread %d\n", > + odp_event_to_u64(ev), > + odp_queue_to_u64(queue_next), thr); > + > + odp_atomic_inc_u32(&gbls->iteration_counter); > + } > +} > + > +/** > + * @internal Worker thread > + * > + * @param ptr Pointer to test arguments > + * > + * @return Pointer to exit status > + */ > +static void *run_thread(void *ptr) > +{ > + int thr; > + uint32_t id; > + odp_event_t ev; > + odp_buffer_t buf; > + test_globals_t *gbls; > + odp_pool_t buffer_pool; > + odp_queue_param_t qparams; > + odp_queue_t queue, queue_next; > + timestamp_event_t *timestamp_ev; > + char queue_name[sizeof(QUEUE_NAME_PREFIX) + 2]; > + > + gbls = ptr; > + thr = odp_thread_id(); > + printf("Thread %i starts on cpu %i\n", thr, odp_cpu_id()); > + > + /* > + * Allocate own queue for receiving timestamps. > + * Own queue is needed to guarantee that next thread for receiving > + * buffer is not the same thread. > + */ > + odp_queue_param_init(&qparams); > + qparams.sched.prio = ODP_SCHED_PRIO_LOWEST; > + qparams.sched.sync = ODP_SCHED_SYNC_NONE; > + qparams.sched.group = ODP_SCHED_GROUP_WORKER; > + > + id = odp_atomic_fetch_inc_u32(&gbls->id_counter); > + sprintf(queue_name, QUEUE_NAME_PREFIX "%d", id); > + queue = odp_queue_create(queue_name, ODP_QUEUE_TYPE_POLL, > &qparams); > + if (queue == ODP_QUEUE_INVALID) > + EXAMPLE_ABORT("Cannot create thread queue, thread %d", > thr); > + > + /* allocate buffer for timestamp */ > + buffer_pool = odp_pool_lookup("time buffers pool"); > + if (buffer_pool == ODP_POOL_INVALID) > + EXAMPLE_ABORT("Buffer pool was not found, thread %d\n", > thr); > + > + buf = odp_buffer_alloc(buffer_pool); > + if (buf == ODP_BUFFER_INVALID) > + EXAMPLE_ABORT("Buffer was not allocated, thread %d\n", > thr); > + > + /* wait all threads allocated their queues */ > + odp_barrier_wait(&gbls->start_barrier); > + > + /* enqueue global timestamp to some queue of some other thread */ > + generate_next_queue(gbls, &queue_next, id); > + > + /* save global timestamp and id for tracing */ > + ev = odp_buffer_to_event(buf); > + timestamp_ev = (timestamp_event_t *)odp_buffer_addr(buf); > + timestamp_ev->id = id; > + timestamp_ev->timestamp = odp_time_global(); > + if (odp_queue_enq(queue_next, ev)) > + EXAMPLE_ABORT("Cannot enqueue timestamp event %" PRIu64 " > on queue %" PRIu64 ", thread %d", > + odp_event_to_u64(ev), > + odp_queue_to_u64(queue_next), thr); > + > + test_global_timestamps(gbls, queue, id); > + > + /* wait all threads are finished their jobs */ > + odp_barrier_wait(&gbls->end_barrier); > + > + /* free all events on the allocated queue */ > + while (1) { > + ev = odp_queue_deq(queue); > + if (ev == ODP_EVENT_INVALID) > + break; > + > + buf = odp_buffer_from_event(ev); > + odp_buffer_free(buf); > + } > + > + /* free allocated queue */ > + if (odp_queue_destroy(queue)) > + EXAMPLE_ABORT("Cannot destroy queue %" PRIu64 "", > + odp_queue_to_u64(queue)); > + > + printf("Thread %i exits\n", thr); > + fflush(NULL); > + return NULL; > +} > + > +int main(void) > +{ > + int err = 0; > + odp_pool_t pool = ODP_POOL_INVALID; > + int num_workers; > + test_globals_t *gbls; > + odp_cpumask_t cpumask; > + odp_pool_param_t params; > + odp_shm_t shm_glbls = ODP_SHM_INVALID; > + odp_shm_t shm_log = ODP_SHM_INVALID; > + int log_size, log_enries_num; > + odph_linux_pthread_t thread_tbl[MAX_WORKERS]; > + > + printf("\nODP global time test starts\n"); > + > + if (odp_init_global(NULL, NULL)) { > + err = 1; > + EXAMPLE_ERR("ODP global init failed.\n"); > + goto end; > + } > + > + /* Init this thread. */ > + if (odp_init_local(ODP_THREAD_CONTROL)) { > + err = 1; > + EXAMPLE_ERR("ODP local init failed.\n"); > + goto err_global; > + } > + > + num_workers = MAX_WORKERS; > + num_workers = odp_cpumask_default_worker(&cpumask, num_workers); > + > + shm_glbls = odp_shm_reserve("test_globals", sizeof(test_globals_t), > + ODP_CACHE_LINE_SIZE, 0); > + if (ODP_SHM_INVALID == shm_glbls) { > + err = 1; > + EXAMPLE_ERR("Error: shared mem reserve failed.\n"); > + goto err; > + } > + > + log_enries_num = num_workers * (ITERATION_NUM + num_workers); > + log_size = sizeof(log_entry_t) * log_enries_num; > + shm_log = odp_shm_reserve("test_log", log_size, > ODP_CACHE_LINE_SIZE, 0); > + if (ODP_SHM_INVALID == shm_log) { > + err = 1; > + EXAMPLE_ERR("Error: shared mem reserve failed.\n"); > + goto err; > + } > + > + gbls = odp_shm_addr(shm_glbls); > + gbls->thread_num = num_workers; > + gbls->iteration_num = ITERATION_NUM; > + odp_atomic_store_u32(&gbls->iteration_counter, 0); > + odp_atomic_store_u32(&gbls->id_counter, 1); > + odp_atomic_store_u32(&gbls->log_counter, 0); > + odp_atomic_store_u32(&gbls->err_counter, 0); > + gbls->log_enries_num = log_enries_num; > + gbls->log = odp_shm_addr(shm_log); > + odp_barrier_init(&gbls->start_barrier, num_workers); > + odp_barrier_init(&gbls->end_barrier, num_workers); > + memset(gbls->log, 0, log_size); > + > + params.buf.size = sizeof(timestamp_event_t); > + params.buf.align = ODP_CACHE_LINE_SIZE; > + params.buf.num = num_workers; > + params.type = ODP_POOL_BUFFER; > + > + pool = odp_pool_create("time buffers pool", ¶ms); > + if (pool == ODP_POOL_INVALID) { > + err = 1; > + EXAMPLE_ERR("Pool create failed.\n"); > + goto err; > + } > + > + /* Create and launch worker threads */ > + odph_linux_pthread_create(thread_tbl, &cpumask, > + run_thread, gbls, ODP_THREAD_WORKER); > + > + /* Wait for worker threads to exit */ > + odph_linux_pthread_join(thread_tbl, num_workers); > + > + print_log(gbls); > + > +err: > + if (pool != ODP_POOL_INVALID) > + if (odp_pool_destroy(pool)) > + err = 1; > + > + if (shm_log != ODP_SHM_INVALID) > + if (odp_shm_free(shm_log)) > + err = 1; > + > + if (shm_glbls != ODP_SHM_INVALID) > + if (odp_shm_free(shm_glbls)) > + err = 1; > + > + if (odp_term_local()) > + err = 1; > +err_global: > + if (odp_term_global()) > + err = 1; > +end: > + if (err) { > + EXAMPLE_ERR("Err: ODP global time test failed\n\n"); > + return -1; > + } > + > + printf("ODP global time test complete\n\n"); > + return 0; > +} > -- > 1.9.1 > > _______________________________________________ > lng-odp mailing list > lng-odp@lists.linaro.org > https://lists.linaro.org/mailman/listinfo/lng-odp >
sent v3 with your reviewed-by. On 03.01.16 15:11, Bill Fischofer wrote: > You need to create a .gitignore file in example/time and add the output binary (odp_global_time_test) to it for this patch to be complete. Other than that: > > On Thu, Dec 31, 2015 at 5:46 AM, Ivan Khoronzhuk <ivan.khoronzhuk@linaro.org <mailto:ivan.khoronzhuk@linaro.org>> wrote: > > This test example uses linux helper functions to schedule worker > threads in order to check global time monotony between the threads. > Each thread receives event from it's own queue, checks global time on > correctness, writes in the buffer new current global time, > chooses randomly next thread queue, sends event and waits on next > event on it's own queue. Using for each thread it's own queue guarantees > that event is read by not the same thread that sends it. When global time > source is correct the time received with event by a thread must be less > than read on the moment of receiving. > > Signed-off-by: Ivan Khoronzhuk <ivan.khoronzhuk@linaro.org <mailto:ivan.khoronzhuk@linaro.org>> > > > Reviewed-and-tested-by: Bill Fischofer <bill.fischofer@linaro.org <mailto:bill.fischofer@linaro.org>> > > --- > > Since v2: > - replaced many includes on odp.h > - corrected a little error path > - increased num of iterations > - rebased on ODPv1.6 > > configure.ac <http://configure.ac> | 1 + > example/Makefile.am | 2 +- > example/time/Makefile.am | 10 ++ > example/time/time_global_test.c | 352 ++++++++++++++++++++++++++++++++++++++++ > 4 files changed, 364 insertions(+), 1 deletion(-) > create mode 100644 example/time/Makefile.am > create mode 100644 example/time/time_global_test.c > > diff --git a/configure.ac <http://configure.ac> b/configure.ac <http://configure.ac> > index 2626fd7..b284384 100644 > --- a/configure.ac <http://configure.ac> > +++ b/configure.ac <http://configure.ac> > @@ -338,6 +338,7 @@ AC_CONFIG_FILES([Makefile > example/generator/Makefile > example/ipsec/Makefile > example/packet/Makefile > + example/time/Makefile > example/timer/Makefile > helper/Makefile > helper/test/Makefile > diff --git a/example/Makefile.am b/example/Makefile.am > index 353f397..39d9b01 100644 > --- a/example/Makefile.am > +++ b/example/Makefile.am > @@ -1 +1 @@ > -SUBDIRS = classifier generator ipsec packet timer > +SUBDIRS = classifier generator ipsec packet time timer > diff --git a/example/time/Makefile.am b/example/time/Makefile.am > new file mode 100644 > index 0000000..915593a > --- /dev/null > +++ b/example/time/Makefile.am > @@ -0,0 +1,10 @@ > +include $(top_srcdir)/example/Makefile.inc > + > +bin_PROGRAMS = odp_time_global_test$(EXEEXT) > +odp_time_global_test_LDFLAGS = $(AM_LDFLAGS) -static > +odp_time_global_test_CFLAGS = $(AM_CFLAGS) -I${top_srcdir}/example > + > +noinst_HEADERS = \ > + $(top_srcdir)/example/example_debug.h > + > +dist_odp_time_global_test_SOURCES = time_global_test.c > diff --git a/example/time/time_global_test.c b/example/time/time_global_test.c > new file mode 100644 > index 0000000..de0ab47 > --- /dev/null > +++ b/example/time/time_global_test.c > @@ -0,0 +1,352 @@ > +/* Copyright (c) 2015, Linaro Limited > + * All rights reserved. > + * > + * SPDX-License-Identifier: BSD-3-Clause > + */ > + > +#include <odp.h> > +#include <example_debug.h> > +#include <odp/helper/linux.h> > + > +#define MAX_WORKERS 32 > +#define ITERATION_NUM 2048 > +#define LOG_LINE_SIZE 153 > +#define LOG_BASE 8 > + > +#define QUEUE_NAME_PREFIX "thread_queue_" > + > +typedef struct { > + odp_time_t timestamp; > + int id; > +} timestamp_event_t; > + > +typedef struct { > + uint8_t thr; > + uint8_t id; > + uint64_t ns; > +} log_entry_t; > + > +typedef struct { > + uint32_t iteration_num; > + odp_atomic_u32_t iteration_counter; > + odp_atomic_u32_t id_counter; > + odp_atomic_u32_t log_counter; > + odp_atomic_u32_t err_counter; > + odp_barrier_t start_barrier; > + odp_barrier_t end_barrier; > + int thread_num; > + log_entry_t *log; > + int log_enries_num; > +} test_globals_t; > + > +static void print_log(test_globals_t *gbls) > +{ > + uint32_t err_num; > + int i, j, k, padding; > + char line[LOG_LINE_SIZE]; > + > + memset(line, '-', LOG_LINE_SIZE - 1); > + line[LOG_LINE_SIZE - 1] = 0; > + for (i = 1; i <= gbls->thread_num; i++) { > + printf("\n==== history of %d buffer, time,ns (thread) ====\n%s\n", > + i, line); > + > + /* print log for buffer */ > + k = 0; > + for (j = 0; j < gbls->log_enries_num; j++) > + if (gbls->log[j].id == i) { > + printf("%10" PRIu64 " (%-3d)", gbls->log[j].ns, > + gbls->log[j].thr); > + > + if (!(++k % LOG_BASE)) > + printf(" |\n"); > + else > + printf(" =>"); > + } > + > + if ((k % 8)) { > + padding = (LOG_BASE - k % LOG_BASE) * 19 - 4; > + printf(" end%*c\n%s\n", padding, '|', line); > + } else { > + printf("%s\n", line); > + } > + } > + > + printf("\n\n"); > + > + err_num = odp_atomic_load_u32(&gbls->err_counter); > + if (err_num) > + printf("Number of errors: %u\n", err_num); > +} > + > +static void > +generate_next_queue(test_globals_t *gbls, odp_queue_t *queue, unsigned int id) > +{ > + int thr; > + unsigned int rand_id; > + char queue_name[sizeof(QUEUE_NAME_PREFIX) + 2]; > + > + thr = odp_thread_id(); > + > + /* generate next random id */ > + do { > + odp_random_data((uint8_t *)&rand_id, sizeof(rand_id), 1); > + rand_id = rand_id % gbls->thread_num + 1; > + } while (rand_id == id); > + > + sprintf(queue_name, QUEUE_NAME_PREFIX "%d", rand_id); > + *queue = odp_queue_lookup(queue_name); > + > + if (ODP_QUEUE_INVALID == *queue) > + EXAMPLE_ABORT("Cannot lookup thread queue \"%s\", thread %d\n", > + queue_name, thr); > +} > + > +static void test_global_timestamps(test_globals_t *gbls, > + odp_queue_t queue, unsigned int id) > +{ > + int thr; > + int log_entry; > + odp_event_t ev; > + odp_time_t time; > + odp_buffer_t buf; > + odp_queue_t queue_next; > + timestamp_event_t *timestamp_ev; > + > + thr = odp_thread_id(); > + while (odp_atomic_load_u32(&gbls->iteration_counter) < > + gbls->iteration_num) { > + ev = odp_queue_deq(queue); > + > + if (ev == ODP_EVENT_INVALID) > + continue; > + > + buf = odp_buffer_from_event(ev); > + timestamp_ev = (timestamp_event_t *)odp_buffer_addr(buf); > + > + time = odp_time_global(); > + if (odp_time_cmp(time, timestamp_ev->timestamp) < 0) { > + EXAMPLE_ERR("timestamp is less than previous time_prev=%" PRIu64 "ns, time_next=%" PRIu64 "ns, thread %d\n", > + odp_time_to_ns(timestamp_ev->timestamp), > + odp_time_to_ns(time), thr); > + odp_atomic_inc_u32(&gbls->err_counter); > + } > + > + /* update the log */ > + log_entry = odp_atomic_fetch_inc_u32(&gbls->log_counter); > + gbls->log[log_entry].ns = > + odp_time_to_ns(timestamp_ev->timestamp); > + gbls->log[log_entry].id = timestamp_ev->id; > + gbls->log[log_entry].thr = thr; > + > + /* assign new current time and send */ > + generate_next_queue(gbls, &queue_next, id); > + timestamp_ev->timestamp = time; > + if (odp_queue_enq(queue_next, ev)) > + EXAMPLE_ABORT("Cannot enqueue event %" PRIu64 " on queue %" PRIu64 ", thread %d\n", > + odp_event_to_u64(ev), > + odp_queue_to_u64(queue_next), thr); > + > + odp_atomic_inc_u32(&gbls->iteration_counter); > + } > +} > + > +/** > + * @internal Worker thread > + * > + * @param ptr Pointer to test arguments > + * > + * @return Pointer to exit status > + */ > +static void *run_thread(void *ptr) > +{ > + int thr; > + uint32_t id; > + odp_event_t ev; > + odp_buffer_t buf; > + test_globals_t *gbls; > + odp_pool_t buffer_pool; > + odp_queue_param_t qparams; > + odp_queue_t queue, queue_next; > + timestamp_event_t *timestamp_ev; > + char queue_name[sizeof(QUEUE_NAME_PREFIX) + 2]; > + > + gbls = ptr; > + thr = odp_thread_id(); > + printf("Thread %i starts on cpu %i\n", thr, odp_cpu_id()); > + > + /* > + * Allocate own queue for receiving timestamps. > + * Own queue is needed to guarantee that next thread for receiving > + * buffer is not the same thread. > + */ > + odp_queue_param_init(&qparams); > + qparams.sched.prio = ODP_SCHED_PRIO_LOWEST; > + qparams.sched.sync = ODP_SCHED_SYNC_NONE; > + qparams.sched.group = ODP_SCHED_GROUP_WORKER; > + > + id = odp_atomic_fetch_inc_u32(&gbls->id_counter); > + sprintf(queue_name, QUEUE_NAME_PREFIX "%d", id); > + queue = odp_queue_create(queue_name, ODP_QUEUE_TYPE_POLL, &qparams); > + if (queue == ODP_QUEUE_INVALID) > + EXAMPLE_ABORT("Cannot create thread queue, thread %d", thr); > + > + /* allocate buffer for timestamp */ > + buffer_pool = odp_pool_lookup("time buffers pool"); > + if (buffer_pool == ODP_POOL_INVALID) > + EXAMPLE_ABORT("Buffer pool was not found, thread %d\n", thr); > + > + buf = odp_buffer_alloc(buffer_pool); > + if (buf == ODP_BUFFER_INVALID) > + EXAMPLE_ABORT("Buffer was not allocated, thread %d\n", thr); > + > + /* wait all threads allocated their queues */ > + odp_barrier_wait(&gbls->start_barrier); > + > + /* enqueue global timestamp to some queue of some other thread */ > + generate_next_queue(gbls, &queue_next, id); > + > + /* save global timestamp and id for tracing */ > + ev = odp_buffer_to_event(buf); > + timestamp_ev = (timestamp_event_t *)odp_buffer_addr(buf); > + timestamp_ev->id = id; > + timestamp_ev->timestamp = odp_time_global(); > + if (odp_queue_enq(queue_next, ev)) > + EXAMPLE_ABORT("Cannot enqueue timestamp event %" PRIu64 " on queue %" PRIu64 ", thread %d", > + odp_event_to_u64(ev), > + odp_queue_to_u64(queue_next), thr); > + > + test_global_timestamps(gbls, queue, id); > + > + /* wait all threads are finished their jobs */ > + odp_barrier_wait(&gbls->end_barrier); > + > + /* free all events on the allocated queue */ > + while (1) { > + ev = odp_queue_deq(queue); > + if (ev == ODP_EVENT_INVALID) > + break; > + > + buf = odp_buffer_from_event(ev); > + odp_buffer_free(buf); > + } > + > + /* free allocated queue */ > + if (odp_queue_destroy(queue)) > + EXAMPLE_ABORT("Cannot destroy queue %" PRIu64 "", > + odp_queue_to_u64(queue)); > + > + printf("Thread %i exits\n", thr); > + fflush(NULL); > + return NULL; > +} > + > +int main(void) > +{ > + int err = 0; > + odp_pool_t pool = ODP_POOL_INVALID; > + int num_workers; > + test_globals_t *gbls; > + odp_cpumask_t cpumask; > + odp_pool_param_t params; > + odp_shm_t shm_glbls = ODP_SHM_INVALID; > + odp_shm_t shm_log = ODP_SHM_INVALID; > + int log_size, log_enries_num; > + odph_linux_pthread_t thread_tbl[MAX_WORKERS]; > + > + printf("\nODP global time test starts\n"); > + > + if (odp_init_global(NULL, NULL)) { > + err = 1; > + EXAMPLE_ERR("ODP global init failed.\n"); > + goto end; > + } > + > + /* Init this thread. */ > + if (odp_init_local(ODP_THREAD_CONTROL)) { > + err = 1; > + EXAMPLE_ERR("ODP local init failed.\n"); > + goto err_global; > + } > + > + num_workers = MAX_WORKERS; > + num_workers = odp_cpumask_default_worker(&cpumask, num_workers); > + > + shm_glbls = odp_shm_reserve("test_globals", sizeof(test_globals_t), > + ODP_CACHE_LINE_SIZE, 0); > + if (ODP_SHM_INVALID == shm_glbls) { > + err = 1; > + EXAMPLE_ERR("Error: shared mem reserve failed.\n"); > + goto err; > + } > + > + log_enries_num = num_workers * (ITERATION_NUM + num_workers); > + log_size = sizeof(log_entry_t) * log_enries_num; > + shm_log = odp_shm_reserve("test_log", log_size, ODP_CACHE_LINE_SIZE, 0); > + if (ODP_SHM_INVALID == shm_log) { > + err = 1; > + EXAMPLE_ERR("Error: shared mem reserve failed.\n"); > + goto err; > + } > + > + gbls = odp_shm_addr(shm_glbls); > + gbls->thread_num = num_workers; > + gbls->iteration_num = ITERATION_NUM; > + odp_atomic_store_u32(&gbls->iteration_counter, 0); > + odp_atomic_store_u32(&gbls->id_counter, 1); > + odp_atomic_store_u32(&gbls->log_counter, 0); > + odp_atomic_store_u32(&gbls->err_counter, 0); > + gbls->log_enries_num = log_enries_num; > + gbls->log = odp_shm_addr(shm_log); > + odp_barrier_init(&gbls->start_barrier, num_workers); > + odp_barrier_init(&gbls->end_barrier, num_workers); > + memset(gbls->log, 0, log_size); > + > + params.buf.size = sizeof(timestamp_event_t); > + params.buf.align = ODP_CACHE_LINE_SIZE; > + params.buf.num = num_workers; > + params.type = ODP_POOL_BUFFER; > + > + pool = odp_pool_create("time buffers pool", ¶ms); > + if (pool == ODP_POOL_INVALID) { > + err = 1; > + EXAMPLE_ERR("Pool create failed.\n"); > + goto err; > + } > + > + /* Create and launch worker threads */ > + odph_linux_pthread_create(thread_tbl, &cpumask, > + run_thread, gbls, ODP_THREAD_WORKER); > + > + /* Wait for worker threads to exit */ > + odph_linux_pthread_join(thread_tbl, num_workers); > + > + print_log(gbls); > + > +err: > + if (pool != ODP_POOL_INVALID) > + if (odp_pool_destroy(pool)) > + err = 1; > + > + if (shm_log != ODP_SHM_INVALID) > + if (odp_shm_free(shm_log)) > + err = 1; > + > + if (shm_glbls != ODP_SHM_INVALID) > + if (odp_shm_free(shm_glbls)) > + err = 1; > + > + if (odp_term_local()) > + err = 1; > +err_global: > + if (odp_term_global()) > + err = 1; > +end: > + if (err) { > + EXAMPLE_ERR("Err: ODP global time test failed\n\n"); > + return -1; > + } > + > + printf("ODP global time test complete\n\n"); > + return 0; > +} > -- > 1.9.1 > > _______________________________________________ > lng-odp mailing list > lng-odp@lists.linaro.org <mailto:lng-odp@lists.linaro.org> > https://lists.linaro.org/mailman/listinfo/lng-odp > >
diff --git a/configure.ac b/configure.ac index 2626fd7..b284384 100644 --- a/configure.ac +++ b/configure.ac @@ -338,6 +338,7 @@ AC_CONFIG_FILES([Makefile example/generator/Makefile example/ipsec/Makefile example/packet/Makefile + example/time/Makefile example/timer/Makefile helper/Makefile helper/test/Makefile diff --git a/example/Makefile.am b/example/Makefile.am index 353f397..39d9b01 100644 --- a/example/Makefile.am +++ b/example/Makefile.am @@ -1 +1 @@ -SUBDIRS = classifier generator ipsec packet timer +SUBDIRS = classifier generator ipsec packet time timer diff --git a/example/time/Makefile.am b/example/time/Makefile.am new file mode 100644 index 0000000..915593a --- /dev/null +++ b/example/time/Makefile.am @@ -0,0 +1,10 @@ +include $(top_srcdir)/example/Makefile.inc + +bin_PROGRAMS = odp_time_global_test$(EXEEXT) +odp_time_global_test_LDFLAGS = $(AM_LDFLAGS) -static +odp_time_global_test_CFLAGS = $(AM_CFLAGS) -I${top_srcdir}/example + +noinst_HEADERS = \ + $(top_srcdir)/example/example_debug.h + +dist_odp_time_global_test_SOURCES = time_global_test.c diff --git a/example/time/time_global_test.c b/example/time/time_global_test.c new file mode 100644 index 0000000..de0ab47 --- /dev/null +++ b/example/time/time_global_test.c @@ -0,0 +1,352 @@ +/* Copyright (c) 2015, Linaro Limited + * All rights reserved. + * + * SPDX-License-Identifier: BSD-3-Clause + */ + +#include <odp.h> +#include <example_debug.h> +#include <odp/helper/linux.h> + +#define MAX_WORKERS 32 +#define ITERATION_NUM 2048 +#define LOG_LINE_SIZE 153 +#define LOG_BASE 8 + +#define QUEUE_NAME_PREFIX "thread_queue_" + +typedef struct { + odp_time_t timestamp; + int id; +} timestamp_event_t; + +typedef struct { + uint8_t thr; + uint8_t id; + uint64_t ns; +} log_entry_t; + +typedef struct { + uint32_t iteration_num; + odp_atomic_u32_t iteration_counter; + odp_atomic_u32_t id_counter; + odp_atomic_u32_t log_counter; + odp_atomic_u32_t err_counter; + odp_barrier_t start_barrier; + odp_barrier_t end_barrier; + int thread_num; + log_entry_t *log; + int log_enries_num; +} test_globals_t; + +static void print_log(test_globals_t *gbls) +{ + uint32_t err_num; + int i, j, k, padding; + char line[LOG_LINE_SIZE]; + + memset(line, '-', LOG_LINE_SIZE - 1); + line[LOG_LINE_SIZE - 1] = 0; + for (i = 1; i <= gbls->thread_num; i++) { + printf("\n==== history of %d buffer, time,ns (thread) ====\n%s\n", + i, line); + + /* print log for buffer */ + k = 0; + for (j = 0; j < gbls->log_enries_num; j++) + if (gbls->log[j].id == i) { + printf("%10" PRIu64 " (%-3d)", gbls->log[j].ns, + gbls->log[j].thr); + + if (!(++k % LOG_BASE)) + printf(" |\n"); + else + printf(" =>"); + } + + if ((k % 8)) { + padding = (LOG_BASE - k % LOG_BASE) * 19 - 4; + printf(" end%*c\n%s\n", padding, '|', line); + } else { + printf("%s\n", line); + } + } + + printf("\n\n"); + + err_num = odp_atomic_load_u32(&gbls->err_counter); + if (err_num) + printf("Number of errors: %u\n", err_num); +} + +static void +generate_next_queue(test_globals_t *gbls, odp_queue_t *queue, unsigned int id) +{ + int thr; + unsigned int rand_id; + char queue_name[sizeof(QUEUE_NAME_PREFIX) + 2]; + + thr = odp_thread_id(); + + /* generate next random id */ + do { + odp_random_data((uint8_t *)&rand_id, sizeof(rand_id), 1); + rand_id = rand_id % gbls->thread_num + 1; + } while (rand_id == id); + + sprintf(queue_name, QUEUE_NAME_PREFIX "%d", rand_id); + *queue = odp_queue_lookup(queue_name); + + if (ODP_QUEUE_INVALID == *queue) + EXAMPLE_ABORT("Cannot lookup thread queue \"%s\", thread %d\n", + queue_name, thr); +} + +static void test_global_timestamps(test_globals_t *gbls, + odp_queue_t queue, unsigned int id) +{ + int thr; + int log_entry; + odp_event_t ev; + odp_time_t time; + odp_buffer_t buf; + odp_queue_t queue_next; + timestamp_event_t *timestamp_ev; + + thr = odp_thread_id(); + while (odp_atomic_load_u32(&gbls->iteration_counter) < + gbls->iteration_num) { + ev = odp_queue_deq(queue); + + if (ev == ODP_EVENT_INVALID) + continue; + + buf = odp_buffer_from_event(ev); + timestamp_ev = (timestamp_event_t *)odp_buffer_addr(buf); + + time = odp_time_global(); + if (odp_time_cmp(time, timestamp_ev->timestamp) < 0) { + EXAMPLE_ERR("timestamp is less than previous time_prev=%" PRIu64 "ns, time_next=%" PRIu64 "ns, thread %d\n", + odp_time_to_ns(timestamp_ev->timestamp), + odp_time_to_ns(time), thr); + odp_atomic_inc_u32(&gbls->err_counter); + } + + /* update the log */ + log_entry = odp_atomic_fetch_inc_u32(&gbls->log_counter); + gbls->log[log_entry].ns = + odp_time_to_ns(timestamp_ev->timestamp); + gbls->log[log_entry].id = timestamp_ev->id; + gbls->log[log_entry].thr = thr; + + /* assign new current time and send */ + generate_next_queue(gbls, &queue_next, id); + timestamp_ev->timestamp = time; + if (odp_queue_enq(queue_next, ev)) + EXAMPLE_ABORT("Cannot enqueue event %" PRIu64 " on queue %" PRIu64 ", thread %d\n", + odp_event_to_u64(ev), + odp_queue_to_u64(queue_next), thr); + + odp_atomic_inc_u32(&gbls->iteration_counter); + } +} + +/** + * @internal Worker thread + * + * @param ptr Pointer to test arguments + * + * @return Pointer to exit status + */ +static void *run_thread(void *ptr) +{ + int thr; + uint32_t id; + odp_event_t ev; + odp_buffer_t buf; + test_globals_t *gbls; + odp_pool_t buffer_pool; + odp_queue_param_t qparams; + odp_queue_t queue, queue_next; + timestamp_event_t *timestamp_ev; + char queue_name[sizeof(QUEUE_NAME_PREFIX) + 2]; + + gbls = ptr; + thr = odp_thread_id(); + printf("Thread %i starts on cpu %i\n", thr, odp_cpu_id()); + + /* + * Allocate own queue for receiving timestamps. + * Own queue is needed to guarantee that next thread for receiving + * buffer is not the same thread. + */ + odp_queue_param_init(&qparams); + qparams.sched.prio = ODP_SCHED_PRIO_LOWEST; + qparams.sched.sync = ODP_SCHED_SYNC_NONE; + qparams.sched.group = ODP_SCHED_GROUP_WORKER; + + id = odp_atomic_fetch_inc_u32(&gbls->id_counter); + sprintf(queue_name, QUEUE_NAME_PREFIX "%d", id); + queue = odp_queue_create(queue_name, ODP_QUEUE_TYPE_POLL, &qparams); + if (queue == ODP_QUEUE_INVALID) + EXAMPLE_ABORT("Cannot create thread queue, thread %d", thr); + + /* allocate buffer for timestamp */ + buffer_pool = odp_pool_lookup("time buffers pool"); + if (buffer_pool == ODP_POOL_INVALID) + EXAMPLE_ABORT("Buffer pool was not found, thread %d\n", thr); + + buf = odp_buffer_alloc(buffer_pool); + if (buf == ODP_BUFFER_INVALID) + EXAMPLE_ABORT("Buffer was not allocated, thread %d\n", thr); + + /* wait all threads allocated their queues */ + odp_barrier_wait(&gbls->start_barrier); + + /* enqueue global timestamp to some queue of some other thread */ + generate_next_queue(gbls, &queue_next, id); + + /* save global timestamp and id for tracing */ + ev = odp_buffer_to_event(buf); + timestamp_ev = (timestamp_event_t *)odp_buffer_addr(buf); + timestamp_ev->id = id; + timestamp_ev->timestamp = odp_time_global(); + if (odp_queue_enq(queue_next, ev)) + EXAMPLE_ABORT("Cannot enqueue timestamp event %" PRIu64 " on queue %" PRIu64 ", thread %d", + odp_event_to_u64(ev), + odp_queue_to_u64(queue_next), thr); + + test_global_timestamps(gbls, queue, id); + + /* wait all threads are finished their jobs */ + odp_barrier_wait(&gbls->end_barrier); + + /* free all events on the allocated queue */ + while (1) { + ev = odp_queue_deq(queue); + if (ev == ODP_EVENT_INVALID) + break; + + buf = odp_buffer_from_event(ev); + odp_buffer_free(buf); + } + + /* free allocated queue */ + if (odp_queue_destroy(queue)) + EXAMPLE_ABORT("Cannot destroy queue %" PRIu64 "", + odp_queue_to_u64(queue)); + + printf("Thread %i exits\n", thr); + fflush(NULL); + return NULL; +} + +int main(void) +{ + int err = 0; + odp_pool_t pool = ODP_POOL_INVALID; + int num_workers; + test_globals_t *gbls; + odp_cpumask_t cpumask; + odp_pool_param_t params; + odp_shm_t shm_glbls = ODP_SHM_INVALID; + odp_shm_t shm_log = ODP_SHM_INVALID; + int log_size, log_enries_num; + odph_linux_pthread_t thread_tbl[MAX_WORKERS]; + + printf("\nODP global time test starts\n"); + + if (odp_init_global(NULL, NULL)) { + err = 1; + EXAMPLE_ERR("ODP global init failed.\n"); + goto end; + } + + /* Init this thread. */ + if (odp_init_local(ODP_THREAD_CONTROL)) { + err = 1; + EXAMPLE_ERR("ODP local init failed.\n"); + goto err_global; + } + + num_workers = MAX_WORKERS; + num_workers = odp_cpumask_default_worker(&cpumask, num_workers); + + shm_glbls = odp_shm_reserve("test_globals", sizeof(test_globals_t), + ODP_CACHE_LINE_SIZE, 0); + if (ODP_SHM_INVALID == shm_glbls) { + err = 1; + EXAMPLE_ERR("Error: shared mem reserve failed.\n"); + goto err; + } + + log_enries_num = num_workers * (ITERATION_NUM + num_workers); + log_size = sizeof(log_entry_t) * log_enries_num; + shm_log = odp_shm_reserve("test_log", log_size, ODP_CACHE_LINE_SIZE, 0); + if (ODP_SHM_INVALID == shm_log) { + err = 1; + EXAMPLE_ERR("Error: shared mem reserve failed.\n"); + goto err; + } + + gbls = odp_shm_addr(shm_glbls); + gbls->thread_num = num_workers; + gbls->iteration_num = ITERATION_NUM; + odp_atomic_store_u32(&gbls->iteration_counter, 0); + odp_atomic_store_u32(&gbls->id_counter, 1); + odp_atomic_store_u32(&gbls->log_counter, 0); + odp_atomic_store_u32(&gbls->err_counter, 0); + gbls->log_enries_num = log_enries_num; + gbls->log = odp_shm_addr(shm_log); + odp_barrier_init(&gbls->start_barrier, num_workers); + odp_barrier_init(&gbls->end_barrier, num_workers); + memset(gbls->log, 0, log_size); + + params.buf.size = sizeof(timestamp_event_t); + params.buf.align = ODP_CACHE_LINE_SIZE; + params.buf.num = num_workers; + params.type = ODP_POOL_BUFFER; + + pool = odp_pool_create("time buffers pool", ¶ms); + if (pool == ODP_POOL_INVALID) { + err = 1; + EXAMPLE_ERR("Pool create failed.\n"); + goto err; + } + + /* Create and launch worker threads */ + odph_linux_pthread_create(thread_tbl, &cpumask, + run_thread, gbls, ODP_THREAD_WORKER); + + /* Wait for worker threads to exit */ + odph_linux_pthread_join(thread_tbl, num_workers); + + print_log(gbls); + +err: + if (pool != ODP_POOL_INVALID) + if (odp_pool_destroy(pool)) + err = 1; + + if (shm_log != ODP_SHM_INVALID) + if (odp_shm_free(shm_log)) + err = 1; + + if (shm_glbls != ODP_SHM_INVALID) + if (odp_shm_free(shm_glbls)) + err = 1; + + if (odp_term_local()) + err = 1; +err_global: + if (odp_term_global()) + err = 1; +end: + if (err) { + EXAMPLE_ERR("Err: ODP global time test failed\n\n"); + return -1; + } + + printf("ODP global time test complete\n\n"); + return 0; +}
This test example uses linux helper functions to schedule worker threads in order to check global time monotony between the threads. Each thread receives event from it's own queue, checks global time on correctness, writes in the buffer new current global time, chooses randomly next thread queue, sends event and waits on next event on it's own queue. Using for each thread it's own queue guarantees that event is read by not the same thread that sends it. When global time source is correct the time received with event by a thread must be less than read on the moment of receiving. Signed-off-by: Ivan Khoronzhuk <ivan.khoronzhuk@linaro.org> --- Since v2: - replaced many includes on odp.h - corrected a little error path - increased num of iterations - rebased on ODPv1.6 configure.ac | 1 + example/Makefile.am | 2 +- example/time/Makefile.am | 10 ++ example/time/time_global_test.c | 352 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 364 insertions(+), 1 deletion(-) create mode 100644 example/time/Makefile.am create mode 100644 example/time/time_global_test.c