app/test/test_distributor.c

   1 /*-
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
   5  *   All rights reserved.
   6  *
   7  *   Redistribution and use in source and binary forms, with or without
   8  *   modification, are permitted provided that the following conditions
   9  *   are met:
  10  *
  11  *     * Redistributions of source code must retain the above copyright
  12  *       notice, this list of conditions and the following disclaimer.
  13  *     * Redistributions in binary form must reproduce the above copyright
  14  *       notice, this list of conditions and the following disclaimer in
  15  *       the documentation and/or other materials provided with the
  16  *       distribution.
  17  *     * Neither the name of Intel Corporation nor the names of its
  18  *       contributors may be used to endorse or promote products derived
  19  *       from this software without specific prior written permission.
  20  *
  21  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  24  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  25  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  26  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  27  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  28  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  29  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  30  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  31  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  32  */
  33
  34 #include "test.h"
  35
  36 #include <unistd.h>
  37 #include <string.h>
  38 #include <rte_cycles.h>
  39 #include <rte_errno.h>
  40 #include <rte_distributor.h>
  41
  42 #define ITER_POWER 20 /* log 2 of how many iterations we do when timing. */
  43 #define BURST 32
  44 #define BIG_BATCH 1024
  45
  46 /* statics - all zero-initialized by default */
  47 static volatile int quit;      /**< general quit variable for all threads */
  48 static volatile int zero_quit; /**< var for when we just want thr0 to quit*/
  49 static volatile unsigned worker_idx;
  50
  51 struct worker_stats {
  52         volatile unsigned handled_packets;
  53 } __rte_cache_aligned;
  54 struct worker_stats worker_stats[RTE_MAX_LCORE];
  55
  56 /* returns the total count of the number of packets handled by the worker
  57  * functions given below.
  58  */
  59 static inline unsigned
  60 total_packet_count(void)
  61 {
  62         unsigned i, count = 0;
  63         for (i = 0; i < worker_idx; i++)
  64                 count += worker_stats[i].handled_packets;
  65         return count;
  66 }
  67
  68 /* resets the packet counts for a new test */
  69 static inline void
  70 clear_packet_count(void)
  71 {
  72         memset(&worker_stats, 0, sizeof(worker_stats));
  73 }
  74
  75 /* this is the basic worker function for sanity test
  76  * it does nothing but return packets and count them.
  77  */
  78 static int
  79 handle_work(void *arg)
  80 {
  81         struct rte_mbuf *pkt = NULL;
  82         struct rte_distributor *d = arg;
  83         unsigned count = 0;
  84         unsigned id = __sync_fetch_and_add(&worker_idx, 1);
  85
  86         pkt = rte_distributor_get_pkt(d, id, NULL);
  87         while (!quit) {
  88                 worker_stats[id].handled_packets++, count++;
  89                 pkt = rte_distributor_get_pkt(d, id, pkt);
  90         }
  91         worker_stats[id].handled_packets++, count++;
  92         rte_distributor_return_pkt(d, id, pkt);
  93         return 0;
  94 }
  95
  96 /* do basic sanity testing of the distributor. This test tests the following:
  97  * - send 32 packets through distributor with the same tag and ensure they
  98  *   all go to the one worker
  99  * - send 32 packets throught the distributor with two different tags and
 100  *   verify that they go equally to two different workers.
 101  * - send 32 packets with different tags through the distributors and
 102  *   just verify we get all packets back.
 103  * - send 1024 packets through the distributor, gathering the returned packets
 104  *   as we go. Then verify that we correctly got all 1024 pointers back again,
 105  *   not necessarily in the same order (as different flows).
 106  */
 107 static int
 108 sanity_test(struct rte_distributor *d, struct rte_mempool *p)
 109 {
 110         struct rte_mbuf *bufs[BURST];
 111         unsigned i;
 112
 113         printf("=== Basic distributor sanity tests ===\n");
 114         clear_packet_count();
 115         if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) {
 116                 printf("line %d: Error getting mbufs from pool\n", __LINE__);
 117                 return -1;
 118         }
 119
 120         /* now set all hash values in all buffers to zero, so all pkts go to the
 121          * one worker thread */
 122         for (i = 0; i < BURST; i++)
 123                 bufs[i]->hash.usr = 0;
 124
 125         rte_distributor_process(d, bufs, BURST);
 126         rte_distributor_flush(d);
 127         if (total_packet_count() != BURST) {
 128                 printf("Line %d: Error, not all packets flushed. "
 129                                 "Expected %u, got %u\n",
 130                                 __LINE__, BURST, total_packet_count());
 131                 return -1;
 132         }
 133
 134         for (i = 0; i < rte_lcore_count() - 1; i++)
 135                 printf("Worker %u handled %u packets\n", i,
 136                                 worker_stats[i].handled_packets);
 137         printf("Sanity test with all zero hashes done.\n");
 138         if (worker_stats[0].handled_packets != BURST)
 139                 return -1;
 140
 141         /* pick two flows and check they go correctly */
 142         if (rte_lcore_count() >= 3) {
 143                 clear_packet_count();
 144                 for (i = 0; i < BURST; i++)
 145                         bufs[i]->hash.usr = (i & 1) << 8;
 146
 147                 rte_distributor_process(d, bufs, BURST);
 148                 rte_distributor_flush(d);
 149                 if (total_packet_count() != BURST) {
 150                         printf("Line %d: Error, not all packets flushed. "
 151                                         "Expected %u, got %u\n",
 152                                         __LINE__, BURST, total_packet_count());
 153                         return -1;
 154                 }
 155
 156                 for (i = 0; i < rte_lcore_count() - 1; i++)
 157                         printf("Worker %u handled %u packets\n", i,
 158                                         worker_stats[i].handled_packets);
 159                 printf("Sanity test with two hash values done\n");
 160
 161                 if (worker_stats[0].handled_packets != 16 ||
 162                                 worker_stats[1].handled_packets != 16)
 163                         return -1;
 164         }
 165
 166         /* give a different hash value to each packet,
 167          * so load gets distributed */
 168         clear_packet_count();
 169         for (i = 0; i < BURST; i++)
 170                 bufs[i]->hash.usr = i;
 171
 172         rte_distributor_process(d, bufs, BURST);
 173         rte_distributor_flush(d);
 174         if (total_packet_count() != BURST) {
 175                 printf("Line %d: Error, not all packets flushed. "
 176                                 "Expected %u, got %u\n",
 177                                 __LINE__, BURST, total_packet_count());
 178                 return -1;
 179         }
 180
 181         for (i = 0; i < rte_lcore_count() - 1; i++)
 182                 printf("Worker %u handled %u packets\n", i,
 183                                 worker_stats[i].handled_packets);
 184         printf("Sanity test with non-zero hashes done\n");
 185
 186         rte_mempool_put_bulk(p, (void *)bufs, BURST);
 187
 188         /* sanity test with BIG_BATCH packets to ensure they all arrived back
 189          * from the returned packets function */
 190         clear_packet_count();
 191         struct rte_mbuf *many_bufs[BIG_BATCH], *return_bufs[BIG_BATCH];
 192         unsigned num_returned = 0;
 193
 194         /* flush out any remaining packets */
 195         rte_distributor_flush(d);
 196         rte_distributor_clear_returns(d);
 197         if (rte_mempool_get_bulk(p, (void *)many_bufs, BIG_BATCH) != 0) {
 198                 printf("line %d: Error getting mbufs from pool\n", __LINE__);
 199                 return -1;
 200         }
 201         for (i = 0; i < BIG_BATCH; i++)
 202                 many_bufs[i]->hash.usr = i << 2;
 203
 204         for (i = 0; i < BIG_BATCH/BURST; i++) {
 205                 rte_distributor_process(d, &many_bufs[i*BURST], BURST);
 206                 num_returned += rte_distributor_returned_pkts(d,
 207                                 &return_bufs[num_returned],
 208                                 BIG_BATCH - num_returned);
 209         }
 210         rte_distributor_flush(d);
 211         num_returned += rte_distributor_returned_pkts(d,
 212                         &return_bufs[num_returned], BIG_BATCH - num_returned);
 213
 214         if (num_returned != BIG_BATCH) {
 215                 printf("line %d: Number returned is not the same as "
 216                                 "number sent\n", __LINE__);
 217                 return -1;
 218         }
 219         /* big check -  make sure all packets made it back!! */
 220         for (i = 0; i < BIG_BATCH; i++) {
 221                 unsigned j;
 222                 struct rte_mbuf *src = many_bufs[i];
 223                 for (j = 0; j < BIG_BATCH; j++)
 224                         if (return_bufs[j] == src)
 225                                 break;
 226
 227                 if (j == BIG_BATCH) {
 228                         printf("Error: could not find source packet #%u\n", i);
 229                         return -1;
 230                 }
 231         }
 232         printf("Sanity test of returned packets done\n");
 233
 234         rte_mempool_put_bulk(p, (void *)many_bufs, BIG_BATCH);
 235
 236         printf("\n");
 237         return 0;
 238 }
 239
 240
 241 /* to test that the distributor does not lose packets, we use this worker
 242  * function which frees mbufs when it gets them. The distributor thread does
 243  * the mbuf allocation. If distributor drops packets we'll eventually run out
 244  * of mbufs.
 245  */
 246 static int
 247 handle_work_with_free_mbufs(void *arg)
 248 {
 249         struct rte_mbuf *pkt = NULL;
 250         struct rte_distributor *d = arg;
 251         unsigned count = 0;
 252         unsigned id = __sync_fetch_and_add(&worker_idx, 1);
 253
 254         pkt = rte_distributor_get_pkt(d, id, NULL);
 255         while (!quit) {
 256                 worker_stats[id].handled_packets++, count++;
 257                 rte_pktmbuf_free(pkt);
 258                 pkt = rte_distributor_get_pkt(d, id, pkt);
 259         }
 260         worker_stats[id].handled_packets++, count++;
 261         rte_distributor_return_pkt(d, id, pkt);
 262         return 0;
 263 }
 264
 265 /* Perform a sanity test of the distributor with a large number of packets,
 266  * where we allocate a new set of mbufs for each burst. The workers then
 267  * free the mbufs. This ensures that we don't have any packet leaks in the
 268  * library.
 269  */
 270 static int
 271 sanity_test_with_mbuf_alloc(struct rte_distributor *d, struct rte_mempool *p)
 272 {
 273         unsigned i;
 274         struct rte_mbuf *bufs[BURST];
 275
 276         printf("=== Sanity test with mbuf alloc/free  ===\n");
 277         clear_packet_count();
 278         for (i = 0; i < ((1<<ITER_POWER)); i += BURST) {
 279                 unsigned j;
 280                 while (rte_mempool_get_bulk(p, (void *)bufs, BURST) < 0)
 281                         rte_distributor_process(d, NULL, 0);
 282                 for (j = 0; j < BURST; j++) {
 283                         bufs[j]->hash.usr = (i+j) << 1;
 284                         rte_mbuf_refcnt_set(bufs[j], 1);
 285                 }
 286
 287                 rte_distributor_process(d, bufs, BURST);
 288         }
 289
 290         rte_distributor_flush(d);
 291         if (total_packet_count() < (1<<ITER_POWER)) {
 292                 printf("Line %u: Packet count is incorrect, %u, expected %u\n",
 293                                 __LINE__, total_packet_count(),
 294                                 (1<<ITER_POWER));
 295                 return -1;
 296         }
 297
 298         printf("Sanity test with mbuf alloc/free passed\n\n");
 299         return 0;
 300 }
 301
 302 static int
 303 handle_work_for_shutdown_test(void *arg)
 304 {
 305         struct rte_mbuf *pkt = NULL;
 306         struct rte_distributor *d = arg;
 307         unsigned count = 0;
 308         const unsigned id = __sync_fetch_and_add(&worker_idx, 1);
 309
 310         pkt = rte_distributor_get_pkt(d, id, NULL);
 311         /* wait for quit single globally, or for worker zero, wait
 312          * for zero_quit */
 313         while (!quit && !(id == 0 && zero_quit)) {
 314                 worker_stats[id].handled_packets++, count++;
 315                 rte_pktmbuf_free(pkt);
 316                 pkt = rte_distributor_get_pkt(d, id, NULL);
 317         }
 318         worker_stats[id].handled_packets++, count++;
 319         rte_distributor_return_pkt(d, id, pkt);
 320
 321         if (id == 0) {
 322                 /* for worker zero, allow it to restart to pick up last packet
 323                  * when all workers are shutting down.
 324                  */
 325                 while (zero_quit)
 326                         usleep(100);
 327                 pkt = rte_distributor_get_pkt(d, id, NULL);
 328                 while (!quit) {
 329                         worker_stats[id].handled_packets++, count++;
 330                         rte_pktmbuf_free(pkt);
 331                         pkt = rte_distributor_get_pkt(d, id, NULL);
 332                 }
 333                 rte_distributor_return_pkt(d, id, pkt);
 334         }
 335         return 0;
 336 }
 337
 338
 339 /* Perform a sanity test of the distributor with a large number of packets,
 340  * where we allocate a new set of mbufs for each burst. The workers then
 341  * free the mbufs. This ensures that we don't have any packet leaks in the
 342  * library.
 343  */
 344 static int
 345 sanity_test_with_worker_shutdown(struct rte_distributor *d,
 346                 struct rte_mempool *p)
 347 {
 348         struct rte_mbuf *bufs[BURST];
 349         unsigned i;
 350
 351         printf("=== Sanity test of worker shutdown ===\n");
 352
 353         clear_packet_count();
 354         if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) {
 355                 printf("line %d: Error getting mbufs from pool\n", __LINE__);
 356                 return -1;
 357         }
 358
 359         /* now set all hash values in all buffers to zero, so all pkts go to the
 360          * one worker thread */
 361         for (i = 0; i < BURST; i++)
 362                 bufs[i]->hash.usr = 0;
 363
 364         rte_distributor_process(d, bufs, BURST);
 365         /* at this point, we will have processed some packets and have a full
 366          * backlog for the other ones at worker 0.
 367          */
 368
 369         /* get more buffers to queue up, again setting them to the same flow */
 370         if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) {
 371                 printf("line %d: Error getting mbufs from pool\n", __LINE__);
 372                 return -1;
 373         }
 374         for (i = 0; i < BURST; i++)
 375                 bufs[i]->hash.usr = 0;
 376
 377         /* get worker zero to quit */
 378         zero_quit = 1;
 379         rte_distributor_process(d, bufs, BURST);
 380
 381         /* flush the distributor */
 382         rte_distributor_flush(d);
 383         if (total_packet_count() != BURST * 2) {
 384                 printf("Line %d: Error, not all packets flushed. "
 385                                 "Expected %u, got %u\n",
 386                                 __LINE__, BURST * 2, total_packet_count());
 387                 return -1;
 388         }
 389
 390         for (i = 0; i < rte_lcore_count() - 1; i++)
 391                 printf("Worker %u handled %u packets\n", i,
 392                                 worker_stats[i].handled_packets);
 393
 394         printf("Sanity test with worker shutdown passed\n\n");
 395         return 0;
 396 }
 397
 398 /* Test that the flush function is able to move packets between workers when
 399  * one worker shuts down..
 400  */
 401 static int
 402 test_flush_with_worker_shutdown(struct rte_distributor *d,
 403                 struct rte_mempool *p)
 404 {
 405         struct rte_mbuf *bufs[BURST];
 406         unsigned i;
 407
 408         printf("=== Test flush fn with worker shutdown ===\n");
 409
 410         clear_packet_count();
 411         if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) {
 412                 printf("line %d: Error getting mbufs from pool\n", __LINE__);
 413                 return -1;
 414         }
 415
 416         /* now set all hash values in all buffers to zero, so all pkts go to the
 417          * one worker thread */
 418         for (i = 0; i < BURST; i++)
 419                 bufs[i]->hash.usr = 0;
 420
 421         rte_distributor_process(d, bufs, BURST);
 422         /* at this point, we will have processed some packets and have a full
 423          * backlog for the other ones at worker 0.
 424          */
 425
 426         /* get worker zero to quit */
 427         zero_quit = 1;
 428
 429         /* flush the distributor */
 430         rte_distributor_flush(d);
 431
 432         zero_quit = 0;
 433         if (total_packet_count() != BURST) {
 434                 printf("Line %d: Error, not all packets flushed. "
 435                                 "Expected %u, got %u\n",
 436                                 __LINE__, BURST, total_packet_count());
 437                 return -1;
 438         }
 439
 440         for (i = 0; i < rte_lcore_count() - 1; i++)
 441                 printf("Worker %u handled %u packets\n", i,
 442                                 worker_stats[i].handled_packets);
 443
 444         printf("Flush test with worker shutdown passed\n\n");
 445         return 0;
 446 }
 447
 448 static
 449 int test_error_distributor_create_name(void)
 450 {
 451         struct rte_distributor *d = NULL;
 452         char *name = NULL;
 453
 454         d = rte_distributor_create(name, rte_socket_id(),
 455                         rte_lcore_count() - 1);
 456         if (d != NULL || rte_errno != EINVAL) {
 457                 printf("ERROR: No error on create() with NULL name param\n");
 458                 return -1;
 459         }
 460
 461         return 0;
 462 }
 463
 464
 465 static
 466 int test_error_distributor_create_numworkers(void)
 467 {
 468         struct rte_distributor *d = NULL;
 469         d = rte_distributor_create("test_numworkers", rte_socket_id(),
 470                         RTE_MAX_LCORE + 10);
 471         if (d != NULL || rte_errno != EINVAL) {
 472                 printf("ERROR: No error on create() with num_workers > MAX\n");
 473                 return -1;
 474         }
 475         return 0;
 476 }
 477
 478
 479 /* Useful function which ensures that all worker functions terminate */
 480 static void
 481 quit_workers(struct rte_distributor *d, struct rte_mempool *p)
 482 {
 483         const unsigned num_workers = rte_lcore_count() - 1;
 484         unsigned i;
 485         struct rte_mbuf *bufs[RTE_MAX_LCORE];
 486         rte_mempool_get_bulk(p, (void *)bufs, num_workers);
 487
 488         zero_quit = 0;
 489         quit = 1;
 490         for (i = 0; i < num_workers; i++)
 491                 bufs[i]->hash.usr = i << 1;
 492         rte_distributor_process(d, bufs, num_workers);
 493
 494         rte_mempool_put_bulk(p, (void *)bufs, num_workers);
 495
 496         rte_distributor_process(d, NULL, 0);
 497         rte_distributor_flush(d);
 498         rte_eal_mp_wait_lcore();
 499         quit = 0;
 500         worker_idx = 0;
 501 }
 502
 503 #define MBUF_SIZE (2048 + sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM)
 504
 505 static int
 506 test_distributor(void)
 507 {
 508         static struct rte_distributor *d;
 509         static struct rte_mempool *p;
 510
 511         if (rte_lcore_count() < 2) {
 512                 printf("ERROR: not enough cores to test distributor\n");
 513                 return -1;
 514         }
 515
 516         if (d == NULL) {
 517                 d = rte_distributor_create("Test_distributor", rte_socket_id(),
 518                                 rte_lcore_count() - 1);
 519                 if (d == NULL) {
 520                         printf("Error creating distributor\n");
 521                         return -1;
 522                 }
 523         } else {
 524                 rte_distributor_flush(d);
 525                 rte_distributor_clear_returns(d);
 526         }
 527
 528         const unsigned nb_bufs = (511 * rte_lcore_count()) < BIG_BATCH ?
 529                         (BIG_BATCH * 2) - 1 : (511 * rte_lcore_count());
 530         if (p == NULL) {
 531                 p = rte_mempool_create("DT_MBUF_POOL", nb_bufs,
 532                                 MBUF_SIZE, BURST,
 533                                 sizeof(struct rte_pktmbuf_pool_private),
 534                                 rte_pktmbuf_pool_init, NULL,
 535                                 rte_pktmbuf_init, NULL,
 536                                 rte_socket_id(), 0);
 537                 if (p == NULL) {
 538                         printf("Error creating mempool\n");
 539                         return -1;
 540                 }
 541         }
 542
 543         rte_eal_mp_remote_launch(handle_work, d, SKIP_MASTER);
 544         if (sanity_test(d, p) < 0)
 545                 goto err;
 546         quit_workers(d, p);
 547
 548         rte_eal_mp_remote_launch(handle_work_with_free_mbufs, d, SKIP_MASTER);
 549         if (sanity_test_with_mbuf_alloc(d, p) < 0)
 550                 goto err;
 551         quit_workers(d, p);
 552
 553         if (rte_lcore_count() > 2) {
 554                 rte_eal_mp_remote_launch(handle_work_for_shutdown_test, d,
 555                                 SKIP_MASTER);
 556                 if (sanity_test_with_worker_shutdown(d, p) < 0)
 557                         goto err;
 558                 quit_workers(d, p);
 559
 560                 rte_eal_mp_remote_launch(handle_work_for_shutdown_test, d,
 561                                 SKIP_MASTER);
 562                 if (test_flush_with_worker_shutdown(d, p) < 0)
 563                         goto err;
 564                 quit_workers(d, p);
 565
 566         } else {
 567                 printf("Not enough cores to run tests for worker shutdown\n");
 568         }
 569
 570         if (test_error_distributor_create_numworkers() == -1 ||
 571                         test_error_distributor_create_name() == -1) {
 572                 printf("rte_distributor_create parameter check tests failed");
 573                 return -1;
 574         }
 575
 576         return 0;
 577
 578 err:
 579         quit_workers(d, p);
 580         return -1;
 581 }
 582
 583 static struct test_command distributor_cmd = {
 584         .command = "distributor_autotest",
 585         .callback = test_distributor,
 586 };
 587 REGISTER_TEST_COMMAND(distributor_cmd);