WIP: work on per driver graphs

Make a graph per driver node and let nodes that link to this join this
graph
Work on an idea for an even simpler scheduling mechanism.

meson.build

@@ -165,6 +165,8 @@ endif
 subdir('spa')
 subdir('src')
 subdir('pkgconfig')
+subdir('pipewire-jack')
+subdir('alsa-plugins')
 
 if get_option('enable_docs')
   doxygen = find_program('doxygen', required : false)

spa/include/spa/graph/graph-scheduler1.h

@@ -36,15 +36,12 @@ static inline void spa_graph_data_init(struct spa_graph_data *data,
 	data->graph = graph;
 }
 
-static inline int spa_graph_impl_process(void *data, struct spa_graph_node *node)
+static inline int spa_graph_trigger(struct spa_graph *g, struct spa_graph_node *node)
 {
-	struct spa_graph_data *d = (struct spa_graph_data *) data;
-	struct spa_graph *g = d->graph;
-	int old = node->state->status, res = 0;
 	uint32_t val;
 
-	spa_debug("node %p: pending %d required %d %d", node,
+	spa_debug("node %p: pending %d required %d", node,
-                        node->state->pending, node->state->required, old);
+                        node->state->pending, node->state->required);
 
 	if (node->state->pending == 0) {
 		spa_debug("node %p: nothing pending", node);
@@ -52,25 +49,10 @@ static inline int spa_graph_impl_process(void *data, struct spa_graph_node *node
 	}
 
 	val = __atomic_sub_fetch(&node->state->pending, 1, __ATOMIC_SEQ_CST);
-        if (val == 0) {
+        if (val == 0)
-		if (old == SPA_STATUS_NEED_BUFFER &&
+		spa_graph_node_process(node);
-		    node->implementation->process_input) {
-			res = spa_node_process_input(node->implementation);
-		}
-		else {
-			res = spa_node_process_output(node->implementation);
-		}
-		spa_debug("node %p: process %d", node, res);
 
-		if (res == SPA_STATUS_HAVE_BUFFER)
-			spa_graph_have_output(g, node);
-
-		node->state->status = res;
-
-                spa_debug("node %p: end %d", node, res);
-        }
         return node->state->status;
-
 }
 
 static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
@@ -118,8 +100,9 @@ static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *n
                         if (pn->sched_link.next != NULL)
                                 continue;
 
-                        if (pp->io->status == SPA_STATUS_NEED_BUFFER) {
+                        if (pp->io->status == SPA_STATUS_NEED_BUFFER &&
-				pn->state->status = spa_node_process_output(pn->implementation);
+			    pn->state->status == SPA_STATUS_HAVE_BUFFER) {
+				pn->state->status = spa_graph_node_process(pn);
                         } else {
                                 n->state->pending--;
                         }
@@ -132,7 +115,7 @@ static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *n
                 n->sched_link.next = NULL;
 
                 spa_debug("schedule node %p: %d", n, n->state->status);
-		spa_graph_process(d->graph, n);
+		spa_graph_trigger(d->graph, n);
         }
 	return 0;
 }
@@ -145,19 +128,22 @@ static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *
 	spa_debug("node %p start push", node);
 
 	spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link)
-		spa_graph_process(d->graph, p->peer->node);
+		spa_graph_trigger(d->graph, p->peer->node);
 
 	return 0;
 }
 
+static inline int spa_graph_impl_run(void *data)
+{
+	return 0;
+}
+
 static const struct spa_graph_callbacks spa_graph_impl_default = {
 	SPA_VERSION_GRAPH_CALLBACKS,
 	.need_input = spa_graph_impl_need_input,
 	.have_output = spa_graph_impl_have_output,
-	.process = spa_graph_impl_process,
 };
 
-
 #ifdef __cplusplus
 }  /* extern "C" */
 #endif

spa/include/spa/graph/graph-scheduler2.h

