[dpdk-dev] [PATCH v2] net/sfc: allow control threads for counter queue polling

[Andrew Rybchenko <andrew.rybchenko@oktetlabs.ru>]

From: Viacheslav Galaktionov <viacheslav.galaktionov@oktetlabs.ru>

MAE counters can be polled from a control thread if no service core is
allocated for this.

Signed-off-by: Viacheslav Galaktionov <viacheslav.galaktionov@oktetlabs.ru>
Signed-off-by: Andrew Rybchenko <andrew.rybchenko@oktetlabs.ru>
Reviewed-by: Andy Moreton <amoreton@xilinx.com>
---
The problem to require service cores for HW offload was raised by
David on review in 21.08 release cycle.

v2:
    - add sleep in the case of empty queue to avoid CPU burn
    - pair release/acquire for thread run/stop condition

 drivers/net/sfc/sfc_mae.h         |  26 ++++-
 drivers/net/sfc/sfc_mae_counter.c | 185 ++++++++++++++++++++++++++----
 2 files changed, 183 insertions(+), 28 deletions(-)

diff --git a/drivers/net/sfc/sfc_mae.h b/drivers/net/sfc/sfc_mae.h
index 23dcf1e482..2750fbb9b1 100644
--- a/drivers/net/sfc/sfc_mae.h
+++ b/drivers/net/sfc/sfc_mae.h
@@ -127,6 +127,13 @@ struct sfc_mae_counters {
 	unsigned int			n_mae_counters;
 };
 
+/** Options for MAE counter polling mode */
+enum sfc_mae_counter_polling_mode {
+	SFC_MAE_COUNTER_POLLING_OFF = 0,
+	SFC_MAE_COUNTER_POLLING_SERVICE,
+	SFC_MAE_COUNTER_POLLING_THREAD,
+};
+
 struct sfc_mae_counter_registry {
 	/* Common counter information */
 	/** Counters collection */
@@ -143,10 +150,21 @@ struct sfc_mae_counter_registry {
 	bool				use_credits;
 
 	/* Information used by configuration routines */
-	/** Counter service core ID */
-	uint32_t			service_core_id;
-	/** Counter service ID */
-	uint32_t			service_id;
+	enum sfc_mae_counter_polling_mode polling_mode;
+	union {
+		struct {
+			/** Counter service core ID */
+			uint32_t			core_id;
+			/** Counter service ID */
+			uint32_t			id;
+		} service;
+		struct {
+			/** Counter thread ID */
+			pthread_t			id;
+			/** The thread should keep running */
+			bool				run;
+		} thread;
+	} polling;
 };
 
 /**
diff --git a/drivers/net/sfc/sfc_mae_counter.c b/drivers/net/sfc/sfc_mae_counter.c
index 418caffe59..e208e97fbd 100644
--- a/drivers/net/sfc/sfc_mae_counter.c
+++ b/drivers/net/sfc/sfc_mae_counter.c
@@ -15,6 +15,33 @@
 #include "sfc_mae_counter.h"
 #include "sfc_service.h"
 
+/**
+ * Approximate maximum number of counters per packet.
+ * In fact maximum depends on per-counter data offset which is specified
+ * in counter packet header.
+ */
+#define SFC_MAE_COUNTERS_PER_PACKET_MAX \
+	((SFC_MAE_COUNTER_STREAM_PACKET_SIZE - \
+	  ER_RX_SL_PACKETISER_HEADER_WORD_SIZE) / \
+	  ER_RX_SL_PACKETISER_PAYLOAD_WORD_SIZE)
+
+/**
+ * Minimum number of Rx buffers in counters only Rx queue.
+ */
+#define SFC_MAE_COUNTER_RXQ_BUFS_MIN \
+	(SFC_COUNTER_RXQ_RX_DESC_COUNT - SFC_COUNTER_RXQ_REFILL_LEVEL)
+
+/**
+ * Approximate number of counter updates fit in counters only Rx queue.
+ * The number is inaccurate since SFC_MAE_COUNTERS_PER_PACKET_MAX is
+ * inaccurate (see above). However, it provides the gist for a number of
+ * counter updates which can fit in an Rx queue after empty poll.
+ *
+ * The define is not actually used, but provides calculations details.
+ */
+#define SFC_MAE_COUNTERS_RXQ_SPACE \
+	(SFC_MAE_COUNTER_RXQ_BUFS_MIN * SFC_MAE_COUNTERS_PER_PACKET_MAX)
+
 static uint32_t
 sfc_mae_counter_get_service_lcore(struct sfc_adapter *sa)
 {
@@ -45,9 +72,6 @@ sfc_mae_counter_rxq_required(struct sfc_adapter *sa)
 	if (encp->enc_mae_supported == B_FALSE)
 		return false;
 
-	if (sfc_mae_counter_get_service_lcore(sa) == RTE_MAX_LCORE)
-		return false;
-
 	return true;
 }
 
@@ -358,9 +382,8 @@ sfc_mae_parse_counter_packet(struct sfc_adapter *sa,
 }
 
 static int32_t
-sfc_mae_counter_routine(void *arg)
+sfc_mae_counter_poll_packets(struct sfc_adapter *sa)
 {
-	struct sfc_adapter *sa = arg;
 	struct sfc_mae_counter_registry *counter_registry =
 		&sa->mae.counter_registry;
 	struct rte_mbuf *mbufs[SFC_MAE_COUNTER_RX_BURST];
@@ -379,7 +402,7 @@ sfc_mae_counter_routine(void *arg)
 	rte_pktmbuf_free_bulk(mbufs, n);
 
 	if (!counter_registry->use_credits)
-		return 0;
+		return n;
 
 	pushed = sfc_rx_get_pushed(sa, counter_registry->rx_dp);
 	pushed_diff = pushed - counter_registry->pushed_n_buffers;
@@ -399,9 +422,57 @@ sfc_mae_counter_routine(void *arg)
 		}
 	}
 
