rx,tx: use one node per port queue

Instead of dealing with a list of queues to poll packets from,
instantiate one port_rx node clone for each configured port RX queue.

Do the same thing for port_tx and each configured port TX queue.

Depending on worker RX queue mapping, include the associated
port_rx-p*q* clones and initialize their context data accordingly.

Always include *all* port_tx-p*q* clones in all graphs, regardless of
worker TX queue mappings. Instead, configure the port_output node
context to steer packets to the proper port_tx-p*q* clone.

All of this requires delicate juggling with multiple node names lists to
ensure that nodes are cloned when needed before they are affected to
a graph. Secondly, we need to call rte_node_free() on those which are
not used in any graph after all of them have been reloaded.

Signed-off-by: Robin Jarry <[email protected]>

Author: rjarry

docs/graph.svg

@@ -57,7 +57,8 @@ static gr_vec struct gr_infra_stat *graph_stats(uint16_t cpu_id) {
 				memccpy(stat.name, name, 0, sizeof(stat.name));
 				gr_vec_add(stats, stat);
 			}
-			if (strcmp(name, "port_rx") == 0 || strcmp(name, "control_input") == 0)
+			if (strncmp(name, "port_rx-", strlen("port_rx-")) == 0
+			    || strcmp(name, "control_input") == 0)
 				pkts += n->packets;
 			node_cycles += n->cycles;
 		}

modules/infra/api/stats.c

@@ -11,6 +11,7 @@
 #include <gr_port.h>
 #include <gr_queue.h>
 #include <gr_rxtx.h>
+#include <gr_string.h>
 #include <gr_vec.h>
 #include <gr_worker.h>
 
@@ -25,7 +26,12 @@
 #include <unistd.h>
 
 struct node_infos node_infos = STAILQ_HEAD_INITIALIZER(node_infos);
-static gr_vec const char **node_names;
+static gr_vec const char **base_node_names;
+static gr_vec char **rx_node_names;
+static gr_vec char **tx_node_names;
+static rte_node_t port_rx_node;
+static rte_node_t port_tx_node;
+static rte_node_t port_output_node;
 
 rte_edge_t gr_node_attach_parent(const char *parent, const char *node) {
 	rte_node_t parent_id;
@@ -68,15 +74,17 @@ void worker_graph_free(struct worker *worker) {
 	}
 }
 
