context: Only activate runnable nodes

Rework the runnable state calculation. It works in 2 steps now:

1. Collect all nodes linked to a driver in some way. Mark nodes that
   are reachable with a non-passive link as runnable.
2, Go through all runnable nodes and set all linked nodes to the
   runnable state as well, up to the driver.

Step 2 is new. Previously if there was just one runnable node, *all*
nodes would be set to runnable. With the addition of step 2, some nodes
might remain idle when they are not used.

This has the effect that virtual sinks without inputs stay idle when
the driver is otherwise running. Grouped nodes (like the RTP session)
will now also only run the linked nodes.

src/pipewire/context.c

@@ -768,13 +768,89 @@ error:
 static int ensure_state(struct pw_impl_node *node, bool running)
 {
 	enum pw_node_state state = node->info.state;
-	if (node->active && !SPA_FLAG_IS_SET(node->spa_flags, SPA_NODE_FLAG_NEED_CONFIGURE) && running)
+	if (node->active && node->runnable &&
+	    !SPA_FLAG_IS_SET(node->spa_flags, SPA_NODE_FLAG_NEED_CONFIGURE) && running)
 		state = PW_NODE_STATE_RUNNING;
 	else if (state > PW_NODE_STATE_IDLE)
 		state = PW_NODE_STATE_IDLE;
 	return pw_impl_node_set_state(node, state);
 }
 
+/* From a node (that is runnable) follow all prepared links and groups to
+ * active nodes up to the driver and make them recursively runnable as well.
+ *
+ * We stop at driver nodes so that other paths linked to the driver will stay
+ * unrunnable when no other runnable path exists.
+ */
+static inline int run_nodes(struct pw_context *context, struct pw_impl_node *node, struct spa_list *nodes)
+{
+	struct pw_impl_node *t;
+	struct pw_impl_port *p;
+	struct pw_impl_link *l;
+
+	if (!node->runnable)
+		return 0;
+
+	pw_log_debug("node %p: '%s'", node, node->name);
+
+	spa_list_for_each(p, &node->input_ports, link) {
+		spa_list_for_each(l, &p->links, input_link) {
+			t = l->output->node;
+
+			if (!t->active || !l->prepared || t->runnable)
+				continue;
+
+			pw_log_debug("  peer %p: '%s'", t, t->name);
+			t->runnable = true;
+			if (!t->driver)
+				run_nodes(context, t, nodes);
+		}
+	}
+	spa_list_for_each(p, &node->output_ports, link) {
+		spa_list_for_each(l, &p->links, output_link) {
+			t = l->input->node;
+
+			if (!t->active || !l->prepared || t->runnable)
+				continue;
+
+			pw_log_debug("  peer %p: '%s'", t, t->name);
+			t->runnable = true;
+			if (!t->driver)
+				run_nodes(context, t, nodes);
+		}
+	}
+	/* now go through all the nodes that have the same link group and
+	 * that are not yet visited. Note how nodes with the same group
+	 * don't get included here. They were added to the same driver but
+	 * need to otherwise stay idle unless some non-passive link activates
+	 * them. */
+	if (node->link_group != NULL) {
+		spa_list_for_each(t, nodes, sort_link) {
+			if (t->exported || !t->active || t->runnable)
+				continue;
+			if (!spa_streq(t->link_group, node->link_group))
+				continue;
+
+			pw_log_debug("  group %p: '%s'", t, t->name);
+			t->runnable = true;
+			if (!t->driver)
+				run_nodes(context, t, nodes);
+		}
+	}
+	return 0;
+}
+
+/* Follow all prepared links and groups from node, activate the links.
+ * If a non-passive link is found, we set the peer runnable flag.
+ *
+ * After this is done, we end up with a list of nodes in collect that are all
+ * linked to node.
+ * Some of the nodes have the runnable flag set. We then start from those nodes
+ * and make all linked nodes and groups runnable as well. (see run_nodes).
+ *
+ * This ensures that we only activate the paths from the runnable nodes to the
+ * driver nodes and leave the other nodes idle.
+ */
 static int collect_nodes(struct pw_context *context, struct pw_impl_node *node, struct spa_list *collect)
 {
 	struct spa_list queue;
@@ -809,16 +885,13 @@ static int collect_nodes(struct pw_context *context, struct pw_impl_node *node,
 
 				pw_impl_link_prepare(l);
 
-				if (!l->prepared)
+				if (!l->prepared || t->visited)
 					continue;
 
 				if (!l->passive)
-					n->runnable = true;
-
-				if (!t->visited) {
-					t->visited = true;
-					spa_list_append(&queue, &t->sort_link);
-				}
+					t->runnable = true;
+				t->visited = true;
+				spa_list_append(&queue, &t->sort_link);
 			}
 		}
 		spa_list_for_each(p, &n->output_ports, link) {
@@ -830,16 +903,13 @@ static int collect_nodes(struct pw_context *context, struct pw_impl_node *node,
 
 				pw_impl_link_prepare(l);
 
-				if (!l->prepared)
+				if (!l->prepared || t->visited)
 					continue;
 
 				if (!l->passive)
-					n->runnable = true;
-
-				if (!t->visited) {
-					t->visited = true;
-					spa_list_append(&queue, &t->sort_link);
-				}
+					t->runnable = true;
+				t->visited = true;
+				spa_list_append(&queue, &t->sort_link);
 			}
 		}
 		/* now go through all the nodes that have the same group and
@@ -857,7 +927,11 @@ static int collect_nodes(struct pw_context *context, struct pw_impl_node *node,
 				spa_list_append(&queue, &t->sort_link);
 			}
 		}
+		pw_log_debug(" next node %p: '%s' runnable:%u", n, n->name, n->runnable);
 	}
+	spa_list_for_each(n, collect, sort_link)
+		run_nodes(context, n, collect);
+
 	return 0;
 }
 
@@ -868,8 +942,7 @@ static void move_to_driver(struct pw_context *context, struct spa_list *nodes,
 	pw_log_debug("driver: %p %s runnable:%u", driver, driver->name, driver->runnable);
 	spa_list_consume(n, nodes, sort_link) {
 		spa_list_remove(&n->sort_link);
-		if (n->runnable)
-			driver->runnable = true;
+
 		pw_log_debug(" follower: %p %s runnable:%u driver-runnable:%u", n, n->name,
 				n->runnable, driver->runnable);
 		pw_impl_node_set_driver(n, driver);
@@ -1187,6 +1260,7 @@ again:
 			/* is any active and want a driver */
 			if (t->want_driver && t->active && t->runnable) {
 				driver = target;
+				driver->runnable = true;
 				break;
 			}
 		}