infra: add TX queue sharing support

Some network devices report fewer TX queues than the number of datapath
workers. Cap the TX queue count to the device maximum and distribute
workers across available queues using modulo assignment.

When multiple workers share the same TX queue, protect rte_eth_tx_burst()
with a per-queue spinlock stored in iface_info_port. The lock pointer in
tx_node_ctx is only set for shared queues to avoid locking overhead when
a worker has exclusive access.

Update unit tests to set max_tx_queues in the mock device info and
adjust expected TX queue assignments for modulo wrapping.

Signed-off-by: Robin Jarry <rjarry@redhat.com>

Author: rjarry

modules/infra/control/gr_port.h

@@ -9,6 +9,7 @@
 #include <rte_ethdev.h>
 #include <rte_ether.h>
 #include <rte_mempool.h>
+#include <rte_spinlock.h>
 
 #include <stdint.h>
 #include <sys/queue.h>
@@ -32,6 +33,7 @@ GR_IFACE_INFO(GR_IFACE_TYPE_PORT, iface_info_port, {
 	struct rte_mempool *pool;
 	char *devargs;
 	uint32_t pool_size;
+	rte_spinlock_t txq_locks[RTE_MAX_QUEUES_PER_PORT];
 	struct {
 		mac_filter_flags_t flags;
 		unsigned hw_limit;

modules/infra/control/graph.c

@@ -164,6 +164,7 @@ worker_graph_new(struct worker *worker, uint8_t index, gr_vec struct iface_info_
 		struct tx_node_ctx *ctx = tx_node_ctx(node);
 		ctx->txq.port_id = UINT16_MAX;
 		ctx->txq.queue_id = UINT16_MAX;
+		ctx->lock = NULL;
 	}
 
 	// initialize the port_output node context to point to invalid edges
@@ -191,6 +192,26 @@ worker_graph_new(struct worker *worker, uint8_t index, gr_vec struct iface_info_
 		ctx->txq.port_id = qmap->port_id;
 		ctx->txq.queue_id = qmap->queue_id;
 
+		// set spinlock only if multiple workers share this TX queue
+		unsigned txq_users = 0;
+		struct worker *w = NULL;
+		STAILQ_FOREACH (w, &workers, next) {
+			gr_vec_foreach_ref (struct queue_map *q, w->txqs) {
+				if (q->port_id == qmap->port_id && q->queue_id == qmap->queue_id)
+					txq_users++;
+			}
+		}
+		if (txq_users > 1) {
+			struct iface_info_port *port = find_port(ports, qmap->port_id);
+			ctx->lock = &port->txq_locks[qmap->queue_id];
+			LOG(WARNING,
+			    "[CPU %d] port %s txq %u shared by %u workers",
+			    worker->cpu_id,
+			    port->devargs,
+			    qmap->queue_id,
+			    txq_users);
+		}
+
 		for (rte_edge_t edge = 0; edge < gr_vec_len(tx_node_names); edge++) {
 			if (strcmp(tx_node_names[edge], node_name) == 0) {
 				// update the port_output context data to map this port to the

modules/infra/control/port.c

@@ -91,19 +91,26 @@ int port_configure(struct iface_info_port *p, uint16_t n_txq_min) {
 	if (numa_available() != -1)
 		socket_id = rte_eth_dev_socket_id(p->port_id);
 
-	// FIXME: deal with drivers that do not support more than 1 (or N) tx queues
-	p->n_txq = n_txq_min;
+	if ((ret = rte_eth_dev_info_get(p->port_id, &info)) < 0)
+		return errno_log(-ret, "rte_eth_dev_info_get");
+
 	if (p->n_rxq == 0)
 		p->n_rxq = 1;
 
-	if ((ret = rte_eth_dev_info_get(p->port_id, &info)) < 0)
-		return errno_log(-ret, "rte_eth_dev_info_get");
+	// cap number of queues to device maximum
+	p->n_txq = RTE_MIN(n_txq_min, info.max_tx_queues);
 
 	if (strcmp(info.driver_name, "net_tap") == 0) {
 		p->n_txq = RTE_MAX(p->n_txq, p->n_rxq);
 		p->n_rxq = p->n_txq;
 	}
 
+	if (p->n_txq < n_txq_min)
+		LOG(NOTICE, "port %s TX queues limited to %u", p->devargs, p->n_txq);
+
+	for (uint16_t q = 0; q < p->n_txq; q++)
+		rte_spinlock_init(&p->txq_locks[q]);
+
 	rxq_size = get_rxq_size(p, &info);
 	txq_size = get_txq_size(p, &info);
 

modules/infra/control/worker.c

@@ -236,9 +236,10 @@ static void worker_txq_distribute(gr_vec struct iface_info_port **ports) {
 		STAILQ_FOREACH (worker, &workers, next) {
 			if (gr_vec_len(worker->rxqs) == 0)
 				continue;
+			assert(port->n_txq > 0);
 			struct queue_map txq = {
 				.port_id = port->port_id,
-				.queue_id = txq_idx,
+				.queue_id = txq_idx % port->n_txq,
 				.enabled = port->started,
 			};
 			gr_vec_add(worker->txqs, txq);

modules/infra/control/worker_test.c

@@ -26,7 +26,12 @@ static struct worker w2 = {.cpu_id = 2};
 static struct worker w3 = {.cpu_id = 3};
 static struct worker w4 = {.cpu_id = 4};
 static struct worker w5 = {.cpu_id = 5};
-static struct rte_eth_dev_info dev_info = {.driver_name = "net_null", .nb_rx_queues = 2};
+static struct rte_eth_dev_info dev_info = {
+	.driver_name = "net_null",
+	.nb_rx_queues = 2,
+	.max_rx_queues = 4,
+	.max_tx_queues = 4,
+};
 
 // mocked types/functions
 int gr_rte_log_type;
@@ -167,6 +172,7 @@ static int setup(void **) {
 		port->started = true;
 		port->port_id = i;
 		port->n_rxq = 2;
+		port->n_txq = 3;
 		ifaces[i] = iface;
 	}
 	STAILQ_INSERT_TAIL(&workers, &w1, next);
@@ -401,14 +407,9 @@ static void queue_distribute_increase(void **) {
 	assert_qmaps(w5.rxqs, q(2, 0));
 	assert_qmaps(w1.txqs, q(0, 2), q(1, 2), q(2, 2));
 	assert_qmaps(w2.txqs, q(0, 3), q(1, 3), q(2, 3));
-	assert_qmaps(w3.txqs, q(0, 4), q(1, 4), q(2, 4));
+	assert_qmaps(w3.txqs, q(0, 0), q(1, 0), q(2, 0));
 	assert_qmaps(w4.txqs, q(0, 0), q(1, 0), q(2, 0));
 	assert_qmaps(w5.txqs, q(0, 1), q(1, 1), q(2, 1));
-
-	for (unsigned i = 0; i < ARRAY_DIM(ifaces); i++) {
-		struct iface_info_port *p = iface_info_port(ifaces[i]);
-		assert_int_equal(p->n_txq, CPU_COUNT(&affinity));
-	}
 }
 
 int main(void) {

modules/infra/datapath/gr_rxtx.h

@@ -8,6 +8,7 @@
 
 #include <rte_build_config.h>
 #include <rte_graph.h>
+#include <rte_spinlock.h>
 
 #include <stddef.h>
 #include <stdint.h>
@@ -28,7 +29,10 @@ GR_NODE_CTX_TYPE(rx_node_ctx, {
 	uint16_t burst_size;
 });
 
-GR_NODE_CTX_TYPE(tx_node_ctx, { struct port_queue txq; });
+GR_NODE_CTX_TYPE(tx_node_ctx, {
+	struct port_queue txq;
+	rte_spinlock_t *lock;
+});
 
 struct port_output_edges {
 	rte_edge_t edges[RTE_MAX_ETHPORTS];

modules/infra/datapath/port_tx.c

@@ -13,6 +13,7 @@
 #include <rte_build_config.h>
 #include <rte_ethdev.h>
 #include <rte_malloc.h>
+#include <rte_spinlock.h>
 
 #include <stdint.h>
 
@@ -53,7 +54,14 @@ tx_process(struct rte_graph *graph, struct rte_node *node, void **objs, uint16_t
 		}
 	}
 
-	tx_ok = rte_eth_tx_burst(ctx->txq.port_id, ctx->txq.queue_id, mbufs, nb_objs);
+	if (likely(ctx->lock == NULL)) {
+		tx_ok = rte_eth_tx_burst(ctx->txq.port_id, ctx->txq.queue_id, mbufs, nb_objs);
+	} else {
+		rte_spinlock_lock(ctx->lock);
+		tx_ok = rte_eth_tx_burst(ctx->txq.port_id, ctx->txq.queue_id, mbufs, nb_objs);
+		rte_spinlock_unlock(ctx->lock);
+	}
+
 	if (tx_ok < nb_objs)
 		rte_node_enqueue(graph, node, TX_ERROR, &objs[tx_ok], nb_objs - tx_ok);