@@ -26,158 +26,143 @@ extern "C" {
 
 #include <spa/graph/graph.h>
 
-static inline int spa_graph_scheduler_default(struct spa_graph_node *node)
+struct spa_graph_data {
+	struct spa_graph *graph;
+};
+
+static inline void spa_graph_data_init(struct spa_graph_data *data,
+				       struct spa_graph *graph)
 {
-	int res;
+	data->graph = graph;
-	struct spa_node *n = node->user_data;
-
-	if (node->action == SPA_GRAPH_ACTION_IN)
-		res = spa_node_process_input(n);
-	else if (node->action == SPA_GRAPH_ACTION_OUT)
-		res = spa_node_process_output(n);
-	else
-		res = -EBADF;
-
-	return res;
 }
 
-static inline void spa_graph_port_check(struct spa_graph *graph, struct spa_graph_port *port)
+static inline int spa_graph_trigger(struct spa_graph *g, struct spa_graph_node *node)
 {
-	struct spa_graph_node *node = port->node;
+	uint32_t val;
 
-	if (port->io->status == SPA_STATUS_HAVE_BUFFER)
+	spa_debug("node %p: pending %d required %d", node,
-		node->ready++;
+                        node->state->pending, node->state->required);
 
-	spa_debug("port %p node %p check %d %d %d", port, node, port->io->status, node->ready, node->required);
+	if (node->state->pending == 0) {
-
+		spa_debug("node %p: nothing pending", node);
-	if (node->required > 0 && node->ready == node->required) {
+		return node->state->status;
-		node->action = SPA_GRAPH_ACTION_IN;
-		if (node->ready_link.next == NULL)
-			spa_list_append(&graph->ready, &node->ready_link);
-	} else if (node->ready_link.next) {
-		spa_list_remove(&node->ready_link);
-		node->ready_link.next = NULL;
 	}
+
+	val = __atomic_sub_fetch(&node->state->pending, 1, __ATOMIC_SEQ_CST);
+        if (val == 0)
+		spa_graph_node_process(node);
+
+        return node->state->status;
 }
 
-static inline void spa_graph_node_update(struct spa_graph *graph, struct spa_graph_node *node) {
+static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
-	struct spa_graph_port *p;
-
-	node->ready = 0;
-	spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
-		if (p->io->status == SPA_STATUS_OK && !(node->flags & SPA_GRAPH_NODE_FLAG_ASYNC))
-			node->ready++;
-	}
-	spa_debug("node %p update %d ready", node, node->ready);
-}
-
-static inline bool spa_graph_scheduler_iterate(struct spa_graph *graph)
 {
-	bool empty;
+#if 0
-	struct spa_graph_port *p;
+	struct spa_graph_data *d = (struct spa_graph_data *) data;
-	struct spa_graph_node *n;
+	struct spa_list queue, pending;
-	int iter = 1;
+	struct spa_graph_node *n, *pn;
-	uint32_t state;
+	struct spa_graph_port *p, *pp;
 
-next:
-	empty = spa_list_is_empty(&graph->ready);
-	if (empty && !spa_list_is_empty(&graph->pending)) {
-		spa_debug("copy pending");
-		spa_list_insert_list(&graph->ready, &graph->pending);
-		spa_list_init(&graph->pending);
-		empty = false;
-	}
-	if (iter-- == 0 || empty)
-		return !empty;
-
-	n = spa_list_first(&graph->ready, struct spa_graph_node, ready_link);
-	spa_list_remove(&n->ready_link);
-	n->ready_link.next = NULL;
-
-	spa_debug("node %p state %d", n, n->state);
-
-	switch (n->state) {
-	case SPA_GRAPH_STATE_IN:
-	case SPA_GRAPH_STATE_OUT:
-	case SPA_GRAPH_STATE_END:
-		if (n->state == SPA_GRAPH_STATE_END)
-			n->state = SPA_GRAPH_STATE_OUT;
-
-		state = n->schedule(n);
-		spa_debug("node %p schedule %d res %d", n, action, state);
-
-		if (n->state == SPA_GRAPH_STATE_IN && n == graph->node)
-			break;
-
-		if (n->state != SPA_GRAPH_STATE_END) {
-			spa_debug("node %p add ready for CHECK", n);
-			if (state == SPA_STATUS_NEED_BUFFER)
-				n->state = SPA_GRAPH_STATE_CHECK_IN;
-			else if (state == SPA_STATUS_HAVE_BUFFER)
-				n->state = SPA_GRAPH_STATE_CHECK_OUT;
-			else if (state == SPA_STATUS_OK)
-				n->state = SPA_GRAPH_STATE_CHECK_OK;
-			spa_list_append(&graph->ready, &n->ready_link);
-		}
-		else {
-			spa_graph_node_update(graph, n);
-		}
-		break;
-
-	case SPA_GRAPH_STATE_CHECK_IN:
-		n->ready = 0;
-		spa_list_for_each(p, &n->ports[SPA_DIRECTION_INPUT], link) {
-			struct spa_graph_node *pn = p->peer->node;
-			if (p->io->status == SPA_STATUS_NEED_BUFFER) {
-				if (pn != graph->node
-				    || pn->flags & SPA_GRAPH_NODE_FLAG_ASYNC) {
-					pn->state = SPA_GRAPH_STATE_OUT;
-					spa_debug("node %p add ready OUT", n);
-					spa_list_append(&graph->ready, &pn->ready_link);
-				}
-			} else if (p->io->status == SPA_STATUS_OK)
-				n->ready++;
-		}
-		break;
-
-	case SPA_GRAPH_STATE_CHECK_OUT:
-		spa_list_for_each(p, &n->ports[SPA_DIRECTION_OUTPUT], link)
-			spa_graph_port_check(graph, p->peer);
-
-		spa_debug("node %p add pending", n);
-		n->state = SPA_GRAPH_STATE_END;
-		spa_list_insert(&graph->pending, &n->ready_link);
-		break;
-
-	case SPA_GRAPH_STATE_CHECK_OK:
-		spa_graph_node_update(graph, n);
-		break;
-
-	default:
-		break;
-	}
-	goto next;
-}
-
-static inline void spa_graph_scheduler_pull(struct spa_graph *graph, struct spa_graph_node *node)
-{
-	node->action = SPA_GRAPH_ACTION_CHECK;
-	node->state = SPA_STATUS_NEED_BUFFER;
-	graph->node = node;
 	spa_debug("node %p start pull", node);
-	if (node->ready_link.next == NULL)
+
-		spa_list_append(&graph->ready, &node->ready_link);
+	spa_list_init(&queue);
+	spa_list_init(&pending);
+
+	node->state->status = SPA_STATUS_NEED_BUFFER;
+
+	if (node->sched_link.next == NULL)
+		spa_list_append(&queue, &node->sched_link);
+
+	while (!spa_list_is_empty(&queue)) {
+
+		n = spa_list_first(&queue, struct spa_graph_node, sched_link);
+		spa_list_remove(&n->sched_link);
+		n->sched_link.next = NULL;
+
+		n->state->pending = n->state->required + 1;
+                spa_debug("node %p: add %d %d status %d", n,
+                                n->state->pending, n->state->required,
+                                n->state->status);
+
+                spa_list_prepend(&pending, &n->sched_link);
+
+                if (n->state->status == SPA_STATUS_HAVE_BUFFER)
+                        continue;
+
+		spa_list_for_each(p, &n->ports[SPA_DIRECTION_INPUT], link) {
+			pp = p->peer;
+			if (pp == NULL)
+				continue;
+			pn = pp->node;
+
+                        spa_debug("node %p: %p in io:%d state:%d %p", n, pn, pp->io->status,
+                                        pn->state->status, pn->sched_link.next);
+
+                        if (pn->sched_link.next != NULL)
+                                continue;
+
+                        if (pp->io->status == SPA_STATUS_NEED_BUFFER &&
+			    pn->state->status == SPA_STATUS_HAVE_BUFFER) {
+				pn->state->status = spa_graph_node_process(pn);
+                        } else {
+                                n->state->pending--;
+                        }
+                        spa_list_append(&queue, &pn->sched_link);
+		}
+	}
+	while (!spa_list_is_empty(&pending)) {
+		n = spa_list_first(&pending, struct spa_graph_node, sched_link);
+		spa_list_remove(&n->sched_link);
+                n->sched_link.next = NULL;
+
+                spa_debug("schedule node %p: %d", n, n->state->status);
+		spa_graph_trigger(d->graph, n);
+        }
+#endif
+	return 0;
 }
 
-static inline void spa_graph_scheduler_push(struct spa_graph *graph, struct spa_graph_node *node)
+static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node)
 {
-	node->action = SPA_GRAPH_ACTION_OUT;
+	struct spa_graph_data *d = (struct spa_graph_data *) data;
-	graph->node = node;
+	struct spa_graph_port *p;
+
 	spa_debug("node %p start push", node);
-	if (node->ready_link.next == NULL)
+
-		spa_list_append(&graph->ready, &node->ready_link);
+	spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link)
+		if (p->peer)
+			spa_graph_trigger(d->graph, p->peer->node);
+
+	return 0;
 }
 
