2 * SPDX-License-Identifier: BSD-3-Clause
3 * Copyright 2017 Cavium, Inc.
6 #include "test_pipeline_common.h"
9 pipeline_event_tx_burst_service_func(void *args)
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];
18 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, ev, BURST_SIZE, 0);
21 for (i = 0; i < tx->nb_ethports; i++)
22 rte_eth_tx_buffer_flush(i, 0, tx->tx_buf[i]);
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);
30 tx->processed_pkts += nb_rx;
36 pipeline_event_tx_service_func(void *args)
40 struct tx_service_data *tx = args;
41 const uint8_t dev = tx->dev_id;
42 const uint8_t port = tx->port_id;
45 uint16_t nb_rx = rte_event_dequeue_burst(dev, port, &ev, 1, 0);
48 for (i = 0; i < tx->nb_ethports; i++)
49 rte_eth_tx_buffer_flush(i, 0, tx->tx_buf[i]);
53 struct rte_mbuf *m = ev.mbuf;
54 rte_eth_tx_buffer(m->port, 0, tx->tx_buf[m->port], m);
61 pipeline_test_result(struct evt_test *test, struct evt_options *opt)
66 struct test_pipeline *t = evt_test_priv(test);
68 printf("Packet distribution across worker cores :\n");
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 printf("Worker %d packets: "CLGRN"%"PRIx64" "CLNRM"percentage:"
73 CLGRN" %3.2f\n"CLNRM, i,
74 t->worker[i].processed_pkts,
75 (((double)t->worker[i].processed_pkts)/total)
81 pipeline_opt_dump(struct evt_options *opt, uint8_t nb_queues)
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);
93 static inline uint64_t
94 processed_pkts(struct test_pipeline *t)
101 total = t->tx_service.processed_pkts;
103 for (i = 0; i < t->nb_workers; i++)
104 total += t->worker[i].processed_pkts;
110 pipeline_launch_lcores(struct evt_test *test, struct evt_options *opt,
111 int (*worker)(void *))
114 struct test_pipeline *t = evt_test_priv(test);
118 RTE_LCORE_FOREACH_SLAVE(lcore_id) {
119 if (!(opt->wlcores[lcore_id]))
122 ret = rte_eal_remote_launch(worker,
123 &t->worker[port_idx], lcore_id);
125 evt_err("failed to launch worker %d", lcore_id);
131 uint64_t perf_cycles = rte_get_timer_cycles();
132 const uint64_t perf_sample = rte_get_timer_hz();
134 static float total_mpps;
135 static uint64_t samples;
137 uint64_t prev_pkts = 0;
139 while (t->done == false) {
140 const uint64_t new_cycles = rte_get_timer_cycles();
142 if ((new_cycles - perf_cycles) > perf_sample) {
143 const uint64_t curr_pkts = processed_pkts(t);
145 float mpps = (float)(curr_pkts - prev_pkts)/1000000;
147 prev_pkts = curr_pkts;
148 perf_cycles = new_cycles;
151 printf(CLGRN"\r%.3f mpps avg %.3f mpps"CLNRM,
152 mpps, total_mpps/samples);
161 pipeline_opt_check(struct evt_options *opt, uint64_t nb_queues)
165 * N worker + 1 master
169 if (!rte_eth_dev_count_avail()) {
170 evt_err("test needs minimum 1 ethernet dev");
174 if (rte_lcore_count() < lcores) {
175 evt_err("test need minimum %d lcores", lcores);
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");
184 if (evt_has_disabled_lcore(opt->wlcores)) {
185 evt_err("one or more workers lcores are not enabled");
188 if (!evt_has_active_lcore(opt->wlcores)) {
189 evt_err("minimum one worker is required");
193 if (nb_queues > EVT_MAX_QUEUES) {
194 evt_err("number of queues exceeds %d", EVT_MAX_QUEUES);
197 if (pipeline_nb_event_ports(opt) > EVT_MAX_PORTS) {
198 evt_err("number of ports exceeds %d", EVT_MAX_PORTS);
202 if (evt_has_invalid_stage(opt))
205 if (evt_has_invalid_sched_type(opt))
211 #define NB_RX_DESC 128
212 #define NB_TX_DESC 512
214 pipeline_ethdev_setup(struct evt_test *test, struct evt_options *opt)
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 = {
223 .mq_mode = ETH_MQ_RX_RSS,
224 .max_rx_pkt_len = ETHER_MAX_LEN,
225 .offloads = DEV_RX_OFFLOAD_CRC_STRIP,
230 .rss_hf = ETH_RSS_IP,
236 if (!rte_eth_dev_count_avail()) {
237 evt_err("No ethernet ports found.\n");
241 RTE_ETH_FOREACH_DEV(i) {
242 struct rte_eth_dev_info dev_info;
243 struct rte_eth_conf local_port_conf = port_conf;
245 rte_eth_dev_info_get(i, &dev_info);
246 mt_state = !(dev_info.tx_offload_capa &
247 DEV_TX_OFFLOAD_MT_LOCKFREE);
248 rx_conf = dev_info.default_rxconf;
249 rx_conf.offloads = port_conf.rxmode.offloads;
251 local_port_conf.rx_adv_conf.rss_conf.rss_hf &=
252 dev_info.flow_type_rss_offloads;
253 if (local_port_conf.rx_adv_conf.rss_conf.rss_hf !=
254 port_conf.rx_adv_conf.rss_conf.rss_hf) {
255 evt_info("Port %u modified RSS hash function based on hardware support,"
256 "requested:%#"PRIx64" configured:%#"PRIx64"\n",
258 port_conf.rx_adv_conf.rss_conf.rss_hf,
259 local_port_conf.rx_adv_conf.rss_conf.rss_hf);
262 if (rte_eth_dev_configure(i, nb_queues, nb_queues,
265 evt_err("Failed to configure eth port [%d]\n", i);
269 if (rte_eth_rx_queue_setup(i, 0, NB_RX_DESC,
270 rte_socket_id(), &rx_conf, t->pool) < 0) {
271 evt_err("Failed to setup eth port [%d] rx_queue: %d.\n",
275 if (rte_eth_tx_queue_setup(i, 0, NB_TX_DESC,
276 rte_socket_id(), NULL) < 0) {
277 evt_err("Failed to setup eth port [%d] tx_queue: %d.\n",
282 t->mt_unsafe |= mt_state;
283 t->tx_service.tx_buf[i] =
284 rte_malloc(NULL, RTE_ETH_TX_BUFFER_SIZE(BURST_SIZE), 0);
285 if (t->tx_service.tx_buf[i] == NULL)
286 rte_panic("Unable to allocate Tx buffer memory.");
287 rte_eth_promiscuous_enable(i);
294 pipeline_event_port_setup(struct evt_test *test, struct evt_options *opt,
295 uint8_t *queue_arr, uint8_t nb_queues,
296 const struct rte_event_port_conf p_conf)
301 struct test_pipeline *t = evt_test_priv(test);
304 /* setup one port per worker, linking to all queues */
305 for (port = 0; port < evt_nr_active_lcores(opt->wlcores); port++) {
306 struct worker_data *w = &t->worker[port];
308 w->dev_id = opt->dev_id;
311 w->processed_pkts = 0;
313 ret = rte_event_port_setup(opt->dev_id, port, &p_conf);
315 evt_err("failed to setup port %d", port);
319 if (queue_arr == NULL) {
320 if (rte_event_port_link(opt->dev_id, port, NULL, NULL,
324 for (i = 0; i < nb_queues; i++) {
325 if (rte_event_port_link(opt->dev_id, port,
326 &queue_arr[i], NULL, 1) != 1)
335 evt_err("failed to link all queues to port %d", port);
340 pipeline_event_rx_adapter_setup(struct evt_options *opt, uint8_t stride,
341 struct rte_event_port_conf prod_conf)
345 struct rte_event_eth_rx_adapter_queue_conf queue_conf;
347 memset(&queue_conf, 0,
348 sizeof(struct rte_event_eth_rx_adapter_queue_conf));
349 queue_conf.ev.sched_type = opt->sched_type_list[0];
350 RTE_ETH_FOREACH_DEV(prod) {
353 ret = rte_event_eth_rx_adapter_caps_get(opt->dev_id,
356 evt_err("failed to get event rx adapter[%d]"
361 queue_conf.ev.queue_id = prod * stride;
362 ret = rte_event_eth_rx_adapter_create(prod, opt->dev_id,
365 evt_err("failed to create rx adapter[%d]", prod);
368 ret = rte_event_eth_rx_adapter_queue_add(prod, prod, -1,
371 evt_err("failed to add rx queues to adapter[%d]", prod);
375 if (!(cap & RTE_EVENT_ETH_RX_ADAPTER_CAP_INTERNAL_PORT)) {
378 rte_event_eth_rx_adapter_service_id_get(prod,
380 ret = evt_service_setup(service_id);
382 evt_err("Failed to setup service core"
383 " for Rx adapter\n");
388 ret = rte_eth_dev_start(prod);
390 evt_err("Ethernet dev [%d] failed to start."
391 " Using synthetic producer", prod);
395 ret = rte_event_eth_rx_adapter_start(prod);
397 evt_err("Rx adapter[%d] start failed", prod);
400 printf("%s: Port[%d] using Rx adapter[%d] started\n", __func__,
408 pipeline_event_tx_service_setup(struct evt_test *test, struct evt_options *opt,
409 uint8_t tx_queue_id, uint8_t tx_port_id,
410 const struct rte_event_port_conf p_conf)
413 struct rte_service_spec serv;
414 struct test_pipeline *t = evt_test_priv(test);
415 struct tx_service_data *tx = &t->tx_service;
417 ret = rte_event_port_setup(opt->dev_id, tx_port_id, &p_conf);
419 evt_err("failed to setup port %d", tx_port_id);
423 if (rte_event_port_link(opt->dev_id, tx_port_id, &tx_queue_id,
425 evt_err("failed to link queues to port %d", tx_port_id);
429 tx->dev_id = opt->dev_id;
430 tx->queue_id = tx_queue_id;
431 tx->port_id = tx_port_id;
432 tx->nb_ethports = rte_eth_dev_count_avail();
435 /* Register Tx service */
436 memset(&serv, 0, sizeof(struct rte_service_spec));
437 snprintf(serv.name, sizeof(serv.name), "Tx_service");
439 if (evt_has_burst_mode(opt->dev_id))
440 serv.callback = pipeline_event_tx_burst_service_func;
442 serv.callback = pipeline_event_tx_service_func;
444 serv.callback_userdata = (void *)tx;
445 ret = rte_service_component_register(&serv, &tx->service_id);
447 evt_err("failed to register Tx service");
451 ret = evt_service_setup(tx->service_id);
453 evt_err("Failed to setup service core for Tx service\n");
457 rte_service_runstate_set(tx->service_id, 1);
464 pipeline_ethdev_destroy(struct evt_test *test, struct evt_options *opt)
469 struct test_pipeline *t = evt_test_priv(test);
472 rte_service_component_runstate_set(t->tx_service.service_id, 0);
473 rte_service_runstate_set(t->tx_service.service_id, 0);
474 rte_service_component_unregister(t->tx_service.service_id);
477 RTE_ETH_FOREACH_DEV(i) {
478 rte_event_eth_rx_adapter_stop(i);
480 rte_eth_dev_close(i);
485 pipeline_eventdev_destroy(struct evt_test *test, struct evt_options *opt)
489 rte_event_dev_stop(opt->dev_id);
490 rte_event_dev_close(opt->dev_id);
494 pipeline_mempool_setup(struct evt_test *test, struct evt_options *opt)
496 struct test_pipeline *t = evt_test_priv(test);
498 t->pool = rte_pktmbuf_pool_create(test->name, /* mempool name */
499 opt->pool_sz, /* number of elements*/
502 RTE_MBUF_DEFAULT_BUF_SIZE,
503 opt->socket_id); /* flags */
505 if (t->pool == NULL) {
506 evt_err("failed to create mempool");
514 pipeline_mempool_destroy(struct evt_test *test, struct evt_options *opt)
517 struct test_pipeline *t = evt_test_priv(test);
519 rte_mempool_free(t->pool);
523 pipeline_test_setup(struct evt_test *test, struct evt_options *opt)
527 test_pipeline = rte_zmalloc_socket(test->name,
528 sizeof(struct test_pipeline), RTE_CACHE_LINE_SIZE,
530 if (test_pipeline == NULL) {
531 evt_err("failed to allocate test_pipeline memory");
534 test->test_priv = test_pipeline;
536 struct test_pipeline *t = evt_test_priv(test);
538 t->nb_workers = evt_nr_active_lcores(opt->wlcores);
539 t->outstand_pkts = opt->nb_pkts * evt_nr_active_lcores(opt->wlcores);
541 t->nb_flows = opt->nb_flows;
542 t->result = EVT_TEST_FAILED;
544 opt->prod_type = EVT_PROD_TYPE_ETH_RX_ADPTR;
545 memcpy(t->sched_type_list, opt->sched_type_list,
546 sizeof(opt->sched_type_list));
553 pipeline_test_destroy(struct evt_test *test, struct evt_options *opt)
557 rte_free(test->test_priv);