-static int worker_graph_new(struct worker *worker, uint8_t index) {
-	struct rx_node_queues *rx = NULL;
-	struct tx_node_queues *tx = NULL;
-	char name[RTE_GRAPH_NAMESIZE];
+static int
+worker_graph_new(struct worker *worker, uint8_t index, gr_vec struct iface_info_port **ports) {
+	gr_vec const char **graph_nodes = NULL;
+	char graph_name[RTE_GRAPH_NAMESIZE];
+	char node_name[RTE_NODE_NAMESIZE];
+	gr_vec char **rx_nodes = NULL;
 	struct queue_map *qmap;
+	struct rte_node *node;
 	uint16_t graph_uid;
 	unsigned n_rxqs;
-	size_t len;
-	int ret;
+	int ret = 0;
 
 	n_rxqs = 0;
 	gr_vec_foreach_ref (qmap, worker->rxqs) {
@@ -90,16 +98,9 @@ static int worker_graph_new(struct worker *worker, uint8_t index) {
 
 	// unique suffix for this graph
 	graph_uid = (worker->cpu_id << 1) | (0x1 & index);
-	snprintf(name, sizeof(name), "gr-%04x", graph_uid);
+	snprintf(graph_name, sizeof(graph_name), "gr-%04x", graph_uid);
 
-	// build rx & tx nodes data
-	len = sizeof(*rx) + n_rxqs * sizeof(struct rx_port_queue);
-	rx = rte_malloc(__func__, len, RTE_CACHE_LINE_SIZE);
-	if (rx == NULL) {
-		ret = -ENOMEM;
-		goto err;
-	}
-	rx->n_queues = 0;
+	// generate graph nodes list
 	gr_vec_foreach_ref (qmap, worker->rxqs) {
 		if (!qmap->enabled)
 			continue;
@@ -108,19 +109,61 @@ static int worker_graph_new(struct worker *worker, uint8_t index) {
 		    worker->cpu_id,
 		    qmap->port_id,
 		    qmap->queue_id);
-		rx->queues[rx->n_queues].port_id = qmap->port_id;
-		rx->queues[rx->n_queues].rxq_id = qmap->queue_id;
-		rx->n_queues++;
+
+		char *name = astrcat(NULL, RX_NODE_FMT, qmap->port_id, qmap->queue_id);
+		gr_vec_add(rx_nodes, name);
+		gr_vec_add(graph_nodes, name);
 	}
-	rx->burst_size = RTE_GRAPH_BURST_SIZE / rx->n_queues;
 
-	tx = rte_malloc(__func__, sizeof(*tx), RTE_CACHE_LINE_SIZE);
-	if (tx == NULL) {
-		ret = -ENOMEM;
-		goto err;
+	gr_vec_extend(graph_nodes, base_node_names);
+	gr_vec_extend(graph_nodes, tx_node_names);
+
+	// graph init
+	struct rte_graph_param params = {
+		.socket_id = rte_lcore_to_socket_id(worker->lcore_id),
+		.nb_node_patterns = gr_vec_len(graph_nodes),
+		.node_patterns = graph_nodes,
+	};
+	if (rte_graph_create(graph_name, &params) == RTE_GRAPH_ID_INVALID) {
+		if (rte_errno == 0)
+			rte_errno = EINVAL;
+		ret = -rte_errno;
+		goto out;
 	}
-	// initialize all to invalid queue_ids
-	memset(tx, 0xff, sizeof(*tx));
+	worker->graph[index] = rte_graph_lookup(graph_name);
+
+	// set rx nodes context data
+	gr_vec_foreach_ref (qmap, worker->rxqs) {
+		if (!qmap->enabled)
+			continue;
+		snprintf(node_name, sizeof(node_name), RX_NODE_FMT, qmap->port_id, qmap->queue_id);
+		node = rte_graph_node_get_by_name(graph_name, node_name);
+		struct rx_node_ctx *ctx = (struct rx_node_ctx *)node->ctx;
+		gr_vec_foreach (struct iface_info_port *p, ports) {
+			if (p->port_id == qmap->port_id) {
+				ctx->iface = RTE_PTR_SUB(p, offsetof(struct iface, info));
+				break;
+			}
+		}
+		assert(ctx->iface != NULL);
+		ctx->rxq.port_id = qmap->port_id;
+		ctx->rxq.queue_id = qmap->queue_id;
+		ctx->burst_size = RTE_GRAPH_BURST_SIZE / gr_vec_len(worker->rxqs);
+	}
+
+	// initialize all tx nodes context to invalid ports and queues
+	gr_vec_foreach (const char *name, tx_node_names) {
+		node = rte_graph_node_get_by_name(graph_name, name);
+		struct port_queue *ctx = (struct port_queue *)node->ctx;
+		ctx->port_id = UINT16_MAX;
+		ctx->queue_id = UINT16_MAX;
+	}
+
+	// initialize the port_output node context to point to invalid edges
+	struct port_output_edges *out = rte_malloc(__func__, sizeof(*out), RTE_CACHE_LINE_SIZE);
+	for (unsigned i = 0; i < ARRAY_DIM(out->edges); i++)
+		out->edges[i] = RTE_EDGE_ID_INVALID;
+
 	gr_vec_foreach_ref (qmap, worker->txqs) {
 		if (!qmap->enabled)
 			continue;
@@ -129,38 +172,39 @@ static int worker_graph_new(struct worker *worker, uint8_t index) {
 		    worker->cpu_id,
 		    qmap->port_id,
 		    qmap->queue_id);
-		tx->txq_ids[qmap->port_id] = qmap->queue_id;
+		// find the corresponding port_tx clone for this port-txq pair
+		snprintf(node_name, sizeof(node_name), TX_NODE_FMT, qmap->port_id, qmap->queue_id);
+		node = rte_graph_node_get_by_name(graph_name, node_name);
+		// and update its context data to correct values
+		struct port_queue *ctx = (struct port_queue *)node->ctx;
+		ctx->port_id = qmap->port_id;
+		ctx->queue_id = qmap->queue_id;
+
+		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
+				// correct edge
+				out->edges[ctx->port_id] = edge;
+				break;
+			}
+		}
 	}
 
-	// graph init
-	struct rte_graph_param params = {
-		.socket_id = rte_lcore_to_socket_id(worker->lcore_id),
-		.nb_node_patterns = gr_vec_len(node_names),
-		.node_patterns = (const char **)node_names,
-	};
-	if (rte_graph_create(name, &params) == RTE_GRAPH_ID_INVALID) {
-		if (rte_errno == 0)
-			rte_errno = EINVAL;
-		ret = -rte_errno;
-		goto err;
-	}
-	worker->graph[index] = rte_graph_lookup(name);
+	// finally, set the port_output context data
+	rte_graph_node_get_by_name(graph_name, "port_output")->ctx_ptr = out;
 
-	rte_graph_node_get_by_name(name, "port_rx")->ctx_ptr = rx;
-	rte_graph_node_get_by_name(name, "port_tx")->ctx_ptr = tx;
+out:
+	gr_vec_free(graph_nodes);
+	gr_strvec_free(rx_nodes);
 
-	return 0;
-err:
-	rte_free(rx);
-	rte_free(tx);
 	return errno_set(-ret);
 }
 
-int worker_graph_reload(struct worker *worker) {
+int worker_graph_reload(struct worker *worker, gr_vec struct iface_info_port **ports) {
 	unsigned next = !atomic_load(&worker->cur_config);
 	int ret;
 
-	if ((ret = worker_graph_new(worker, next)) < 0)
+	if ((ret = worker_graph_new(worker, next, ports)) < 0)
 		return errno_log(-ret, "worker_graph_new");
 
 	// wait for datapath worker to pickup the config update
@@ -181,15 +225,95 @@ int worker_graph_reload(struct worker *worker) {
 	return 0;
 }
 
-int worker_graph_reload_all(void) {
+static char *ensure_queue_node(
+	rte_node_t base_node,
+	const char *name_fmt,
+	uint16_t port_id,
+	uint16_t queue_id,
+	gr_vec char **old_list
+) {
+	char *name = astrcat(NULL, name_fmt, port_id, queue_id);
+	assert(name != NULL);
+
+	// remove node from the old list so that only unused nodes remain
+	for (unsigned i = 0; i < gr_vec_len(old_list); i++) {
+		if (strcmp(old_list[i], name) == 0) {
+			free(old_list[i]);
+			gr_vec_del(old_list, i);
+			break;
+		}
+	}
+	if (rte_node_from_name(name) == RTE_NODE_ID_INVALID) {
+		// node does not exist yet, clone it from the base
+		rte_node_t node = rte_node_clone(base_node, strstr(name, "-") + 1);
+		assert(node != RTE_NODE_ID_INVALID);
+	}
+
+	return name;
+}
+
+static gr_vec rte_node_t *worker_graph_nodes_add_missing(gr_vec struct iface_info_port **ports) {
+	gr_vec char **old_rx = gr_vec_clone(rx_node_names);
+	gr_vec char **old_tx = gr_vec_clone(tx_node_names);
+	rte_node_t *unused_nodes = NULL;
+	rte_edge_t edge;
+	char *name;
+
+	gr_vec_free(rx_node_names);
+	gr_vec_free(tx_node_names);
+
+	// clone port_rx and port_tx to match all possible port-queue pairs
+	gr_vec_foreach (struct iface_info_port *port, ports) {
+		for (uint16_t rxq = 0; rxq < port->n_rxq; rxq++) {
+			name = ensure_queue_node(
+				port_rx_node, RX_NODE_FMT, port->port_id, rxq, old_rx
+			);
+			gr_vec_add(rx_node_names, name);
+		}
+		for (uint16_t txq = 0; txq < port->n_txq; txq++) {
+			name = ensure_queue_node(
+				port_tx_node, TX_NODE_FMT, port->port_id, txq, old_tx
+			);
+			gr_vec_add(tx_node_names, name);
+		}
+	}
+
+	// update the port_output edges to allow reaching all possible port_tx clones
+	edge = rte_node_edge_update(
+		port_output_node, 0, (const char **)tx_node_names, gr_vec_len(tx_node_names)
+	);
+	assert(edge != RTE_EDGE_ID_INVALID);
+	edge = rte_node_edge_shrink(port_output_node, gr_vec_len(tx_node_names));
+	assert(edge != RTE_EDGE_ID_INVALID);
+
+	// store all unused node_ids in a list to be returned to the caller
+	gr_vec_foreach (name, old_rx)
+		gr_vec_add(unused_nodes, rte_node_from_name(name));
+	gr_vec_foreach (name, old_tx)
+		gr_vec_add(unused_nodes, rte_node_from_name(name));
+	gr_strvec_free(old_rx);
+	gr_strvec_free(old_tx);
+
+	return unused_nodes;
+}
+
+int worker_graph_reload_all(gr_vec struct iface_info_port **ports) {
 	struct worker *worker;
 	int ret;
 
+	gr_vec rte_node_t *unused_nodes = worker_graph_nodes_add_missing(ports);
+
 	STAILQ_FOREACH (worker, &workers, next) {
-		if ((ret = worker_graph_reload(worker)) < 0)
+		if ((ret = worker_graph_reload(worker, ports)) < 0)
 			return ret;
 	}
 
+	// these port_rx and port_tx clones are now not referenced in any graph
+	// we can safely delete them
+	gr_vec_foreach (rte_node_t node, unused_nodes)
+		rte_node_free(node);
+	gr_vec_free(unused_nodes);
+
 	return 0;
 }
 
@@ -227,6 +351,9 @@ static struct api_out graph_dump(const void *request, struct api_ctx *) {
 		const char *name = info->node->name;
 		const char *attrs = "";
 
+		if (node_id == port_output_node)
+			nb_edges = info->node->nb_edges;
+
 		if (!errors) {
 			if (nb_edges == 0)
 				continue;
@@ -252,7 +379,10 @@ static struct api_out graph_dump(const void *request, struct api_ctx *) {
 			continue;
 		if ((edges = calloc(nb_edges, sizeof(char *))) == NULL)
 			goto err;
-		if (rte_node_edge_get(node_id, edges) == RTE_EDGE_ID_INVALID)
+		if (node_id == port_output_node) {
+			for (unsigned i = 0; i < nb_edges; i++)
+				edges[i] = (char *)info->node->next_nodes[i];
+		} else if (rte_node_edge_get(node_id, edges) == RTE_EDGE_ID_INVALID)
 			goto err;
 
 		for (unsigned i = 0; i < nb_edges; i++) {
@@ -270,7 +400,7 @@ static struct api_out graph_dump(const void *request, struct api_ctx *) {
 					continue;
 
 				rte_node_t id = rte_node_from_name(n->node->name);
-				if (rte_node_edge_count(id) == 0) {
+				if (id != port_output_node && rte_node_edge_count(id) == 0) {
 					if (!errors)
 						goto skip;
 					node_attrs = " [color=darkorange]";
@@ -326,7 +456,16 @@ static void graph_init(struct event_base *) {
 		reg->id = __rte_node_register(reg);
 		if (reg->id == RTE_NODE_ID_INVALID)
 			ABORT("__rte_node_register(%s): %s", reg->name, rte_strerror(rte_errno));
-		gr_vec_add(node_names, reg->name);
+
+		if (strcmp(reg->name, RX_NODE_BASE) == 0)
+			port_rx_node = reg->id;
+		else if (strcmp(reg->name, TX_NODE_BASE) == 0)
+			port_tx_node = reg->id;
+		else
+			gr_vec_add(base_node_names, reg->name);
+
+		if (strcmp(reg->name, "port_output") == 0)
+			port_output_node = reg->id;
 	}
 
 	// then, invoke all registration callbacks where applicable
@@ -346,8 +485,9 @@ static void graph_fini(struct event_base *) {
 		}
 	}
 
-	gr_vec_free(node_names);
-	node_names = NULL;
+	gr_vec_free(base_node_names);
+	gr_strvec_free(rx_node_names);
+	gr_strvec_free(tx_node_names);
 }
 
 static struct gr_module graph_module = {

modules/infra/control/graph.c

@@ -3,8 +3,13 @@
 
 #pragma once
 
+#include <gr_port.h>
+#include <gr_vec.h>
 #include <gr_worker.h>
 
-int worker_graph_reload(struct worker *);
-int worker_graph_reload_all(void);
+#include <rte_graph.h>
+
+int worker_graph_reload(struct worker *, gr_vec struct iface_info_port **);
+int worker_graph_reload_all(gr_vec struct iface_info_port **);
+void worker_graph_nodes_del_unused(gr_vec rte_node_t *);
 void worker_graph_free(struct worker *);