xref: /dpdk/examples/eventdev_pipeline/pipeline_worker_generic.c (revision 79ca24a41c16445594303a62151ee68156a5a320)
1 /*
2  * SPDX-License-Identifier: BSD-3-Clause
3  * Copyright 2016 Intel Corporation.
4  * Copyright 2017 Cavium, Inc.
5  */
6 
7 #include <stdlib.h>
8 
9 #include "pipeline_common.h"
10 
11 static __rte_always_inline int
12 worker_generic(void *arg)
13 {
14 	struct rte_event ev;
15 
16 	struct worker_data *data = (struct worker_data *)arg;
17 	uint8_t dev_id = data->dev_id;
18 	uint8_t port_id = data->port_id;
19 	size_t sent = 0, received = 0;
20 	unsigned int lcore_id = rte_lcore_id();
21 	uint16_t nb_rx = 0, nb_tx = 0;
22 
23 	while (!fdata->done) {
24 
25 		if (fdata->cap.scheduler)
26 			fdata->cap.scheduler(lcore_id);
27 
28 		if (!fdata->worker_core[lcore_id]) {
29 			rte_pause();
30 			continue;
31 		}
32 
33 		nb_rx = rte_event_dequeue_burst(dev_id, port_id, &ev, 1, 0);
34 
35 		if (nb_rx == 0) {
36 			rte_pause();
37 			continue;
38 		}
39 		received++;
40 
41 		/* The first worker stage does classification and sets txq. */
42 		if (ev.queue_id == cdata.qid[0]) {
43 			ev.flow_id = ev.mbuf->hash.rss
44 						% cdata.num_fids;
45 			rte_event_eth_tx_adapter_txq_set(ev.mbuf, 0);
46 		}
47 
48 		ev.queue_id = cdata.next_qid[ev.queue_id];
49 		ev.op = RTE_EVENT_OP_FORWARD;
50 		ev.sched_type = cdata.queue_type;
51 
52 		work();
53 
54 		do {
55 			nb_tx = rte_event_enqueue_burst(dev_id, port_id, &ev,
56 							1);
57 		} while (!nb_tx && !fdata->done);
58 		sent++;
59 	}
60 
61 	worker_cleanup(dev_id, port_id, &ev, nb_tx, nb_rx);
62 	if (!cdata.quiet)
63 		printf("  worker %u thread done. RX=%zu TX=%zu\n",
64 				rte_lcore_id(), received, sent);
65 
66 	return 0;
67 }
68 
69 static int
70 worker_generic_burst(void *arg)
71 {
72 	struct rte_event events[BATCH_SIZE];
73 
74 	struct worker_data *data = (struct worker_data *)arg;
75 	uint8_t dev_id = data->dev_id;
76 	uint8_t port_id = data->port_id;
77 	size_t sent = 0, received = 0;
78 	unsigned int lcore_id = rte_lcore_id();
79 	uint16_t i, nb_rx = 0, nb_tx = 0;
80 
81 	while (!fdata->done) {
82 		if (fdata->cap.scheduler)
83 			fdata->cap.scheduler(lcore_id);
84 
85 		if (!fdata->worker_core[lcore_id]) {
86 			rte_pause();
87 			continue;
88 		}
89 
90 		nb_rx = rte_event_dequeue_burst(dev_id, port_id, events,
91 						RTE_DIM(events), 0);
92 
93 		if (nb_rx == 0) {
94 			rte_pause();
95 			continue;
96 		}
97 		received += nb_rx;
98 
99 		for (i = 0; i < nb_rx; i++) {
100 
101 			/* The first worker stage does classification and sets txq. */
102 			if (events[i].queue_id == cdata.qid[0]) {
103 				events[i].flow_id = events[i].mbuf->hash.rss
104 							% cdata.num_fids;
105 				rte_event_eth_tx_adapter_txq_set(events[i].mbuf, 0);
106 			}
107 
108 			events[i].queue_id = cdata.next_qid[events[i].queue_id];
109 			events[i].op = RTE_EVENT_OP_FORWARD;
110 			events[i].sched_type = cdata.queue_type;
111 
112 			work();
113 		}
114 		nb_tx = rte_event_enqueue_burst(dev_id, port_id, events, nb_rx);
115 		while (nb_tx < nb_rx && !fdata->done)
116 			nb_tx += rte_event_enqueue_burst(dev_id, port_id,
117 							events + nb_tx,
118 							nb_rx - nb_tx);
119 		sent += nb_tx;
120 	}
121 
122 	worker_cleanup(dev_id, port_id, events, nb_tx, nb_rx);
123 
124 	if (!cdata.quiet)
125 		printf("  worker %u thread done. RX=%zu TX=%zu\n",
126 				rte_lcore_id(), received, sent);
127 
128 	return 0;
129 }
130 
131 static int
132 setup_eventdev_generic(struct worker_data *worker_data)
133 {
134 	const uint8_t dev_id = 0;
135 	/* +1 stages is for a SINGLE_LINK TX stage */
136 	const uint8_t nb_queues = cdata.num_stages + 1;
137 	const uint8_t nb_ports = cdata.num_workers;
138 	struct rte_event_dev_config config = {
139 			.nb_event_queues = nb_queues,
140 			.nb_event_ports = nb_ports,
141 			.nb_single_link_event_port_queues = 1,
142 			.nb_events_limit  = 4096,
143 			.nb_event_queue_flows = 1024,
144 			.nb_event_port_dequeue_depth = 128,
145 			.nb_event_port_enqueue_depth = 128,
146 	};
147 	struct rte_event_port_conf wkr_p_conf = {
148 			.dequeue_depth = cdata.worker_cq_depth,
149 			.enqueue_depth = 64,
150 			.new_event_threshold = 4096,
151 			.event_port_cfg = RTE_EVENT_PORT_CFG_HINT_WORKER,
152 	};
153 	struct rte_event_queue_conf wkr_q_conf = {
154 			.schedule_type = cdata.queue_type,
155 			.priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
156 			.nb_atomic_flows = 1024,
157 			.nb_atomic_order_sequences = 1024,
158 	};
159 	struct rte_event_queue_conf tx_q_conf = {
160 			.priority = RTE_EVENT_DEV_PRIORITY_HIGHEST,
161 			.event_queue_cfg = RTE_EVENT_QUEUE_CFG_SINGLE_LINK,
162 	};
163 
164 	struct port_link worker_queues[MAX_NUM_STAGES];
165 	uint8_t disable_implicit_release;
166 	unsigned int i;
167 
168 	int ret, ndev = rte_event_dev_count();
169 	if (ndev < 1) {
170 		printf("%d: No Eventdev Devices Found\n", __LINE__);
171 		return -1;
172 	}
173 
174 	struct rte_event_dev_info dev_info;
175 	ret = rte_event_dev_info_get(dev_id, &dev_info);
176 	printf("\tEventdev %d: %s\n", dev_id, dev_info.driver_name);
177 
178 	disable_implicit_release = (dev_info.event_dev_cap &
179 			RTE_EVENT_DEV_CAP_IMPLICIT_RELEASE_DISABLE);
180 
181 	wkr_p_conf.event_port_cfg = disable_implicit_release ?
182 		RTE_EVENT_PORT_CFG_DISABLE_IMPL_REL : 0;
183 
184 	if (dev_info.max_num_events < config.nb_events_limit)
185 		config.nb_events_limit = dev_info.max_num_events;
186 	if (dev_info.max_event_port_dequeue_depth <
187 			config.nb_event_port_dequeue_depth)
188 		config.nb_event_port_dequeue_depth =
189 				dev_info.max_event_port_dequeue_depth;
190 	if (dev_info.max_event_port_enqueue_depth <
191 			config.nb_event_port_enqueue_depth)
192 		config.nb_event_port_enqueue_depth =
193 				dev_info.max_event_port_enqueue_depth;
194 
195 	ret = rte_event_dev_configure(dev_id, &config);
196 	if (ret < 0) {
197 		printf("%d: Error configuring device\n", __LINE__);
198 		return -1;
199 	}
200 
201 	/* Q creation - one load balanced per pipeline stage*/
202 	printf("  Stages:\n");
203 	for (i = 0; i < cdata.num_stages; i++) {
204 		if (rte_event_queue_setup(dev_id, i, &wkr_q_conf) < 0) {
205 			printf("%d: error creating qid %d\n", __LINE__, i);
206 			return -1;
207 		}
208 		cdata.qid[i] = i;
209 		cdata.next_qid[i] = i+1;
210 		worker_queues[i].queue_id = i;
211 		if (cdata.enable_queue_priorities) {
212 			/* calculate priority stepping for each stage, leaving
213 			 * headroom of 1 for the SINGLE_LINK TX below
214 			 */
215 			const uint32_t prio_delta =
216 				(RTE_EVENT_DEV_PRIORITY_LOWEST-1) /  nb_queues;
217 
218 			/* higher priority for queues closer to tx */
219 			wkr_q_conf.priority =
220 				RTE_EVENT_DEV_PRIORITY_LOWEST - prio_delta * i;
221 		}
222 
223 		const char *type_str = "Atomic";
224 		switch (wkr_q_conf.schedule_type) {
225 		case RTE_SCHED_TYPE_ORDERED:
226 			type_str = "Ordered";
227 			break;
228 		case RTE_SCHED_TYPE_PARALLEL:
229 			type_str = "Parallel";
230 			break;
231 		}
232 		printf("\tStage %d, Type %s\tPriority = %d\n", i, type_str,
233 				wkr_q_conf.priority);
234 	}
235 	printf("\n");
236 
237 	/* final queue for sending to TX core */
238 	if (rte_event_queue_setup(dev_id, i, &tx_q_conf) < 0) {
239 		printf("%d: error creating qid %d\n", __LINE__, i);
240 		return -1;
241 	}
242 	cdata.tx_queue_id = i;
243 
244 	if (wkr_p_conf.new_event_threshold > config.nb_events_limit)
245 		wkr_p_conf.new_event_threshold = config.nb_events_limit;
246 	if (wkr_p_conf.dequeue_depth > config.nb_event_port_dequeue_depth)
247 		wkr_p_conf.dequeue_depth = config.nb_event_port_dequeue_depth;
248 	if (wkr_p_conf.enqueue_depth > config.nb_event_port_enqueue_depth)
249 		wkr_p_conf.enqueue_depth = config.nb_event_port_enqueue_depth;
250 
251 	/* set up one port per worker, linking to all stage queues */
252 	for (i = 0; i < cdata.num_workers; i++) {
253 		struct worker_data *w = &worker_data[i];
254 		w->dev_id = dev_id;
255 		if (rte_event_port_setup(dev_id, i, &wkr_p_conf) < 0) {
256 			printf("Error setting up port %d\n", i);
257 			return -1;
258 		}
259 
260 		uint32_t s;
261 		for (s = 0; s < cdata.num_stages; s++) {
262 			if (rte_event_port_link(dev_id, i,
263 						&worker_queues[s].queue_id,
264 						&worker_queues[s].priority,
265 						1) != 1) {
266 				printf("%d: error creating link for port %d\n",
267 						__LINE__, i);
268 				return -1;
269 			}
270 		}
271 		w->port_id = i;
272 	}
273 
274 	ret = rte_event_dev_service_id_get(dev_id,
275 				&fdata->evdev_service_id);
276 	if (ret != -ESRCH && ret != 0) {
277 		printf("Error getting the service ID for sw eventdev\n");
278 		return -1;
279 	}
280 	rte_service_runstate_set(fdata->evdev_service_id, 1);
281 	rte_service_set_runstate_mapped_check(fdata->evdev_service_id, 0);
282 
283 	return dev_id;
284 }
285 
286 /*
287  * Initializes a given port using global settings and with the RX buffers
288  * coming from the mbuf_pool passed as a parameter.
289  */
290 static inline int
291 port_init(uint8_t port, struct rte_mempool *mbuf_pool)
292 {
293 	struct rte_eth_rxconf rx_conf;
294 	static const struct rte_eth_conf port_conf_default = {
295 		.rxmode = {
296 			.mq_mode = RTE_ETH_MQ_RX_RSS,
297 		},
298 		.rx_adv_conf = {
299 			.rss_conf = {
300 				.rss_hf = RTE_ETH_RSS_IP |
301 					  RTE_ETH_RSS_TCP |
302 					  RTE_ETH_RSS_UDP,
303 			}
304 		}
305 	};
306 	const uint16_t rx_rings = 1, tx_rings = 1;
307 	const uint16_t rx_ring_size = 512, tx_ring_size = 512;
308 	struct rte_eth_conf port_conf = port_conf_default;
309 	int retval;
310 	uint16_t q;
311 	struct rte_eth_dev_info dev_info;
312 	struct rte_eth_txconf txconf;
313 
314 	if (!rte_eth_dev_is_valid_port(port))
315 		return -1;
316 
317 	retval = rte_eth_dev_info_get(port, &dev_info);
318 	if (retval != 0) {
319 		printf("Error during getting device (port %u) info: %s\n",
320 				port, strerror(-retval));
321 		return retval;
322 	}
323 
324 	if (dev_info.tx_offload_capa & RTE_ETH_TX_OFFLOAD_MBUF_FAST_FREE)
325 		port_conf.txmode.offloads |=
326 			RTE_ETH_TX_OFFLOAD_MBUF_FAST_FREE;
327 
328 	if (dev_info.rx_offload_capa & RTE_ETH_RX_OFFLOAD_RSS_HASH)
329 		port_conf.rxmode.offloads |= RTE_ETH_RX_OFFLOAD_RSS_HASH;
330 
331 	rx_conf = dev_info.default_rxconf;
332 	rx_conf.offloads = port_conf.rxmode.offloads;
333 
334 	port_conf.rx_adv_conf.rss_conf.rss_hf &=
335 		dev_info.flow_type_rss_offloads;
336 	if (port_conf.rx_adv_conf.rss_conf.rss_hf !=
337 			port_conf_default.rx_adv_conf.rss_conf.rss_hf) {
338 		printf("Port %u modified RSS hash function based on hardware support,"
339 			"requested:%#"PRIx64" configured:%#"PRIx64"\n",
340 			port,
341 			port_conf_default.rx_adv_conf.rss_conf.rss_hf,
342 			port_conf.rx_adv_conf.rss_conf.rss_hf);
343 	}
344 
345 	/* Configure the Ethernet device. */
346 	retval = rte_eth_dev_configure(port, rx_rings, tx_rings, &port_conf);
347 	if (retval != 0)
348 		return retval;
349 
350 	/* Allocate and set up 1 RX queue per Ethernet port. */
351 	for (q = 0; q < rx_rings; q++) {
352 		retval = rte_eth_rx_queue_setup(port, q, rx_ring_size,
353 				rte_eth_dev_socket_id(port), &rx_conf,
354 				mbuf_pool);
355 		if (retval < 0)
356 			return retval;
357 	}
358 
359 	txconf = dev_info.default_txconf;
360 	txconf.offloads = port_conf_default.txmode.offloads;
361 	/* Allocate and set up 1 TX queue per Ethernet port. */
362 	for (q = 0; q < tx_rings; q++) {
363 		retval = rte_eth_tx_queue_setup(port, q, tx_ring_size,
364 				rte_eth_dev_socket_id(port), &txconf);
365 		if (retval < 0)
366 			return retval;
367 	}
368 
369 	/* Display the port MAC address. */
370 	struct rte_ether_addr addr;
371 	retval = rte_eth_macaddr_get(port, &addr);
372 	if (retval != 0) {
373 		printf("Failed to get MAC address (port %u): %s\n",
374 				port, rte_strerror(-retval));
375 		return retval;
376 	}
377 
378 	printf("Port %u MAC: %02" PRIx8 " %02" PRIx8 " %02" PRIx8
379 			" %02" PRIx8 " %02" PRIx8 " %02" PRIx8 "\n",
380 			(unsigned int)port, RTE_ETHER_ADDR_BYTES(&addr));
381 
382 	/* Enable RX in promiscuous mode for the Ethernet device. */
383 	retval = rte_eth_promiscuous_enable(port);
384 	if (retval != 0)
385 		return retval;
386 
387 	return 0;
388 }
389 
390 static int
391 init_ports(uint16_t num_ports)
392 {
393 	uint16_t portid;
394 
395 	if (!cdata.num_mbuf)
396 		cdata.num_mbuf = 16384 * num_ports;
397 
398 	struct rte_mempool *mp = rte_pktmbuf_pool_create("packet_pool",
399 			/* mbufs */ cdata.num_mbuf,
400 			/* cache_size */ 512,
401 			/* priv_size*/ 0,
402 			/* data_room_size */ RTE_MBUF_DEFAULT_BUF_SIZE,
403 			rte_socket_id());
404 
405 	RTE_ETH_FOREACH_DEV(portid)
406 		if (port_init(portid, mp) != 0)
407 			rte_exit(EXIT_FAILURE, "Cannot init port %"PRIu16 "\n",
408 					portid);
409 
410 	return 0;
411 }
412 
413 static void
414 init_adapters(uint16_t nb_ports)
415 {
416 	int i;
417 	int ret;
418 	uint8_t tx_port_id = 0;
419 	uint8_t evdev_id = 0;
420 	struct rte_event_dev_info dev_info;
421 
422 	ret = rte_event_dev_info_get(evdev_id, &dev_info);
423 
424 	struct rte_event_port_conf adptr_p_conf = {
425 		.dequeue_depth = cdata.worker_cq_depth,
426 		.enqueue_depth = 64,
427 		.new_event_threshold = 4096,
428 		.event_port_cfg = RTE_EVENT_PORT_CFG_HINT_PRODUCER,
429 	};
430 
431 	if (adptr_p_conf.new_event_threshold > dev_info.max_num_events)
432 		adptr_p_conf.new_event_threshold = dev_info.max_num_events;
433 	if (adptr_p_conf.dequeue_depth > dev_info.max_event_port_dequeue_depth)
434 		adptr_p_conf.dequeue_depth =
435 			dev_info.max_event_port_dequeue_depth;
436 	if (adptr_p_conf.enqueue_depth > dev_info.max_event_port_enqueue_depth)
437 		adptr_p_conf.enqueue_depth =
438 			dev_info.max_event_port_enqueue_depth;
439 
440 	init_ports(nb_ports);
441 	/* Create one adapter for all the ethernet ports. */
442 	ret = rte_event_eth_rx_adapter_create(cdata.rx_adapter_id, evdev_id,
443 			&adptr_p_conf);
444 	if (ret)
445 		rte_exit(EXIT_FAILURE, "failed to create rx adapter[%d]",
446 				cdata.rx_adapter_id);
447 
448 	ret = rte_event_eth_tx_adapter_create(cdata.tx_adapter_id, evdev_id,
449 			&adptr_p_conf);
450 	if (ret)
451 		rte_exit(EXIT_FAILURE, "failed to create tx adapter[%d]",
452 				cdata.tx_adapter_id);
453 
454 	struct rte_event_eth_rx_adapter_queue_conf queue_conf;
455 	memset(&queue_conf, 0, sizeof(queue_conf));
456 	queue_conf.ev.sched_type = cdata.queue_type;
457 	queue_conf.ev.queue_id = cdata.qid[0];
458 
459 	for (i = 0; i < nb_ports; i++) {
460 		ret = rte_event_eth_rx_adapter_queue_add(cdata.rx_adapter_id, i,
461 				-1, &queue_conf);
462 		if (ret)
463 			rte_exit(EXIT_FAILURE,
464 					"Failed to add queues to Rx adapter");
465 
466 		ret = rte_event_eth_tx_adapter_queue_add(cdata.tx_adapter_id, i,
467 				-1);
468 		if (ret)
469 			rte_exit(EXIT_FAILURE,
470 					"Failed to add queues to Tx adapter");
471 	}
472 
473 	ret = rte_event_eth_tx_adapter_event_port_get(cdata.tx_adapter_id,
474 			&tx_port_id);
475 	if (ret)
476 		rte_exit(EXIT_FAILURE,
477 				"Failed to get Tx adapter port id");
478 	ret = rte_event_port_link(evdev_id, tx_port_id, &cdata.tx_queue_id,
479 			NULL, 1);
480 	if (ret != 1)
481 		rte_exit(EXIT_FAILURE,
482 				"Unable to link Tx adapter port to Tx queue");
483 
484 	ret = rte_event_eth_rx_adapter_service_id_get(cdata.rx_adapter_id,
485 				&fdata->rxadptr_service_id);
486 	if (ret != -ESRCH && ret != 0) {
487 		rte_exit(EXIT_FAILURE,
488 			"Error getting the service ID for Rx adapter\n");
489 	}
490 	rte_service_runstate_set(fdata->rxadptr_service_id, 1);
491 	rte_service_set_runstate_mapped_check(fdata->rxadptr_service_id, 0);
492 
493 	ret = rte_event_eth_tx_adapter_service_id_get(cdata.tx_adapter_id,
494 				&fdata->txadptr_service_id);
495 	if (ret != -ESRCH && ret != 0) {
496 		rte_exit(EXIT_FAILURE,
497 			"Error getting the service ID for Tx adapter\n");
498 	}
499 	rte_service_runstate_set(fdata->txadptr_service_id, 1);
500 	rte_service_set_runstate_mapped_check(fdata->txadptr_service_id, 0);
501 
502 	ret = rte_event_eth_rx_adapter_start(cdata.rx_adapter_id);
503 	if (ret)
504 		rte_exit(EXIT_FAILURE, "Rx adapter[%d] start failed",
505 				cdata.rx_adapter_id);
506 
507 	ret = rte_event_eth_tx_adapter_start(cdata.tx_adapter_id);
508 	if (ret)
509 		rte_exit(EXIT_FAILURE, "Tx adapter[%d] start failed",
510 				cdata.tx_adapter_id);
511 
512 	if (rte_event_dev_start(evdev_id) < 0)
513 		rte_exit(EXIT_FAILURE, "Error starting eventdev");
514 }
515 
516 static void
517 generic_opt_check(void)
518 {
519 	int i;
520 	int ret;
521 	uint32_t cap = 0;
522 	uint8_t rx_needed = 0;
523 	uint8_t sched_needed = 0;
524 	struct rte_event_dev_info eventdev_info;
525 
526 	memset(&eventdev_info, 0, sizeof(struct rte_event_dev_info));
527 	rte_event_dev_info_get(0, &eventdev_info);
528 
529 	if (cdata.all_type_queues && !(eventdev_info.event_dev_cap &
530 				RTE_EVENT_DEV_CAP_QUEUE_ALL_TYPES))
531 		rte_exit(EXIT_FAILURE,
532 				"Event dev doesn't support all type queues\n");
533 	sched_needed = !(eventdev_info.event_dev_cap &
534 		RTE_EVENT_DEV_CAP_DISTRIBUTED_SCHED);
535 
536 	RTE_ETH_FOREACH_DEV(i) {
537 		ret = rte_event_eth_rx_adapter_caps_get(0, i, &cap);
538 		if (ret)
539 			rte_exit(EXIT_FAILURE,
540 				"failed to get event rx adapter capabilities");
541 		rx_needed |=
542 			!(cap & RTE_EVENT_ETH_RX_ADAPTER_CAP_INTERNAL_PORT);
543 	}
544 
545 	if (cdata.worker_lcore_mask == 0 ||
546 			(rx_needed && cdata.rx_lcore_mask == 0) ||
547 			(cdata.tx_lcore_mask == 0) ||
548 			(sched_needed && cdata.sched_lcore_mask == 0)) {
549 		printf("Core part of pipeline was not assigned any cores. "
550 			"This will stall the pipeline, please check core masks "
551 			"(use -h for details on setting core masks):\n"
552 			"\trx: %"PRIu64"\n\ttx: %"PRIu64"\n\tsched: %"PRIu64
553 			"\n\tworkers: %"PRIu64"\n",
554 			cdata.rx_lcore_mask, cdata.tx_lcore_mask,
555 			cdata.sched_lcore_mask,
556 			cdata.worker_lcore_mask);
557 		rte_exit(-1, "Fix core masks\n");
558 	}
559 
560 	if (!sched_needed)
561 		memset(fdata->sched_core, 0,
562 				sizeof(unsigned int) * MAX_NUM_CORE);
563 	if (!rx_needed)
564 		memset(fdata->rx_core, 0,
565 				sizeof(unsigned int) * MAX_NUM_CORE);
566 }
567 
568 void
569 set_worker_generic_setup_data(struct setup_data *caps, bool burst)
570 {
571 	if (burst) {
572 		caps->worker = worker_generic_burst;
573 	} else {
574 		caps->worker = worker_generic;
575 	}
576 
577 	caps->adptr_setup = init_adapters;
578 	caps->scheduler = schedule_devices;
579 	caps->evdev_setup = setup_eventdev_generic;
580 	caps->check_opt = generic_opt_check;
581 }
582