+static inline int spa_graph_impl_run(void *data)
+{
+	struct spa_graph_data *d = (struct spa_graph_data *) data;
+	struct spa_graph *g = d->graph;
+	struct spa_graph_node *n;
+
+	spa_debug("graph %p run", d->graph);
+
+	spa_list_for_each(n, &g->nodes, link) {
+		n->state->pending = n->state->required + 1;
+                spa_debug("node %p: add %d %d status %d", n,
+                                n->state->pending, n->state->required,
+                                n->state->status);
+	}
+	spa_list_for_each(n, &g->nodes, link)
+		spa_graph_trigger(d->graph, n);
+
+	return 0;
+}
+
+static const struct spa_graph_callbacks spa_graph_impl_default = {
+	SPA_VERSION_GRAPH_CALLBACKS,
+	.need_input = spa_graph_impl_need_input,
+	.have_output = spa_graph_impl_have_output,
+	.run = spa_graph_impl_run,
+};
+
 #ifdef __cplusplus
 }  /* extern "C" */
 #endif

spa/include/spa/graph/graph.h

@@ -43,9 +43,7 @@ struct spa_graph_callbacks {
 
 	int (*need_input) (void *data, struct spa_graph_node *node);
 	int (*have_output) (void *data, struct spa_graph_node *node);
-	int (*process) (void *data, struct spa_graph_node *node);
+	int (*run) (void *data);
-	int (*reuse_buffer) (void *data, struct spa_graph_node *node,
-			uint32_t port_id, uint32_t buffer_id);
 };
 
 struct spa_graph {
@@ -56,8 +54,7 @@ struct spa_graph {
 
 #define spa_graph_need_input(g,n)	((g)->callbacks->need_input((g)->callbacks_data, (n)))
 #define spa_graph_have_output(g,n)	((g)->callbacks->have_output((g)->callbacks_data, (n)))
-#define spa_graph_process(g,n)		((g)->callbacks->process((g)->callbacks_data, (n)))
+#define spa_graph_run(g)		((g)->callbacks->run((g)->callbacks_data))
-#define spa_graph_reuse_buffer(g,n,p,i)	((g)->callbacks->reuse_buffer((g)->callbacks_data, (n),(p),(i)))
 
 struct spa_graph_state {
 	int status;			/**< status of the node */
@@ -65,18 +62,31 @@ struct spa_graph_state {
 	uint32_t pending;		/**< number of input nodes pending */
 };
 
+struct spa_graph_node_callbacks {
+#define SPA_VERSION_GRAPH_NODE_CALLBACKS	0
+	uint32_t version;
+
+	int (*process) (void *data, struct spa_graph_node *node);
+	int (*reuse_buffer) (void *data, struct spa_graph_node *node,
+			uint32_t port_id, uint32_t buffer_id);
+};
+
 struct spa_graph_node {
 	struct spa_list link;		/**< link in graph nodes list */
 	struct spa_graph *graph;	/**< owner graph */
 	struct spa_list ports[2];	/**< list of input and output ports */
 #define SPA_GRAPH_NODE_FLAG_ASYNC	(1 << 0)
 	uint32_t flags;			/**< node flags */
-	struct spa_node *implementation;/**< node implementation */
 	struct spa_graph_state *state;	/**< state of the node */
+	const struct spa_graph_node_callbacks *callbacks;
+	void *callbacks_data;
 	struct spa_list sched_link;	/**< link for scheduler */
 	void *scheduler_data;		/**< scheduler private data */
 };
 
+#define spa_graph_node_process(n)	((n)->callbacks->process((n)->callbacks_data, (n)))
+#define spa_graph_node_reuse_buffer(n,p,i) ((n)->callbacks->reuse_buffer((n)->callbacks_data, (n),(p),(i)))
+
 struct spa_graph_port {
 	struct spa_list link;		/**< link in node port list */
 	struct spa_graph_node *node;	/**< owner node */
@@ -115,10 +125,12 @@ spa_graph_node_init(struct spa_graph_node *node, struct spa_graph_state *state)
 }
 
 static inline void
-spa_graph_node_set_implementation(struct spa_graph_node *node,
+spa_graph_node_set_callbacks(struct spa_graph_node *node,
-				  struct spa_node *implementation)
+		const struct spa_graph_node_callbacks *callbacks,
+		void *callbacks_data)
 {
-	node->implementation = implementation;
+	node->callbacks = callbacks;
+	node->callbacks_data = callbacks_data;
 }
 
 static inline void
@@ -202,6 +214,45 @@ spa_graph_port_unlink(struct spa_graph_port *port)
 	}
 }
 
+static inline int spa_graph_node_impl_process(void *data, struct spa_graph_node *node)
+{
+	struct spa_graph *g = node->graph;
+	struct spa_node *n = data;
+	//int old = node->state->status, res = 0;
+	int res = 0;
+
+//	if (old == SPA_STATUS_NEED_BUFFER && n->process_input &&
+	if (n->process_input &&
+	    !spa_list_is_empty(&node->ports[SPA_DIRECTION_INPUT]))
+		res = spa_node_process_input(n);
+	else
+		res = spa_node_process_output(n);
+
+	spa_debug("node %p: process %d", node, res);
+
+	node->state->status = res;
+
+	if (res == SPA_STATUS_HAVE_BUFFER)
+		spa_graph_have_output(g, node);
+
+	spa_debug("node %p: end %d", node, res);
+
+        return res;
+}
+
+static inline int spa_graph_node_impl_reuse_buffer(void *data, struct spa_graph_node *node,
+		uint32_t port_id, uint32_t buffer_id)
+{
+	struct spa_node *n = data;
+	return spa_node_port_reuse_buffer(n, port_id, buffer_id);
+}
+
+static const struct spa_graph_node_callbacks spa_graph_node_impl_default = {
+	SPA_VERSION_GRAPH_NODE_CALLBACKS,
+	.process = spa_graph_node_impl_process,
+	.reuse_buffer = spa_graph_node_impl_reuse_buffer,
+};
+
 #ifdef __cplusplus
 }  /* extern "C" */
 #endif

