[PATCH 00/10] port: implement output port non-blocking behavior

[PATCH 00/10] port: implement output port non-blocking behavior

[Cristian Dumitrescu]

In case of blocking behavior, the output port retries sending the
packets that could not be sent successfully. The retry can take place
potentially forever in case the Ethernet device or the ring consumer
are down, which leads to deadlock.

In case of the non-blocking behavior introduced by this series, the
packets that could not be sent successfully are dropped and the
associated drop statistics counters are incremented.

Depends-on: series-24205 ("net/softnic: replace the legacy pipeline with SWX pipeline")

Cristian Dumitrescu (10):
  port: add output port packet drop statistics couters
  port: adjust the sink port counters
  port: rework the Ethernet device output port behavior to non-blocking
  port: free buffered packets on Ethernet device output port free
  port: prevent unnecessary flush for the Ethernet device output port
  port: rework the ring output port behavior to non-blocking
  port: free buffered packets on ring output port free
  port: prevent unnecessary flush for the ring output port
  examples/pipeline: print the output port packet drop counters
  net/softnic: print the output port packet drop counters

 drivers/net/softnic/rte_eth_softnic_cli.c |   4 +
 examples/pipeline/cli.c                   |   4 +
 lib/port/rte_swx_port.h                   |  10 +-
 lib/port/rte_swx_port_ethdev.c            | 117 +++++++++++++++-------
 lib/port/rte_swx_port_ring.c              | 113 ++++++++++++++-------
 lib/port/rte_swx_port_source_sink.c       |   8 +-
 6 files changed, 177 insertions(+), 79 deletions(-)

-- 
2.34.1

[PATCH 01/10] port: add output port packet drop statistics couters

[Cristian Dumitrescu]