+	return n;
+}
+
+static int32_t
+sfc_mae_counter_service_routine(void *arg)
+{
+	struct sfc_adapter *sa = arg;
+
+	/*
+	 * We cannot propagate any errors and we don't need to know
+	 * the number of packets we've received.
+	 */
+	(void)sfc_mae_counter_poll_packets(sa);
+
 	return 0;
 }
 
+static void *
+sfc_mae_counter_thread(void *data)
+{
+	struct sfc_adapter *sa = data;
+	struct sfc_mae_counter_registry *counter_registry =
+		&sa->mae.counter_registry;
+	int32_t rc;
+
+	while (__atomic_load_n(&counter_registry->polling.thread.run,
+			       __ATOMIC_ACQUIRE)) {
+		rc = sfc_mae_counter_poll_packets(sa);
+		if (rc == 0) {
+			/*
+			 * The queue is empty. Do not burn CPU.
+			 * An empty queue has just enough space for about
+			 * SFC_MAE_COUNTERS_RXQ_SPACE counter updates which is
+			 * more than 100K, so we can sleep a bit. The queue uses
+			 * a credit-based flow control anyway, so firmware will
+			 * not enqueue more counter updates until the host
+			 * supplies it with additional credits. The counters are
+			 * 48bits wide, so the timeout need only be short enough
+			 * to ensure that the counter values do not overflow
+			 * before the next counter update. Also we should not
+			 * delay counter updates for a long time, otherwise
+			 * application may decide that flow is idle and should
+			 * be removed.
+			 */
+			rte_delay_ms(1);
+		}
+	}
+
+	return NULL;
+}
+
 static void
 sfc_mae_counter_service_unregister(struct sfc_adapter *sa)
 {
@@ -410,15 +481,15 @@ sfc_mae_counter_service_unregister(struct sfc_adapter *sa)
 	const unsigned int wait_ms = 10000;
 	unsigned int i;
 
-	rte_service_runstate_set(registry->service_id, 0);
-	rte_service_component_runstate_set(registry->service_id, 0);
+	rte_service_runstate_set(registry->polling.service.id, 0);
+	rte_service_component_runstate_set(registry->polling.service.id, 0);
 
 	/*
 	 * Wait for the counter routine to finish the last iteration.
 	 * Give up on timeout.
 	 */
 	for (i = 0; i < wait_ms; i++) {
-		if (rte_service_may_be_active(registry->service_id) == 0)
+		if (rte_service_may_be_active(registry->polling.service.id) == 0)
 			break;
 
 		rte_delay_ms(1);
@@ -426,10 +497,10 @@ sfc_mae_counter_service_unregister(struct sfc_adapter *sa)
 	if (i == wait_ms)
 		sfc_warn(sa, "failed to wait for counter service to stop");
 
-	rte_service_map_lcore_set(registry->service_id,
-				  registry->service_core_id, 0);
+	rte_service_map_lcore_set(registry->polling.service.id,
+				  registry->polling.service.core_id, 0);
 
-	rte_service_component_unregister(registry->service_id);
+	rte_service_component_unregister(registry->polling.service.id);
 }
 
 static struct sfc_rxq_info *
@@ -438,6 +509,18 @@ sfc_counter_rxq_info_get(struct sfc_adapter *sa)
 	return &sfc_sa2shared(sa)->rxq_info[sa->counter_rxq.sw_index];
 }
 
+static void
+sfc_mae_counter_registry_prepare(struct sfc_mae_counter_registry *registry,
+				 struct sfc_adapter *sa,
+				 uint32_t counter_stream_flags)
+{
+	registry->rx_pkt_burst = sa->eth_dev->rx_pkt_burst;
+	registry->rx_dp = sfc_counter_rxq_info_get(sa)->dp;
+	registry->pushed_n_buffers = 0;
+	registry->use_credits = counter_stream_flags &
+		EFX_MAE_COUNTERS_STREAM_OUT_USES_CREDITS;
+}
+
 static int
 sfc_mae_counter_service_register(struct sfc_adapter *sa,
 				 uint32_t counter_stream_flags)
