examples/eventdev_pipeline/main.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2016-2017 Intel Corporation
   3  */
   4
   5 #include <getopt.h>
   6 #include <stdint.h>
   7 #include <stdio.h>
   8 #include <signal.h>
   9 #include <sched.h>
  10
  11 #include "pipeline_common.h"
  12
  13 struct fastpath_data *fdata;
  14
  15 struct config_data cdata = {
  16         .num_packets = (1L << 25), /* do ~32M packets */
  17         .num_fids = 512,
  18         .queue_type = RTE_SCHED_TYPE_ATOMIC,
  19         .next_qid = {-1},
  20         .qid = {-1},
  21         .num_stages = 1,
  22         .worker_cq_depth = 16
  23 };
  24
  25 static bool
  26 core_in_use(unsigned int lcore_id) {
  27         return (fdata->rx_core[lcore_id] || fdata->sched_core[lcore_id] ||
  28                 fdata->tx_core[lcore_id] || fdata->worker_core[lcore_id]);
  29 }
  30
  31 /*
  32  * Parse the coremask given as argument (hexadecimal string) and fill
  33  * the global configuration (core role and core count) with the parsed
  34  * value.
  35  */
  36 static int xdigit2val(unsigned char c)
  37 {
  38         int val;
  39
  40         if (isdigit(c))
  41                 val = c - '0';
  42         else if (isupper(c))
  43                 val = c - 'A' + 10;
  44         else
  45                 val = c - 'a' + 10;
  46         return val;
  47 }
  48
  49 static uint64_t
  50 parse_coremask(const char *coremask)
  51 {
  52         int i, j, idx = 0;
  53         unsigned int count = 0;
  54         char c;
  55         int val;
  56         uint64_t mask = 0;
  57         const int32_t BITS_HEX = 4;
  58
  59         if (coremask == NULL)
  60                 return -1;
  61         /* Remove all blank characters ahead and after .
  62          * Remove 0x/0X if exists.
  63          */
  64         while (isblank(*coremask))
  65                 coremask++;
  66         if (coremask[0] == '0' && ((coremask[1] == 'x')
  67                 || (coremask[1] == 'X')))
  68                 coremask += 2;
  69         i = strlen(coremask);
  70         while ((i > 0) && isblank(coremask[i - 1]))
  71                 i--;
  72         if (i == 0)
  73                 return -1;
  74
  75         for (i = i - 1; i >= 0 && idx < MAX_NUM_CORE; i--) {
  76                 c = coremask[i];
  77                 if (isxdigit(c) == 0) {
  78                         /* invalid characters */
  79                         return -1;
  80                 }
  81                 val = xdigit2val(c);
  82                 for (j = 0; j < BITS_HEX && idx < MAX_NUM_CORE; j++, idx++) {
  83                         if ((1 << j) & val) {
  84                                 mask |= (1ULL << idx);
  85                                 count++;
  86                         }
  87                 }
  88         }
  89         for (; i >= 0; i--)
  90                 if (coremask[i] != '0')
  91                         return -1;
  92         if (count == 0)
  93                 return -1;
  94         return mask;
  95 }
  96
  97 static struct option long_options[] = {
  98         {"workers", required_argument, 0, 'w'},
  99         {"packets", required_argument, 0, 'n'},
 100         {"atomic-flows", required_argument, 0, 'f'},
 101         {"num_stages", required_argument, 0, 's'},
 102         {"rx-mask", required_argument, 0, 'r'},
 103         {"tx-mask", required_argument, 0, 't'},
 104         {"sched-mask", required_argument, 0, 'e'},
 105         {"cq-depth", required_argument, 0, 'c'},
 106         {"work-cycles", required_argument, 0, 'W'},
 107         {"mempool-size", required_argument, 0, 'm'},
 108         {"queue-priority", no_argument, 0, 'P'},
 109         {"parallel", no_argument, 0, 'p'},
 110         {"ordered", no_argument, 0, 'o'},
 111         {"quiet", no_argument, 0, 'q'},
 112         {"use-atq", no_argument, 0, 'a'},
 113         {"dump", no_argument, 0, 'D'},
 114         {0, 0, 0, 0}
 115 };
 116
 117 static void
 118 usage(void)
 119 {
 120         const char *usage_str =
 121                 "  Usage: eventdev_demo [options]\n"
 122                 "  Options:\n"
 123                 "  -n, --packets=N              Send N packets (default ~32M), 0 implies no limit\n"
 124                 "  -f, --atomic-flows=N         Use N random flows from 1 to N (default 16)\n"
 125                 "  -s, --num_stages=N           Use N atomic stages (default 1)\n"
 126                 "  -r, --rx-mask=core mask      Run NIC rx on CPUs in core mask\n"
 127                 "  -w, --worker-mask=core mask  Run worker on CPUs in core mask\n"
 128                 "  -t, --tx-mask=core mask      Run NIC tx on CPUs in core mask\n"
 129                 "  -e  --sched-mask=core mask   Run scheduler on CPUs in core mask\n"
 130                 "  -c  --cq-depth=N             Worker CQ depth (default 16)\n"
 131                 "  -W  --work-cycles=N          Worker cycles (default 0)\n"
 132                 "  -P  --queue-priority         Enable scheduler queue prioritization\n"
 133                 "  -o, --ordered                Use ordered scheduling\n"
 134                 "  -p, --parallel               Use parallel scheduling\n"
 135                 "  -q, --quiet                  Minimize printed output\n"
 136                 "  -a, --use-atq                Use all type queues\n"
 137                 "  -m, --mempool-size=N         Dictate the mempool size\n"
 138                 "  -D, --dump                   Print detailed statistics before exit"
 139                 "\n";
 140         fprintf(stderr, "%s", usage_str);
 141         exit(1);
 142 }
 143
 144 static void
 145 parse_app_args(int argc, char **argv)
 146 {
 147         /* Parse cli options*/
 148         int option_index;
 149         int c;
 150         opterr = 0;
 151         uint64_t rx_lcore_mask = 0;
 152         uint64_t tx_lcore_mask = 0;
 153         uint64_t sched_lcore_mask = 0;
 154         uint64_t worker_lcore_mask = 0;
 155         int i;
 156
 157         for (;;) {
 158                 c = getopt_long(argc, argv, "r:t:e:c:w:n:f:s:m:paoPqDW:",
 159                                 long_options, &option_index);
 160                 if (c == -1)
 161                         break;
 162
 163                 int popcnt = 0;
 164                 switch (c) {
 165                 case 'n':
 166                         cdata.num_packets = (int64_t)atol(optarg);
 167                         if (cdata.num_packets == 0)
 168                                 cdata.num_packets = INT64_MAX;
 169                         break;
 170                 case 'f':
 171                         cdata.num_fids = (unsigned int)atoi(optarg);
 172                         break;
 173                 case 's':
 174                         cdata.num_stages = (unsigned int)atoi(optarg);
 175                         break;
 176                 case 'c':
 177                         cdata.worker_cq_depth = (unsigned int)atoi(optarg);
 178                         break;
 179                 case 'W':
 180                         cdata.worker_cycles = (unsigned int)atoi(optarg);
 181                         break;
 182                 case 'P':
 183                         cdata.enable_queue_priorities = 1;
 184                         break;
 185                 case 'o':
 186                         cdata.queue_type = RTE_SCHED_TYPE_ORDERED;
 187                         break;
 188                 case 'p':
 189                         cdata.queue_type = RTE_SCHED_TYPE_PARALLEL;
 190                         break;
 191                 case 'a':
 192                         cdata.all_type_queues = 1;
 193                         break;
 194                 case 'q':
 195                         cdata.quiet = 1;
 196                         break;
 197                 case 'D':
 198                         cdata.dump_dev = 1;
 199                         break;
 200                 case 'w':
 201                         worker_lcore_mask = parse_coremask(optarg);
 202                         break;
 203                 case 'r':
 204                         rx_lcore_mask = parse_coremask(optarg);
 205                         popcnt = __builtin_popcountll(rx_lcore_mask);
 206                         fdata->rx_single = (popcnt == 1);
 207                         break;
 208                 case 't':
 209                         tx_lcore_mask = parse_coremask(optarg);
 210                         popcnt = __builtin_popcountll(tx_lcore_mask);
 211                         fdata->tx_single = (popcnt == 1);
 212                         break;
 213                 case 'e':
 214                         sched_lcore_mask = parse_coremask(optarg);
 215                         popcnt = __builtin_popcountll(sched_lcore_mask);
 216                         fdata->sched_single = (popcnt == 1);
 217                         break;
 218                 case 'm':
 219                         cdata.num_mbuf = (uint64_t)atol(optarg);
 220                         break;
 221                 default:
 222                         usage();
 223                 }
 224         }
 225
 226         cdata.worker_lcore_mask = worker_lcore_mask;
 227         cdata.sched_lcore_mask = sched_lcore_mask;
 228         cdata.rx_lcore_mask = rx_lcore_mask;
 229         cdata.tx_lcore_mask = tx_lcore_mask;
 230
 231         if (cdata.num_stages == 0 || cdata.num_stages > MAX_NUM_STAGES)
 232                 usage();
 233
 234         for (i = 0; i < MAX_NUM_CORE; i++) {
 235                 fdata->rx_core[i] = !!(rx_lcore_mask & (1ULL << i));
 236                 fdata->tx_core[i] = !!(tx_lcore_mask & (1ULL << i));
 237                 fdata->sched_core[i] = !!(sched_lcore_mask & (1ULL << i));
 238                 fdata->worker_core[i] = !!(worker_lcore_mask & (1ULL << i));
 239
 240                 if (fdata->worker_core[i])
 241                         cdata.num_workers++;
 242                 if (core_in_use(i)) {
 243                         if (!rte_lcore_is_enabled(i)) {
 244                                 printf("lcore %d is not enabled in lcore list\n",
 245                                         i);
 246                                 rte_exit(EXIT_FAILURE,
 247                                         "check lcore params failed\n");
 248                         }
 249                         cdata.active_cores++;
 250                 }
 251         }
 252 }
 253
 254 static void
 255 do_capability_setup(uint8_t eventdev_id)
 256 {
 257         int ret;
 258         uint16_t i;
 259         uint8_t generic_pipeline = 0;
 260         uint8_t burst = 0;
 261
 262         RTE_ETH_FOREACH_DEV(i) {
 263                 uint32_t caps = 0;
 264
 265                 ret = rte_event_eth_tx_adapter_caps_get(eventdev_id, i, &caps);
 266                 if (ret)
 267                         rte_exit(EXIT_FAILURE,
 268                                 "Invalid capability for Tx adptr port %d\n", i);
 269                 generic_pipeline |= !(caps &
 270                                 RTE_EVENT_ETH_TX_ADAPTER_CAP_INTERNAL_PORT);
 271         }
 272
 273         struct rte_event_dev_info eventdev_info;
 274         memset(&eventdev_info, 0, sizeof(struct rte_event_dev_info));
 275
 276         rte_event_dev_info_get(eventdev_id, &eventdev_info);
 277         burst = eventdev_info.event_dev_cap & RTE_EVENT_DEV_CAP_BURST_MODE ? 1 :
 278                 0;
 279
 280         if (generic_pipeline)
 281                 set_worker_generic_setup_data(&fdata->cap, burst);
 282         else
 283                 set_worker_tx_enq_setup_data(&fdata->cap, burst);
 284 }
 285
 286 static void
 287 signal_handler(int signum)
 288 {
 289         static uint8_t once;
 290         uint16_t portid;
 291
 292         if (fdata->done)
 293                 rte_exit(1, "Exiting on signal %d\n", signum);
 294         if ((signum == SIGINT || signum == SIGTERM) && !once) {
 295                 printf("\n\nSignal %d received, preparing to exit...\n",
 296                                 signum);
 297                 if (cdata.dump_dev)
 298                         rte_event_dev_dump(0, stdout);
 299                 once = 1;
 300                 fdata->done = 1;
 301                 rte_smp_wmb();
 302
 303                 RTE_ETH_FOREACH_DEV(portid) {
 304                         rte_event_eth_rx_adapter_stop(portid);
 305                         rte_event_eth_tx_adapter_stop(portid);
 306                         if (rte_eth_dev_stop(portid) < 0)
 307                                 printf("Failed to stop port %u", portid);
 308                 }
 309
 310                 rte_eal_mp_wait_lcore();
 311
 312         }
 313         if (signum == SIGTSTP)
 314                 rte_event_dev_dump(0, stdout);
 315 }
 316
 317 static inline uint64_t
 318 port_stat(int dev_id, int32_t p)
 319 {
 320         char statname[64];
 321         snprintf(statname, sizeof(statname), "port_%u_rx", p);
 322         return rte_event_dev_xstats_by_name_get(dev_id, statname, NULL);
 323 }
 324
 325 int
 326 main(int argc, char **argv)
 327 {
 328         struct worker_data *worker_data;
 329         uint16_t num_ports;
 330         uint16_t portid;
 331         int lcore_id;
 332         int err;
 333
 334         signal(SIGINT, signal_handler);
 335         signal(SIGTERM, signal_handler);
 336         signal(SIGTSTP, signal_handler);
 337
 338         err = rte_eal_init(argc, argv);
 339         if (err < 0)
 340                 rte_panic("Invalid EAL arguments\n");
 341
 342         argc -= err;
 343         argv += err;
 344
 345         fdata = rte_malloc(NULL, sizeof(struct fastpath_data), 0);
 346         if (fdata == NULL)
 347                 rte_panic("Out of memory\n");
 348
 349         /* Parse cli options*/
 350         parse_app_args(argc, argv);
 351
 352         num_ports = rte_eth_dev_count_avail();
 353         if (num_ports == 0)
 354                 rte_panic("No ethernet ports found\n");
 355
 356         const unsigned int cores_needed = cdata.active_cores;
 357
 358         if (!cdata.quiet) {
 359                 printf("  Config:\n");
 360                 printf("\tports: %u\n", num_ports);
 361                 printf("\tworkers: %u\n", cdata.num_workers);
 362                 printf("\tpackets: %"PRIi64"\n", cdata.num_packets);
 363                 printf("\tQueue-prio: %u\n", cdata.enable_queue_priorities);
 364                 if (cdata.queue_type == RTE_SCHED_TYPE_ORDERED)
 365                         printf("\tqid0 type: ordered\n");
 366                 if (cdata.queue_type == RTE_SCHED_TYPE_ATOMIC)
 367                         printf("\tqid0 type: atomic\n");
 368                 printf("\tCores available: %u\n", rte_lcore_count());
 369                 printf("\tCores used: %u\n", cores_needed);
 370         }
 371
 372         if (rte_lcore_count() < cores_needed)
 373                 rte_panic("Too few cores (%d < %d)\n", rte_lcore_count(),
 374                                 cores_needed);
 375
 376         const unsigned int ndevs = rte_event_dev_count();
 377         if (ndevs == 0)
 378                 rte_panic("No dev_id devs found. Pasl in a --vdev eventdev.\n");
 379         if (ndevs > 1)
 380                 fprintf(stderr, "Warning: More than one eventdev, using idx 0");
 381
 382
 383         do_capability_setup(0);
 384         fdata->cap.check_opt();
 385
 386         worker_data = rte_calloc(0, cdata.num_workers,
 387                         sizeof(worker_data[0]), 0);
 388         if (worker_data == NULL)
 389                 rte_panic("rte_calloc failed\n");
 390
 391         int dev_id = fdata->cap.evdev_setup(worker_data);
 392         if (dev_id < 0)
 393                 rte_exit(EXIT_FAILURE, "Error setting up eventdev\n");
 394
 395         fdata->cap.adptr_setup(num_ports);
 396
 397         /* Start the Ethernet port. */
 398         RTE_ETH_FOREACH_DEV(portid) {
 399                 err = rte_eth_dev_start(portid);
 400                 if (err < 0)
 401                         rte_exit(EXIT_FAILURE, "Error starting ethdev %d\n",
 402                                         portid);
 403         }
 404
 405         int worker_idx = 0;
 406         RTE_LCORE_FOREACH_WORKER(lcore_id) {
 407                 if (lcore_id >= MAX_NUM_CORE)
 408                         break;
 409
 410                 if (!fdata->rx_core[lcore_id] &&
 411                         !fdata->worker_core[lcore_id] &&
 412                         !fdata->tx_core[lcore_id] &&
 413                         !fdata->sched_core[lcore_id])
 414                         continue;
 415
 416                 if (fdata->rx_core[lcore_id])
 417                         printf(
 418                                 "[%s()] lcore %d executing NIC Rx\n",
 419                                 __func__, lcore_id);
 420
 421                 if (fdata->tx_core[lcore_id])
 422                         printf(
 423                                 "[%s()] lcore %d executing NIC Tx\n",
 424                                 __func__, lcore_id);
 425
 426                 if (fdata->sched_core[lcore_id])
 427                         printf("[%s()] lcore %d executing scheduler\n",
 428                                         __func__, lcore_id);
 429
 430                 if (fdata->worker_core[lcore_id])
 431                         printf(
 432                                 "[%s()] lcore %d executing worker, using eventdev port %u\n",
 433                                 __func__, lcore_id,
 434                                 worker_data[worker_idx].port_id);
 435
 436                 err = rte_eal_remote_launch(fdata->cap.worker,
 437                                 &worker_data[worker_idx], lcore_id);
 438                 if (err) {
 439                         rte_panic("Failed to launch worker on core %d\n",
 440                                         lcore_id);
 441                         continue;
 442                 }
 443                 if (fdata->worker_core[lcore_id])
 444                         worker_idx++;
 445         }
 446
 447         lcore_id = rte_lcore_id();
 448
 449         if (core_in_use(lcore_id))
 450                 fdata->cap.worker(&worker_data[worker_idx++]);
 451
 452         rte_eal_mp_wait_lcore();
 453
 454         if (!cdata.quiet && (port_stat(dev_id, worker_data[0].port_id) !=
 455                         (uint64_t)-ENOTSUP)) {
 456                 printf("\nPort Workload distribution:\n");
 457                 uint32_t i;
 458                 uint64_t tot_pkts = 0;
 459                 uint64_t pkts_per_wkr[RTE_MAX_LCORE] = {0};
 460                 for (i = 0; i < cdata.num_workers; i++) {
 461                         pkts_per_wkr[i] =
 462                                 port_stat(dev_id, worker_data[i].port_id);
 463                         tot_pkts += pkts_per_wkr[i];
 464                 }
 465                 for (i = 0; i < cdata.num_workers; i++) {
 466                         float pc = pkts_per_wkr[i]  * 100 /
 467                                 ((float)tot_pkts);
 468                         printf("worker %i :\t%.1f %% (%"PRIu64" pkts)\n",
 469                                         i, pc, pkts_per_wkr[i]);
 470                 }
 471
 472         }
 473
 474         RTE_ETH_FOREACH_DEV(portid) {
 475                 rte_eth_dev_close(portid);
 476         }
 477
 478         rte_event_dev_stop(0);
 479         rte_event_dev_close(0);
 480
 481         rte_eal_cleanup();
 482
 483         return 0;
 484 }