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