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 *
rte_port_ring_reader_create_internal(void * params,int socket_id,uint32_t is_multi)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 *
rte_port_ring_reader_create(void * params,int socket_id)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 *
rte_port_ring_multi_reader_create(void * params,int socket_id)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
rte_port_ring_reader_rx(void * port,struct rte_mbuf ** pkts,uint32_t n_pkts)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
rte_port_ring_multi_reader_rx(void * port,struct rte_mbuf ** pkts,uint32_t n_pkts)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
rte_port_ring_reader_free(void * port)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
rte_port_ring_reader_stats_read(void * port,struct rte_port_in_stats * stats,int clear)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 *
rte_port_ring_writer_create_internal(void * params,int socket_id,uint32_t is_multi)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 *
rte_port_ring_writer_create(void * params,int socket_id)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 *
rte_port_ring_multi_writer_create(void * params,int socket_id)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
send_burst(struct rte_port_ring_writer * p)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
send_burst_mp(struct rte_port_ring_writer * p)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
rte_port_ring_writer_tx(void * port,struct rte_mbuf * pkt)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
rte_port_ring_multi_writer_tx(void * port,struct rte_mbuf * pkt)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
rte_port_ring_writer_tx_bulk_internal(void * port,struct rte_mbuf ** pkts,uint64_t pkts_mask,uint32_t is_multi)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
rte_port_ring_writer_tx_bulk(void * port,struct rte_mbuf ** pkts,uint64_t pkts_mask)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
rte_port_ring_multi_writer_tx_bulk(void * port,struct rte_mbuf ** pkts,uint64_t pkts_mask)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
rte_port_ring_writer_flush(void * port)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
rte_port_ring_multi_writer_flush(void * port)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
rte_port_ring_writer_free(void * port)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
rte_port_ring_writer_stats_read(void * port,struct rte_port_out_stats * stats,int clear)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 *
rte_port_ring_writer_nodrop_create_internal(void * params,int socket_id,uint32_t is_multi)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 *
rte_port_ring_writer_nodrop_create(void * params,int socket_id)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 *
rte_port_ring_multi_writer_nodrop_create(void * params,int socket_id)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
send_burst_nodrop(struct rte_port_ring_writer_nodrop * p)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
send_burst_mp_nodrop(struct rte_port_ring_writer_nodrop * p)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
rte_port_ring_writer_nodrop_tx(void * port,struct rte_mbuf * pkt)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
rte_port_ring_multi_writer_nodrop_tx(void * port,struct rte_mbuf * pkt)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
rte_port_ring_writer_nodrop_tx_bulk_internal(void * port,struct rte_mbuf ** pkts,uint64_t pkts_mask,uint32_t is_multi)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
rte_port_ring_writer_nodrop_tx_bulk(void * port,struct rte_mbuf ** pkts,uint64_t pkts_mask)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
rte_port_ring_multi_writer_nodrop_tx_bulk(void * port,struct rte_mbuf ** pkts,uint64_t pkts_mask)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
rte_port_ring_writer_nodrop_flush(void * port)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
rte_port_ring_multi_writer_nodrop_flush(void * port)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
rte_port_ring_writer_nodrop_free(void * port)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
rte_port_ring_writer_nodrop_stats_read(void * port,struct rte_port_out_stats * stats,int clear)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