@@ -456,13 +539,10 @@ sfc_mae_counter_service_register(struct sfc_adapter *sa,
 	memset(&service, 0, sizeof(service));
 	rte_strscpy(service.name, counter_service_name, sizeof(service.name));
 	service.socket_id = sa->socket_id;
-	service.callback = sfc_mae_counter_routine;
+	service.callback = sfc_mae_counter_service_routine;
 	service.callback_userdata = sa;
-	counter_registry->rx_pkt_burst = sa->eth_dev->rx_pkt_burst;
-	counter_registry->rx_dp = sfc_counter_rxq_info_get(sa)->dp;
-	counter_registry->pushed_n_buffers = 0;
-	counter_registry->use_credits = counter_stream_flags &
-		EFX_MAE_COUNTERS_STREAM_OUT_USES_CREDITS;
+	sfc_mae_counter_registry_prepare(counter_registry, sa,
+					 counter_stream_flags);
 
 	cid = sfc_get_service_lcore(sa->socket_id);
 	if (cid == RTE_MAX_LCORE && sa->socket_id != SOCKET_ID_ANY) {
@@ -520,8 +600,9 @@ sfc_mae_counter_service_register(struct sfc_adapter *sa,
 		goto fail_runstate_set;
 	}
 
-	counter_registry->service_core_id = cid;
-	counter_registry->service_id = sid;
+	counter_registry->polling_mode = SFC_MAE_COUNTER_POLLING_SERVICE;
+	counter_registry->polling.service.core_id = cid;
+	counter_registry->polling.service.id = sid;
 
 	sfc_log_init(sa, "done");
 
@@ -544,6 +625,47 @@ sfc_mae_counter_service_register(struct sfc_adapter *sa,
 	return rc;
 }
 
+static void
+sfc_mae_counter_thread_stop(struct sfc_adapter *sa)
+{
+	struct sfc_mae_counter_registry *counter_registry =
+		&sa->mae.counter_registry;
+	int rc;
+
+	/* Ensure that flag is set before attempting to join thread */
+	__atomic_store_n(&counter_registry->polling.thread.run, false,
+			 __ATOMIC_RELEASE);
+
+	rc = pthread_join(counter_registry->polling.thread.id, NULL);
+	if (rc != 0)
+		sfc_err(sa, "failed to join the MAE counter polling thread");
+
+	counter_registry->polling_mode = SFC_MAE_COUNTER_POLLING_OFF;
+}
+
+static int
+sfc_mae_counter_thread_spawn(struct sfc_adapter *sa,
+			     uint32_t counter_stream_flags)
+{
+	struct sfc_mae_counter_registry *counter_registry =
+		&sa->mae.counter_registry;
+	int rc;
+
+	sfc_log_init(sa, "entry");
+
+	sfc_mae_counter_registry_prepare(counter_registry, sa,
+					 counter_stream_flags);
+
+	counter_registry->polling_mode = SFC_MAE_COUNTER_POLLING_THREAD;
+	counter_registry->polling.thread.run = true;
+
+	rc = rte_ctrl_thread_create(&sa->mae.counter_registry.polling.thread.id,
+				    "mae_counter_thread", NULL,
+				    sfc_mae_counter_thread, sa);
+
+	return rc;
+}
+
 int
 sfc_mae_counters_init(struct sfc_mae_counters *counters,
 		      uint32_t nb_counters_max)
@@ -754,7 +876,15 @@ sfc_mae_counter_stop(struct sfc_adapter *sa)
 		return;
 	}
 
-	sfc_mae_counter_service_unregister(sa);
+	SFC_ASSERT(mae->counter_registry.polling_mode !=
+			SFC_MAE_COUNTER_POLLING_OFF);
+
+	if (mae->counter_registry.polling_mode ==
+			SFC_MAE_COUNTER_POLLING_SERVICE)
+		sfc_mae_counter_service_unregister(sa);
+	else
+		sfc_mae_counter_thread_stop(sa);
+
 	efx_mae_counters_stream_stop(sa->nic, sa->counter_rxq.sw_index, NULL);
 
 	mae->counter_rxq_running = false;
@@ -787,15 +917,22 @@ sfc_mae_counter_start(struct sfc_adapter *sa)
 
 	sfc_log_init(sa, "stream start flags: 0x%x", flags);
 
-	rc = sfc_mae_counter_service_register(sa, flags);
-	if (rc != 0)
-		goto fail_service_register;
+	if (sfc_mae_counter_get_service_lcore(sa) != RTE_MAX_LCORE) {
+		rc = sfc_mae_counter_service_register(sa, flags);
+		if (rc != 0)
+			goto fail_service_register;
+	} else {
+		rc = sfc_mae_counter_thread_spawn(sa, flags);
+		if (rc != 0)
+			goto fail_thread_spawn;
+	}
 
 	mae->counter_rxq_running = true;
 
 	return 0;
 
 fail_service_register:
+fail_thread_spawn:
 	efx_mae_counters_stream_stop(sa->nic, sa->counter_rxq.sw_index, NULL);
 
 fail_counter_stream:
-- 
2.30.2