spa/plugins/v4l2/v4l2-source.c

@@ -862,6 +862,7 @@ static int impl_node_process_output(struct spa_node *node)
 static const struct spa_dict_item info_items[] = {
 	{ "media.class", "Video/Source" },
 	{ "node.pause-on-idle", "false" },
+	{ "node.driver", "true" },
 };
 
 static const struct spa_dict info = {

spa/tests/test-control.c

@@ -399,13 +399,13 @@ static int make_nodes(struct data *data, const char *device)
 			     &data->source_sink_io[0], sizeof(data->source_sink_io[0]));
 
 	spa_graph_node_init(&data->source_node, &data->source_state);
-	spa_graph_node_set_implementation(&data->source_node, data->source);
+	spa_graph_node_set_callbacks(&data->source_node, &spa_graph_node_impl_default, data->source);
 	spa_graph_node_add(&data->graph, &data->source_node);
 	spa_graph_port_init(&data->source_out, SPA_DIRECTION_OUTPUT, 0, 0, &data->source_sink_io[0]);
 	spa_graph_port_add(&data->source_node, &data->source_out);
 
 	spa_graph_node_init(&data->sink_node, &data->sink_state);
-	spa_graph_node_set_implementation(&data->sink_node, data->sink);
+	spa_graph_node_set_callbacks(&data->sink_node, &spa_graph_node_impl_default, data->sink);
 	spa_graph_node_add(&data->graph, &data->sink_node);
 	spa_graph_port_init(&data->sink_in, SPA_DIRECTION_INPUT, 0, 0, &data->source_sink_io[0]);
 	spa_graph_port_add(&data->sink_node, &data->sink_in);

spa/tests/test-graph.c

@@ -357,13 +357,13 @@ static int make_nodes(struct data *data, const char *device)
 			     &data->volume_sink_io[0], sizeof(data->volume_sink_io[0]));
 
 	spa_graph_node_init(&data->source_node, &data->source_state);
-	spa_graph_node_set_implementation(&data->source_node, data->source);
+	spa_graph_node_set_callbacks(&data->source_node, &spa_graph_node_impl_default, data->source);
 	spa_graph_node_add(&data->graph, &data->source_node);
 	spa_graph_port_init(&data->source_out, SPA_DIRECTION_OUTPUT, 0, 0, &data->source_volume_io[0]);
 	spa_graph_port_add(&data->source_node, &data->source_out);
 
 	spa_graph_node_init(&data->volume_node, &data->volume_state);
-	spa_graph_node_set_implementation(&data->volume_node, data->volume);
+	spa_graph_node_set_callbacks(&data->volume_node, &spa_graph_node_impl_default, data->volume);
 	spa_graph_node_add(&data->graph, &data->volume_node);
 	spa_graph_port_init(&data->volume_in, SPA_DIRECTION_INPUT, 0, 0, &data->source_volume_io[0]);
 	spa_graph_port_add(&data->volume_node, &data->volume_in);
@@ -374,7 +374,7 @@ static int make_nodes(struct data *data, const char *device)
 	spa_graph_port_add(&data->volume_node, &data->volume_out);
 
 	spa_graph_node_init(&data->sink_node, &data->sink_state);
-	spa_graph_node_set_implementation(&data->sink_node, data->sink);
+	spa_graph_node_set_callbacks(&data->sink_node, &spa_graph_node_impl_default, data->sink);
 	spa_graph_node_add(&data->graph, &data->sink_node);
 	spa_graph_port_init(&data->sink_in, SPA_DIRECTION_INPUT, 0, 0, &data->volume_sink_io[0]);
 	spa_graph_port_add(&data->sink_node, &data->sink_in);

spa/tests/test-mixer.c

@@ -472,19 +472,19 @@ static int make_nodes(struct data *data, const char *device)
 
 #ifdef USE_GRAPH
 	spa_graph_node_init(&data->source1_node, &data->source1_state);
-	spa_graph_node_set_implementation(&data->source1_node, data->source1);
+	spa_graph_node_set_callbacks(&data->source1_node, &spa_graph_node_impl_default, data->source1);
 	spa_graph_port_init(&data->source1_out, SPA_DIRECTION_OUTPUT, 0, 0, &data->source1_mix_io[0]);
 	spa_graph_port_add(&data->source1_node, &data->source1_out);
 	spa_graph_node_add(&data->graph, &data->source1_node);
 
 	spa_graph_node_init(&data->source2_node, &data->source2_state);
