1 /* SPDX-License-Identifier: BSD-3-Clause 2 * Copyright(c) 2010-2017 Intel Corporation 3 */ 4 5 #include "test.h" 6 7 #include <unistd.h> 8 #include <string.h> 9 #include <rte_cycles.h> 10 #include <rte_errno.h> 11 #include <rte_mempool.h> 12 #include <rte_mbuf.h> 13 #include <rte_distributor.h> 14 #include <rte_string_fns.h> 15 16 #define ITER_POWER 20 /* log 2 of how many iterations we do when timing. */ 17 #define BURST 32 18 #define BIG_BATCH 1024 19 20 struct worker_params { 21 char name[64]; 22 struct rte_distributor *dist; 23 }; 24 25 struct worker_params worker_params; 26 27 /* statics - all zero-initialized by default */ 28 static volatile int quit; /**< general quit variable for all threads */ 29 static volatile int zero_quit; /**< var for when we just want thr0 to quit*/ 30 static volatile int zero_sleep; /**< thr0 has quit basic loop and is sleeping*/ 31 static volatile unsigned worker_idx; 32 static volatile unsigned zero_idx; 33 34 struct worker_stats { 35 volatile unsigned handled_packets; 36 } __rte_cache_aligned; 37 struct worker_stats worker_stats[RTE_MAX_LCORE]; 38 39 /* returns the total count of the number of packets handled by the worker 40 * functions given below. 41 */ 42 static inline unsigned 43 total_packet_count(void) 44 { 45 unsigned i, count = 0; 46 for (i = 0; i < worker_idx; i++) 47 count += __atomic_load_n(&worker_stats[i].handled_packets, 48 __ATOMIC_RELAXED); 49 return count; 50 } 51 52 /* resets the packet counts for a new test */ 53 static inline void 54 clear_packet_count(void) 55 { 56 unsigned int i; 57 for (i = 0; i < RTE_MAX_LCORE; i++) 58 __atomic_store_n(&worker_stats[i].handled_packets, 0, 59 __ATOMIC_RELAXED); 60 } 61 62 /* this is the basic worker function for sanity test 63 * it does nothing but return packets and count them. 64 */ 65 static int 66 handle_work(void *arg) 67 { 68 struct rte_mbuf *buf[8] __rte_cache_aligned; 69 struct worker_params *wp = arg; 70 struct rte_distributor *db = wp->dist; 71 unsigned int num; 72 unsigned int id = __atomic_fetch_add(&worker_idx, 1, __ATOMIC_RELAXED); 73 74 num = rte_distributor_get_pkt(db, id, buf, NULL, 0); 75 while (!quit) { 76 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 77 __ATOMIC_RELAXED); 78 num = rte_distributor_get_pkt(db, id, 79 buf, buf, num); 80 } 81 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 82 __ATOMIC_RELAXED); 83 rte_distributor_return_pkt(db, id, buf, num); 84 return 0; 85 } 86 87 /* do basic sanity testing of the distributor. This test tests the following: 88 * - send 32 packets through distributor with the same tag and ensure they 89 * all go to the one worker 90 * - send 32 packets through the distributor with two different tags and 91 * verify that they go equally to two different workers. 92 * - send 32 packets with different tags through the distributors and 93 * just verify we get all packets back. 94 * - send 1024 packets through the distributor, gathering the returned packets 95 * as we go. Then verify that we correctly got all 1024 pointers back again, 96 * not necessarily in the same order (as different flows). 97 */ 98 static int 99 sanity_test(struct worker_params *wp, struct rte_mempool *p) 100 { 101 struct rte_distributor *db = wp->dist; 102 struct rte_mbuf *bufs[BURST]; 103 struct rte_mbuf *returns[BURST*2]; 104 unsigned int i, count; 105 unsigned int retries; 106 unsigned int processed; 107 108 printf("=== Basic distributor sanity tests ===\n"); 109 clear_packet_count(); 110 if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) { 111 printf("line %d: Error getting mbufs from pool\n", __LINE__); 112 return -1; 113 } 114 115 /* now set all hash values in all buffers to zero, so all pkts go to the 116 * one worker thread */ 117 for (i = 0; i < BURST; i++) 118 bufs[i]->hash.usr = 0; 119 120 processed = 0; 121 while (processed < BURST) 122 processed += rte_distributor_process(db, &bufs[processed], 123 BURST - processed); 124 125 count = 0; 126 do { 127 128 rte_distributor_flush(db); 129 count += rte_distributor_returned_pkts(db, 130 returns, BURST*2); 131 } while (count < BURST); 132 133 if (total_packet_count() != BURST) { 134 printf("Line %d: Error, not all packets flushed. " 135 "Expected %u, got %u\n", 136 __LINE__, BURST, total_packet_count()); 137 rte_mempool_put_bulk(p, (void *)bufs, BURST); 138 return -1; 139 } 140 141 for (i = 0; i < rte_lcore_count() - 1; i++) 142 printf("Worker %u handled %u packets\n", i, 143 __atomic_load_n(&worker_stats[i].handled_packets, 144 __ATOMIC_RELAXED)); 145 printf("Sanity test with all zero hashes done.\n"); 146 147 /* pick two flows and check they go correctly */ 148 if (rte_lcore_count() >= 3) { 149 clear_packet_count(); 150 for (i = 0; i < BURST; i++) 151 bufs[i]->hash.usr = (i & 1) << 8; 152 153 rte_distributor_process(db, bufs, BURST); 154 count = 0; 155 do { 156 rte_distributor_flush(db); 157 count += rte_distributor_returned_pkts(db, 158 returns, BURST*2); 159 } while (count < BURST); 160 if (total_packet_count() != BURST) { 161 printf("Line %d: Error, not all packets flushed. " 162 "Expected %u, got %u\n", 163 __LINE__, BURST, total_packet_count()); 164 rte_mempool_put_bulk(p, (void *)bufs, BURST); 165 return -1; 166 } 167 168 for (i = 0; i < rte_lcore_count() - 1; i++) 169 printf("Worker %u handled %u packets\n", i, 170 __atomic_load_n( 171 &worker_stats[i].handled_packets, 172 __ATOMIC_RELAXED)); 173 printf("Sanity test with two hash values done\n"); 174 } 175 176 /* give a different hash value to each packet, 177 * so load gets distributed */ 178 clear_packet_count(); 179 for (i = 0; i < BURST; i++) 180 bufs[i]->hash.usr = i+1; 181 182 rte_distributor_process(db, bufs, BURST); 183 count = 0; 184 do { 185 rte_distributor_flush(db); 186 count += rte_distributor_returned_pkts(db, 187 returns, BURST*2); 188 } while (count < BURST); 189 if (total_packet_count() != BURST) { 190 printf("Line %d: Error, not all packets flushed. " 191 "Expected %u, got %u\n", 192 __LINE__, BURST, total_packet_count()); 193 rte_mempool_put_bulk(p, (void *)bufs, BURST); 194 return -1; 195 } 196 197 for (i = 0; i < rte_lcore_count() - 1; i++) 198 printf("Worker %u handled %u packets\n", i, 199 __atomic_load_n(&worker_stats[i].handled_packets, 200 __ATOMIC_RELAXED)); 201 printf("Sanity test with non-zero hashes done\n"); 202 203 rte_mempool_put_bulk(p, (void *)bufs, BURST); 204 205 /* sanity test with BIG_BATCH packets to ensure they all arrived back 206 * from the returned packets function */ 207 clear_packet_count(); 208 struct rte_mbuf *many_bufs[BIG_BATCH], *return_bufs[BIG_BATCH]; 209 unsigned num_returned = 0; 210 211 /* flush out any remaining packets */ 212 rte_distributor_flush(db); 213 rte_distributor_clear_returns(db); 214 215 if (rte_mempool_get_bulk(p, (void *)many_bufs, BIG_BATCH) != 0) { 216 printf("line %d: Error getting mbufs from pool\n", __LINE__); 217 return -1; 218 } 219 for (i = 0; i < BIG_BATCH; i++) 220 many_bufs[i]->hash.usr = i << 2; 221 222 printf("=== testing big burst (%s) ===\n", wp->name); 223 for (i = 0; i < BIG_BATCH/BURST; i++) { 224 rte_distributor_process(db, 225 &many_bufs[i*BURST], BURST); 226 count = rte_distributor_returned_pkts(db, 227 &return_bufs[num_returned], 228 BIG_BATCH - num_returned); 229 num_returned += count; 230 } 231 rte_distributor_flush(db); 232 count = rte_distributor_returned_pkts(db, 233 &return_bufs[num_returned], 234 BIG_BATCH - num_returned); 235 num_returned += count; 236 retries = 0; 237 do { 238 rte_distributor_flush(db); 239 count = rte_distributor_returned_pkts(db, 240 &return_bufs[num_returned], 241 BIG_BATCH - num_returned); 242 num_returned += count; 243 retries++; 244 } while ((num_returned < BIG_BATCH) && (retries < 100)); 245 246 if (num_returned != BIG_BATCH) { 247 printf("line %d: Missing packets, expected %d\n", 248 __LINE__, num_returned); 249 rte_mempool_put_bulk(p, (void *)many_bufs, BIG_BATCH); 250 return -1; 251 } 252 253 /* big check - make sure all packets made it back!! */ 254 for (i = 0; i < BIG_BATCH; i++) { 255 unsigned j; 256 struct rte_mbuf *src = many_bufs[i]; 257 for (j = 0; j < BIG_BATCH; j++) { 258 if (return_bufs[j] == src) 259 break; 260 } 261 262 if (j == BIG_BATCH) { 263 printf("Error: could not find source packet #%u\n", i); 264 rte_mempool_put_bulk(p, (void *)many_bufs, BIG_BATCH); 265 return -1; 266 } 267 } 268 printf("Sanity test of returned packets done\n"); 269 270 rte_mempool_put_bulk(p, (void *)many_bufs, BIG_BATCH); 271 272 printf("\n"); 273 return 0; 274 } 275 276 277 /* to test that the distributor does not lose packets, we use this worker 278 * function which frees mbufs when it gets them. The distributor thread does 279 * the mbuf allocation. If distributor drops packets we'll eventually run out 280 * of mbufs. 281 */ 282 static int 283 handle_work_with_free_mbufs(void *arg) 284 { 285 struct rte_mbuf *buf[8] __rte_cache_aligned; 286 struct worker_params *wp = arg; 287 struct rte_distributor *d = wp->dist; 288 unsigned int i; 289 unsigned int num; 290 unsigned int id = __atomic_fetch_add(&worker_idx, 1, __ATOMIC_RELAXED); 291 292 num = rte_distributor_get_pkt(d, id, buf, NULL, 0); 293 while (!quit) { 294 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 295 __ATOMIC_RELAXED); 296 for (i = 0; i < num; i++) 297 rte_pktmbuf_free(buf[i]); 298 num = rte_distributor_get_pkt(d, id, buf, NULL, 0); 299 } 300 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 301 __ATOMIC_RELAXED); 302 rte_distributor_return_pkt(d, id, buf, num); 303 return 0; 304 } 305 306 /* Perform a sanity test of the distributor with a large number of packets, 307 * where we allocate a new set of mbufs for each burst. The workers then 308 * free the mbufs. This ensures that we don't have any packet leaks in the 309 * library. 310 */ 311 static int 312 sanity_test_with_mbuf_alloc(struct worker_params *wp, struct rte_mempool *p) 313 { 314 struct rte_distributor *d = wp->dist; 315 unsigned i; 316 struct rte_mbuf *bufs[BURST]; 317 unsigned int processed; 318 319 printf("=== Sanity test with mbuf alloc/free (%s) ===\n", wp->name); 320 321 clear_packet_count(); 322 for (i = 0; i < ((1<<ITER_POWER)); i += BURST) { 323 unsigned j; 324 while (rte_mempool_get_bulk(p, (void *)bufs, BURST) < 0) 325 rte_distributor_process(d, NULL, 0); 326 for (j = 0; j < BURST; j++) { 327 bufs[j]->hash.usr = (i+j) << 1; 328 } 329 330 processed = 0; 331 while (processed < BURST) 332 processed += rte_distributor_process(d, 333 &bufs[processed], BURST - processed); 334 } 335 336 rte_distributor_flush(d); 337 338 rte_delay_us(10000); 339 340 if (total_packet_count() < (1<<ITER_POWER)) { 341 printf("Line %u: Packet count is incorrect, %u, expected %u\n", 342 __LINE__, total_packet_count(), 343 (1<<ITER_POWER)); 344 return -1; 345 } 346 347 printf("Sanity test with mbuf alloc/free passed\n\n"); 348 return 0; 349 } 350 351 static int 352 handle_work_for_shutdown_test(void *arg) 353 { 354 struct rte_mbuf *buf[8] __rte_cache_aligned; 355 struct worker_params *wp = arg; 356 struct rte_distributor *d = wp->dist; 357 unsigned int num; 358 unsigned int zero_id = 0; 359 unsigned int zero_unset; 360 const unsigned int id = __atomic_fetch_add(&worker_idx, 1, 361 __ATOMIC_RELAXED); 362 363 num = rte_distributor_get_pkt(d, id, buf, NULL, 0); 364 365 if (num > 0) { 366 zero_unset = RTE_MAX_LCORE; 367 __atomic_compare_exchange_n(&zero_idx, &zero_unset, id, 368 false, __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE); 369 } 370 zero_id = __atomic_load_n(&zero_idx, __ATOMIC_ACQUIRE); 371 372 /* wait for quit single globally, or for worker zero, wait 373 * for zero_quit */ 374 while (!quit && !(id == zero_id && zero_quit)) { 375 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 376 __ATOMIC_RELAXED); 377 num = rte_distributor_get_pkt(d, id, buf, NULL, 0); 378 379 if (num > 0) { 380 zero_unset = RTE_MAX_LCORE; 381 __atomic_compare_exchange_n(&zero_idx, &zero_unset, id, 382 false, __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE); 383 } 384 zero_id = __atomic_load_n(&zero_idx, __ATOMIC_ACQUIRE); 385 } 386 387 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 388 __ATOMIC_RELAXED); 389 if (id == zero_id) { 390 rte_distributor_return_pkt(d, id, NULL, 0); 391 392 /* for worker zero, allow it to restart to pick up last packet 393 * when all workers are shutting down. 394 */ 395 __atomic_store_n(&zero_sleep, 1, __ATOMIC_RELEASE); 396 while (zero_quit) 397 usleep(100); 398 __atomic_store_n(&zero_sleep, 0, __ATOMIC_RELEASE); 399 400 num = rte_distributor_get_pkt(d, id, buf, NULL, 0); 401 402 while (!quit) { 403 __atomic_fetch_add(&worker_stats[id].handled_packets, 404 num, __ATOMIC_RELAXED); 405 num = rte_distributor_get_pkt(d, id, buf, NULL, 0); 406 } 407 } 408 rte_distributor_return_pkt(d, id, buf, num); 409 return 0; 410 } 411 412 413 /* Perform a sanity test of the distributor with a large number of packets, 414 * where we allocate a new set of mbufs for each burst. The workers then 415 * free the mbufs. This ensures that we don't have any packet leaks in the 416 * library. 417 */ 418 static int 419 sanity_test_with_worker_shutdown(struct worker_params *wp, 420 struct rte_mempool *p) 421 { 422 struct rte_distributor *d = wp->dist; 423 struct rte_mbuf *bufs[BURST]; 424 struct rte_mbuf *bufs2[BURST]; 425 unsigned int i; 426 unsigned int failed = 0; 427 unsigned int processed = 0; 428 429 printf("=== Sanity test of worker shutdown ===\n"); 430 431 clear_packet_count(); 432 433 if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) { 434 printf("line %d: Error getting mbufs from pool\n", __LINE__); 435 return -1; 436 } 437 438 /* 439 * Now set all hash values in all buffers to same value so all 440 * pkts go to the one worker thread 441 */ 442 for (i = 0; i < BURST; i++) 443 bufs[i]->hash.usr = 1; 444 445 processed = 0; 446 while (processed < BURST) 447 processed += rte_distributor_process(d, &bufs[processed], 448 BURST - processed); 449 rte_distributor_flush(d); 450 451 /* at this point, we will have processed some packets and have a full 452 * backlog for the other ones at worker 0. 453 */ 454 455 /* get more buffers to queue up, again setting them to the same flow */ 456 if (rte_mempool_get_bulk(p, (void *)bufs2, BURST) != 0) { 457 printf("line %d: Error getting mbufs from pool\n", __LINE__); 458 rte_mempool_put_bulk(p, (void *)bufs, BURST); 459 return -1; 460 } 461 for (i = 0; i < BURST; i++) 462 bufs2[i]->hash.usr = 1; 463 464 /* get worker zero to quit */ 465 zero_quit = 1; 466 rte_distributor_process(d, bufs2, BURST); 467 468 /* flush the distributor */ 469 rte_distributor_flush(d); 470 while (!__atomic_load_n(&zero_sleep, __ATOMIC_ACQUIRE)) 471 rte_distributor_flush(d); 472 473 zero_quit = 0; 474 while (__atomic_load_n(&zero_sleep, __ATOMIC_ACQUIRE)) 475 rte_delay_us(100); 476 477 for (i = 0; i < rte_lcore_count() - 1; i++) 478 printf("Worker %u handled %u packets\n", i, 479 __atomic_load_n(&worker_stats[i].handled_packets, 480 __ATOMIC_RELAXED)); 481 482 if (total_packet_count() != BURST * 2) { 483 printf("Line %d: Error, not all packets flushed. " 484 "Expected %u, got %u\n", 485 __LINE__, BURST * 2, total_packet_count()); 486 failed = 1; 487 } 488 489 rte_mempool_put_bulk(p, (void *)bufs, BURST); 490 rte_mempool_put_bulk(p, (void *)bufs2, BURST); 491 492 if (failed) 493 return -1; 494 495 printf("Sanity test with worker shutdown passed\n\n"); 496 return 0; 497 } 498 499 /* Test that the flush function is able to move packets between workers when 500 * one worker shuts down.. 501 */ 502 static int 503 test_flush_with_worker_shutdown(struct worker_params *wp, 504 struct rte_mempool *p) 505 { 506 struct rte_distributor *d = wp->dist; 507 struct rte_mbuf *bufs[BURST]; 508 unsigned int i; 509 unsigned int failed = 0; 510 unsigned int processed; 511 512 printf("=== Test flush fn with worker shutdown (%s) ===\n", wp->name); 513 514 clear_packet_count(); 515 if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) { 516 printf("line %d: Error getting mbufs from pool\n", __LINE__); 517 return -1; 518 } 519 520 /* now set all hash values in all buffers to zero, so all pkts go to the 521 * one worker thread */ 522 for (i = 0; i < BURST; i++) 523 bufs[i]->hash.usr = 0; 524 525 processed = 0; 526 while (processed < BURST) 527 processed += rte_distributor_process(d, &bufs[processed], 528 BURST - processed); 529 /* at this point, we will have processed some packets and have a full 530 * backlog for the other ones at worker 0. 531 */ 532 533 /* get worker zero to quit */ 534 zero_quit = 1; 535 536 /* flush the distributor */ 537 rte_distributor_flush(d); 538 539 while (!__atomic_load_n(&zero_sleep, __ATOMIC_ACQUIRE)) 540 rte_distributor_flush(d); 541 542 zero_quit = 0; 543 544 while (__atomic_load_n(&zero_sleep, __ATOMIC_ACQUIRE)) 545 rte_delay_us(100); 546 547 for (i = 0; i < rte_lcore_count() - 1; i++) 548 printf("Worker %u handled %u packets\n", i, 549 __atomic_load_n(&worker_stats[i].handled_packets, 550 __ATOMIC_RELAXED)); 551 552 if (total_packet_count() != BURST) { 553 printf("Line %d: Error, not all packets flushed. " 554 "Expected %u, got %u\n", 555 __LINE__, BURST, total_packet_count()); 556 failed = 1; 557 } 558 559 rte_mempool_put_bulk(p, (void *)bufs, BURST); 560 561 if (failed) 562 return -1; 563 564 printf("Flush test with worker shutdown passed\n\n"); 565 return 0; 566 } 567 568 static int 569 handle_and_mark_work(void *arg) 570 { 571 struct rte_mbuf *buf[8] __rte_cache_aligned; 572 struct worker_params *wp = arg; 573 struct rte_distributor *db = wp->dist; 574 unsigned int num, i; 575 unsigned int id = __atomic_fetch_add(&worker_idx, 1, __ATOMIC_RELAXED); 576 num = rte_distributor_get_pkt(db, id, buf, NULL, 0); 577 while (!quit) { 578 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 579 __ATOMIC_RELAXED); 580 for (i = 0; i < num; i++) 581 buf[i]->udata64 += id + 1; 582 num = rte_distributor_get_pkt(db, id, 583 buf, buf, num); 584 } 585 __atomic_fetch_add(&worker_stats[id].handled_packets, num, 586 __ATOMIC_RELAXED); 587 rte_distributor_return_pkt(db, id, buf, num); 588 return 0; 589 } 590 591 /* sanity_mark_test sends packets to workers which mark them. 592 * Every packet has also encoded sequence number. 593 * The returned packets are sorted and verified if they were handled 594 * by proper workers. 595 */ 596 static int 597 sanity_mark_test(struct worker_params *wp, struct rte_mempool *p) 598 { 599 const unsigned int buf_count = 24; 600 const unsigned int burst = 8; 601 const unsigned int shift = 12; 602 const unsigned int seq_shift = 10; 603 604 struct rte_distributor *db = wp->dist; 605 struct rte_mbuf *bufs[buf_count]; 606 struct rte_mbuf *returns[buf_count]; 607 unsigned int i, count, id; 608 unsigned int sorted[buf_count], seq; 609 unsigned int failed = 0; 610 unsigned int processed; 611 612 printf("=== Marked packets test ===\n"); 613 clear_packet_count(); 614 if (rte_mempool_get_bulk(p, (void *)bufs, buf_count) != 0) { 615 printf("line %d: Error getting mbufs from pool\n", __LINE__); 616 return -1; 617 } 618 619 /* bufs' hashes will be like these below, but shifted left. 620 * The shifting is for avoiding collisions with backlogs 621 * and in-flight tags left by previous tests. 622 * [1, 1, 1, 1, 1, 1, 1, 1 623 * 1, 1, 1, 1, 2, 2, 2, 2 624 * 2, 2, 2, 2, 1, 1, 1, 1] 625 */ 626 for (i = 0; i < burst; i++) { 627 bufs[0 * burst + i]->hash.usr = 1 << shift; 628 bufs[1 * burst + i]->hash.usr = ((i < burst / 2) ? 1 : 2) 629 << shift; 630 bufs[2 * burst + i]->hash.usr = ((i < burst / 2) ? 2 : 1) 631 << shift; 632 } 633 /* Assign a sequence number to each packet. The sequence is shifted, 634 * so that lower bits of the udate64 will hold mark from worker. 635 */ 636 for (i = 0; i < buf_count; i++) 637 bufs[i]->udata64 = i << seq_shift; 638 639 count = 0; 640 for (i = 0; i < buf_count/burst; i++) { 641 processed = 0; 642 while (processed < burst) 643 processed += rte_distributor_process(db, 644 &bufs[i * burst + processed], 645 burst - processed); 646 count += rte_distributor_returned_pkts(db, &returns[count], 647 buf_count - count); 648 } 649 650 do { 651 rte_distributor_flush(db); 652 count += rte_distributor_returned_pkts(db, &returns[count], 653 buf_count - count); 654 } while (count < buf_count); 655 656 for (i = 0; i < rte_lcore_count() - 1; i++) 657 printf("Worker %u handled %u packets\n", i, 658 __atomic_load_n(&worker_stats[i].handled_packets, 659 __ATOMIC_RELAXED)); 660 661 /* Sort returned packets by sent order (sequence numbers). */ 662 for (i = 0; i < buf_count; i++) { 663 seq = returns[i]->udata64 >> seq_shift; 664 id = returns[i]->udata64 - (seq << seq_shift); 665 sorted[seq] = id; 666 } 667 668 /* Verify that packets [0-11] and [20-23] were processed 669 * by the same worker 670 */ 671 for (i = 1; i < 12; i++) { 672 if (sorted[i] != sorted[0]) { 673 printf("Packet number %u processed by worker %u," 674 " but should be processes by worker %u\n", 675 i, sorted[i], sorted[0]); 676 failed = 1; 677 } 678 } 679 for (i = 20; i < 24; i++) { 680 if (sorted[i] != sorted[0]) { 681 printf("Packet number %u processed by worker %u," 682 " but should be processes by worker %u\n", 683 i, sorted[i], sorted[0]); 684 failed = 1; 685 } 686 } 687 /* And verify that packets [12-19] were processed 688 * by the another worker 689 */ 690 for (i = 13; i < 20; i++) { 691 if (sorted[i] != sorted[12]) { 692 printf("Packet number %u processed by worker %u," 693 " but should be processes by worker %u\n", 694 i, sorted[i], sorted[12]); 695 failed = 1; 696 } 697 } 698 699 rte_mempool_put_bulk(p, (void *)bufs, buf_count); 700 701 if (failed) 702 return -1; 703 704 printf("Marked packets test passed\n"); 705 return 0; 706 } 707 708 static 709 int test_error_distributor_create_name(void) 710 { 711 struct rte_distributor *d = NULL; 712 struct rte_distributor *db = NULL; 713 char *name = NULL; 714 715 d = rte_distributor_create(name, rte_socket_id(), 716 rte_lcore_count() - 1, 717 RTE_DIST_ALG_SINGLE); 718 if (d != NULL || rte_errno != EINVAL) { 719 printf("ERROR: No error on create() with NULL name param\n"); 720 return -1; 721 } 722 723 db = rte_distributor_create(name, rte_socket_id(), 724 rte_lcore_count() - 1, 725 RTE_DIST_ALG_BURST); 726 if (db != NULL || rte_errno != EINVAL) { 727 printf("ERROR: No error on create() with NULL param\n"); 728 return -1; 729 } 730 731 return 0; 732 } 733 734 735 static 736 int test_error_distributor_create_numworkers(void) 737 { 738 struct rte_distributor *ds = NULL; 739 struct rte_distributor *db = NULL; 740 741 ds = rte_distributor_create("test_numworkers", rte_socket_id(), 742 RTE_MAX_LCORE + 10, 743 RTE_DIST_ALG_SINGLE); 744 if (ds != NULL || rte_errno != EINVAL) { 745 printf("ERROR: No error on create() with num_workers > MAX\n"); 746 return -1; 747 } 748 749 db = rte_distributor_create("test_numworkers", rte_socket_id(), 750 RTE_MAX_LCORE + 10, 751 RTE_DIST_ALG_BURST); 752 if (db != NULL || rte_errno != EINVAL) { 753 printf("ERROR: No error on create() num_workers > MAX\n"); 754 return -1; 755 } 756 757 return 0; 758 } 759 760 761 /* Useful function which ensures that all worker functions terminate */ 762 static void 763 quit_workers(struct worker_params *wp, struct rte_mempool *p) 764 { 765 struct rte_distributor *d = wp->dist; 766 const unsigned num_workers = rte_lcore_count() - 1; 767 unsigned i; 768 struct rte_mbuf *bufs[RTE_MAX_LCORE]; 769 struct rte_mbuf *returns[RTE_MAX_LCORE]; 770 if (rte_mempool_get_bulk(p, (void *)bufs, num_workers) != 0) { 771 printf("line %d: Error getting mbufs from pool\n", __LINE__); 772 return; 773 } 774 775 zero_quit = 0; 776 quit = 1; 777 for (i = 0; i < num_workers; i++) { 778 bufs[i]->hash.usr = i << 1; 779 rte_distributor_process(d, &bufs[i], 1); 780 } 781 782 rte_distributor_process(d, NULL, 0); 783 rte_distributor_flush(d); 784 rte_eal_mp_wait_lcore(); 785 786 while (rte_distributor_returned_pkts(d, returns, RTE_MAX_LCORE)) 787 ; 788 789 rte_distributor_clear_returns(d); 790 rte_mempool_put_bulk(p, (void *)bufs, num_workers); 791 792 quit = 0; 793 worker_idx = 0; 794 zero_idx = RTE_MAX_LCORE; 795 zero_quit = 0; 796 zero_sleep = 0; 797 } 798 799 static int 800 test_distributor(void) 801 { 802 static struct rte_distributor *ds; 803 static struct rte_distributor *db; 804 static struct rte_distributor *dist[2]; 805 static struct rte_mempool *p; 806 int i; 807 808 if (rte_lcore_count() < 2) { 809 printf("Not enough cores for distributor_autotest, expecting at least 2\n"); 810 return TEST_SKIPPED; 811 } 812 813 if (db == NULL) { 814 db = rte_distributor_create("Test_dist_burst", rte_socket_id(), 815 rte_lcore_count() - 1, 816 RTE_DIST_ALG_BURST); 817 if (db == NULL) { 818 printf("Error creating burst distributor\n"); 819 return -1; 820 } 821 } else { 822 rte_distributor_flush(db); 823 rte_distributor_clear_returns(db); 824 } 825 826 if (ds == NULL) { 827 ds = rte_distributor_create("Test_dist_single", 828 rte_socket_id(), 829 rte_lcore_count() - 1, 830 RTE_DIST_ALG_SINGLE); 831 if (ds == NULL) { 832 printf("Error creating single distributor\n"); 833 return -1; 834 } 835 } else { 836 rte_distributor_flush(ds); 837 rte_distributor_clear_returns(ds); 838 } 839 840 const unsigned nb_bufs = (511 * rte_lcore_count()) < BIG_BATCH ? 841 (BIG_BATCH * 2) - 1 : (511 * rte_lcore_count()); 842 if (p == NULL) { 843 p = rte_pktmbuf_pool_create("DT_MBUF_POOL", nb_bufs, BURST, 844 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id()); 845 if (p == NULL) { 846 printf("Error creating mempool\n"); 847 return -1; 848 } 849 } 850 851 dist[0] = ds; 852 dist[1] = db; 853 854 for (i = 0; i < 2; i++) { 855 856 worker_params.dist = dist[i]; 857 if (i) 858 strlcpy(worker_params.name, "burst", 859 sizeof(worker_params.name)); 860 else 861 strlcpy(worker_params.name, "single", 862 sizeof(worker_params.name)); 863 864 rte_eal_mp_remote_launch(handle_work, 865 &worker_params, SKIP_MASTER); 866 if (sanity_test(&worker_params, p) < 0) 867 goto err; 868 quit_workers(&worker_params, p); 869 870 rte_eal_mp_remote_launch(handle_work_with_free_mbufs, 871 &worker_params, SKIP_MASTER); 872 if (sanity_test_with_mbuf_alloc(&worker_params, p) < 0) 873 goto err; 874 quit_workers(&worker_params, p); 875 876 if (rte_lcore_count() > 2) { 877 rte_eal_mp_remote_launch(handle_work_for_shutdown_test, 878 &worker_params, 879 SKIP_MASTER); 880 if (sanity_test_with_worker_shutdown(&worker_params, 881 p) < 0) 882 goto err; 883 quit_workers(&worker_params, p); 884 885 rte_eal_mp_remote_launch(handle_work_for_shutdown_test, 886 &worker_params, 887 SKIP_MASTER); 888 if (test_flush_with_worker_shutdown(&worker_params, 889 p) < 0) 890 goto err; 891 quit_workers(&worker_params, p); 892 893 rte_eal_mp_remote_launch(handle_and_mark_work, 894 &worker_params, SKIP_MASTER); 895 if (sanity_mark_test(&worker_params, p) < 0) 896 goto err; 897 quit_workers(&worker_params, p); 898 899 } else { 900 printf("Too few cores to run worker shutdown test\n"); 901 } 902 903 } 904 905 if (test_error_distributor_create_numworkers() == -1 || 906 test_error_distributor_create_name() == -1) { 907 printf("rte_distributor_create parameter check tests failed"); 908 return -1; 909 } 910 911 return 0; 912 913 err: 914 quit_workers(&worker_params, p); 915 return -1; 916 } 917 918 REGISTER_TEST_COMMAND(distributor_autotest, test_distributor); 919