1 /* 2 * SPDX-License-Identifier: BSD-3-Clause 3 * Copyright 2017 Cavium, Inc. 4 */ 5 6 #include "test_pipeline_common.h" 7 8 /* See http://doc.dpdk.org/guides/tools/testeventdev.html for test details */ 9 10 static __rte_always_inline int 11 pipeline_atq_nb_event_queues(struct evt_options *opt) 12 { 13 RTE_SET_USED(opt); 14 15 return rte_eth_dev_count_avail(); 16 } 17 18 static __rte_noinline int 19 pipeline_atq_worker_single_stage_tx(void *arg) 20 { 21 PIPELINE_WORKER_SINGLE_STAGE_INIT; 22 23 while (t->done == false) { 24 uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0); 25 26 if (!event) { 27 rte_pause(); 28 continue; 29 } 30 31 pipeline_event_tx(dev, port, &ev); 32 w->processed_pkts++; 33 } 34 35 return 0; 36 } 37 38 static __rte_noinline int 39 pipeline_atq_worker_single_stage_fwd(void *arg) 40 { 41 PIPELINE_WORKER_SINGLE_STAGE_INIT; 42 const uint8_t *tx_queue = t->tx_evqueue_id; 43 44 while (t->done == false) { 45 uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0); 46 47 if (!event) { 48 rte_pause(); 49 continue; 50 } 51 52 ev.queue_id = tx_queue[ev.mbuf->port]; 53 pipeline_fwd_event(&ev, RTE_SCHED_TYPE_ATOMIC); 54 pipeline_event_enqueue(dev, port, &ev); 55 w->processed_pkts++; 56 } 57 58 return 0; 59 } 60 61 static __rte_noinline int 62 pipeline_atq_worker_single_stage_burst_tx(void *arg) 63 { 64 PIPELINE_WORKER_SINGLE_STAGE_BURST_INIT; 65 66 while (t->done == false) { 67 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev, 68 BURST_SIZE, 0); 69 70 if (!nb_rx) { 71 rte_pause(); 72 continue; 73 } 74 75 for (i = 0; i < nb_rx; i++) { 76 rte_prefetch0(ev[i + 1].mbuf); 77 rte_event_eth_tx_adapter_txq_set(ev[i].mbuf, 0); 78 } 79 80 pipeline_event_tx_burst(dev, port, ev, nb_rx); 81 w->processed_pkts += nb_rx; 82 } 83 84 return 0; 85 } 86 87 static __rte_noinline int 88 pipeline_atq_worker_single_stage_burst_fwd(void *arg) 89 { 90 PIPELINE_WORKER_SINGLE_STAGE_BURST_INIT; 91 const uint8_t *tx_queue = t->tx_evqueue_id; 92 93 while (t->done == false) { 94 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev, 95 BURST_SIZE, 0); 96 97 if (!nb_rx) { 98 rte_pause(); 99 continue; 100 } 101 102 for (i = 0; i < nb_rx; i++) { 103 rte_prefetch0(ev[i + 1].mbuf); 104 rte_event_eth_tx_adapter_txq_set(ev[i].mbuf, 0); 105 ev[i].queue_id = tx_queue[ev[i].mbuf->port]; 106 pipeline_fwd_event(&ev[i], RTE_SCHED_TYPE_ATOMIC); 107 } 108 109 pipeline_event_enqueue_burst(dev, port, ev, nb_rx); 110 w->processed_pkts += nb_rx; 111 } 112 113 return 0; 114 } 115 116 static __rte_noinline int 117 pipeline_atq_worker_multi_stage_tx(void *arg) 118 { 119 PIPELINE_WORKER_MULTI_STAGE_INIT; 120 121 while (t->done == false) { 122 uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0); 123 124 if (!event) { 125 rte_pause(); 126 continue; 127 } 128 129 cq_id = ev.sub_event_type % nb_stages; 130 131 if (cq_id == last_queue) { 132 pipeline_event_tx(dev, port, &ev); 133 w->processed_pkts++; 134 continue; 135 } 136 137 ev.sub_event_type++; 138 pipeline_fwd_event(&ev, sched_type_list[cq_id]); 139 pipeline_event_enqueue(dev, port, &ev); 140 } 141 142 return 0; 143 } 144 145 static __rte_noinline int 146 pipeline_atq_worker_multi_stage_fwd(void *arg) 147 { 148 PIPELINE_WORKER_MULTI_STAGE_INIT; 149 const uint8_t *tx_queue = t->tx_evqueue_id; 150 151 while (t->done == false) { 152 uint16_t event = rte_event_dequeue_burst(dev, port, &ev, 1, 0); 153 154 if (!event) { 155 rte_pause(); 156 continue; 157 } 158 159 cq_id = ev.sub_event_type % nb_stages; 160 161 if (cq_id == last_queue) { 162 ev.queue_id = tx_queue[ev.mbuf->port]; 163 pipeline_fwd_event(&ev, RTE_SCHED_TYPE_ATOMIC); 164 w->processed_pkts++; 165 } else { 166 ev.sub_event_type++; 167 pipeline_fwd_event(&ev, sched_type_list[cq_id]); 168 } 169 170 pipeline_event_enqueue(dev, port, &ev); 171 } 172 173 return 0; 174 } 175 176 static __rte_noinline int 177 pipeline_atq_worker_multi_stage_burst_tx(void *arg) 178 { 179 PIPELINE_WORKER_MULTI_STAGE_BURST_INIT; 180 181 while (t->done == false) { 182 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev, 183 BURST_SIZE, 0); 184 185 if (!nb_rx) { 186 rte_pause(); 187 continue; 188 } 189 190 for (i = 0; i < nb_rx; i++) { 191 rte_prefetch0(ev[i + 1].mbuf); 192 cq_id = ev[i].sub_event_type % nb_stages; 193 194 if (cq_id == last_queue) { 195 pipeline_event_tx(dev, port, &ev[i]); 196 ev[i].op = RTE_EVENT_OP_RELEASE; 197 w->processed_pkts++; 198 continue; 199 } 200 201 ev[i].sub_event_type++; 202 pipeline_fwd_event(&ev[i], sched_type_list[cq_id]); 203 } 204 205 pipeline_event_enqueue_burst(dev, port, ev, nb_rx); 206 } 207 208 return 0; 209 } 210 211 static __rte_noinline int 212 pipeline_atq_worker_multi_stage_burst_fwd(void *arg) 213 { 214 PIPELINE_WORKER_MULTI_STAGE_BURST_INIT; 215 const uint8_t *tx_queue = t->tx_evqueue_id; 216 217 while (t->done == false) { 218 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev, 219 BURST_SIZE, 0); 220 221 if (!nb_rx) { 222 rte_pause(); 223 continue; 224 } 225 226 for (i = 0; i < nb_rx; i++) { 227 rte_prefetch0(ev[i + 1].mbuf); 228 cq_id = ev[i].sub_event_type % nb_stages; 229 230 if (cq_id == last_queue) { 231 w->processed_pkts++; 232 ev[i].queue_id = tx_queue[ev[i].mbuf->port]; 233 pipeline_fwd_event(&ev[i], 234 RTE_SCHED_TYPE_ATOMIC); 235 } else { 236 ev[i].sub_event_type++; 237 pipeline_fwd_event(&ev[i], 238 sched_type_list[cq_id]); 239 } 240 } 241 242 pipeline_event_enqueue_burst(dev, port, ev, nb_rx); 243 } 244 245 return 0; 246 } 247 248 static int 249 worker_wrapper(void *arg) 250 { 251 struct worker_data *w = arg; 252 struct evt_options *opt = w->t->opt; 253 const bool burst = evt_has_burst_mode(w->dev_id); 254 const bool internal_port = w->t->internal_port; 255 const uint8_t nb_stages = opt->nb_stages; 256 RTE_SET_USED(opt); 257 258 if (nb_stages == 1) { 259 if (!burst && internal_port) 260 return pipeline_atq_worker_single_stage_tx(arg); 261 else if (!burst && !internal_port) 262 return pipeline_atq_worker_single_stage_fwd(arg); 263 else if (burst && internal_port) 264 return pipeline_atq_worker_single_stage_burst_tx(arg); 265 else if (burst && !internal_port) 266 return pipeline_atq_worker_single_stage_burst_fwd(arg); 267 } else { 268 if (!burst && internal_port) 269 return pipeline_atq_worker_multi_stage_tx(arg); 270 else if (!burst && !internal_port) 271 return pipeline_atq_worker_multi_stage_fwd(arg); 272 if (burst && internal_port) 273 return pipeline_atq_worker_multi_stage_burst_tx(arg); 274 else if (burst && !internal_port) 275 return pipeline_atq_worker_multi_stage_burst_fwd(arg); 276 } 277 278 rte_panic("invalid worker\n"); 279 } 280 281 static int 282 pipeline_atq_launch_lcores(struct evt_test *test, struct evt_options *opt) 283 { 284 return pipeline_launch_lcores(test, opt, worker_wrapper); 285 } 286 287 static int 288 pipeline_atq_eventdev_setup(struct evt_test *test, struct evt_options *opt) 289 { 290 int ret; 291 int nb_ports; 292 int nb_queues; 293 uint8_t queue; 294 uint8_t tx_evqueue_id[RTE_MAX_ETHPORTS]; 295 uint8_t queue_arr[RTE_EVENT_MAX_QUEUES_PER_DEV]; 296 uint8_t nb_worker_queues = 0; 297 uint8_t tx_evport_id = 0; 298 uint16_t prod = 0; 299 struct rte_event_dev_info info; 300 struct test_pipeline *t = evt_test_priv(test); 301 302 nb_ports = evt_nr_active_lcores(opt->wlcores); 303 nb_queues = rte_eth_dev_count_avail(); 304 305 memset(tx_evqueue_id, 0, sizeof(uint8_t) * RTE_MAX_ETHPORTS); 306 memset(queue_arr, 0, sizeof(uint8_t) * RTE_EVENT_MAX_QUEUES_PER_DEV); 307 /* One queue for Tx adapter per port */ 308 if (!t->internal_port) { 309 RTE_ETH_FOREACH_DEV(prod) { 310 tx_evqueue_id[prod] = nb_queues; 311 nb_queues++; 312 } 313 } 314 315 rte_event_dev_info_get(opt->dev_id, &info); 316 317 const struct rte_event_dev_config config = { 318 .nb_event_queues = nb_queues, 319 .nb_event_ports = nb_ports, 320 .nb_events_limit = info.max_num_events, 321 .nb_event_queue_flows = opt->nb_flows, 322 .nb_event_port_dequeue_depth = 323 info.max_event_port_dequeue_depth, 324 .nb_event_port_enqueue_depth = 325 info.max_event_port_enqueue_depth, 326 }; 327 ret = rte_event_dev_configure(opt->dev_id, &config); 328 if (ret) { 329 evt_err("failed to configure eventdev %d", opt->dev_id); 330 return ret; 331 } 332 333 struct rte_event_queue_conf q_conf = { 334 .priority = RTE_EVENT_DEV_PRIORITY_NORMAL, 335 .nb_atomic_flows = opt->nb_flows, 336 .nb_atomic_order_sequences = opt->nb_flows, 337 }; 338 /* queue configurations */ 339 for (queue = 0; queue < nb_queues; queue++) { 340 q_conf.event_queue_cfg = RTE_EVENT_QUEUE_CFG_ALL_TYPES; 341 342 if (!t->internal_port) { 343 RTE_ETH_FOREACH_DEV(prod) { 344 if (queue == tx_evqueue_id[prod]) { 345 q_conf.event_queue_cfg = 346 RTE_EVENT_QUEUE_CFG_SINGLE_LINK; 347 } else { 348 queue_arr[nb_worker_queues] = queue; 349 nb_worker_queues++; 350 } 351 } 352 } 353 354 ret = rte_event_queue_setup(opt->dev_id, queue, &q_conf); 355 if (ret) { 356 evt_err("failed to setup queue=%d", queue); 357 return ret; 358 } 359 } 360 361 if (opt->wkr_deq_dep > info.max_event_port_dequeue_depth) 362 opt->wkr_deq_dep = info.max_event_port_dequeue_depth; 363 364 /* port configuration */ 365 const struct rte_event_port_conf p_conf = { 366 .dequeue_depth = opt->wkr_deq_dep, 367 .enqueue_depth = info.max_event_port_dequeue_depth, 368 .new_event_threshold = info.max_num_events, 369 }; 370 371 if (!t->internal_port) 372 ret = pipeline_event_port_setup(test, opt, queue_arr, 373 nb_worker_queues, p_conf); 374 else 375 ret = pipeline_event_port_setup(test, opt, NULL, nb_queues, 376 p_conf); 377 378 if (ret) 379 return ret; 380 381 /* 382 * The pipelines are setup in the following manner: 383 * 384 * eth_dev_count = 2, nb_stages = 2, atq mode 385 * 386 * eth0, eth1 have Internal port capability : 387 * queues = 2 388 * stride = 1 389 * 390 * event queue pipelines: 391 * eth0 -> q0 ->Tx 392 * eth1 -> q1 ->Tx 393 * 394 * q0, q1 are configured as ATQ so, all the different stages can 395 * be enqueued on the same queue. 396 * 397 * eth0, eth1 use Tx adapters service core : 398 * queues = 4 399 * stride = 1 400 * 401 * event queue pipelines: 402 * eth0 -> q0 -> q2 -> Tx 403 * eth1 -> q1 -> q3 -> Tx 404 * 405 * q0, q1 are configured as stated above. 406 * q2, q3 configured as SINGLE_LINK. 407 */ 408 ret = pipeline_event_rx_adapter_setup(opt, 1, p_conf); 409 if (ret) 410 return ret; 411 ret = pipeline_event_tx_adapter_setup(opt, p_conf); 412 if (ret) 413 return ret; 414 415 if (!evt_has_distributed_sched(opt->dev_id)) { 416 uint32_t service_id; 417 rte_event_dev_service_id_get(opt->dev_id, &service_id); 418 ret = evt_service_setup(service_id); 419 if (ret) { 420 evt_err("No service lcore found to run event dev."); 421 return ret; 422 } 423 } 424 425 /* Connect the tx_evqueue_id to the Tx adapter port */ 426 if (!t->internal_port) { 427 RTE_ETH_FOREACH_DEV(prod) { 428 ret = rte_event_eth_tx_adapter_event_port_get(prod, 429 &tx_evport_id); 430 if (ret) { 431 evt_err("Unable to get Tx adapter[%d]", prod); 432 return ret; 433 } 434 435 if (rte_event_port_link(opt->dev_id, tx_evport_id, 436 &tx_evqueue_id[prod], 437 NULL, 1) != 1) { 438 evt_err("Unable to link Tx adptr[%d] evprt[%d]", 439 prod, tx_evport_id); 440 return ret; 441 } 442 } 443 } 444 445 RTE_ETH_FOREACH_DEV(prod) { 446 ret = rte_eth_dev_start(prod); 447 if (ret) { 448 evt_err("Ethernet dev [%d] failed to start." 449 " Using synthetic producer", prod); 450 return ret; 451 } 452 } 453 454 ret = rte_event_dev_start(opt->dev_id); 455 if (ret) { 456 evt_err("failed to start eventdev %d", opt->dev_id); 457 return ret; 458 } 459 460 RTE_ETH_FOREACH_DEV(prod) { 461 ret = rte_event_eth_rx_adapter_start(prod); 462 if (ret) { 463 evt_err("Rx adapter[%d] start failed", prod); 464 return ret; 465 } 466 467 ret = rte_event_eth_tx_adapter_start(prod); 468 if (ret) { 469 evt_err("Tx adapter[%d] start failed", prod); 470 return ret; 471 } 472 } 473 474 memcpy(t->tx_evqueue_id, tx_evqueue_id, sizeof(uint8_t) * 475 RTE_MAX_ETHPORTS); 476 477 return 0; 478 } 479 480 static void 481 pipeline_atq_opt_dump(struct evt_options *opt) 482 { 483 pipeline_opt_dump(opt, pipeline_atq_nb_event_queues(opt)); 484 } 485 486 static int 487 pipeline_atq_opt_check(struct evt_options *opt) 488 { 489 return pipeline_opt_check(opt, pipeline_atq_nb_event_queues(opt)); 490 } 491 492 static bool 493 pipeline_atq_capability_check(struct evt_options *opt) 494 { 495 struct rte_event_dev_info dev_info; 496 497 rte_event_dev_info_get(opt->dev_id, &dev_info); 498 if (dev_info.max_event_queues < pipeline_atq_nb_event_queues(opt) || 499 dev_info.max_event_ports < 500 evt_nr_active_lcores(opt->wlcores)) { 501 evt_err("not enough eventdev queues=%d/%d or ports=%d/%d", 502 pipeline_atq_nb_event_queues(opt), 503 dev_info.max_event_queues, 504 evt_nr_active_lcores(opt->wlcores), 505 dev_info.max_event_ports); 506 } 507 508 return true; 509 } 510 511 static const struct evt_test_ops pipeline_atq = { 512 .cap_check = pipeline_atq_capability_check, 513 .opt_check = pipeline_atq_opt_check, 514 .opt_dump = pipeline_atq_opt_dump, 515 .test_setup = pipeline_test_setup, 516 .mempool_setup = pipeline_mempool_setup, 517 .ethdev_setup = pipeline_ethdev_setup, 518 .eventdev_setup = pipeline_atq_eventdev_setup, 519 .launch_lcores = pipeline_atq_launch_lcores, 520 .eventdev_destroy = pipeline_eventdev_destroy, 521 .mempool_destroy = pipeline_mempool_destroy, 522 .ethdev_destroy = pipeline_ethdev_destroy, 523 .test_result = pipeline_test_result, 524 .test_destroy = pipeline_test_destroy, 525 }; 526 527 EVT_TEST_REGISTER(pipeline_atq); 528