xref: /dpdk/lib/port/rte_port_ring.c (revision e9fd1ebf981f361844aea9ec94e17f4bda5e1479) 
1 /* SPDX-License-Identifier: BSD-3-Clause
2  * Copyright(c) 2010-2014 Intel Corporation
3  */
4 #include <string.h>
5 #include <stdint.h>
6 
7 #include <rte_mbuf.h>
8 #include <rte_ring.h>
9 #include <rte_malloc.h>
10 
11 #include "rte_port_ring.h"
12 
13 #include "port_log.h"
14 
15 /*
16  * Port RING Reader
17  */
18 #ifdef RTE_PORT_STATS_COLLECT
19 
20 #define RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(port, val) \
21 	port->stats.n_pkts_in += val
22 #define RTE_PORT_RING_READER_STATS_PKTS_DROP_ADD(port, val) \
23 	port->stats.n_pkts_drop += val
24 
25 #else
26 
27 #define RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(port, val)
28 #define RTE_PORT_RING_READER_STATS_PKTS_DROP_ADD(port, val)
29 
30 #endif
31 
32 struct rte_port_ring_reader {
33 	struct rte_port_in_stats stats;
34 
35 	struct rte_ring *ring;
36 };
37 
38 static void *
39 rte_port_ring_reader_create_internal(void *params, int socket_id,
40 	uint32_t is_multi)
41 {
42 	struct rte_port_ring_reader_params *conf =
43 			params;
44 	struct rte_port_ring_reader *port;
45 
46 	/* Check input parameters */
47 	if ((conf == NULL) ||
48 		(conf->ring == NULL) ||
49 		(rte_ring_is_cons_single(conf->ring) && is_multi) ||
50 		(!rte_ring_is_cons_single(conf->ring) && !is_multi)) {
51 		PORT_LOG(ERR, "%s: Invalid Parameters", __func__);
52 		return NULL;
53 	}
54 
55 	/* Memory allocation */
56 	port = rte_zmalloc_socket("PORT", sizeof(*port),
57 			RTE_CACHE_LINE_SIZE, socket_id);
58 	if (port == NULL) {
59 		PORT_LOG(ERR, "%s: Failed to allocate port", __func__);
60 		return NULL;
61 	}
62 
63 	/* Initialization */
64 	port->ring = conf->ring;
65 
66 	return port;
67 }
68 
69 static void *
70 rte_port_ring_reader_create(void *params, int socket_id)
71 {
72 	return rte_port_ring_reader_create_internal(params, socket_id, 0);
73 }
74 
75 static void *
76 rte_port_ring_multi_reader_create(void *params, int socket_id)
77 {
78 	return rte_port_ring_reader_create_internal(params, socket_id, 1);
79 }
80 
81 static int
82 rte_port_ring_reader_rx(void *port, struct rte_mbuf **pkts, uint32_t n_pkts)
83 {
84 	struct rte_port_ring_reader *p = port;
85 	uint32_t nb_rx;
86 
87 	nb_rx = rte_ring_sc_dequeue_burst(p->ring, (void **) pkts,
88 			n_pkts, NULL);
89 	RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx);
90 
91 	return nb_rx;
92 }
93 
94 static int
95 rte_port_ring_multi_reader_rx(void *port, struct rte_mbuf **pkts,
96 	uint32_t n_pkts)
97 {
98 	struct rte_port_ring_reader *p = port;
99 	uint32_t nb_rx;
100 
101 	nb_rx = rte_ring_mc_dequeue_burst(p->ring, (void **) pkts,
102 			n_pkts, NULL);
103 	RTE_PORT_RING_READER_STATS_PKTS_IN_ADD(p, nb_rx);
104 
105 	return nb_rx;
106 }
107 
108 static int
109 rte_port_ring_reader_free(void *port)
110 {
111 	if (port == NULL) {
112 		PORT_LOG(ERR, "%s: port is NULL", __func__);
113 		return -EINVAL;
114 	}
115 
116 	rte_free(port);
117 
118 	return 0;
119 }
120 
121 static int
122 rte_port_ring_reader_stats_read(void *port,
123 		struct rte_port_in_stats *stats, int clear)
124 {
125 	struct rte_port_ring_reader *p =
126 		port;
127 
128 	if (stats != NULL)
129 		memcpy(stats, &p->stats, sizeof(p->stats));
130 
131 	if (clear)
132 		memset(&p->stats, 0, sizeof(p->stats));
133 
134 	return 0;
135 }
136 
137 /*
138  * Port RING Writer
139  */
140 #ifdef RTE_PORT_STATS_COLLECT
141 
142 #define RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(port, val) \
143 	port->stats.n_pkts_in += val
144 #define RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(port, val) \
145 	port->stats.n_pkts_drop += val
146 
147 #else
148 
149 #define RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(port, val)
150 #define RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(port, val)
151 
152 #endif
153 
154 struct rte_port_ring_writer {
155 	struct rte_port_out_stats stats;
156 
157 	struct rte_mbuf *tx_buf[2 * RTE_PORT_IN_BURST_SIZE_MAX];
158 	struct rte_ring *ring;
159 	uint32_t tx_burst_sz;
160 	uint32_t tx_buf_count;
161 	uint64_t bsz_mask;
162 	uint32_t is_multi;
163 };
164 
165 static void *
166 rte_port_ring_writer_create_internal(void *params, int socket_id,
167 	uint32_t is_multi)
168 {
169 	struct rte_port_ring_writer_params *conf =
170 			params;
171 	struct rte_port_ring_writer *port;
172 
173 	/* Check input parameters */
174 	if ((conf == NULL) ||
175 		(conf->ring == NULL) ||
176 		(rte_ring_is_prod_single(conf->ring) && is_multi) ||
177 		(!rte_ring_is_prod_single(conf->ring) && !is_multi) ||
178 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
179 		PORT_LOG(ERR, "%s: Invalid Parameters", __func__);
180 		return NULL;
181 	}
182 
183 	/* Memory allocation */
184 	port = rte_zmalloc_socket("PORT", sizeof(*port),
185 			RTE_CACHE_LINE_SIZE, socket_id);
186 	if (port == NULL) {
187 		PORT_LOG(ERR, "%s: Failed to allocate port", __func__);
188 		return NULL;
189 	}
190 
191 	/* Initialization */
192 	port->ring = conf->ring;
193 	port->tx_burst_sz = conf->tx_burst_sz;
194 	port->tx_buf_count = 0;
195 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
196 	port->is_multi = is_multi;
197 
198 	return port;
199 }
200 
201 static void *
202 rte_port_ring_writer_create(void *params, int socket_id)
203 {
204 	return rte_port_ring_writer_create_internal(params, socket_id, 0);
205 }
206 
207 static void *
208 rte_port_ring_multi_writer_create(void *params, int socket_id)
209 {
210 	return rte_port_ring_writer_create_internal(params, socket_id, 1);
211 }
212 
213 static inline void
214 send_burst(struct rte_port_ring_writer *p)
215 {
216 	uint32_t nb_tx;
217 
218 	nb_tx = rte_ring_sp_enqueue_burst(p->ring, (void **)p->tx_buf,
219 			p->tx_buf_count, NULL);
220 
221 	RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
222 	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
223 		rte_pktmbuf_free(p->tx_buf[nb_tx]);
224 
225 	p->tx_buf_count = 0;
226 }
227 
228 static inline void
229 send_burst_mp(struct rte_port_ring_writer *p)
230 {
231 	uint32_t nb_tx;
232 
233 	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
234 			p->tx_buf_count, NULL);
235 
236 	RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
237 	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
238 		rte_pktmbuf_free(p->tx_buf[nb_tx]);
239 
240 	p->tx_buf_count = 0;
241 }
242 
243 static int
244 rte_port_ring_writer_tx(void *port, struct rte_mbuf *pkt)
245 {
246 	struct rte_port_ring_writer *p = port;
247 
248 	p->tx_buf[p->tx_buf_count++] = pkt;
249 	RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
250 	if (p->tx_buf_count >= p->tx_burst_sz)
251 		send_burst(p);
252 
253 	return 0;
254 }
255 
256 static int
257 rte_port_ring_multi_writer_tx(void *port, struct rte_mbuf *pkt)
258 {
259 	struct rte_port_ring_writer *p = port;
260 
261 	p->tx_buf[p->tx_buf_count++] = pkt;
262 	RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
263 	if (p->tx_buf_count >= p->tx_burst_sz)
264 		send_burst_mp(p);
265 
266 	return 0;
267 }
268 
269 static __rte_always_inline int
270 rte_port_ring_writer_tx_bulk_internal(void *port,
271 		struct rte_mbuf **pkts,
272 		uint64_t pkts_mask,
273 		uint32_t is_multi)
274 {
275 	struct rte_port_ring_writer *p =
276 		port;
277 
278 	uint64_t bsz_mask = p->bsz_mask;
279 	uint32_t tx_buf_count = p->tx_buf_count;
280 	uint64_t expr = (pkts_mask & (pkts_mask + 1)) |
281 			((pkts_mask & bsz_mask) ^ bsz_mask);
282 
283 	if (expr == 0) {
284 		uint64_t n_pkts = rte_popcount64(pkts_mask);
285 		uint32_t n_pkts_ok;
286 
287 		if (tx_buf_count) {
288 			if (is_multi)
289 				send_burst_mp(p);
290 			else
291 				send_burst(p);
292 		}
293 
294 		RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, n_pkts);
295 		if (is_multi)
296 			n_pkts_ok = rte_ring_mp_enqueue_burst(p->ring,
297 					(void **)pkts, n_pkts, NULL);
298 		else
299 			n_pkts_ok = rte_ring_sp_enqueue_burst(p->ring,
300 					(void **)pkts, n_pkts, NULL);
301 
302 		RTE_PORT_RING_WRITER_STATS_PKTS_DROP_ADD(p, n_pkts - n_pkts_ok);
303 		for ( ; n_pkts_ok < n_pkts; n_pkts_ok++) {
304 			struct rte_mbuf *pkt = pkts[n_pkts_ok];
305 
306 			rte_pktmbuf_free(pkt);
307 		}
308 	} else {
309 		for ( ; pkts_mask; ) {
310 			uint32_t pkt_index = rte_ctz64(pkts_mask);
311 			uint64_t pkt_mask = 1LLU << pkt_index;
312 			struct rte_mbuf *pkt = pkts[pkt_index];
313 
314 			p->tx_buf[tx_buf_count++] = pkt;
315 			RTE_PORT_RING_WRITER_STATS_PKTS_IN_ADD(p, 1);
316 			pkts_mask &= ~pkt_mask;
317 		}
318 
319 		p->tx_buf_count = tx_buf_count;
320 		if (tx_buf_count >= p->tx_burst_sz) {
321 			if (is_multi)
322 				send_burst_mp(p);
323 			else
324 				send_burst(p);
325 		}
326 	}
327 
328 	return 0;
329 }
330 
331 static int
332 rte_port_ring_writer_tx_bulk(void *port,
333 		struct rte_mbuf **pkts,
334 		uint64_t pkts_mask)
335 {
336 	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 0);
337 }
338 
339 static int
340 rte_port_ring_multi_writer_tx_bulk(void *port,
341 		struct rte_mbuf **pkts,
342 		uint64_t pkts_mask)
343 {
344 	return rte_port_ring_writer_tx_bulk_internal(port, pkts, pkts_mask, 1);
345 }
346 
347 static int
348 rte_port_ring_writer_flush(void *port)
349 {
350 	struct rte_port_ring_writer *p = port;
351 
352 	if (p->tx_buf_count > 0)
353 		send_burst(p);
354 
355 	return 0;
356 }
357 
358 static int
359 rte_port_ring_multi_writer_flush(void *port)
360 {
361 	struct rte_port_ring_writer *p = port;
362 
363 	if (p->tx_buf_count > 0)
364 		send_burst_mp(p);
365 
366 	return 0;
367 }
368 
369 static int
370 rte_port_ring_writer_free(void *port)
371 {
372 	struct rte_port_ring_writer *p = port;
373 
374 	if (port == NULL) {
375 		PORT_LOG(ERR, "%s: Port is NULL", __func__);
376 		return -EINVAL;
377 	}
378 
379 	if (p->is_multi)
380 		rte_port_ring_multi_writer_flush(port);
381 	else
382 		rte_port_ring_writer_flush(port);
383 
384 	rte_free(port);
385 
386 	return 0;
387 }
388 
389 static int
390 rte_port_ring_writer_stats_read(void *port,
391 		struct rte_port_out_stats *stats, int clear)
392 {
393 	struct rte_port_ring_writer *p =
394 		port;
395 
396 	if (stats != NULL)
397 		memcpy(stats, &p->stats, sizeof(p->stats));
398 
399 	if (clear)
400 		memset(&p->stats, 0, sizeof(p->stats));
401 
402 	return 0;
403 }
404 
405 /*
406  * Port RING Writer Nodrop
407  */
408 #ifdef RTE_PORT_STATS_COLLECT
409 
410 #define RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(port, val) \
411 	port->stats.n_pkts_in += val
412 #define RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(port, val) \
413 	port->stats.n_pkts_drop += val
414 
415 #else
416 
417 #define RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(port, val)
418 #define RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(port, val)
419 
420 #endif
421 
422 struct rte_port_ring_writer_nodrop {
423 	struct rte_port_out_stats stats;
424 
425 	struct rte_mbuf *tx_buf[2 * RTE_PORT_IN_BURST_SIZE_MAX];
426 	struct rte_ring *ring;
427 	uint32_t tx_burst_sz;
428 	uint32_t tx_buf_count;
429 	uint64_t bsz_mask;
430 	uint64_t n_retries;
431 	uint32_t is_multi;
432 };
433 
434 static void *
435 rte_port_ring_writer_nodrop_create_internal(void *params, int socket_id,
436 	uint32_t is_multi)
437 {
438 	struct rte_port_ring_writer_nodrop_params *conf =
439 			params;
440 	struct rte_port_ring_writer_nodrop *port;
441 
442 	/* Check input parameters */
443 	if ((conf == NULL) ||
444 		(conf->ring == NULL) ||
445 		(rte_ring_is_prod_single(conf->ring) && is_multi) ||
446 		(!rte_ring_is_prod_single(conf->ring) && !is_multi) ||
447 		(conf->tx_burst_sz > RTE_PORT_IN_BURST_SIZE_MAX)) {
448 		PORT_LOG(ERR, "%s: Invalid Parameters", __func__);
449 		return NULL;
450 	}
451 
452 	/* Memory allocation */
453 	port = rte_zmalloc_socket("PORT", sizeof(*port),
454 			RTE_CACHE_LINE_SIZE, socket_id);
455 	if (port == NULL) {
456 		PORT_LOG(ERR, "%s: Failed to allocate port", __func__);
457 		return NULL;
458 	}
459 
460 	/* Initialization */
461 	port->ring = conf->ring;
462 	port->tx_burst_sz = conf->tx_burst_sz;
463 	port->tx_buf_count = 0;
464 	port->bsz_mask = 1LLU << (conf->tx_burst_sz - 1);
465 	port->is_multi = is_multi;
466 
467 	/*
468 	 * When n_retries is 0 it means that we should wait for every packet to
469 	 * send no matter how many retries should it take. To limit number of
470 	 * branches in fast path, we use UINT64_MAX instead of branching.
471 	 */
472 	port->n_retries = (conf->n_retries == 0) ? UINT64_MAX : conf->n_retries;
473 
474 	return port;
475 }
476 
477 static void *
478 rte_port_ring_writer_nodrop_create(void *params, int socket_id)
479 {
480 	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 0);
481 }
482 
483 static void *
484 rte_port_ring_multi_writer_nodrop_create(void *params, int socket_id)
485 {
486 	return rte_port_ring_writer_nodrop_create_internal(params, socket_id, 1);
487 }
488 
489 static inline void
490 send_burst_nodrop(struct rte_port_ring_writer_nodrop *p)
491 {
492 	uint32_t nb_tx = 0, i;
493 
494 	nb_tx = rte_ring_sp_enqueue_burst(p->ring, (void **)p->tx_buf,
495 				p->tx_buf_count, NULL);
496 
497 	/* We sent all the packets in a first try */
498 	if (nb_tx >= p->tx_buf_count) {
499 		p->tx_buf_count = 0;
500 		return;
501 	}
502 
503 	for (i = 0; i < p->n_retries; i++) {
504 		nb_tx += rte_ring_sp_enqueue_burst(p->ring,
505 				(void **) (p->tx_buf + nb_tx),
506 				p->tx_buf_count - nb_tx, NULL);
507 
508 		/* We sent all the packets in more than one try */
509 		if (nb_tx >= p->tx_buf_count) {
510 			p->tx_buf_count = 0;
511 			return;
512 		}
513 	}
514 
515 	/* We didn't send the packets in maximum allowed attempts */
516 	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
517 	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
518 		rte_pktmbuf_free(p->tx_buf[nb_tx]);
519 
520 	p->tx_buf_count = 0;
521 }
522 
523 static inline void
524 send_burst_mp_nodrop(struct rte_port_ring_writer_nodrop *p)
525 {
526 	uint32_t nb_tx = 0, i;
527 
528 	nb_tx = rte_ring_mp_enqueue_burst(p->ring, (void **)p->tx_buf,
529 				p->tx_buf_count, NULL);
530 
531 	/* We sent all the packets in a first try */
532 	if (nb_tx >= p->tx_buf_count) {
533 		p->tx_buf_count = 0;
534 		return;
535 	}
536 
537 	for (i = 0; i < p->n_retries; i++) {
538 		nb_tx += rte_ring_mp_enqueue_burst(p->ring,
539 				(void **) (p->tx_buf + nb_tx),
540 				p->tx_buf_count - nb_tx, NULL);
541 
542 		/* We sent all the packets in more than one try */
543 		if (nb_tx >= p->tx_buf_count) {
544 			p->tx_buf_count = 0;
545 			return;
546 		}
547 	}
548 
549 	/* We didn't send the packets in maximum allowed attempts */
550 	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_DROP_ADD(p, p->tx_buf_count - nb_tx);
551 	for ( ; nb_tx < p->tx_buf_count; nb_tx++)
552 		rte_pktmbuf_free(p->tx_buf[nb_tx]);
553 
554 	p->tx_buf_count = 0;
555 }
556 
557 static int
558 rte_port_ring_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
559 {
560 	struct rte_port_ring_writer_nodrop *p =
561 			port;
562 
563 	p->tx_buf[p->tx_buf_count++] = pkt;
564 	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
565 	if (p->tx_buf_count >= p->tx_burst_sz)
566 		send_burst_nodrop(p);
567 
568 	return 0;
569 }
570 
571 static int
572 rte_port_ring_multi_writer_nodrop_tx(void *port, struct rte_mbuf *pkt)
573 {
574 	struct rte_port_ring_writer_nodrop *p =
575 			port;
576 
577 	p->tx_buf[p->tx_buf_count++] = pkt;
578 	RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
579 	if (p->tx_buf_count >= p->tx_burst_sz)
580 		send_burst_mp_nodrop(p);
581 
582 	return 0;
583 }
584 
585 static __rte_always_inline int
586 rte_port_ring_writer_nodrop_tx_bulk_internal(void *port,
587 		struct rte_mbuf **pkts,
588 		uint64_t pkts_mask,
589 		uint32_t is_multi)
590 {
591 	struct rte_port_ring_writer_nodrop *p =
592 		port;
593 
594 	uint64_t bsz_mask = p->bsz_mask;
595 	uint32_t tx_buf_count = p->tx_buf_count;
596 	uint64_t expr = (pkts_mask & (pkts_mask + 1)) |
597 			((pkts_mask & bsz_mask) ^ bsz_mask);
598 
599 	if (expr == 0) {
600 		uint64_t n_pkts = rte_popcount64(pkts_mask);
601 		uint32_t n_pkts_ok;
602 
603 		if (tx_buf_count) {
604 			if (is_multi)
605 				send_burst_mp_nodrop(p);
606 			else
607 				send_burst_nodrop(p);
608 		}
609 
610 		RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, n_pkts);
611 		if (is_multi)
612 			n_pkts_ok =
613 				rte_ring_mp_enqueue_burst(p->ring,
614 						(void **)pkts, n_pkts, NULL);
615 		else
616 			n_pkts_ok =
617 				rte_ring_sp_enqueue_burst(p->ring,
618 						(void **)pkts, n_pkts, NULL);
619 
620 		if (n_pkts_ok >= n_pkts)
621 			return 0;
622 
623 		/*
624 		 * If we didn't manage to send all packets in single burst, move
625 		 * remaining packets to the buffer and call send burst.
626 		 */
627 		for (; n_pkts_ok < n_pkts; n_pkts_ok++) {
628 			struct rte_mbuf *pkt = pkts[n_pkts_ok];
629 
630 			p->tx_buf[p->tx_buf_count++] = pkt;
631 		}
632 		if (is_multi)
633 			send_burst_mp_nodrop(p);
634 		else
635 			send_burst_nodrop(p);
636 	} else {
637 		for ( ; pkts_mask; ) {
638 			uint32_t pkt_index = rte_ctz64(pkts_mask);
639 			uint64_t pkt_mask = 1LLU << pkt_index;
640 			struct rte_mbuf *pkt = pkts[pkt_index];
641 
642 			p->tx_buf[tx_buf_count++] = pkt;
643 			RTE_PORT_RING_WRITER_NODROP_STATS_PKTS_IN_ADD(p, 1);
644 			pkts_mask &= ~pkt_mask;
645 		}
646 
647 		p->tx_buf_count = tx_buf_count;
648 		if (tx_buf_count >= p->tx_burst_sz) {
649 			if (is_multi)
650 				send_burst_mp_nodrop(p);
651 			else
652 				send_burst_nodrop(p);
653 		}
654 	}
655 
656 	return 0;
657 }
658 
659 static int
660 rte_port_ring_writer_nodrop_tx_bulk(void *port,
661 		struct rte_mbuf **pkts,
662 		uint64_t pkts_mask)
663 {
664 	return
665 		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 0);
666 }
667 
668 static int
669 rte_port_ring_multi_writer_nodrop_tx_bulk(void *port,
670 		struct rte_mbuf **pkts,
671 		uint64_t pkts_mask)
672 {
673 	return
674 		rte_port_ring_writer_nodrop_tx_bulk_internal(port, pkts, pkts_mask, 1);
675 }
676 
677 static int
678 rte_port_ring_writer_nodrop_flush(void *port)
679 {
680 	struct rte_port_ring_writer_nodrop *p =
681 			port;
682 
683 	if (p->tx_buf_count > 0)
684 		send_burst_nodrop(p);
685 
686 	return 0;
687 }
688 
689 static int
690 rte_port_ring_multi_writer_nodrop_flush(void *port)
691 {
692 	struct rte_port_ring_writer_nodrop *p =
693 			port;
694 
695 	if (p->tx_buf_count > 0)
696 		send_burst_mp_nodrop(p);
697 
698 	return 0;
699 }
700 
701 static int
702 rte_port_ring_writer_nodrop_free(void *port)
703 {
704 	struct rte_port_ring_writer_nodrop *p =
705 			port;
706 
707 	if (port == NULL) {
708 		PORT_LOG(ERR, "%s: Port is NULL", __func__);
709 		return -EINVAL;
710 	}
711 
712 	if (p->is_multi)
713 		rte_port_ring_multi_writer_nodrop_flush(port);
714 	else
715 		rte_port_ring_writer_nodrop_flush(port);
716 
717 	rte_free(port);
718 
719 	return 0;
720 }
721 
722 static int
723 rte_port_ring_writer_nodrop_stats_read(void *port,
724 		struct rte_port_out_stats *stats, int clear)
725 {
726 	struct rte_port_ring_writer_nodrop *p =
727 		port;
728 
729 	if (stats != NULL)
730 		memcpy(stats, &p->stats, sizeof(p->stats));
731 
732 	if (clear)
733 		memset(&p->stats, 0, sizeof(p->stats));
734 
735 	return 0;
736 }
737 
738 /*
739  * Summary of port operations
740  */
741 struct rte_port_in_ops rte_port_ring_reader_ops = {
742 	.f_create = rte_port_ring_reader_create,
743 	.f_free = rte_port_ring_reader_free,
744 	.f_rx = rte_port_ring_reader_rx,
745 	.f_stats = rte_port_ring_reader_stats_read,
746 };
747 
748 struct rte_port_out_ops rte_port_ring_writer_ops = {
749 	.f_create = rte_port_ring_writer_create,
750 	.f_free = rte_port_ring_writer_free,
751 	.f_tx = rte_port_ring_writer_tx,
752 	.f_tx_bulk = rte_port_ring_writer_tx_bulk,
753 	.f_flush = rte_port_ring_writer_flush,
754 	.f_stats = rte_port_ring_writer_stats_read,
755 };
756 
757 struct rte_port_out_ops rte_port_ring_writer_nodrop_ops = {
758 	.f_create = rte_port_ring_writer_nodrop_create,
759 	.f_free = rte_port_ring_writer_nodrop_free,
760 	.f_tx = rte_port_ring_writer_nodrop_tx,
761 	.f_tx_bulk = rte_port_ring_writer_nodrop_tx_bulk,
762 	.f_flush = rte_port_ring_writer_nodrop_flush,
763 	.f_stats = rte_port_ring_writer_nodrop_stats_read,
764 };
765 
766 struct rte_port_in_ops rte_port_ring_multi_reader_ops = {
767 	.f_create = rte_port_ring_multi_reader_create,
768 	.f_free = rte_port_ring_reader_free,
769 	.f_rx = rte_port_ring_multi_reader_rx,
770 	.f_stats = rte_port_ring_reader_stats_read,
771 };
772 
773 struct rte_port_out_ops rte_port_ring_multi_writer_ops = {
774 	.f_create = rte_port_ring_multi_writer_create,
775 	.f_free = rte_port_ring_writer_free,
776 	.f_tx = rte_port_ring_multi_writer_tx,
777 	.f_tx_bulk = rte_port_ring_multi_writer_tx_bulk,
778 	.f_flush = rte_port_ring_multi_writer_flush,
779 	.f_stats = rte_port_ring_writer_stats_read,
780 };
781 
782 struct rte_port_out_ops rte_port_ring_multi_writer_nodrop_ops = {
783 	.f_create = rte_port_ring_multi_writer_nodrop_create,
784 	.f_free = rte_port_ring_writer_nodrop_free,
785 	.f_tx = rte_port_ring_multi_writer_nodrop_tx,
786 	.f_tx_bulk = rte_port_ring_multi_writer_nodrop_tx_bulk,
787 	.f_flush = rte_port_ring_multi_writer_nodrop_flush,
788 	.f_stats = rte_port_ring_writer_nodrop_stats_read,
789 };
790