-	spa_graph_node_set_implementation(&data->source2_node, data->source2);
+	spa_graph_node_set_callbacks(&data->source2_node, &spa_graph_node_impl_default, data->source2);
 	spa_graph_port_init(&data->source2_out, SPA_DIRECTION_OUTPUT, 0, 0, &data->source2_mix_io[0]);
 	spa_graph_port_add(&data->source2_node, &data->source2_out);
 	spa_graph_node_add(&data->graph, &data->source2_node);
 
 	spa_graph_node_init(&data->mix_node, &data->mix_state);
-	spa_graph_node_set_implementation(&data->mix_node, data->mix);
+	spa_graph_node_set_callbacks(&data->mix_node, &spa_graph_node_impl_default, data->mix);
 	spa_graph_port_init(&data->mix_in[0], SPA_DIRECTION_INPUT,
 			    data->mix_ports[0], 0, &data->source1_mix_io[0]);
 	spa_graph_port_add(&data->mix_node, &data->mix_in[0]);
@@ -500,7 +500,7 @@ static int make_nodes(struct data *data, const char *device)
 	spa_graph_port_add(&data->mix_node, &data->mix_out);
 
 	spa_graph_node_init(&data->sink_node, &data->sink_state);
-	spa_graph_node_set_implementation(&data->sink_node, data->sink);
+	spa_graph_node_set_callbacks(&data->sink_node, &spa_graph_node_impl_default, data->sink);
 	spa_graph_port_init(&data->sink_in, SPA_DIRECTION_INPUT, 0, 0, &data->mix_sink_io[0]);
 	spa_graph_port_add(&data->sink_node, &data->sink_in);
 	spa_graph_node_add(&data->graph, &data->sink_node);

spa/tests/test-perf.c

@@ -375,7 +375,7 @@ static int make_nodes(struct data *data)
 			     &data->source_sink_io[0], sizeof(data->source_sink_io[0]));
 
 	spa_graph_node_init(&data->source_node, &data->source_state);
-	spa_graph_node_set_implementation(&data->source_node, data->source);
+	spa_graph_node_set_callbacks(&data->source_node, &spa_graph_node_impl_default, data->source);
 	spa_graph_node_add(&data->graph, &data->source_node);
 
 	data->source_node.flags = (data->mode & MODE_ASYNC_PUSH) ? SPA_GRAPH_NODE_FLAG_ASYNC : 0;
@@ -383,7 +383,7 @@ static int make_nodes(struct data *data)
 	spa_graph_port_add(&data->source_node, &data->source_out);
 
 	spa_graph_node_init(&data->sink_node, &data->sink_state);
-	spa_graph_node_set_implementation(&data->sink_node, data->sink);
+	spa_graph_node_set_callbacks(&data->sink_node, &spa_graph_node_impl_default, data->sink);
 	spa_graph_node_add(&data->graph, &data->sink_node);
 
 	data->sink_node.flags = (data->mode & MODE_ASYNC_PULL) ? SPA_GRAPH_NODE_FLAG_ASYNC : 0;

src/modules/module-audio-dsp.c

@@ -685,11 +685,14 @@ static int schedule_mix_output(struct spa_node *_node)
 	struct spa_graph_port *gp;
 	struct spa_io_buffers *io = port->rt.mix_port.io;
 
+	pw_log_trace("port %p", port);
+
 	spa_list_for_each(gp, &node->ports[SPA_DIRECTION_INPUT], link) {
 		pw_log_trace("port %p: port %d %d %p->%p %d %d", port,
 				gp->port_id, gp->flags, io, gp->io, io->status, io->buffer_id);
 		gp->io->status = io->status;
 	}
+	io->status =  SPA_STATUS_HAVE_BUFFER;
 	return io->status;
 }
 

src/modules/module-client-node/client-node.c

@@ -866,7 +866,7 @@ static int impl_node_process_input(struct spa_node *node)
 
 			/* explicitly recycle buffers when the client is not going to do it */
 			if (!client_reuse && (pp = p->peer))
-		                spa_node_port_reuse_buffer(pp->node->implementation,
+		                spa_graph_node_reuse_buffer(pp->node,
 						pp->port_id, io->buffer_id);
 		}
 		pw_log_trace("client-node %p: send process input", this);

src/pipewire/core.c

@@ -21,8 +21,6 @@
 #include <time.h>
 #include <stdio.h>
 
-#define spa_debug pw_log_trace
-
 #include <spa/lib/debug.h>
 #include <spa/support/dbus.h>
 
@@ -33,13 +31,9 @@
 #include <pipewire/core.h>
 #include <pipewire/data-loop.h>
 
-#include <spa/graph/graph-scheduler1.h>
-
 /** \cond */
 struct impl {
 	struct pw_core this;
-
-	struct spa_graph_data data;
 };
 
 
