Control Plane Refactor and Pipeline Mode Integration

modules/infra/api/affinity.c

@@ -3,7 +3,6 @@
 
 #include <gr_api.h>
 #include <gr_config.h>
-#include <gr_control_output.h>
 #include <gr_infra.h>
 #include <gr_module.h>
 #include <gr_port.h>
@@ -38,10 +37,6 @@ static struct api_out affinity_set(const void *request, void ** /*response*/) {
 		if (ret < 0)
 			goto out;
 
-		ret = -control_output_set_affinity(CPU_SETSIZE, &req->control_cpus);
-		if (ret < 0)
-			goto out;
-
 		gr_config.control_cpus = req->control_cpus;
 	}
 	if (CPU_COUNT(&req->datapath_cpus) > 0) {

modules/infra/api/stats.c

@@ -50,7 +50,17 @@ static struct api_out stats_get(const void *request, void **response) {
 				continue;
 			for (unsigned i = 0; i < w_stats->n_stats; i++) {
 				const struct node_stats *n = &w_stats->stats[i];
-				const char *name = rte_node_id_to_name(n->node_id);
+				const struct rte_node_register *nr = gr_node_info_get(n->node_id)
+									     ->node;
+				char name[RTE_NODE_NAMESIZE];
+				snprintf(
+					name,
+					RTE_NODE_NAMESIZE,
+					"%s%c",
+					rte_node_id_to_name(n->node_id),
+					nr->flags & GR_NODE_FLAG_CONTROL_PLANE ? '*' : '\0'
+				);
+
 				s = find_stat(stats, name);
 				if (s != NULL) {
 					s->objs += n->objs;
@@ -66,6 +76,39 @@ static struct api_out stats_get(const void *request, void **response) {
 					gr_vec_add(stats, stat);
 				}
 			}
+
+			if (worker->stats_ctl)
+				for (unsigned i = 0; i < worker->stats_ctl->n_stats; i++) {
+					const struct node_stats *n = &w_stats->stats[i];
+					const struct rte_node_register *nr = gr_node_info_get(
+										     n->node_id
+					)
+										     ->node;
+					char name[RTE_NODE_NAMESIZE];
+					snprintf(
+						name,
+						RTE_NODE_NAMESIZE,
+						"%s%c",
+						rte_node_id_to_name(n->node_id),
+						nr->flags & GR_NODE_FLAG_CONTROL_PLANE ? '*' : '\0'
+					);
+
+					s = find_stat(stats, name);
+					if (s != NULL) {
+						s->objs += n->objs;
+						s->calls += n->calls;
+						s->cycles += n->cycles;
+					} else {
+						struct stat stat = {
+							.objs = n->objs,
+							.calls = n->calls,
+							.cycles = n->cycles,
+						};
+						memccpy(stat.name, name, 0, sizeof(stat.name));
+						gr_vec_add(stats, stat);
+					}
+				}
+
 			s = find_stat(stats, "idle");
 			if (s != NULL) {
 				s->calls += w_stats->n_sleeps;

modules/infra/control/gr_graph.h

@@ -10,6 +10,8 @@
 
 #include <sys/queue.h>
 
+#define RTE_GRAPH_MODEL_SELECT RTE_GRAPH_MODEL_MCORE_DISPATCH
+
 #ifdef __GROUT_UNIT_TEST__
 #include <gr_cmocka.h>
 
@@ -39,6 +41,8 @@ int drop_format(char *buf, size_t buf_len, const void *data, size_t data_len);
 typedef void (*gr_node_register_cb_t)(void);
 
 struct gr_node_info {
+// Flag used in node->flags to specify that a node is running on a control plane thread
+#define GR_NODE_FLAG_CONTROL_PLANE (1 << 31)
 	struct rte_node_register *node;
 	gr_node_register_cb_t register_callback;
 	gr_node_register_cb_t unregister_callback;

modules/infra/control/gr_loopback.h

@@ -7,6 +7,5 @@
 
 #include <rte_byteorder.h>
 
-void loopback_tx(struct rte_mbuf *m);
 control_input_t loopback_get_control_id(void);
 void loopback_input_add_type(rte_be16_t eth_type, const char *next_node);

modules/infra/control/gr_worker.h

@@ -62,6 +62,11 @@ struct worker {
 	pthread_t thread;
 	struct queue_map *rxqs;
 	struct queue_map *txqs;
+
+	struct rte_graph *base[2]; // Base graph, not walked
+	struct rte_graph *ctl_graph[2]; // graph used to process ctl packets
+	_Atomic(const struct worker_stats *) stats_ctl;
+
 	STAILQ_ENTRY(worker) next;
 } __rte_cache_aligned;