[PATCH 2/6] examples/l3fwd-power: check Rx interrupt enable errors

rx_interrupt_wait() enabled the Rx queue interrupts of an lcore through
turn_on_off_intr() and ignored the return of
rte_eth_dev_rx_intr_enable(). Some PMDs report a transient condition
there: for example dpaa2 returns -EAGAIN when it finds traffic already
queued while arming, meaning the queue was not armed and the lcore must
poll rather than sleep.

Split the helper into rx_intr_enable_all(), which returns the error and
unwinds the queues it already armed, and rx_intr_disable_all(). Return
the error from rx_interrupt_wait(); in the sleep path, on -EAGAIN go back
to polling, and on any other error give up interrupt mode for the lcore
instead of sleeping on unarmed queues.

Signed-off-by: Maxime Leroy <[email protected]>
---
 examples/l3fwd-power/main.c | 74 ++++++++++++++++++++++++++++++++-----
 1 file changed, 64 insertions(+), 10 deletions(-)

diff --git a/examples/l3fwd-power/main.c b/examples/l3fwd-power/main.c
index 764899217c..70f76d202e 100644
--- a/examples/l3fwd-power/main.c
+++ b/examples/l3fwd-power/main.c
@@ -863,12 +863,32 @@ sleep_until_rx_interrupt(int num, int lcore)
 	return 0;
 }
 
-static void turn_on_off_intr(struct lcore_conf *qconf, bool on)
+static void
+rx_intr_disable_all(struct lcore_conf *qconf)
 {
+	struct lcore_rx_queue *rx_queue;
+	uint16_t queue_id;
+	uint16_t port_id;
 	int i;
+
+	for (i = 0; i < qconf->n_rx_queue; ++i) {
+		rx_queue = &(qconf->rx_queue_list[i]);
+		port_id = rx_queue->port_id;
+		queue_id = rx_queue->queue_id;
+
+		rte_spinlock_lock(&(locks[port_id]));
+		rte_eth_dev_rx_intr_disable(port_id, queue_id);
+		rte_spinlock_unlock(&(locks[port_id]));
+	}
+}
+
+static int
+rx_intr_enable_all(struct lcore_conf *qconf)
+{
 	struct lcore_rx_queue *rx_queue;
 	uint16_t queue_id;
 	uint16_t port_id;
+	int i, ret;
 
 	for (i = 0; i < qconf->n_rx_queue; ++i) {
 		rx_queue = &(qconf->rx_queue_list[i]);
@@ -876,12 +896,26 @@ static void turn_on_off_intr(struct lcore_conf *qconf, bool on)
 		queue_id = rx_queue->queue_id;
 
 		rte_spinlock_lock(&(locks[port_id]));
-		if (on)
-			rte_eth_dev_rx_intr_enable(port_id, queue_id);
-		else
-			rte_eth_dev_rx_intr_disable(port_id, queue_id);
+		ret = rte_eth_dev_rx_intr_enable(port_id, queue_id);
 		rte_spinlock_unlock(&(locks[port_id]));
+		if (ret != 0)
+			goto fail;
 	}
+
+	return 0;
+
+fail:
+	while (--i >= 0) {
+		rx_queue = &(qconf->rx_queue_list[i]);
+		port_id = rx_queue->port_id;
+		queue_id = rx_queue->queue_id;
+
+		rte_spinlock_lock(&(locks[port_id]));
+		rte_eth_dev_rx_intr_disable(port_id, queue_id);
+		rte_spinlock_unlock(&(locks[port_id]));
+	}
+
+	return ret;
 }
 
 static int event_register(struct lcore_conf *qconf)
@@ -910,12 +944,18 @@ static int event_register(struct lcore_conf *qconf)
 	return 0;
 }
 
-static void
+static int
 rx_interrupt_wait(struct lcore_conf *qconf)
 {
-	turn_on_off_intr(qconf, 1);
+	int ret;
+
+	ret = rx_intr_enable_all(qconf);
+	if (ret != 0)
+		return ret;
+
 	sleep_until_rx_interrupt(qconf->n_rx_queue, rte_lcore_id());
-	turn_on_off_intr(qconf, 0);
+	rx_intr_disable_all(qconf);
+	return 0;
 }
 
 /* Main processing loop. 8< */
@@ -931,6 +971,7 @@ static int main_intr_loop(__rte_unused void *dummy)
 	uint32_t lcore_rx_idle_count = 0;
 	uint32_t lcore_idle_hint = 0;
 	int intr_en = 0;
+	int ret;
 
 	const uint64_t drain_tsc = (rte_get_tsc_hz() + US_PER_S - 1) /
 				   US_PER_S * BURST_TX_DRAIN_US;
@@ -1062,7 +1103,13 @@ static int main_intr_loop(__rte_unused void *dummy)
 			else {
 				/* suspend until rx interrupt triggers */
 				if (intr_en) {
-					rx_interrupt_wait(qconf);
+					ret = rx_interrupt_wait(qconf);
+					if (ret == -EAGAIN)
+						goto start_rx;
+					if (ret != 0) {
+						intr_en = 0;
+						continue;
+					}
 					/**
 					 * start receiving packets immediately
 					 */
@@ -1214,6 +1261,7 @@ main_legacy_loop(__rte_unused void *dummy)
 	uint32_t lcore_rx_idle_count = 0;
 	uint32_t lcore_idle_hint = 0;
 	int intr_en = 0;
+	int ret;
 
 	const uint64_t drain_tsc = (rte_get_tsc_hz() + US_PER_S - 1) / US_PER_S * BURST_TX_DRAIN_US;
 
@@ -1376,7 +1424,13 @@ main_legacy_loop(__rte_unused void *dummy)
 			else {
 				/* suspend until rx interrupt triggers */
 				if (intr_en) {
-					rx_interrupt_wait(qconf);
+					ret = rx_interrupt_wait(qconf);
+					if (ret == -EAGAIN)
+						goto start_rx;
+					if (ret != 0) {
+						intr_en = 0;
+						continue;
+					}
 					/**
 					 * start receiving packets immediately
 					 */
-- 
2.43.0