@@ -396,10 +390,6 @@ struct pw_core *pw_core_new(struct pw_loop *main_loop, struct pw_properties *pro
 	pw_type_init(&this->type);
 	pw_map_init(&this->globals, 128, 32);
 
-	spa_graph_init(&this->rt.graph);
-	spa_graph_data_init(&impl->data, &this->rt.graph);
-	spa_graph_set_callbacks(&this->rt.graph, &spa_graph_impl_default, &impl->data);
-
 	spa_debug_set_type_map(this->type.map);
 
 	this->support[0] = SPA_SUPPORT_INIT(SPA_TYPE__TypeMap, this->type.map);

src/pipewire/link.c

@@ -925,7 +925,8 @@ static void clear_port_buffers(struct pw_link *link, struct pw_port *port)
 
 	pw_log_debug("%d %p", spa_list_is_empty(&port->links), port->allocation.mem);
 
-	if (port->direction == PW_DIRECTION_OUTPUT && !spa_list_is_empty(&port->links))
+//	if (port->direction == PW_DIRECTION_OUTPUT && !spa_list_is_empty(&port->links))
+	if (port->direction == PW_DIRECTION_OUTPUT)
 		return;
 
 	if ((res = pw_port_use_buffers(port,
@@ -1164,6 +1165,7 @@ struct pw_link *pw_link_new(struct pw_core *core,
 	struct impl *impl;
 	struct pw_link *this;
 	struct pw_node *input_node, *output_node;
+	struct spa_graph *in_graph, *out_graph;
 
 	if (output == input)
 		goto same_ports;
@@ -1171,6 +1173,15 @@ struct pw_link *pw_link_new(struct pw_core *core,
 	if (pw_link_find(output, input))
 		goto link_exists;
 
+	input_node = input->node;
+	output_node = output->node;
+
+	in_graph = input_node->rt.node.graph;
+	out_graph = output_node->rt.node.graph;
+
+	if (in_graph != NULL && out_graph != NULL && in_graph != out_graph)
+		goto link_not_supported;
+
 	impl = calloc(1, sizeof(struct impl) + user_data_size);
 	if (impl == NULL)
 		goto no_mem;
@@ -1190,9 +1201,6 @@ struct pw_link *pw_link_new(struct pw_core *core,
 	this->input = input;
 	this->output = output;
 
-	input_node = input->node;
-	output_node = output->node;
-
 	if (properties) {
 		const char *str = pw_properties_get(properties, PW_LINK_PROP_PASSIVE);
 		if (str && pw_properties_parse_bool(str))
@@ -1235,6 +1243,11 @@ struct pw_link *pw_link_new(struct pw_core *core,
 		     output_node, output->port_id, this->rt.mix[SPA_DIRECTION_OUTPUT].port.port_id,
 		     input_node, input->port_id, this->rt.mix[SPA_DIRECTION_INPUT].port.port_id);
 
+	if (out_graph != NULL)
+		pw_node_join_graph(input_node, out_graph);
+	else if (in_graph != NULL)
+		pw_node_join_graph(output_node, in_graph);
+
 	spa_hook_list_call(&output->listener_list, struct pw_port_events, link_added, this);
 	spa_hook_list_call(&input->listener_list, struct pw_port_events, link_added, this);
 
@@ -1246,6 +1259,9 @@ struct pw_link *pw_link_new(struct pw_core *core,
       link_exists:
 	asprintf(error, "link already exists");
 	return NULL;
+      link_not_supported:
+	asprintf(error, "link between drivers not yet supported");
+	return NULL;
       no_mem:
 	asprintf(error, "no memory");
 	return NULL;

src/pipewire/node.c

@@ -35,6 +35,10 @@
 #include "pipewire/main-loop.h"
 #include "pipewire/work-queue.h"
 
+#define spa_debug pw_log_trace
+
+#include <spa/graph/graph-scheduler2.h>
+
 /** \cond */
 struct impl {
 	struct pw_node this;
@@ -42,6 +46,9 @@ struct impl {
 	struct pw_work_queue *work;
 	bool pause_on_idle;
 
+	struct spa_graph graph_driver;
+	struct spa_graph_data graph_data;
+
 	struct pw_node_activation activation;
 };
 
@@ -318,13 +325,38 @@ static const struct pw_global_events global_events = {
 };
 
 static int
-do_node_add(struct spa_loop *loop,
+do_node_join(struct spa_loop *loop,
 	    bool async, uint32_t seq, const void *data, size_t size, void *user_data)
 {
 	struct pw_node *this = user_data;
+	struct spa_graph *graph = *(struct spa_graph **)data;
+	struct spa_graph_port *p;
 
-	spa_graph_node_add(this->rt.graph, &this->rt.node);
+	if (this->rt.node.graph != NULL) {
+		spa_graph_node_remove(&this->rt.node);
+		spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_INPUT], link)
+			spa_graph_node_remove(p->peer->node);
+		spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_OUTPUT], link)
+			spa_graph_node_remove(p->peer->node);
+	}
 
+	if (graph) {
+		spa_graph_node_add(graph, &this->rt.node);
+		spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_INPUT], link)
+			spa_graph_node_add(graph, p->peer->node);
+		spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_OUTPUT], link)
+			spa_graph_node_add(graph, p->peer->node);
+	}
+	else
+		this->rt.node.graph = NULL;
+
+	return 0;
+}
+
+int pw_node_join_graph(struct pw_node *node, struct spa_graph *graph)
+{
+	pw_loop_invoke(node->data_loop, do_node_join, 1,
+			graph, sizeof(struct spa_graph *), false, node);
 	return 0;
 }
 
@@ -349,8 +381,6 @@ int pw_node_register(struct pw_node *this,
 
 	pw_node_update_ports(this);
 
-	pw_loop_invoke(this->data_loop, do_node_add, 1, NULL, 0, false, this);
-
 	if ((str = pw_properties_get(this->properties, "media.class")) != NULL)
 		pw_properties_set(properties, "media.class", str);
 	pw_properties_set(properties, "node.name", this->info.name);
@@ -393,6 +423,20 @@ static void check_properties(struct pw_node *node)
 		impl->pause_on_idle = pw_properties_parse_bool(str);
 	else
 		impl->pause_on_idle = true;
+
+	if ((str = pw_properties_get(node->properties, "node.driver")))
+		node->driver = pw_properties_parse_bool(str);
+	else
+		node->driver = false;
+
+	if (node->driver) {
+		spa_graph_init(&impl->graph_driver);
+		spa_graph_data_init(&impl->graph_data, &impl->graph_driver);
+		spa_graph_set_callbacks(&impl->graph_driver,
+				&spa_graph_impl_default, &impl->graph_data);
+		pw_node_join_graph(node, &impl->graph_driver);
+	}
+
 }
 
 struct pw_node *pw_node_new(struct pw_core *core,
@@ -429,8 +473,6 @@ struct pw_node *pw_node_new(struct pw_core *core,
 
 	this->data_loop = core->data_loop;
 
-	this->rt.graph = &core->rt.graph;
-
 	spa_list_init(&this->resource_list);
 
 	spa_hook_list_init(&this->listener_list);
@@ -530,21 +572,37 @@ static void node_event(void *data, struct spa_event *event)
 static void node_need_input(void *data)
 {
 	struct pw_node *node = data;
+	struct impl *impl = SPA_CONTAINER_OF(node, struct impl, this);
 
 	pw_log_trace("node %p: need input %d", node, node->rt.activation->state.status);
 
 	spa_hook_list_call(&node->listener_list, struct pw_node_events, need_input);
-	spa_graph_need_input(node->rt.graph, &node->rt.node);
+
+	if (node->driver)
+		spa_graph_run(&impl->graph_driver);
+	else if (node->rt.node.graph)
+		spa_graph_need_input(node->rt.node.graph, &node->rt.node);
+	else
+		pw_log_error("node %p: not added in graph", node);
 }
 
 static void node_have_output(void *data)
 {
 	struct pw_node *node = data;
+	struct impl *impl = SPA_CONTAINER_OF(node, struct impl, this);
 
-	pw_log_trace("node %p: have output", node);
+	pw_log_trace("node %p: have output %d", node, node->driver);
 
 	spa_hook_list_call(&node->listener_list, struct pw_node_events, have_output);
-	spa_graph_have_output(node->rt.graph, &node->rt.node);
+
+	if (node->driver)
+		spa_graph_run(&impl->graph_driver);
+
+	if (node->rt.node.graph)
+		spa_graph_have_output(node->rt.node.graph, &node->rt.node);
+	else
+		pw_log_error("node %p: not added in graph", node);
+
 }
 
 static void node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
@@ -557,7 +615,7 @@ static void node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
 			continue;
 
 		if ((pp = p->peer) != NULL)
-			spa_node_port_reuse_buffer(pp->node->implementation, pp->port_id, buffer_id);
+			spa_graph_node_reuse_buffer(pp->node, pp->port_id, buffer_id);
 		break;
 	}
 }
@@ -576,7 +634,7 @@ void pw_node_set_implementation(struct pw_node *node,
 {
 	node->node = spa_node;
 	spa_node_set_callbacks(node->node, &node_callbacks, node);
-	spa_graph_node_set_implementation(&node->rt.node, spa_node);
+	spa_graph_node_set_callbacks(&node->rt.node, &spa_graph_node_impl_default, spa_node);
 
 	if (spa_node->info)
 		pw_node_update_properties(node, spa_node->info);
@@ -600,11 +658,7 @@ do_node_remove(struct spa_loop *loop,
 	       bool async, uint32_t seq, const void *data, size_t size, void *user_data)
 {
 	struct pw_node *this = user_data;
-
-	pause_node(this);
-
 	spa_graph_node_remove(&this->rt.node);
-
 	return 0;
 }
 

src/pipewire/port.c

@@ -90,7 +90,7 @@ static int schedule_tee_reuse_buffer(struct spa_node *data, uint32_t port_id, ui
 
 	if ((pp = p->peer) != NULL) {
 		pw_log_trace("port %p: tee reuse buffer %d %d", this, port_id, buffer_id);
-		spa_node_port_reuse_buffer(pp->node->implementation, port_id, buffer_id);
+		spa_graph_node_reuse_buffer(pp->node, port_id, buffer_id);
 	}
 	return 0;
 }
@@ -146,7 +146,7 @@ static int schedule_mix_reuse_buffer(struct spa_node *data, uint32_t port_id, ui
 	spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
 		if ((pp = p->peer) != NULL) {
 			pw_log_trace("port %p: reuse buffer %d %d", this, port_id, buffer_id);
-			spa_node_port_reuse_buffer(pp->node->implementation, port_id, buffer_id);
+			spa_graph_node_reuse_buffer(pp->node, port_id, buffer_id);
 		}
 	}
 	return 0;
@@ -250,7 +250,8 @@ struct pw_port *pw_port_new(enum pw_direction direction,
 	this->mix_node = this->direction == PW_DIRECTION_INPUT ?
 				schedule_mix_node :
 				schedule_tee_node;
-	spa_graph_node_set_implementation(&this->rt.mix_node, &this->mix_node);
+	spa_graph_node_set_callbacks(&this->rt.mix_node,
+			&spa_graph_node_impl_default, &this->mix_node);
 	pw_map_init(&this->mix_port_map, 64, 64);
 
 	spa_graph_port_init(&this->rt.mix_port,
@@ -332,10 +333,12 @@ static int do_add_port(struct spa_loop *loop,
 
 	this->rt.port.flags = this->spa_info->flags;
 	spa_graph_port_add(&this->node->rt.node, &this->rt.port);
-	spa_graph_node_add(this->rt.graph, &this->rt.mix_node);
 	spa_graph_port_add(&this->rt.mix_node, &this->rt.mix_port);
 	spa_graph_port_link(&this->rt.port, &this->rt.mix_port);
 
+	if (this->rt.mix_node.graph)
+		spa_graph_node_add(this->rt.mix_node.graph, &this->rt.mix_node);
+
 	return 0;
 }
 
@@ -508,7 +511,7 @@ int pw_port_add(struct pw_port *port, struct pw_node *node)
 		pw_port_register(port, node->global->owner, node->global,
 				pw_properties_copy(port->properties));
 
-	port->rt.graph = node->rt.graph;
+	port->rt.mix_node.graph = node->rt.node.graph;
 	pw_loop_invoke(node->data_loop, do_add_port, SPA_ID_INVALID, NULL, 0, false, port);
 
 	if (port->state <= PW_PORT_STATE_INIT)
@@ -546,7 +549,9 @@ static int do_remove_port(struct spa_loop *loop,
 		spa_graph_port_remove(p);
 
 	spa_graph_port_remove(&this->rt.mix_port);
-	spa_graph_node_remove(&this->rt.mix_node);
+	if (this->rt.mix_node.graph)
+		spa_graph_node_remove(&this->rt.mix_node);
+	this->rt.mix_node.graph = NULL;
 
 	return 0;
 }

src/pipewire/private.h

@@ -163,10 +163,6 @@ struct pw_core {
 	struct pw_client *current_client;	/**< client currently executing code in mainloop */
 
 	long sc_pagesize;
-
-	struct {
-		struct spa_graph graph;
-	} rt;
 };
 
 struct pw_data_loop {
@@ -255,6 +251,7 @@ struct pw_node {
 	bool enabled;			/**< if the node is enabled */
 	bool active;			/**< if the node is active */
 	bool live;			/**< if the node is live */
+	bool driver;			/**< if the node drives the graph */
 	struct spa_clock *clock;	/**< handle to SPA clock if any */
 	struct spa_node *node;		/**< SPA node implementation */
 
@@ -275,7 +272,6 @@ struct pw_node {
 	struct pw_loop *data_loop;		/**< the data loop for this node */
 
 	struct {
-		struct spa_graph *graph;
 		struct spa_graph_node node;
 		struct spa_list links[2];
 		struct pw_node_activation *activation;
@@ -336,7 +332,6 @@ struct pw_port {
 	struct pw_map mix_port_map;	/**< map from port_id from mixer */
 
 	struct {
-		struct spa_graph *graph;
 		struct spa_io_buffers io;	/**< io area of the port */
 		struct spa_graph_port port;	/**< this graph port, linked to mix_port */
 		struct spa_graph_port mix_port;	/**< port from the mixer */
@@ -599,6 +594,8 @@ void pw_node_update_state(struct pw_node *node, enum pw_node_state state, char *
 
 int pw_node_update_ports(struct pw_node *node);
 
+int pw_node_join_graph(struct pw_node *node, struct spa_graph *graph);
+
 /** Activate a link \memberof pw_link
   * Starts the negotiation of formats and buffers on \a link and then
   * starts data streaming */

src/pipewire/remote.c

@@ -486,7 +486,7 @@ static void do_push(struct node_data *data, enum spa_direction direction)
 
 	spa_list_for_each(p, &node->ports[direction], link) {
 		if (p->peer)
-			spa_node_process_input(p->peer->node->implementation);
+			spa_graph_node_process(p->peer->node);
 	}
 }
 
@@ -497,7 +497,7 @@ static void do_pull(struct node_data *data, enum spa_direction direction)
 
 	spa_list_for_each(p, &node->ports[direction], link) {
 		if (p->peer)
-			spa_node_process_output(p->peer->node->implementation);
+			spa_graph_node_process(p->peer->node);
 	}
 }
 
@@ -532,17 +532,14 @@ static int process_input(struct node_data *data)
 	pw_log_trace("remote %p: process input", data->remote);
 	do_push(data, SPA_DIRECTION_INPUT);
 
-	if (node->implementation->process_input == NULL)
+	res = spa_graph_node_process(node);
-		res = SPA_STATUS_HAVE_BUFFER;
-	else
-		res = spa_node_process_input(node->implementation);
 
 	switch (res) {
 	case SPA_STATUS_HAVE_BUFFER:
 		node_have_output(data);
 		break;
 	case SPA_STATUS_NEED_BUFFER:
-		node_need_input(data);
+//		node_need_input(data);
 		break;
 	}
 	return res;
@@ -556,17 +553,14 @@ static int process_output(struct node_data *data)
 	pw_log_trace("remote %p: process output", data->remote);
 	do_pull(data, SPA_DIRECTION_OUTPUT);
 
-	if (node->implementation->process_output == NULL)
+	res = spa_graph_node_process(node);
-		res = SPA_STATUS_NEED_BUFFER;
-	else
-		res = spa_node_process_output(node->implementation);
 
 	switch (res) {
 	case SPA_STATUS_HAVE_BUFFER:
 		node_have_output(data);
 		break;
 	case SPA_STATUS_NEED_BUFFER:
-		node_need_input(data);
+//		node_need_input(data);
 		break;
 	}
 	return res;
@@ -593,7 +587,7 @@ static void handle_rtnode_message(struct pw_proxy *proxy, struct pw_client_node_
 		uint32_t buffer_id = rb->body.buffer_id.value;
 		struct spa_graph_node *node = &data->node->rt.node;
 
-		spa_node_port_reuse_buffer(node->implementation, port_id, buffer_id);
+		spa_graph_node_reuse_buffer(node, port_id, buffer_id);
 		break;
 	}
 	default:
@@ -895,7 +889,6 @@ static void client_node_event(void *object, const struct spa_event *event)
 
 static void do_start(struct node_data *data)
 {
-	uint64_t cmd = 1;
 	struct mix *mix;
 
 	spa_list_for_each(mix, &data->mix[SPA_DIRECTION_INPUT], link) {
@@ -906,12 +899,15 @@ static void do_start(struct node_data *data)
 		mix->mix.port.io->status = SPA_STATUS_NEED_BUFFER;
 		mix->mix.port.io->buffer_id = SPA_ID_INVALID;
 	}
+#if 0
 	if (!spa_list_is_empty(&data->mix[SPA_DIRECTION_INPUT])) {
+		uint64_t cmd = 1;
 		pw_log_trace("remote %p: send need input", data);
 		pw_client_node_transport_add_message(data->trans,
 				&PW_CLIENT_NODE_MESSAGE_INIT(PW_CLIENT_NODE_MESSAGE_NEED_INPUT));
 		write(data->rtwritefd, &cmd, 8);
 	}
+#endif
 }
 
 static void client_node_command(void *object, uint32_t seq, const struct spa_command *command)