node: add group_id property

Nodes with the same group id are scheduled by the same driver, even
if they are not in any way linked to eachother.

src/pipewire/context.c

@@ -797,7 +797,7 @@ static int ensure_state(struct pw_impl_node *node, bool running)
 	return pw_impl_node_set_state(node, state);
 }
 
-static int collect_nodes(struct pw_impl_node *driver)
+static int collect_nodes(struct pw_context *context, struct pw_impl_node *driver)
 {
 	struct spa_list queue;
 	struct pw_impl_node *n, *t;
@@ -811,10 +811,13 @@ static int collect_nodes(struct pw_impl_node *driver)
 
 	pw_log_debug("driver %p: '%s'", driver, driver->name);
 
+	/* start with driver in the queue */
 	spa_list_init(&queue);
 	spa_list_append(&queue, &driver->sort_link);
 	driver->visited = true;
 
+	/* now follow all the links from the nodes in the queue
+	 * and add the peers to the queue. */
 	spa_list_consume(n, &queue, sort_link) {
 		spa_list_remove(&n->sort_link);
 		pw_impl_node_set_driver(n, driver);
@@ -850,6 +853,19 @@ static int collect_nodes(struct pw_impl_node *driver)
 				}
 			}
 		}
+		/* now go through all the followers of this driver and add the
+		 * nodes that have the same group and that are not yet visited */
+		if (n->group_id == SPA_ID_INVALID)
+			continue;
+
+		spa_list_for_each(t, &context->node_list, link) {
+			if (t->exported || t == n || !t->active || t->visited)
+				continue;
+			if (t->group_id != n->group_id)
+				continue;
+			t->visited = true;
+			spa_list_append(&queue, &t->sort_link);
+		}
 	}
 	return 0;
 }
@@ -877,7 +893,7 @@ int pw_context_recalc_graph(struct pw_context *context, const char *reason)
 			continue;
 
 		if (!n->visited)
-			collect_nodes(n);
+			collect_nodes(context, n);
 
 		/* from now on we are only interested in active driving nodes.
 		 * We're going to see if there are active followers. */

src/pipewire/impl-node.c

@@ -774,12 +774,25 @@ static void check_properties(struct pw_impl_node *node)
 	struct pw_context *context = node->context;
 	const char *str;
 	bool driver, do_recalc = false;
+	uint32_t group_id;
 
 	if ((str = pw_properties_get(node->properties, PW_KEY_PRIORITY_DRIVER))) {
 		node->priority_driver = pw_properties_parse_int(str);
 		pw_log_debug(NAME" %p: priority driver %d", node, node->priority_driver);
 	}
 
+	/* group_id defines what nodes are scheduled together */
+	if ((str = pw_properties_get(node->properties, PW_KEY_NODE_GROUP)))
+		group_id = pw_properties_parse_int(str);
+	else
+		group_id = SPA_ID_INVALID;
+
+	if (group_id != node->group_id) {
+		pw_log_debug(NAME" %p: group %u->%u", node, node->group_id, group_id);
+		node->group_id = group_id;
+		do_recalc = true;
+	}
+
 	if ((str = pw_properties_get(node->properties, PW_KEY_NODE_NAME)) &&
 	    (node->name == NULL || strcmp(node->name, str) != 0)) {
 		free(node->name);
@@ -797,11 +810,6 @@ static void check_properties(struct pw_impl_node *node)
 	else
 		driver = false;
 
-	if ((str = pw_properties_get(node->properties, PW_KEY_NODE_ALWAYS_PROCESS)))
-		node->want_driver = pw_properties_parse_bool(str);
-	else
-		node->want_driver = false;
-
 	if (node->driver != driver) {
 		pw_log_debug(NAME" %p: driver %d -> %d", node, node->driver, driver);
 		node->driver = driver;
@@ -811,8 +819,14 @@ static void check_properties(struct pw_impl_node *node)
 			else
 				spa_list_remove(&node->driver_link);
 		}
+		do_recalc = true;
 	}
 
+	if ((str = pw_properties_get(node->properties, PW_KEY_NODE_ALWAYS_PROCESS)))
+		node->want_driver = pw_properties_parse_bool(str);
+	else
+		node->want_driver = false;
+
 	if ((str = pw_properties_get(node->properties, PW_KEY_NODE_LATENCY))) {
 		uint32_t num, denom;
                 if (sscanf(str, "%u/%u", &num, &denom) == 2 && denom != 0) {
@@ -826,13 +840,14 @@ static void check_properties(struct pw_impl_node *node)
 				pw_log_info("(%s-%u) quantum %u/%u", node->name, node->info.id,
 						quantum_size, context->defaults.clock_rate);
 				node->quantum_size = quantum_size;
-				do_recalc |= node->active;
+				do_recalc = true;
 			}
 		}
 	}
-	pw_log_debug(NAME" %p: driver:%d recalc:%d", node, node->driver, do_recalc);
+	pw_log_debug(NAME" %p: driver:%d recalc:%d active:%d", node, node->driver,
+			do_recalc, node->active);
 
-	if (do_recalc)
+	if (do_recalc && node->active)
 		pw_context_recalc_graph(context, "quantum change");
 }
 
@@ -1043,6 +1058,7 @@ struct pw_impl_node *pw_context_create_node(struct pw_context *context,
 	this = &impl->this;
 	this->context = context;
 	this->name = strdup("node");
+	this->group_id = SPA_ID_INVALID;
 
 	if (user_data_size > 0)
                 this->user_data = SPA_MEMBER(impl, sizeof(struct impl), void);

src/pipewire/keys.h

@@ -127,6 +127,9 @@ extern "C" {
 								  *  nanoseconds. */
 
 #define PW_KEY_NODE_SESSION		"node.session"		/**< the session id this node is part of */
+#define PW_KEY_NODE_GROUP		"node.group"		/**< the group id this node is part of. Nodes
+								  *  in the same group are always scheduled
+								  *  with the same driver. */
 #define PW_KEY_NODE_EXCLUSIVE		"node.exclusive"	/**< node wants exclusive access to resources */
 #define PW_KEY_NODE_AUTOCONNECT		"node.autoconnect"	/**< node wants to be automatically connected
 								  *  to a compatible node */

src/pipewire/private.h

@@ -536,6 +536,7 @@ struct pw_impl_node {
 	char *name;				/** for debug */
 
 	uint32_t priority_driver;	/** priority for being driver */
+	uint32_t group_id;		/** group to schedule this node in */
 	uint32_t spa_flags;
 
 	unsigned int registered:1;