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 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 workers */ 352 RTE_LCORE_FOREACH_WORKER(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 workers and collect stats. */ 369 mc = rte_lcore_id(); 370 arg[mc].stats = init_stat; 371 372 rc = 0; 373 RTE_LCORE_FOREACH_WORKER(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 }; 397