Add packet drop statistics counters for the output ports. Required by
the non-blocking output port behavior where the packets that cannot
be sent at the time of the operation are dropped as opposed to the
send operation being retried potentially forever for the same packets.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port.h | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/lib/port/rte_swx_port.h b/lib/port/rte_swx_port.h
index 5e29100f59..1dbd95ae87 100644
--- a/lib/port/rte_swx_port.h
+++ b/lib/port/rte_swx_port.h
@@ -185,12 +185,18 @@ typedef void
 
 /** Output port statistics counters. */
 struct rte_swx_port_out_stats {
-	/** Number of packets. */
+	/** Number of packets successfully transmitted. */
 	uint64_t n_pkts;
 
-	/** Number of bytes. */
+	/** Number of bytes successfully transmitted. */
 	uint64_t n_bytes;
 
+	/** Number of packets dropped. */
+	uint64_t n_pkts_drop;
+
+	/** Number of bytes dropped. */
+	uint64_t n_bytes_drop;
+
 	/** Number of packets cloned successfully. */
 	uint64_t n_pkts_clone;
 
-- 
2.34.1

[PATCH 02/10] port: adjust the sink port counters

[Cristian Dumitrescu]

The sink port is tasked to drop all packets, hence the packet and byte
counters should be named to reflect the drop operation.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port_source_sink.c | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/lib/port/rte_swx_port_source_sink.c b/lib/port/rte_swx_port_source_sink.c
index d3cf42c30b..757a2c4a2e 100644
--- a/lib/port/rte_swx_port_source_sink.c
+++ b/lib/port/rte_swx_port_source_sink.c
@@ -299,8 +299,8 @@ sink_pkt_tx(void *port, struct rte_swx_pkt *pkt)
 	m->pkt_len = pkt->length;
 	m->data_off = (uint16_t)pkt->offset;
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
+	p->stats.n_pkts_drop++;
+	p->stats.n_bytes_drop += pkt->length;
 
 #ifdef RTE_PORT_PCAP
 	if (p->f_dump) {
@@ -335,8 +335,8 @@ __sink_pkt_clone_tx(void *port, struct rte_swx_pkt *pkt, uint32_t truncation_len
 	m->pkt_len = pkt->length;
 	m->data_off = (uint16_t)pkt->offset;
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
+	p->stats.n_pkts_drop++;
+	p->stats.n_bytes_drop += pkt->length;
 	p->stats.n_pkts_clone++;
 
 #ifdef RTE_PORT_PCAP
-- 
2.34.1

[PATCH 03/10] port: rework the Ethernet device output port behavior to non-blocking

[Cristian Dumitrescu]

Drop packets that cannot be sent instead of retry sending the same
packets potentially forever when the Ethernet device that is down.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port_ethdev.c | 103 +++++++++++++++++++++------------
 1 file changed, 67 insertions(+), 36 deletions(-)

diff --git a/lib/port/rte_swx_port_ethdev.c b/lib/port/rte_swx_port_ethdev.c
index ca4a43ac4f..eae20d34ce 100644
--- a/lib/port/rte_swx_port_ethdev.c
+++ b/lib/port/rte_swx_port_ethdev.c
@@ -173,6 +173,7 @@ struct writer {
 
 	struct rte_mbuf **pkts;
 	int n_pkts;
+	uint32_t n_bytes;
 };
 
 static void *
@@ -214,34 +215,56 @@ writer_create(void *args)
 	return p;
 }
 
-static void
+static inline void
 __writer_flush(struct writer *p)
 {
-	int n_pkts;
-
-	for (n_pkts = 0; ; ) {
-		n_pkts += rte_eth_tx_burst(p->params.port_id,
-					   p->params.queue_id,
-					   p->pkts + n_pkts,
-					   p->n_pkts - n_pkts);
-
-		TRACE("[Ethdev TX port %u queue %u] %d packets out\n",
-		      (uint32_t)p->params.port_id,
-		      (uint32_t)p->params.queue_id,
-		      n_pkts);
-
-		if (n_pkts == p->n_pkts)
-			break;
+	struct rte_mbuf **pkts = p->pkts;
+	uint64_t n_pkts_total = p->stats.n_pkts;
+	uint64_t n_bytes_total = p->stats.n_bytes;
+	uint64_t n_pkts_drop_total = p->stats.n_pkts_drop;
+	uint64_t n_bytes_drop_total = p->stats.n_bytes_drop;
+	int n_pkts = p->n_pkts, n_pkts_drop, n_pkts_tx;
+	uint32_t n_bytes = p->n_bytes, n_bytes_drop = 0;
+
+	/* Packet TX. */
+	n_pkts_tx = rte_eth_tx_burst(p->params.port_id,
+				     p->params.queue_id,
+				     pkts,
+				     n_pkts);
+
+	/* Packet drop. */
+	n_pkts_drop = n_pkts - n_pkts_tx;
+
+	for ( ; n_pkts_tx < n_pkts; n_pkts_tx++) {
+		struct rte_mbuf *m = pkts[n_pkts_tx];
+
+		n_bytes_drop += m->pkt_len;
+		rte_pktmbuf_free(m);
 	}
 
+	/* Port update. */
+	p->stats.n_pkts = n_pkts_total + n_pkts - n_pkts_drop;
+	p->stats.n_bytes = n_bytes_total + n_bytes - n_bytes_drop;
+	p->stats.n_pkts_drop = n_pkts_drop_total + n_pkts_drop;
+	p->stats.n_bytes_drop = n_bytes_drop_total + n_bytes_drop;
 	p->n_pkts = 0;
+	p->n_bytes = 0;
+
+	TRACE("[Ethdev TX port %u queue %u] Buffered packets flushed: %d out, %d dropped\n",
+	      (uint32_t)p->params.port_id,
+	      (uint32_t)p->params.queue_id,
+	      n_pkts - n_pkts_drop,
+	      n_pkts_drop);
 }
 
 static void
 writer_pkt_tx(void *port, struct rte_swx_pkt *pkt)
 {
 	struct writer *p = port;
+	int n_pkts = p->n_pkts;
+	uint32_t n_bytes = p->n_bytes;
 	struct rte_mbuf *m = pkt->handle;
+	uint32_t pkt_length = pkt->length;
 
 	TRACE("[Ethdev TX port %u queue %u] Pkt %d (%u bytes at offset %u)\n",
 	      (uint32_t)p->params.port_id,
@@ -252,15 +275,15 @@ writer_pkt_tx(void *port, struct rte_swx_pkt *pkt)
 	if (TRACE_LEVEL)
 		rte_hexdump(stdout, NULL, &pkt->pkt[pkt->offset], pkt->length);
 
-	m->data_len = (uint16_t)(pkt->length + m->data_len - m->pkt_len);
-	m->pkt_len = pkt->length;
+	m->data_len = (uint16_t)(pkt_length + m->data_len - m->pkt_len);
+	m->pkt_len = pkt_length;
 	m->data_off = (uint16_t)pkt->offset;
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
+	p->pkts[n_pkts++] = m;
+	p->n_pkts = n_pkts;
+	p->n_bytes = n_bytes + pkt_length;
 
-	p->pkts[p->n_pkts++] = m;
-	if (p->n_pkts ==  (int)p->params.burst_size)
+	if (n_pkts == (int)p->params.burst_size)
 		__writer_flush(p);
 }
 
@@ -268,7 +291,11 @@ static void
 writer_pkt_fast_clone_tx(void *port, struct rte_swx_pkt *pkt)
 {
 	struct writer *p = port;
+	int n_pkts = p->n_pkts;
+	uint32_t n_bytes = p->n_bytes;
+	uint64_t n_pkts_clone = p->stats.n_pkts_clone;
 	struct rte_mbuf *m = pkt->handle;
+	uint32_t pkt_length = pkt->length;
 
 	TRACE("[Ethdev TX port %u queue %u] Pkt %d (%u bytes at offset %u) (fast clone)\n",
 	      (uint32_t)p->params.port_id,
@@ -279,17 +306,17 @@ writer_pkt_fast_clone_tx(void *port, struct rte_swx_pkt *pkt)
 	if (TRACE_LEVEL)
 		rte_hexdump(stdout, NULL, &pkt->pkt[pkt->offset], pkt->length);
 
-	m->data_len = (uint16_t)(pkt->length + m->data_len - m->pkt_len);
-	m->pkt_len = pkt->length;
+	m->data_len = (uint16_t)(pkt_length + m->data_len - m->pkt_len);
+	m->pkt_len = pkt_length;
 	m->data_off = (uint16_t)pkt->offset;
 	rte_pktmbuf_refcnt_update(m, 1);
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
-	p->stats.n_pkts_clone++;
+	p->pkts[n_pkts++] = m;
+	p->n_pkts = n_pkts;
+	p->n_bytes = n_bytes + pkt_length;
+	p->stats.n_pkts_clone = n_pkts_clone + 1;
 
-	p->pkts[p->n_pkts++] = m;
-	if (p->n_pkts == (int)p->params.burst_size)
+	if (n_pkts == (int)p->params.burst_size)
 		__writer_flush(p);
 }
 
@@ -297,7 +324,11 @@ static void
 writer_pkt_clone_tx(void *port, struct rte_swx_pkt *pkt, uint32_t truncation_length)
 {
 	struct writer *p = port;
+	int n_pkts = p->n_pkts;
+	uint32_t n_bytes = p->n_bytes;
+	uint64_t n_pkts_clone = p->stats.n_pkts_clone;
 	struct rte_mbuf *m = pkt->handle, *m_clone;
+	uint32_t pkt_length = pkt->length;
 
 	TRACE("[Ethdev TX port %u queue %u] Pkt %d (%u bytes at offset %u) (clone)\n",
 	      (uint32_t)p->params.port_id,
@@ -308,8 +339,8 @@ writer_pkt_clone_tx(void *port, struct rte_swx_pkt *pkt, uint32_t truncation_len
 	if (TRACE_LEVEL)
 		rte_hexdump(stdout, NULL, &pkt->pkt[pkt->offset], pkt->length);
 
-	m->data_len = (uint16_t)(pkt->length + m->data_len - m->pkt_len);
-	m->pkt_len = pkt->length;
+	m->data_len = (uint16_t)(pkt_length + m->data_len - m->pkt_len);
+	m->pkt_len = pkt_length;
 	m->data_off = (uint16_t)pkt->offset;
 
 	m_clone = rte_pktmbuf_copy(m, m->pool, 0, truncation_length);
@@ -318,12 +349,12 @@ writer_pkt_clone_tx(void *port, struct rte_swx_pkt *pkt, uint32_t truncation_len
 		return;
 	}
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
-	p->stats.n_pkts_clone++;
+	p->pkts[n_pkts++] = m_clone;
+	p->n_pkts = n_pkts;
+	p->n_bytes = n_bytes + pkt_length;
+	p->stats.n_pkts_clone = n_pkts_clone + 1;
 
-	p->pkts[p->n_pkts++] = m_clone;
-	if (p->n_pkts == (int)p->params.burst_size)
+	if (n_pkts == (int)p->params.burst_size)
 		__writer_flush(p);
 }
 
-- 
2.34.1

[PATCH 04/10] port: free buffered packets on Ethernet device output port free

[Cristian Dumitrescu]

Free the buffered packets as opposed to retrying to send them when the
output port is freed.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port_ethdev.c | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/lib/port/rte_swx_port_ethdev.c b/lib/port/rte_swx_port_ethdev.c
index eae20d34ce..8a1da71003 100644
--- a/lib/port/rte_swx_port_ethdev.c
+++ b/lib/port/rte_swx_port_ethdev.c
@@ -371,11 +371,17 @@ static void
 writer_free(void *port)
 {
 	struct writer *p = port;
+	int i;
 
 	if (!p)
 		return;
 
-	writer_flush(p);
+	for (i = 0; i < p->n_pkts; i++) {
+		struct rte_mbuf *m = p->pkts[i];
+
+		rte_pktmbuf_free(m);
+	}
+
 	free(p->pkts);
 	free(port);
 }
-- 
2.34.1

[PATCH 05/10] port: prevent unnecessary flush for the Ethernet device output port

[Cristian Dumitrescu]

Do not flush the buffered packets unnecessarily when a burst was sent
since the last flush call.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port_ethdev.c | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/lib/port/rte_swx_port_ethdev.c b/lib/port/rte_swx_port_ethdev.c
index 8a1da71003..7cb3c4dfb1 100644
--- a/lib/port/rte_swx_port_ethdev.c
+++ b/lib/port/rte_swx_port_ethdev.c
@@ -174,6 +174,7 @@ struct writer {
 	struct rte_mbuf **pkts;
 	int n_pkts;
 	uint32_t n_bytes;
+	int flush_flag;
 };
 
 static void *
@@ -249,6 +250,7 @@ __writer_flush(struct writer *p)
 	p->stats.n_bytes_drop = n_bytes_drop_total + n_bytes_drop;
 	p->n_pkts = 0;
 	p->n_bytes = 0;
+	p->flush_flag = 0;
 
 	TRACE("[Ethdev TX port %u queue %u] Buffered packets flushed: %d out, %d dropped\n",
 	      (uint32_t)p->params.port_id,
@@ -363,8 +365,10 @@ writer_flush(void *port)
 {
 	struct writer *p = port;
 
-	if (p->n_pkts)
+	if (p->n_pkts && p->flush_flag)
 		__writer_flush(p);
+
+	p->flush_flag = 1;
 }
 
 static void
-- 
2.34.1

[PATCH 06/10] port: rework the ring output port behavior to non-blocking

[Cristian Dumitrescu]

Drop packets that cannot be sent instead of retry sending the same
packets potentially forever when the ring consumer that is down.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port_ring.c | 99 ++++++++++++++++++++++++------------
 1 file changed, 66 insertions(+), 33 deletions(-)

diff --git a/lib/port/rte_swx_port_ring.c b/lib/port/rte_swx_port_ring.c
index c62fb3d8c8..72ec7209bf 100644
--- a/lib/port/rte_swx_port_ring.c
+++ b/lib/port/rte_swx_port_ring.c
@@ -172,6 +172,7 @@ struct writer {
 
 	struct rte_mbuf **pkts;
 	int n_pkts;
+	uint32_t n_bytes;
 };
 
 static void *
@@ -218,31 +219,55 @@ writer_create(void *args)
 	return NULL;
 }
 
-static void
+static inline void
 __writer_flush(struct writer *p)
 {
-	int n_pkts;
-
-	for (n_pkts = 0; ; ) {
-		n_pkts += rte_ring_sp_enqueue_burst(p->params.ring,
-						    (void **)p->pkts + n_pkts,
-						    p->n_pkts - n_pkts,
-						    NULL);
-
-		TRACE("[Ring %s] %d packets out\n", p->params.name, n_pkts);
-
-		if (n_pkts == p->n_pkts)
-			break;
+	struct rte_mbuf **pkts = p->pkts;
+	uint64_t n_pkts_total = p->stats.n_pkts;
+	uint64_t n_bytes_total = p->stats.n_bytes;
+	uint64_t n_pkts_drop_total = p->stats.n_pkts_drop;
+	uint64_t n_bytes_drop_total = p->stats.n_bytes_drop;
+	int n_pkts = p->n_pkts, n_pkts_drop, n_pkts_tx;
+	uint32_t n_bytes = p->n_bytes, n_bytes_drop = 0;
+
+	/* Packet TX. */
+	n_pkts_tx = rte_ring_sp_enqueue_burst(p->params.ring,
+					      (void **)pkts,
+					      n_pkts,
+					      NULL);
+
+	/* Packet drop. */
+	n_pkts_drop = n_pkts - n_pkts_tx;
+
+	for ( ; n_pkts_tx < n_pkts; n_pkts_tx++) {
+		struct rte_mbuf *m = pkts[n_pkts_tx];
+
+		n_bytes_drop += m->pkt_len;
+		rte_pktmbuf_free(m);
 	}
 
+	/* Port update. */
+	p->stats.n_pkts = n_pkts_total + n_pkts - n_pkts_drop;
+	p->stats.n_bytes = n_bytes_total + n_bytes - n_bytes_drop;
+	p->stats.n_pkts_drop = n_pkts_drop_total + n_pkts_drop;
+	p->stats.n_bytes_drop = n_bytes_drop_total + n_bytes_drop;
 	p->n_pkts = 0;
+	p->n_bytes = 0;
+
+	TRACE("[Ring %s] Buffered packets flushed: %d out, %d dropped\n",
+	      p->params.name,
+	      n_pkts - n_pkts_drop,
+	      n_pkts_drop);
 }
 
 static void
 writer_pkt_tx(void *port, struct rte_swx_pkt *pkt)
 {
 	struct writer *p = port;
+	int n_pkts = p->n_pkts;
+	uint32_t n_bytes = p->n_bytes;
 	struct rte_mbuf *m = pkt->handle;
+	uint32_t pkt_length = pkt->length;
 
 	TRACE("[Ring %s] Pkt %d (%u bytes at offset %u)\n",
 	      p->params.name,
@@ -252,15 +277,15 @@ writer_pkt_tx(void *port, struct rte_swx_pkt *pkt)
 	if (TRACE_LEVEL)
 		rte_hexdump(stdout, NULL, &pkt->pkt[pkt->offset], pkt->length);
 
-	m->data_len = (uint16_t)(pkt->length + m->data_len - m->pkt_len);
-	m->pkt_len = pkt->length;
+	m->data_len = (uint16_t)(pkt_length + m->data_len - m->pkt_len);
+	m->pkt_len = pkt_length;
 	m->data_off = (uint16_t)pkt->offset;
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
+	p->pkts[n_pkts++] = m;
+	p->n_pkts = n_pkts;
+	p->n_bytes = n_bytes + pkt_length;
 
-	p->pkts[p->n_pkts++] = m;
-	if (p->n_pkts ==  (int)p->params.burst_size)
+	if (n_pkts == (int)p->params.burst_size)
 		__writer_flush(p);
 }
 
@@ -268,7 +293,11 @@ static void
 writer_pkt_fast_clone_tx(void *port, struct rte_swx_pkt *pkt)
 {
 	struct writer *p = port;
+	int n_pkts = p->n_pkts;
+	uint32_t n_bytes = p->n_bytes;
+	uint64_t n_pkts_clone = p->stats.n_pkts_clone;
 	struct rte_mbuf *m = pkt->handle;
+	uint32_t pkt_length = pkt->length;
 
 	TRACE("[Ring %s] Pkt %d (%u bytes at offset %u) (fast clone)\n",
 	      p->params.name,
@@ -278,17 +307,17 @@ writer_pkt_fast_clone_tx(void *port, struct rte_swx_pkt *pkt)
 	if (TRACE_LEVEL)
 		rte_hexdump(stdout, NULL, &pkt->pkt[pkt->offset], pkt->length);
 
-	m->data_len = (uint16_t)(pkt->length + m->data_len - m->pkt_len);
-	m->pkt_len = pkt->length;
+	m->data_len = (uint16_t)(pkt_length + m->data_len - m->pkt_len);
+	m->pkt_len = pkt_length;
 	m->data_off = (uint16_t)pkt->offset;
 	rte_pktmbuf_refcnt_update(m, 1);
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
-	p->stats.n_pkts_clone++;
+	p->pkts[n_pkts++] = m;
+	p->n_pkts = n_pkts;
+	p->n_bytes = n_bytes + pkt_length;
+	p->stats.n_pkts_clone = n_pkts_clone + 1;
 
-	p->pkts[p->n_pkts++] = m;
-	if (p->n_pkts == (int)p->params.burst_size)
+	if (n_pkts == (int)p->params.burst_size)
 		__writer_flush(p);
 }
 
@@ -296,7 +325,11 @@ static void
 writer_pkt_clone_tx(void *port, struct rte_swx_pkt *pkt, uint32_t truncation_length)
 {
 	struct writer *p = port;
+	int n_pkts = p->n_pkts;
+	uint32_t n_bytes = p->n_bytes;
+	uint64_t n_pkts_clone = p->stats.n_pkts_clone;
 	struct rte_mbuf *m = pkt->handle, *m_clone;
+	uint32_t pkt_length = pkt->length;
 
 	TRACE("[Ring %s] Pkt %d (%u bytes at offset %u) (clone)\n",
 	      p->params.name,
@@ -306,8 +339,8 @@ writer_pkt_clone_tx(void *port, struct rte_swx_pkt *pkt, uint32_t truncation_len
 	if (TRACE_LEVEL)
 		rte_hexdump(stdout, NULL, &pkt->pkt[pkt->offset], pkt->length);
 
-	m->data_len = (uint16_t)(pkt->length + m->data_len - m->pkt_len);
-	m->pkt_len = pkt->length;
+	m->data_len = (uint16_t)(pkt_length + m->data_len - m->pkt_len);
+	m->pkt_len = pkt_length;
 	m->data_off = (uint16_t)pkt->offset;
 
 	m_clone = rte_pktmbuf_copy(m, m->pool, 0, truncation_length);
@@ -316,12 +349,12 @@ writer_pkt_clone_tx(void *port, struct rte_swx_pkt *pkt, uint32_t truncation_len
 		return;
 	}
 
-	p->stats.n_pkts++;
-	p->stats.n_bytes += pkt->length;
-	p->stats.n_pkts_clone++;
+	p->pkts[n_pkts++] = m_clone;
+	p->n_pkts = n_pkts;
+	p->n_bytes = n_bytes + pkt_length;
+	p->stats.n_pkts_clone = n_pkts_clone + 1;
 
-	p->pkts[p->n_pkts++] = m_clone;
-	if (p->n_pkts == (int)p->params.burst_size)
+	if (n_pkts == (int)p->params.burst_size)
 		__writer_flush(p);
 }
 
-- 
2.34.1

[PATCH 07/10] port: free buffered packets on ring output port free

[Cristian Dumitrescu]

Free the buffered packets as opposed to retrying to send them when the
output port is freed.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port_ring.c | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/lib/port/rte_swx_port_ring.c b/lib/port/rte_swx_port_ring.c
index 72ec7209bf..7cdcd4b638 100644
--- a/lib/port/rte_swx_port_ring.c
+++ b/lib/port/rte_swx_port_ring.c
@@ -371,11 +371,17 @@ static void
 writer_free(void *port)
 {
 	struct writer *p = port;
+	int i;
 
 	if (!p)
 		return;
 
-	writer_flush(p);
+	for (i = 0; i < p->n_pkts; i++) {
+		struct rte_mbuf *m = p->pkts[i];
+
+		rte_pktmbuf_free(m);
+	}
+
 	free(p->pkts);
 	free(p->params.name);
 	free(port);
-- 
2.34.1

[PATCH 08/10] port: prevent unnecessary flush for the ring output port

[Cristian Dumitrescu]

Do not flush the buffered packets unnecessarily when a burst was sent
since the last flush call.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 lib/port/rte_swx_port_ring.c | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/lib/port/rte_swx_port_ring.c b/lib/port/rte_swx_port_ring.c
index 7cdcd4b638..d0058087bd 100644
--- a/lib/port/rte_swx_port_ring.c
+++ b/lib/port/rte_swx_port_ring.c
@@ -173,6 +173,7 @@ struct writer {
 	struct rte_mbuf **pkts;
 	int n_pkts;
 	uint32_t n_bytes;
+	int flush_flag;
 };
 
 static void *
@@ -253,6 +254,7 @@ __writer_flush(struct writer *p)
 	p->stats.n_bytes_drop = n_bytes_drop_total + n_bytes_drop;
 	p->n_pkts = 0;
 	p->n_bytes = 0;
+	p->flush_flag = 0;
 
 	TRACE("[Ring %s] Buffered packets flushed: %d out, %d dropped\n",
 	      p->params.name,
@@ -363,8 +365,10 @@ writer_flush(void *port)
 {
 	struct writer *p = port;
 
-	if (p->n_pkts)
+	if (p->n_pkts && p->flush_flag)
 		__writer_flush(p);
+
+	p->flush_flag = 1;
 }
 
 static void
-- 
2.34.1

[PATCH 09/10] examples/pipeline: print the output port packet drop counters

[Cristian Dumitrescu]

Print the output port pacet drop statistics counters.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 examples/pipeline/cli.c | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/examples/pipeline/cli.c b/examples/pipeline/cli.c
index 75c32b9089..2e69698031 100644
--- a/examples/pipeline/cli.c
+++ b/examples/pipeline/cli.c
@@ -2276,10 +2276,14 @@ cmd_pipeline_stats(char **tokens,
 			out_size,
 			" packets %" PRIu64
 			" bytes %" PRIu64
+			" packets dropped %" PRIu64
+			" bytes dropped %" PRIu64
 			" clone %" PRIu64
 			" clonerr %" PRIu64 "\n",
 			stats.n_pkts,
 			stats.n_bytes,
+			stats.n_pkts_drop,
+			stats.n_bytes_drop,
 			stats.n_pkts_clone,
 			stats.n_pkts_clone_err);
 
-- 
2.34.1

[PATCH 10/10] net/softnic: print the output port packet drop counters

[Cristian Dumitrescu]

Print the output port packet drop statistics counters.

Signed-off-by: Cristian Dumitrescu <cristian.dumitrescu@intel.com>
---
 drivers/net/softnic/rte_eth_softnic_cli.c | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/drivers/net/softnic/rte_eth_softnic_cli.c b/drivers/net/softnic/rte_eth_softnic_cli.c
index 61221b2f2e..9f4c44579a 100644
--- a/drivers/net/softnic/rte_eth_softnic_cli.c
+++ b/drivers/net/softnic/rte_eth_softnic_cli.c
@@ -1987,10 +1987,14 @@ cmd_softnic_pipeline_stats(struct pmd_internals *softnic,
 			out_size,
 			" packets %" PRIu64
 			" bytes %" PRIu64
+			" packets dropped %" PRIu64
+			" bytes dropped %" PRIu64
 			" clone %" PRIu64
 			" clonerr %" PRIu64 "\n",
 			stats.n_pkts,
 			stats.n_bytes,
+			stats.n_pkts_drop,
+			stats.n_bytes_drop,
 			stats.n_pkts_clone,
 			stats.n_pkts_clone_err);
 
-- 
2.34.1

Re: [PATCH 00/10] port: implement output port non-blocking behavior

[Thomas Monjalon]

06/08/2022 00:00, Cristian Dumitrescu:
> In case of blocking behavior, the output port retries sending the
> packets that could not be sent successfully. The retry can take place
> potentially forever in case the Ethernet device or the ring consumer
> are down, which leads to deadlock.
> 
> In case of the non-blocking behavior introduced by this series, the
> packets that could not be sent successfully are dropped and the
> associated drop statistics counters are incremented.
> 
> Depends-on: series-24205 ("net/softnic: replace the legacy pipeline with SWX pipeline")
> 
> Cristian Dumitrescu (10):
>   port: add output port packet drop statistics couters
>   port: adjust the sink port counters
>   port: rework the Ethernet device output port behavior to non-blocking
>   port: free buffered packets on Ethernet device output port free
>   port: prevent unnecessary flush for the Ethernet device output port
>   port: rework the ring output port behavior to non-blocking
>   port: free buffered packets on ring output port free
>   port: prevent unnecessary flush for the ring output port
>   examples/pipeline: print the output port packet drop counters
>   net/softnic: print the output port packet drop counters

Applied, thanks.