1 /* 2 * SPDX-License-Identifier: BSD-3-Clause 3 * Copyright 2017 Cavium, Inc. 4 */ 5 6 #include "test_pipeline_common.h" 7 8 static int32_t 9 pipeline_event_tx_burst_service_func(void *args) 10 { 11 12 int i; 13 struct tx_service_data *tx = args; 14 const uint8_t dev = tx->dev_id; 15 const uint8_t port = tx->port_id; 16 struct rte_event ev[BURST_SIZE + 1]; 17 18 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev, BURST_SIZE, 0); 19 20 if (!nb_rx) { 21 for (i = 0; i < tx->nb_ethports; i++) 22 rte_eth_tx_buffer_flush(i, 0, tx->tx_buf[i]); 23 return 0; 24 } 25 26 for (i = 0; i < nb_rx; i++) { 27 struct rte_mbuf *m = ev[i].mbuf; 28 rte_eth_tx_buffer(m->port, 0, tx->tx_buf[m->port], m); 29 } 30 tx->processed_pkts += nb_rx; 31 32 return 0; 33 } 34 35 static int32_t 36 pipeline_event_tx_service_func(void *args) 37 { 38 39 int i; 40 struct tx_service_data *tx = args; 41 const uint8_t dev = tx->dev_id; 42 const uint8_t port = tx->port_id; 43 struct rte_event ev; 44 45 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, &ev, 1, 0); 46 47 if (!nb_rx) { 48 for (i = 0; i < tx->nb_ethports; i++) 49 rte_eth_tx_buffer_flush(i, 0, tx->tx_buf[i]); 50 return 0; 51 } 52 53 struct rte_mbuf *m = ev.mbuf; 54 rte_eth_tx_buffer(m->port, 0, tx->tx_buf[m->port], m); 55 tx->processed_pkts++; 56 57 return 0; 58 } 59 60 int 61 pipeline_test_result(struct evt_test *test, struct evt_options *opt) 62 { 63 RTE_SET_USED(opt); 64 int i; 65 uint64_t total = 0; 66 struct test_pipeline *t = evt_test_priv(test); 67 68 evt_info("Packet distribution across worker cores :"); 69 for (i = 0; i < t->nb_workers; i++) 70 total += t->worker[i].processed_pkts; 71 for (i = 0; i < t->nb_workers; i++) 72 evt_info("Worker %d packets: "CLGRN"%"PRIx64""CLNRM" percentage:" 73 CLGRN" %3.2f"CLNRM, i, 74 t->worker[i].processed_pkts, 75 (((double)t->worker[i].processed_pkts)/total) 76 * 100); 77 return t->result; 78 } 79 80 void 81 pipeline_opt_dump(struct evt_options *opt, uint8_t nb_queues) 82 { 83 evt_dump("nb_worker_lcores", "%d", evt_nr_active_lcores(opt->wlcores)); 84 evt_dump_worker_lcores(opt); 85 evt_dump_nb_stages(opt); 86 evt_dump("nb_evdev_ports", "%d", pipeline_nb_event_ports(opt)); 87 evt_dump("nb_evdev_queues", "%d", nb_queues); 88 evt_dump_queue_priority(opt); 89 evt_dump_sched_type_list(opt); 90 evt_dump_producer_type(opt); 91 } 92 93 static inline uint64_t 94 processed_pkts(struct test_pipeline *t) 95 { 96 uint8_t i; 97 uint64_t total = 0; 98 99 rte_smp_rmb(); 100 if (t->mt_unsafe) 101 total = t->tx_service.processed_pkts; 102 else 103 for (i = 0; i < t->nb_workers; i++) 104 total += t->worker[i].processed_pkts; 105 106 return total; 107 } 108 109 int 110 pipeline_launch_lcores(struct evt_test *test, struct evt_options *opt, 111 int (*worker)(void *)) 112 { 113 int ret, lcore_id; 114 struct test_pipeline *t = evt_test_priv(test); 115 116 int port_idx = 0; 117 /* launch workers */ 118 RTE_LCORE_FOREACH_SLAVE(lcore_id) { 119 if (!(opt->wlcores[lcore_id])) 120 continue; 121 122 ret = rte_eal_remote_launch(worker, 123 &t->worker[port_idx], lcore_id); 124 if (ret) { 125 evt_err("failed to launch worker %d", lcore_id); 126 return ret; 127 } 128 port_idx++; 129 } 130 131 uint64_t perf_cycles = rte_get_timer_cycles(); 132 const uint64_t perf_sample = rte_get_timer_hz(); 133 134 static float total_mpps; 135 static uint64_t samples; 136 137 uint64_t prev_pkts = 0; 138 139 while (t->done == false) { 140 const uint64_t new_cycles = rte_get_timer_cycles(); 141 142 if ((new_cycles - perf_cycles) > perf_sample) { 143 const uint64_t curr_pkts = processed_pkts(t); 144 145 float mpps = (float)(curr_pkts - prev_pkts)/1000000; 146 147 prev_pkts = curr_pkts; 148 perf_cycles = new_cycles; 149 total_mpps += mpps; 150 ++samples; 151 printf(CLGRN"\r%.3f mpps avg %.3f mpps"CLNRM, 152 mpps, total_mpps/samples); 153 fflush(stdout); 154 } 155 } 156 printf("\n"); 157 return 0; 158 } 159 160 int 161 pipeline_opt_check(struct evt_options *opt, uint64_t nb_queues) 162 { 163 unsigned int lcores; 164 /* 165 * N worker + 1 master 166 */ 167 lcores = 2; 168 169 if (!rte_eth_dev_count_avail()) { 170 evt_err("test needs minimum 1 ethernet dev"); 171 return -1; 172 } 173 174 if (rte_lcore_count() < lcores) { 175 evt_err("test need minimum %d lcores", lcores); 176 return -1; 177 } 178 179 /* Validate worker lcores */ 180 if (evt_lcores_has_overlap(opt->wlcores, rte_get_master_lcore())) { 181 evt_err("worker lcores overlaps with master lcore"); 182 return -1; 183 } 184 if (evt_has_disabled_lcore(opt->wlcores)) { 185 evt_err("one or more workers lcores are not enabled"); 186 return -1; 187 } 188 if (!evt_has_active_lcore(opt->wlcores)) { 189 evt_err("minimum one worker is required"); 190 return -1; 191 } 192 193 if (nb_queues > EVT_MAX_QUEUES) { 194 evt_err("number of queues exceeds %d", EVT_MAX_QUEUES); 195 return -1; 196 } 197 if (pipeline_nb_event_ports(opt) > EVT_MAX_PORTS) { 198 evt_err("number of ports exceeds %d", EVT_MAX_PORTS); 199 return -1; 200 } 201 202 if (evt_has_invalid_stage(opt)) 203 return -1; 204 205 if (evt_has_invalid_sched_type(opt)) 206 return -1; 207 208 return 0; 209 } 210 211 #define NB_RX_DESC 128 212 #define NB_TX_DESC 512 213 int 214 pipeline_ethdev_setup(struct evt_test *test, struct evt_options *opt) 215 { 216 uint16_t i; 217 uint8_t nb_queues = 1; 218 uint8_t mt_state = 0; 219 struct test_pipeline *t = evt_test_priv(test); 220 struct rte_eth_rxconf rx_conf; 221 struct rte_eth_conf port_conf = { 222 .rxmode = { 223 .mq_mode = ETH_MQ_RX_RSS, 224 .max_rx_pkt_len = ETHER_MAX_LEN, 225 }, 226 .rx_adv_conf = { 227 .rss_conf = { 228 .rss_key = NULL, 229 .rss_hf = ETH_RSS_IP, 230 }, 231 }, 232 }; 233 234 RTE_SET_USED(opt); 235 if (!rte_eth_dev_count_avail()) { 236 evt_err("No ethernet ports found."); 237 return -ENODEV; 238 } 239 240 RTE_ETH_FOREACH_DEV(i) { 241 struct rte_eth_dev_info dev_info; 242 struct rte_eth_conf local_port_conf = port_conf; 243 244 rte_eth_dev_info_get(i, &dev_info); 245 mt_state = !(dev_info.tx_offload_capa & 246 DEV_TX_OFFLOAD_MT_LOCKFREE); 247 rx_conf = dev_info.default_rxconf; 248 rx_conf.offloads = port_conf.rxmode.offloads; 249 250 local_port_conf.rx_adv_conf.rss_conf.rss_hf &= 251 dev_info.flow_type_rss_offloads; 252 if (local_port_conf.rx_adv_conf.rss_conf.rss_hf != 253 port_conf.rx_adv_conf.rss_conf.rss_hf) { 254 evt_info("Port %u modified RSS hash function based on hardware support," 255 "requested:%#"PRIx64" configured:%#"PRIx64"", 256 i, 257 port_conf.rx_adv_conf.rss_conf.rss_hf, 258 local_port_conf.rx_adv_conf.rss_conf.rss_hf); 259 } 260 261 if (rte_eth_dev_configure(i, nb_queues, nb_queues, 262 &local_port_conf) 263 < 0) { 264 evt_err("Failed to configure eth port [%d]", i); 265 return -EINVAL; 266 } 267 268 if (rte_eth_rx_queue_setup(i, 0, NB_RX_DESC, 269 rte_socket_id(), &rx_conf, t->pool) < 0) { 270 evt_err("Failed to setup eth port [%d] rx_queue: %d.", 271 i, 0); 272 return -EINVAL; 273 } 274 if (rte_eth_tx_queue_setup(i, 0, NB_TX_DESC, 275 rte_socket_id(), NULL) < 0) { 276 evt_err("Failed to setup eth port [%d] tx_queue: %d.", 277 i, 0); 278 return -EINVAL; 279 } 280 281 t->mt_unsafe |= mt_state; 282 t->tx_service.tx_buf[i] = 283 rte_malloc(NULL, RTE_ETH_TX_BUFFER_SIZE(BURST_SIZE), 0); 284 if (t->tx_service.tx_buf[i] == NULL) 285 rte_panic("Unable to allocate Tx buffer memory."); 286 rte_eth_promiscuous_enable(i); 287 } 288 289 return 0; 290 } 291 292 int 293 pipeline_event_port_setup(struct evt_test *test, struct evt_options *opt, 294 uint8_t *queue_arr, uint8_t nb_queues, 295 const struct rte_event_port_conf p_conf) 296 { 297 int i; 298 int ret; 299 uint8_t port; 300 struct test_pipeline *t = evt_test_priv(test); 301 302 303 /* setup one port per worker, linking to all queues */ 304 for (port = 0; port < evt_nr_active_lcores(opt->wlcores); port++) { 305 struct worker_data *w = &t->worker[port]; 306 307 w->dev_id = opt->dev_id; 308 w->port_id = port; 309 w->t = t; 310 w->processed_pkts = 0; 311 312 ret = rte_event_port_setup(opt->dev_id, port, &p_conf); 313 if (ret) { 314 evt_err("failed to setup port %d", port); 315 return ret; 316 } 317 318 if (queue_arr == NULL) { 319 if (rte_event_port_link(opt->dev_id, port, NULL, NULL, 320 0) != nb_queues) 321 goto link_fail; 322 } else { 323 for (i = 0; i < nb_queues; i++) { 324 if (rte_event_port_link(opt->dev_id, port, 325 &queue_arr[i], NULL, 1) != 1) 326 goto link_fail; 327 } 328 } 329 } 330 331 return 0; 332 333 link_fail: 334 evt_err("failed to link all queues to port %d", port); 335 return -EINVAL; 336 } 337 338 int 339 pipeline_event_rx_adapter_setup(struct evt_options *opt, uint8_t stride, 340 struct rte_event_port_conf prod_conf) 341 { 342 int ret = 0; 343 uint16_t prod; 344 struct rte_event_eth_rx_adapter_queue_conf queue_conf; 345 346 memset(&queue_conf, 0, 347 sizeof(struct rte_event_eth_rx_adapter_queue_conf)); 348 queue_conf.ev.sched_type = opt->sched_type_list[0]; 349 RTE_ETH_FOREACH_DEV(prod) { 350 uint32_t cap; 351 352 ret = rte_event_eth_rx_adapter_caps_get(opt->dev_id, 353 prod, &cap); 354 if (ret) { 355 evt_err("failed to get event rx adapter[%d]" 356 " capabilities", 357 opt->dev_id); 358 return ret; 359 } 360 queue_conf.ev.queue_id = prod * stride; 361 ret = rte_event_eth_rx_adapter_create(prod, opt->dev_id, 362 &prod_conf); 363 if (ret) { 364 evt_err("failed to create rx adapter[%d]", prod); 365 return ret; 366 } 367 ret = rte_event_eth_rx_adapter_queue_add(prod, prod, -1, 368 &queue_conf); 369 if (ret) { 370 evt_err("failed to add rx queues to adapter[%d]", prod); 371 return ret; 372 } 373 374 if (!(cap & RTE_EVENT_ETH_RX_ADAPTER_CAP_INTERNAL_PORT)) { 375 uint32_t service_id; 376 377 rte_event_eth_rx_adapter_service_id_get(prod, 378 &service_id); 379 ret = evt_service_setup(service_id); 380 if (ret) { 381 evt_err("Failed to setup service core" 382 " for Rx adapter"); 383 return ret; 384 } 385 } 386 387 ret = rte_eth_dev_start(prod); 388 if (ret) { 389 evt_err("Ethernet dev [%d] failed to start." 390 " Using synthetic producer", prod); 391 return ret; 392 } 393 394 ret = rte_event_eth_rx_adapter_start(prod); 395 if (ret) { 396 evt_err("Rx adapter[%d] start failed", prod); 397 return ret; 398 } 399 evt_info("Port[%d] using Rx adapter[%d] started", prod, prod); 400 } 401 402 return ret; 403 } 404 405 int 406 pipeline_event_tx_service_setup(struct evt_test *test, struct evt_options *opt, 407 uint8_t tx_queue_id, uint8_t tx_port_id, 408 const struct rte_event_port_conf p_conf) 409 { 410 int ret; 411 struct rte_service_spec serv; 412 struct test_pipeline *t = evt_test_priv(test); 413 struct tx_service_data *tx = &t->tx_service; 414 415 ret = rte_event_port_setup(opt->dev_id, tx_port_id, &p_conf); 416 if (ret) { 417 evt_err("failed to setup port %d", tx_port_id); 418 return ret; 419 } 420 421 if (rte_event_port_link(opt->dev_id, tx_port_id, &tx_queue_id, 422 NULL, 1) != 1) { 423 evt_err("failed to link queues to port %d", tx_port_id); 424 return -EINVAL; 425 } 426 427 tx->dev_id = opt->dev_id; 428 tx->queue_id = tx_queue_id; 429 tx->port_id = tx_port_id; 430 tx->nb_ethports = rte_eth_dev_count_avail(); 431 tx->t = t; 432 433 /* Register Tx service */ 434 memset(&serv, 0, sizeof(struct rte_service_spec)); 435 snprintf(serv.name, sizeof(serv.name), "Tx_service"); 436 437 if (evt_has_burst_mode(opt->dev_id)) 438 serv.callback = pipeline_event_tx_burst_service_func; 439 else 440 serv.callback = pipeline_event_tx_service_func; 441 442 serv.callback_userdata = (void *)tx; 443 ret = rte_service_component_register(&serv, &tx->service_id); 444 if (ret) { 445 evt_err("failed to register Tx service"); 446 return ret; 447 } 448 449 ret = evt_service_setup(tx->service_id); 450 if (ret) { 451 evt_err("Failed to setup service core for Tx service\n"); 452 return ret; 453 } 454 455 rte_service_runstate_set(tx->service_id, 1); 456 457 return 0; 458 } 459 460 461 void 462 pipeline_ethdev_destroy(struct evt_test *test, struct evt_options *opt) 463 { 464 uint16_t i; 465 RTE_SET_USED(test); 466 RTE_SET_USED(opt); 467 struct test_pipeline *t = evt_test_priv(test); 468 469 if (t->mt_unsafe) { 470 rte_service_component_runstate_set(t->tx_service.service_id, 0); 471 rte_service_runstate_set(t->tx_service.service_id, 0); 472 rte_service_component_unregister(t->tx_service.service_id); 473 } 474 475 RTE_ETH_FOREACH_DEV(i) { 476 rte_event_eth_rx_adapter_stop(i); 477 rte_eth_dev_stop(i); 478 } 479 } 480 481 void 482 pipeline_eventdev_destroy(struct evt_test *test, struct evt_options *opt) 483 { 484 RTE_SET_USED(test); 485 486 rte_event_dev_stop(opt->dev_id); 487 rte_event_dev_close(opt->dev_id); 488 } 489 490 int 491 pipeline_mempool_setup(struct evt_test *test, struct evt_options *opt) 492 { 493 struct test_pipeline *t = evt_test_priv(test); 494 495 t->pool = rte_pktmbuf_pool_create(test->name, /* mempool name */ 496 opt->pool_sz, /* number of elements*/ 497 512, /* cache size*/ 498 0, 499 RTE_MBUF_DEFAULT_BUF_SIZE, 500 opt->socket_id); /* flags */ 501 502 if (t->pool == NULL) { 503 evt_err("failed to create mempool"); 504 return -ENOMEM; 505 } 506 507 return 0; 508 } 509 510 void 511 pipeline_mempool_destroy(struct evt_test *test, struct evt_options *opt) 512 { 513 RTE_SET_USED(opt); 514 struct test_pipeline *t = evt_test_priv(test); 515 516 rte_mempool_free(t->pool); 517 } 518 519 int 520 pipeline_test_setup(struct evt_test *test, struct evt_options *opt) 521 { 522 void *test_pipeline; 523 524 test_pipeline = rte_zmalloc_socket(test->name, 525 sizeof(struct test_pipeline), RTE_CACHE_LINE_SIZE, 526 opt->socket_id); 527 if (test_pipeline == NULL) { 528 evt_err("failed to allocate test_pipeline memory"); 529 goto nomem; 530 } 531 test->test_priv = test_pipeline; 532 533 struct test_pipeline *t = evt_test_priv(test); 534 535 t->nb_workers = evt_nr_active_lcores(opt->wlcores); 536 t->outstand_pkts = opt->nb_pkts * evt_nr_active_lcores(opt->wlcores); 537 t->done = false; 538 t->nb_flows = opt->nb_flows; 539 t->result = EVT_TEST_FAILED; 540 t->opt = opt; 541 opt->prod_type = EVT_PROD_TYPE_ETH_RX_ADPTR; 542 memcpy(t->sched_type_list, opt->sched_type_list, 543 sizeof(opt->sched_type_list)); 544 return 0; 545 nomem: 546 return -ENOMEM; 547 } 548 549 void 550 pipeline_test_destroy(struct evt_test *test, struct evt_options *opt) 551 { 552 RTE_SET_USED(opt); 553 554 rte_free(test->test_priv); 555 } 556