app/test/test_ring_stress_impl.h

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2020 Intel Corporation
   3  */
   4
   5 #include "test_ring_stress.h"
   6
   7 /**
   8  * Stress test for ring enqueue/dequeue operations.
   9  * Performs the following pattern on each slave worker:
  10  * dequeue/read-write data from the dequeued objects/enqueue.
  11  * Serves as both functional and performance test of ring
  12  * enqueue/dequeue operations under high contention
  13  * (for both over committed and non-over committed scenarios).
  14  */
  15
  16 #define RING_NAME       "RING_STRESS"
  17 #define BULK_NUM        32
  18 #define RING_SIZE       (2 * BULK_NUM * RTE_MAX_LCORE)
  19
  20 enum {
  21         WRK_CMD_STOP,
  22         WRK_CMD_RUN,
  23 };
  24
  25 static volatile uint32_t wrk_cmd __rte_cache_aligned;
  26
  27 /* test run-time in seconds */
  28 static const uint32_t run_time = 60;
  29 static const uint32_t verbose;
  30
  31 struct lcore_stat {
  32         uint64_t nb_cycle;
  33         struct {
  34                 uint64_t nb_call;
  35                 uint64_t nb_obj;
  36                 uint64_t nb_cycle;
  37                 uint64_t max_cycle;
  38                 uint64_t min_cycle;
  39         } op;
  40 };
  41
  42 struct lcore_arg {
  43         struct rte_ring *rng;
  44         struct lcore_stat stats;
  45 } __rte_cache_aligned;
  46
  47 struct ring_elem {
  48         uint32_t cnt[RTE_CACHE_LINE_SIZE / sizeof(uint32_t)];
  49 } __rte_cache_aligned;
  50
  51 /*
  52  * redefinable functions
  53  */
  54 static uint32_t
  55 _st_ring_dequeue_bulk(struct rte_ring *r, void **obj, uint32_t n,
  56         uint32_t *avail);
  57
  58 static uint32_t
  59 _st_ring_enqueue_bulk(struct rte_ring *r, void * const *obj, uint32_t n,
  60         uint32_t *free);
  61
  62 static int
  63 _st_ring_init(struct rte_ring *r, const char *name, uint32_t num);
  64
  65
  66 static void
  67 lcore_stat_update(struct lcore_stat *ls, uint64_t call, uint64_t obj,
  68         uint64_t tm, int32_t prcs)
  69 {
  70         ls->op.nb_call += call;
  71         ls->op.nb_obj += obj;
  72         ls->op.nb_cycle += tm;
  73         if (prcs) {
  74                 ls->op.max_cycle = RTE_MAX(ls->op.max_cycle, tm);
  75                 ls->op.min_cycle = RTE_MIN(ls->op.min_cycle, tm);
  76         }
  77 }
  78
  79 static void
  80 lcore_op_stat_aggr(struct lcore_stat *ms, const struct lcore_stat *ls)
  81 {
  82
  83         ms->op.nb_call += ls->op.nb_call;
  84         ms->op.nb_obj += ls->op.nb_obj;
  85         ms->op.nb_cycle += ls->op.nb_cycle;
  86         ms->op.max_cycle = RTE_MAX(ms->op.max_cycle, ls->op.max_cycle);
  87         ms->op.min_cycle = RTE_MIN(ms->op.min_cycle, ls->op.min_cycle);
  88 }
  89
  90 static void
  91 lcore_stat_aggr(struct lcore_stat *ms, const struct lcore_stat *ls)
  92 {
  93         ms->nb_cycle = RTE_MAX(ms->nb_cycle, ls->nb_cycle);
  94         lcore_op_stat_aggr(ms, ls);
  95 }
  96
  97 static void
  98 lcore_stat_dump(FILE *f, uint32_t lc, const struct lcore_stat *ls)
  99 {
 100         long double st;
 101
 102         st = (long double)rte_get_timer_hz() / US_PER_S;
 103
 104         if (lc == UINT32_MAX)
 105                 fprintf(f, "%s(AGGREGATE)={\n", __func__);
 106         else
 107                 fprintf(f, "%s(lcore=%u)={\n", __func__, lc);
 108
 109         fprintf(f, "\tnb_cycle=%" PRIu64 "(%.2Lf usec),\n",
 110                 ls->nb_cycle, (long double)ls->nb_cycle / st);
 111
 112         fprintf(f, "\tDEQ+ENQ={\n");
 113
 114         fprintf(f, "\t\tnb_call=%" PRIu64 ",\n", ls->op.nb_call);
 115         fprintf(f, "\t\tnb_obj=%" PRIu64 ",\n", ls->op.nb_obj);
 116         fprintf(f, "\t\tnb_cycle=%" PRIu64 ",\n", ls->op.nb_cycle);
 117         fprintf(f, "\t\tobj/call(avg): %.2Lf\n",
 118                 (long double)ls->op.nb_obj / ls->op.nb_call);
 119         fprintf(f, "\t\tcycles/obj(avg): %.2Lf\n",
 120                 (long double)ls->op.nb_cycle / ls->op.nb_obj);
 121         fprintf(f, "\t\tcycles/call(avg): %.2Lf\n",
 122                 (long double)ls->op.nb_cycle / ls->op.nb_call);
 123
 124         /* if min/max cycles per call stats was collected */
 125         if (ls->op.min_cycle != UINT64_MAX) {
 126                 fprintf(f, "\t\tmax cycles/call=%" PRIu64 "(%.2Lf usec),\n",
 127                         ls->op.max_cycle,
 128                         (long double)ls->op.max_cycle / st);
 129                 fprintf(f, "\t\tmin cycles/call=%" PRIu64 "(%.2Lf usec),\n",
 130                         ls->op.min_cycle,
 131                         (long double)ls->op.min_cycle / st);
 132         }
 133
 134         fprintf(f, "\t},\n");
 135         fprintf(f, "};\n");
 136 }
 137
 138 static void
 139 fill_ring_elm(struct ring_elem *elm, uint32_t fill)
 140 {
 141         uint32_t i;
 142
 143         for (i = 0; i != RTE_DIM(elm->cnt); i++)
 144                 elm->cnt[i] = fill;
 145 }
 146
 147 static int32_t
 148 check_updt_elem(struct ring_elem *elm[], uint32_t num,
 149         const struct ring_elem *check, const struct ring_elem *fill)
 150 {
 151         uint32_t i;
 152
 153         static rte_spinlock_t dump_lock;
 154
 155         for (i = 0; i != num; i++) {
 156                 if (memcmp(check, elm[i], sizeof(*check)) != 0) {
 157                         rte_spinlock_lock(&dump_lock);
 158                         printf("%s(lc=%u, num=%u) failed at %u-th iter, "
 159                                 "offending object: %p\n",
 160                                 __func__, rte_lcore_id(), num, i, elm[i]);
 161                         rte_memdump(stdout, "expected", check, sizeof(*check));
 162                         rte_memdump(stdout, "result", elm[i], sizeof(elm[i]));
 163                         rte_spinlock_unlock(&dump_lock);
 164                         return -EINVAL;
 165                 }
 166                 memcpy(elm[i], fill, sizeof(*elm[i]));
 167         }
 168
 169         return 0;
 170 }
 171
 172 static int
 173 check_ring_op(uint32_t exp, uint32_t res, uint32_t lc,
 174         const char *fname, const char *opname)
 175 {
 176         if (exp != res) {
 177                 printf("%s(lc=%u) failure: %s expected: %u, returned %u\n",
 178                         fname, lc, opname, exp, res);
 179                 return -ENOSPC;
 180         }
 181         return 0;
 182 }
 183
 184 static int
 185 test_worker(void *arg, const char *fname, int32_t prcs)
 186 {
 187         int32_t rc;
 188         uint32_t lc, n, num;
 189         uint64_t cl, tm0, tm1;
 190         struct lcore_arg *la;
 191         struct ring_elem def_elm, loc_elm;
 192         struct ring_elem *obj[2 * BULK_NUM];
 193
 194         la = arg;
 195         lc = rte_lcore_id();
 196
 197         fill_ring_elm(&def_elm, UINT32_MAX);
 198         fill_ring_elm(&loc_elm, lc);
 199
 200         while (wrk_cmd != WRK_CMD_RUN) {
 201                 rte_smp_rmb();
 202                 rte_pause();
 203         }
 204
 205         cl = rte_rdtsc_precise();
 206
 207         do {
 208                 /* num in interval [7/8, 11/8] of BULK_NUM */
 209                 num = 7 * BULK_NUM / 8 + rte_rand() % (BULK_NUM / 2);
 210
 211                 /* reset all pointer values */
 212                 memset(obj, 0, sizeof(obj));
 213
 214                 /* dequeue num elems */
 215                 tm0 = (prcs != 0) ? rte_rdtsc_precise() : 0;
 216                 n = _st_ring_dequeue_bulk(la->rng, (void **)obj, num, NULL);
 217                 tm0 = (prcs != 0) ? rte_rdtsc_precise() - tm0 : 0;
 218
 219                 /* check return value and objects */
 220                 rc = check_ring_op(num, n, lc, fname,
 221                         RTE_STR(_st_ring_dequeue_bulk));
 222                 if (rc == 0)
 223                         rc = check_updt_elem(obj, num, &def_elm, &loc_elm);
 224                 if (rc != 0)
 225                         break;
 226
 227                 /* enqueue num elems */
 228                 rte_compiler_barrier();
 229                 rc = check_updt_elem(obj, num, &loc_elm, &def_elm);
 230                 if (rc != 0)
 231                         break;
 232
 233                 tm1 = (prcs != 0) ? rte_rdtsc_precise() : 0;
 234                 n = _st_ring_enqueue_bulk(la->rng, (void **)obj, num, NULL);
 235                 tm1 = (prcs != 0) ? rte_rdtsc_precise() - tm1 : 0;
 236
 237                 /* check return value */
 238                 rc = check_ring_op(num, n, lc, fname,
 239                         RTE_STR(_st_ring_enqueue_bulk));
 240                 if (rc != 0)
 241                         break;
 242
 243                 lcore_stat_update(&la->stats, 1, num, tm0 + tm1, prcs);
 244
 245         } while (wrk_cmd == WRK_CMD_RUN);
 246
 247         cl = rte_rdtsc_precise() - cl;
 248         if (prcs == 0)
 249                 lcore_stat_update(&la->stats, 0, 0, cl, 0);
 250         la->stats.nb_cycle = cl;
 251         return rc;
 252 }
 253 static int
 254 test_worker_prcs(void *arg)
 255 {
 256         return test_worker(arg, __func__, 1);
 257 }
 258
 259 static int
 260 test_worker_avg(void *arg)
 261 {
 262         return test_worker(arg, __func__, 0);
 263 }
 264
 265 static void
 266 mt1_fini(struct rte_ring *rng, void *data)
 267 {
 268         rte_free(rng);
 269         rte_free(data);
 270 }
 271
 272 static int
 273 mt1_init(struct rte_ring **rng, void **data, uint32_t num)
 274 {
 275         int32_t rc;
 276         size_t sz;
 277         uint32_t i, nr;
 278         struct rte_ring *r;
 279         struct ring_elem *elm;
 280         void *p;
 281
 282         *rng = NULL;
 283         *data = NULL;
 284
 285         sz = num * sizeof(*elm);
 286         elm = rte_zmalloc(NULL, sz, __alignof__(*elm));
 287         if (elm == NULL) {
 288                 printf("%s: alloc(%zu) for %u elems data failed",
 289                         __func__, sz, num);
 290                 return -ENOMEM;
 291         }
 292
 293         *data = elm;
 294
 295         /* alloc ring */
 296         nr = 2 * num;
 297         sz = rte_ring_get_memsize(nr);
 298         r = rte_zmalloc(NULL, sz, __alignof__(*r));
 299         if (r == NULL) {
 300                 printf("%s: alloc(%zu) for FIFO with %u elems failed",
 301                         __func__, sz, nr);
 302                 return -ENOMEM;
 303         }
 304
 305         *rng = r;
 306
 307         rc = _st_ring_init(r, RING_NAME, nr);
 308         if (rc != 0) {
 309                 printf("%s: _st_ring_init(%p, %u) failed, error: %d(%s)\n",
 310                         __func__, r, nr, rc, strerror(-rc));
 311                 return rc;
 312         }
 313
 314         for (i = 0; i != num; i++) {
 315                 fill_ring_elm(elm + i, UINT32_MAX);
 316                 p = elm + i;
 317                 if (_st_ring_enqueue_bulk(r, &p, 1, NULL) != 1)
 318                         break;
 319         }
 320
 321         if (i != num) {
 322                 printf("%s: _st_ring_enqueue(%p, %u) returned %u\n",
 323                         __func__, r, num, i);
 324                 return -ENOSPC;
 325         }
 326
 327         return 0;
 328 }
 329
 330 static int
 331 test_mt1(int (*test)(void *))
 332 {
 333         int32_t rc;
 334         uint32_t lc, mc;
 335         struct rte_ring *r;
 336         void *data;
 337         struct lcore_arg arg[RTE_MAX_LCORE];
 338
 339         static const struct lcore_stat init_stat = {
 340                 .op.min_cycle = UINT64_MAX,
 341         };
 342
 343         rc = mt1_init(&r, &data, RING_SIZE);
 344         if (rc != 0) {
 345                 mt1_fini(r, data);
 346                 return rc;
 347         }
 348
 349         memset(arg, 0, sizeof(arg));
 350
 351         /* launch on all slaves */
 352         RTE_LCORE_FOREACH_SLAVE(lc) {
 353                 arg[lc].rng = r;
 354                 arg[lc].stats = init_stat;
 355                 rte_eal_remote_launch(test, &arg[lc], lc);
 356         }
 357
 358         /* signal worker to start test */
 359         wrk_cmd = WRK_CMD_RUN;
 360         rte_smp_wmb();
 361
 362         usleep(run_time * US_PER_S);
 363
 364         /* signal worker to start test */
 365         wrk_cmd = WRK_CMD_STOP;
 366         rte_smp_wmb();
 367
 368         /* wait for slaves and collect stats. */
 369         mc = rte_lcore_id();
 370         arg[mc].stats = init_stat;
 371
 372         rc = 0;
 373         RTE_LCORE_FOREACH_SLAVE(lc) {
 374                 rc |= rte_eal_wait_lcore(lc);
 375                 lcore_stat_aggr(&arg[mc].stats, &arg[lc].stats);
 376                 if (verbose != 0)
 377                         lcore_stat_dump(stdout, lc, &arg[lc].stats);
 378         }
 379
 380         lcore_stat_dump(stdout, UINT32_MAX, &arg[mc].stats);
 381         mt1_fini(r, data);
 382         return rc;
 383 }
 384
 385 static const struct test_case tests[] = {
 386         {
 387                 .name = "MT-WRK_ENQ_DEQ-MST_NONE-PRCS",
 388                 .func = test_mt1,
 389                 .wfunc = test_worker_prcs,
 390         },
 391         {
 392                 .name = "MT-WRK_ENQ_DEQ-MST_NONE-AVG",
 393                 .func = test_mt1,
 394                 .wfunc = test_worker_avg,
 395         },
 396 };