graph: add new scheduler

Move the activation state to the graph
Replace an old scheduler with a new one that works with the new
activation states.
Remove the DISABLED port flags, we explicitly add and remove to
make ports enabled/disabled.
Find first compatible port for link
Remove the node based scheduler and use the spa one.

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

@@ -26,136 +26,126 @@ extern "C" {
 
 #include <spa/graph/graph.h>
 
-#define SPA_GRAPH_STATE_IN		0
-#define SPA_GRAPH_STATE_OUT		1
-#define SPA_GRAPH_STATE_CHECK_IN	2
-#define SPA_GRAPH_STATE_CHECK_OUT	3
-
 struct spa_graph_data {
 	struct spa_graph *graph;
-        struct spa_list ready;
-        struct spa_graph_node *node;
 };
 
 static inline void spa_graph_data_init(struct spa_graph_data *data,
 				       struct spa_graph *graph)
 {
 	data->graph = graph;
-	spa_list_init(&data->ready);
-	data->node = NULL;
 }
 
-static inline void spa_graph_data_port_check(struct spa_graph_data *data, struct spa_graph_port *port)
+static inline int spa_graph_impl_process(void *data, struct spa_graph_node *node)
 {
-	struct spa_graph_node *node = port->node;
+	struct spa_graph_data *d = (struct spa_graph_data *) data;
-	uint32_t required = node->required[SPA_DIRECTION_INPUT];
+	struct spa_graph *g = d->graph;
+	int old = node->state->status, res = 0;
+	uint32_t val;
 
-	if (port->io->status == SPA_STATUS_HAVE_BUFFER)
+	spa_debug("node %p: pending %d required %d %d", node,
-		node->ready[SPA_DIRECTION_INPUT]++;
+                        node->state->pending, node->state->required, old);
 
-	spa_debug("port %p node %p check %d %d %d", port, node,
+	if (node->state->pending == 0) {
-	      port->io->status, node->ready[SPA_DIRECTION_INPUT], required);
+		spa_debug("node %p: nothing pending", node);
-
+		return node->state->status;
-	if (required > 0 && node->ready[SPA_DIRECTION_INPUT] == required) {
-		node->state = SPA_GRAPH_STATE_IN;
-		if (node->ready_link.next == NULL)
-			spa_list_append(&data->ready, &node->ready_link);
-	} else if (node->ready_link.next) {
-		spa_list_remove(&node->ready_link);
-		node->ready_link.next = NULL;
 	}
-}
 
-static inline bool spa_graph_data_iterate(struct spa_graph_data *data)
+	val = __atomic_sub_fetch(&node->state->pending, 1, __ATOMIC_SEQ_CST);
-{
+        if (val == 0) {
-	bool res;
+		if (old == SPA_STATUS_NEED_BUFFER &&
-	int state;
+		    node->implementation->process_input) {
-	struct spa_graph_port *p;
+			res = spa_node_process_input(node->implementation);
-	struct spa_graph_node *n;
-
-	res = !spa_list_is_empty(&data->ready);
-	if (res) {
-		n = spa_list_first(&data->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:
-			state = spa_node_process_input(n->implementation);
-			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;
-			spa_debug("node %p processed input state %d", n, n->state);
-			if (n == data->node)
-				break;
-			spa_list_append(&data->ready, &n->ready_link);
-			break;
-
-		case SPA_GRAPH_STATE_OUT:
-			state = spa_node_process_output(n->implementation);
-			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;
-			spa_debug("node %p processed output state %d", n, n->state);
-			spa_list_append(&data->ready, &n->ready_link);
-			break;
-
-		case SPA_GRAPH_STATE_CHECK_IN:
-			n->ready[SPA_DIRECTION_INPUT] = 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 != data->node
-					    || pn->flags & SPA_GRAPH_NODE_FLAG_ASYNC) {
-						pn->state = SPA_GRAPH_STATE_OUT;
-						spa_list_append(&data->ready,
-								&pn->ready_link);
-					}
-				} else if (p->io->status == SPA_STATUS_OK)
-					n->ready[SPA_DIRECTION_INPUT]++;
-			}
-		case SPA_GRAPH_STATE_CHECK_OUT:
-			spa_list_for_each(p, &n->ports[SPA_DIRECTION_OUTPUT], link)
-				spa_graph_data_port_check(data, p->peer);
-			break;
-
-		default:
-			break;
 		}
-		res = !spa_list_is_empty(&data->ready);
+		else {
-	}
+			res = spa_node_process_output(node->implementation);
-	return res;
+		}
+		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)
 {
 	struct spa_graph_data *d = (struct spa_graph_data *) data;
+	struct spa_list queue, pending;
+	struct spa_graph_node *n, *pn;
+	struct spa_graph_port *p, *pp;
+
 	spa_debug("node %p start pull", node);
-	node->state = SPA_GRAPH_STATE_CHECK_IN;
-	d->node = node;
-	if (node->ready_link.next == NULL)
-		spa_list_append(&d->ready, &node->ready_link);
 
-	while(spa_graph_data_iterate(d));
+	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_node_process_output(pn->implementation);
+                        } 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_process(d->graph, n);
+        }
 	return 0;
 }
 
 static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node)
 {
 	struct spa_graph_data *d = (struct spa_graph_data *) data;
-	spa_debug("node %p start push", node);
+	struct spa_graph_port *p;
-	node->state = SPA_GRAPH_STATE_OUT;
-	d->node = node;
-	if (node->ready_link.next == NULL)
-		spa_list_append(&d->ready, &node->ready_link);
 
-	while(spa_graph_data_iterate(d));
+	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);
 
 	return 0;
 }
@@ -164,6 +154,7 @@ 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,
 };
 
 

spa/include/spa/graph/graph.h

@@ -43,6 +43,9 @@ 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 (*reuse_buffer) (void *data, struct spa_graph_node *node,
+			uint32_t port_id, uint32_t buffer_id);
 };
 
 struct spa_graph {
@@ -53,19 +56,24 @@ 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_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 */
+	uint32_t required;		/**< required number of input nodes */
+	uint32_t pending;		/**< number of input nodes pending */
+};
+
 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 */
-	struct spa_list ready_link;	/**< link for scheduler */
 #define SPA_GRAPH_NODE_FLAG_ASYNC	(1 << 0)
 	uint32_t flags;			/**< node flags */
-	uint32_t required[2];		/**< required number of ports */
-	uint32_t ready[2];		/**< number of ports with data */
-	int state;			/**< state of the node */
 	struct spa_node *implementation;/**< node implementation */
+	struct spa_graph_state *state;	/**< state of the node */
+	struct spa_list sched_link;	/**< link for scheduler */
 	void *scheduler_data;		/**< scheduler private data */
 };
 
@@ -74,7 +82,6 @@ struct spa_graph_port {
 	struct spa_graph_node *node;	/**< owner node */
 	enum spa_direction direction;	/**< port direction */
 	uint32_t port_id;		/**< port id */
-#define SPA_GRAPH_PORT_FLAG_DISABLED	(1 << 0)
 	uint32_t flags;			/**< port flags */
 	struct spa_io_buffers *io;	/**< io area of the port */
 	struct spa_graph_port *peer;	/**< peer */
@@ -96,13 +103,14 @@ spa_graph_set_callbacks(struct spa_graph *graph,
 }
 
 static inline void
-spa_graph_node_init(struct spa_graph_node *node)
+spa_graph_node_init(struct spa_graph_node *node, struct spa_graph_state *state)
 {
 	spa_list_init(&node->ports[SPA_DIRECTION_INPUT]);
 	spa_list_init(&node->ports[SPA_DIRECTION_OUTPUT]);
 	node->flags = 0;
-	node->required[SPA_DIRECTION_INPUT] = node->ready[SPA_DIRECTION_INPUT] = 0;
+	node->state = state;
-	node->required[SPA_DIRECTION_OUTPUT] = node->ready[SPA_DIRECTION_OUTPUT] = 0;
+	node->state->required = node->state->pending = 0;
+	node->state->status = SPA_STATUS_OK;
 	spa_debug("node %p init", node);
 }
 
@@ -118,8 +126,7 @@ spa_graph_node_add(struct spa_graph *graph,
 		   struct spa_graph_node *node)
 {
 	node->graph = graph;
-	node->state = SPA_STATUS_OK;
+	node->sched_link.next = NULL;
-	node->ready_link.next = NULL;
 	spa_list_append(&graph->nodes, &node->link);
 	spa_debug("node %p add", node);
 }
@@ -145,43 +152,53 @@ spa_graph_port_add(struct spa_graph_node *node,
 	spa_debug("port %p add to node %p", port, node);
 	port->node = node;
 	spa_list_append(&node->ports[port->direction], &port->link);
-	if (!(port->flags & SPA_PORT_INFO_FLAG_OPTIONAL))
-		node->required[port->direction]++;
 }
 
 static inline void spa_graph_node_remove(struct spa_graph_node *node)
 {
 	spa_debug("node %p remove", node);
 	spa_list_remove(&node->link);
-	if (node->ready_link.next)
+	if (node->sched_link.next)
-		spa_list_remove(&node->ready_link);
+		spa_list_remove(&node->sched_link);
 }
 
 static inline void spa_graph_port_remove(struct spa_graph_port *port)
 {
 	spa_debug("port %p remove", port);
 	spa_list_remove(&port->link);
-	if (!(port->flags & SPA_PORT_INFO_FLAG_OPTIONAL) &&
+	port->node = NULL;
-	    port->node->required[port->direction] > 0) {
-		port->node->required[port->direction]--;
-	}
 }
 
 static inline void
 spa_graph_port_link(struct spa_graph_port *out, struct spa_graph_port *in)
 {
-	spa_debug("port %p link to %p", out, in);
+	spa_debug("port %p link to %p %p %p", out, in, in->node, in->node->state);
 	out->peer = in;
 	in->peer = out;
+	if (in->direction == SPA_DIRECTION_INPUT)
+		in->node->state->required++;
+	else
+		out->node->state->required++;
 }
 
 static inline void
 spa_graph_port_unlink(struct spa_graph_port *port)
 {
+	struct spa_graph_port *out, *in;
+
 	spa_debug("port %p unlink from %p", port, port->peer);
-	if (port->peer) {
+	if (port->direction == SPA_DIRECTION_INPUT) {
-		port->peer->peer = NULL;
+		in = port;
-		port->peer = NULL;
+		out = port->peer;
+	} else {
+		out = port;
+		in = port->peer;
+	}
+
+	if (out && in) {
+		in->node->state->required--;
+		out->peer = NULL;
+		in->peer = NULL;
 	}
 }
 

spa/tests/test-bluez5.c

@@ -48,7 +48,7 @@ static struct spa_log *logger;
 #define spa_debug(f,...) spa_log_trace(logger, f, __VA_ARGS__)
 
 #include <spa/graph/graph.h>
-#include <spa/graph/graph-scheduler6.h>
+#include <spa/graph/graph-scheduler1.h>
 
 #include <lib/debug.h>
 

spa/tests/test-control.c

@@ -47,7 +47,7 @@ static SPA_LOG_IMPL(default_log);
 #define spa_debug(f,...) spa_log_trace(&default_log.log, f, __VA_ARGS__)
 
 #include <spa/graph/graph.h>
-#include <spa/graph/graph-scheduler6.h>
+#include <spa/graph/graph-scheduler1.h>
 
 #include <lib/debug.h>
 
@@ -116,9 +116,11 @@ struct data {
 	struct spa_graph graph;
 	struct spa_graph_data graph_data;
 	struct spa_graph_node source_node;
+	struct spa_graph_state source_state;
 	struct spa_graph_port source_out;
 	struct spa_graph_port sink_in;
 	struct spa_graph_node sink_node;
+	struct spa_graph_state sink_state;
 
 	struct spa_node *sink;
 
@@ -396,13 +398,13 @@ static int make_nodes(struct data *data, const char *device)
 			     data->type.io.Buffers,
 			     &data->source_sink_io[0], sizeof(data->source_sink_io[0]));
 
-	spa_graph_node_init(&data->source_node);
+	spa_graph_node_init(&data->source_node, &data->source_state);
 	spa_graph_node_set_implementation(&data->source_node, 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);
+	spa_graph_node_init(&data->sink_node, &data->sink_state);
 	spa_graph_node_set_implementation(&data->sink_node, 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/tests/test-graph.c

@@ -42,7 +42,7 @@ static SPA_LOG_IMPL(default_log);
 #define spa_debug(f,...) spa_log_trace(&default_log.log, f, __VA_ARGS__)
 
 #include <spa/graph/graph.h>
-#include <spa/graph/graph-scheduler6.h>
+#include <spa/graph/graph-scheduler1.h>
 
 #include <lib/debug.h>
 
@@ -109,12 +109,15 @@ struct data {
 	struct spa_graph graph;
 	struct spa_graph_data graph_data;
 	struct spa_graph_node source_node;
+	struct spa_graph_state source_state;
 	struct spa_graph_port source_out;
 	struct spa_graph_port volume_in;
 	struct spa_graph_node volume_node;
+	struct spa_graph_state volume_state;
 	struct spa_graph_port volume_out;
 	struct spa_graph_port sink_in;
 	struct spa_graph_node sink_node;
+	struct spa_graph_state sink_state;
 
 	struct spa_node *sink;
 	struct spa_io_buffers volume_sink_io[1];
@@ -141,7 +144,7 @@ struct data {
 
 #define MIN_LATENCY     64
 
-#define BUFFER_SIZE    MIN_LATENCY
+#define BUFFER_SIZE    (MIN_LATENCY * sizeof(int16_t) * 2)
 
 static void
 init_buffer(struct data *data, struct spa_buffer **bufs, struct buffer *ba, int n_buffers,
@@ -353,13 +356,13 @@ static int make_nodes(struct data *data, const char *device)
 			     data->type.io.Buffers,
 			     &data->volume_sink_io[0], sizeof(data->volume_sink_io[0]));
 
-	spa_graph_node_init(&data->source_node);
+	spa_graph_node_init(&data->source_node, &data->source_state);
 	spa_graph_node_set_implementation(&data->source_node, 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);
+	spa_graph_node_init(&data->volume_node, &data->volume_state);
 	spa_graph_node_set_implementation(&data->volume_node, 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]);
@@ -370,7 +373,7 @@ static int make_nodes(struct data *data, const char *device)
 	spa_graph_port_init(&data->volume_out, SPA_DIRECTION_OUTPUT, 0, 0, &data->volume_sink_io[0]);
 	spa_graph_port_add(&data->volume_node, &data->volume_out);
 
-	spa_graph_node_init(&data->sink_node);
+	spa_graph_node_init(&data->sink_node, &data->sink_state);
 	spa_graph_node_set_implementation(&data->sink_node, 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/tests/test-graph2.c

@@ -170,11 +170,11 @@ static int make_graph1(struct data *data)
 {
 	int v = start_write(data);
 
-	spa_graph_node_init(&data->source_node[v]);
+	spa_graph_node_init(&data->source_node[v], NULL);
 	spa_graph_node_add(&data->graph[v], &data->source_node[v]);
 	spa_graph_port_add(&data->source_node[v], &data->source_out[v]);
 
-	spa_graph_node_init(&data->volume_node[v]);
+	spa_graph_node_init(&data->volume_node[v], NULL);
 	spa_graph_node_add(&data->graph[v], &data->volume_node[v]);
 	spa_graph_port_add(&data->volume_node[v], &data->volume_in[v]);
 
@@ -182,7 +182,7 @@ static int make_graph1(struct data *data)
 
 	spa_graph_port_add(&data->volume_node[v], &data->volume_out[v]);
 
-	spa_graph_node_init(&data->sink_node[v]);
+	spa_graph_node_init(&data->sink_node[v], NULL);
 	spa_graph_node_add(&data->graph[v], &data->sink_node[v]);
 	spa_graph_port_add(&data->sink_node[v], &data->sink_in[v]);
 

spa/tests/test-mixer.c

@@ -117,14 +117,18 @@ struct data {
 	struct spa_graph graph;
 	struct spa_graph_data graph_data;
 	struct spa_graph_node source1_node;
+	struct spa_graph_state source1_state;
 	struct spa_graph_port source1_out;
 	struct spa_graph_node source2_node;
+	struct spa_graph_state source2_state;
 	struct spa_graph_port source2_out;
 	struct spa_graph_port mix_in[2];
 	struct spa_graph_node mix_node;
+	struct spa_graph_state mix_state;
 	struct spa_graph_port mix_out;
 	struct spa_graph_port sink_in;
 	struct spa_graph_node sink_node;
+	struct spa_graph_state sink_state;
 
 	struct spa_node *sink;
 	struct spa_io_buffers mix_sink_io[1];
@@ -157,7 +161,7 @@ struct data {
 	unsigned int n_fds;
 };
 
-#define MIN_LATENCY     64
+#define MIN_LATENCY     512
 
 #define BUFFER_SIZE1    MIN_LATENCY
 #define BUFFER_SIZE2    MIN_LATENCY - 4
@@ -467,19 +471,19 @@ static int make_nodes(struct data *data, const char *device)
 
 
 #ifdef USE_GRAPH
-	spa_graph_node_init(&data->source1_node);
+	spa_graph_node_init(&data->source1_node, &data->source1_state);
 	spa_graph_node_set_implementation(&data->source1_node, 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);
+	spa_graph_node_init(&data->source2_node, &data->source2_state);
 	spa_graph_node_set_implementation(&data->source2_node, 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);
+	spa_graph_node_init(&data->mix_node, &data->mix_state);
 	spa_graph_node_set_implementation(&data->mix_node, data->mix);
 	spa_graph_port_init(&data->mix_in[0], SPA_DIRECTION_INPUT,
 			    data->mix_ports[0], 0, &data->source1_mix_io[0]);
@@ -495,7 +499,7 @@ static int make_nodes(struct data *data, const char *device)
 	spa_graph_port_init(&data->mix_out, SPA_DIRECTION_OUTPUT, 0, 0, &data->mix_sink_io[0]);
 	spa_graph_port_add(&data->mix_node, &data->mix_out);
 
-	spa_graph_node_init(&data->sink_node);
+	spa_graph_node_init(&data->sink_node, &data->sink_state);
 	spa_graph_node_set_implementation(&data->sink_node, 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/tests/test-perf.c

@@ -111,9 +111,11 @@ struct data {
 	struct spa_graph graph;
 	struct spa_graph_data graph_data;
 	struct spa_graph_node source_node;
+	struct spa_graph_state source_state;
 	struct spa_graph_port source_out;
 	struct spa_graph_port sink_in;
 	struct spa_graph_node sink_node;
+	struct spa_graph_state sink_state;
 
 	struct spa_node *sink;
 	struct spa_io_buffers source_sink_io[1];
@@ -226,7 +228,6 @@ static int make_node(struct data *data, struct spa_node **node, const char *lib,
 static void on_sink_pull(struct data *data)
 {
 	spa_log_trace(data->log, "do sink pull");
-	data->sink_node.state = SPA_STATUS_NEED_BUFFER;
 	if (data->mode & MODE_DIRECT) {
 		spa_node_process_output(data->source);
 		spa_node_process_input(data->sink);
@@ -373,7 +374,7 @@ static int make_nodes(struct data *data)
 			     data->type.io.Buffers,
 			     &data->source_sink_io[0], sizeof(data->source_sink_io[0]));
 
-	spa_graph_node_init(&data->source_node);
+	spa_graph_node_init(&data->source_node, &data->source_state);
 	spa_graph_node_set_implementation(&data->source_node, data->source);
 	spa_graph_node_add(&data->graph, &data->source_node);
 
@@ -381,7 +382,7 @@ static int make_nodes(struct data *data)
 	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);
+	spa_graph_node_init(&data->sink_node, &data->sink_state);
 	spa_graph_node_set_implementation(&data->sink_node, data->sink);
 	spa_graph_node_add(&data->graph, &data->sink_node);
 

src/modules/module-audio-dsp.c

@@ -43,6 +43,10 @@
 #define MAX_PORTS	256
 #define MAX_BUFFERS	8
 
+#define DEFAULT_CHANNELS	2
+#define DEFAULT_SAMPLE_RATE	44100
+#define DEFAULT_BUFFER_SIZE	1024
+
 struct type {
 	struct spa_type_media_type media_type;
         struct spa_type_media_subtype media_subtype;
@@ -77,7 +81,6 @@ struct impl {
 
 struct buffer {
 #define BUFFER_FLAG_OUT		(1<<0)
-#define BUFFER_FLAG_FILLED	(1<<1)
 	uint32_t flags;
         struct spa_list link;
 	struct spa_buffer *outbuf;
@@ -218,7 +221,7 @@ static void recycle_buffer(struct node *n, struct port *p, uint32_t id)
 	if (SPA_FLAG_CHECK(b->flags, BUFFER_FLAG_OUT)) {
 		pw_log_trace("recycle buffer %d", id);
 	        spa_list_append(&p->queue, &b->link);
-		SPA_FLAG_UNSET(b->flags, BUFFER_FLAG_OUT | BUFFER_FLAG_FILLED);
+		SPA_FLAG_UNSET(b->flags, BUFFER_FLAG_OUT);
 	}
 }
 
@@ -242,7 +245,8 @@ static void conv_f32_s16(int16_t *out, int index, float *in, int n_samples, int
 		else if (in[i] >= 1.0f)
 			*out = 32767;
 		else
-			*out = lrintf(in[i] * 32767.0f);
+			//*out = lrintf(in[i] * 32767.0f);
+			*out = in[i] * 32767.0f;
 		out += stride;
 	}
 }
@@ -294,19 +298,16 @@ static int node_process_input(struct spa_node *node)
 		return -EPIPE;
 	}
 
-	if (SPA_FLAG_CHECK(out->flags, BUFFER_FLAG_FILLED)) {
+	dequeue_buffer(n, out);
-		dequeue_buffer(n, out);
+	outio->buffer_id = out->outbuf->id;
-		outio->buffer_id = out->outbuf->id;
+	outio->status = SPA_STATUS_HAVE_BUFFER;
-		outio->status = SPA_STATUS_HAVE_BUFFER;
+
+	pw_log_trace(NAME " %p: output buffer %d %d", this, out->outbuf->id, out->flags);
+
+	out->outbuf->datas[0].chunk->offset = 0;
+	out->outbuf->datas[0].chunk->size = n->buffer_size * sizeof(int16_t) * n->channels;
+	out->outbuf->datas[0].chunk->stride = 0;
 
-		out->outbuf->datas[0].chunk->offset = 0;
-		out->outbuf->datas[0].chunk->size = n->buffer_size * sizeof(int16_t) * 2;
-		out->outbuf->datas[0].chunk->stride = 0;
-	}
-	else {
-		outio->buffer_id = SPA_ID_INVALID;
-		outio->status = SPA_STATUS_HAVE_BUFFER;
-	}
 	return outio->status;
 }
 
@@ -340,6 +341,7 @@ static int node_process_output(struct spa_node *node)
 				continue;
 
 			inio->status = SPA_STATUS_NEED_BUFFER;
+			pw_log_trace(NAME " %p: port %d %d", this, i, inio->buffer_id);
 		}
 		outio->status = SPA_STATUS_NEED_BUFFER;
 	}
@@ -459,8 +461,8 @@ static int port_enum_params(struct spa_node *node,
 
 		param = spa_pod_builder_object(&b,
 			id, t->param_buffers.Buffers,
-			":", t->param_buffers.size,    "iru", n->buffer_size * sizeof(float),
+			":", t->param_buffers.size,    "i", n->buffer_size * sizeof(float),
-				SPA_POD_PROP_MIN_MAX(24, 4096),
+//				SPA_POD_PROP_MIN_MAX(24, 4096),
 			":", t->param_buffers.stride,  "i", 0,
 			":", t->param_buffers.buffers, "ir", 2,
 				SPA_POD_PROP_MIN_MAX(1, MAX_BUFFERS),
@@ -543,8 +545,6 @@ static int port_use_buffers(struct spa_node *node, enum spa_direction direction,
 		b->flags = 0;
 		b->outbuf = buffers[i];
 
-		n->buffer_size = d[0].maxsize / 4;
-
 		if ((d[0].type == t->data.MemPtr ||
 		     d[0].type == t->data.MemFd ||
 		     d[0].type == t->data.DmaBuf) && d[0].data != NULL) {
@@ -637,18 +637,17 @@ static int schedule_mix_input(struct spa_node *_node)
 	struct spa_graph_node *node = &port->rt.mix_node;
 	struct spa_graph_port *gp;
 	struct spa_io_buffers *io = port->rt.mix_port.io;
-	size_t buffer_size = p->node->buffer_size;
+	size_t buffer_size = n->buffer_size;
 	struct buffer *outb;
 	float *out = NULL;
 	int layer = 0;
 	int stride = n->n_in_ports;
 
+	pw_log_trace("port %p", port);
+
 	spa_list_for_each(gp, &node->ports[SPA_DIRECTION_INPUT], link) {
 		struct pw_port_mix *mix = SPA_CONTAINER_OF(gp, struct pw_port_mix, port);
 
-		if (gp->flags & SPA_GRAPH_PORT_FLAG_DISABLED)
-			continue;
-
 		pw_log_trace("mix %p: input %d %d/%d", node,
 				gp->io->status, gp->io->buffer_id, mix->n_buffers);
 
@@ -656,36 +655,32 @@ static int schedule_mix_input(struct spa_node *_node)
 			continue;
 
 		if (layer++ == 0) {
-			io->buffer_id = gp->io->buffer_id;
+			out = mix->buffers[gp->io->buffer_id]->datas[0].data;
-			io->status = SPA_STATUS_HAVE_BUFFER;
-			out = mix->buffers[io->buffer_id]->datas[0].data;
 		}
 		else {
 			add_f32(out, mix->buffers[gp->io->buffer_id]->datas[0].data, buffer_size);
 		}
 
-		pw_log_trace("mix %p: input %p %p->%p %d %d", node,
+		pw_log_trace("mix %p: input %p %p->%p %d %d %zd", node,
-				gp, gp->io, io, gp->io->status, gp->io->buffer_id);
+				gp, gp->io, io, gp->io->status, gp->io->buffer_id, buffer_size);
 	}
 
 	outb = peek_buffer(n, outp);
 	if (outb == NULL)
 		return -EPIPE;
 
-	if (layer == 0) {
+	if (layer > 0)
+		conv_f32_s16(outb->ptr, port->port_id, out, buffer_size, stride);
+	else {
 		fill_s16(outb->ptr, port->port_id, buffer_size, stride);
 	}
-	else {
+
-		conv_f32_s16(outb->ptr, port->port_id, out, buffer_size, stride);
-		SPA_FLAG_SET(outb->flags, BUFFER_FLAG_FILLED);
-	}
 	return SPA_STATUS_HAVE_BUFFER;
 }
 
 static int schedule_mix_output(struct spa_node *_node)
 {
 	struct pw_port *port = SPA_CONTAINER_OF(_node, struct pw_port, mix_node);
-	struct port *p = port->owner_data;
 	struct spa_graph_node *node = &port->rt.mix_node;
 	struct spa_graph_port *gp;
 	struct spa_io_buffers *io = port->rt.mix_port.io;
@@ -693,9 +688,7 @@ static int schedule_mix_output(struct spa_node *_node)
 	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);
-		if (gp->flags & SPA_GRAPH_PORT_FLAG_DISABLED)
+		gp->io->status = io->status;
-			continue;
-		*gp->io = *io;
 	}
 	return io->status;
 }
@@ -708,9 +701,14 @@ static int schedule_mix_use_buffers(struct spa_node *_node,
 {
 	struct pw_port *port = SPA_CONTAINER_OF(_node, struct pw_port, mix_node);
 	struct port *p = port->owner_data;
+	struct node *n = p->node;
 
 	pw_log_debug("port %p: %d use buffers %d %p", port, port_id, n_buffers, buffers);
 
+	if (n_buffers > 0) {
+		n->buffer_size = buffers[0]->datas[0].maxsize / sizeof(float);
+	}
+
 	return 0;
 }
 
@@ -763,8 +761,10 @@ static struct port *make_port(struct node *n, enum pw_direction direction,
 	if (direction == PW_DIRECTION_INPUT) {
 		n->in_ports[id] = p;
 		n->n_in_ports++;
-		port->mix_node = schedule_mix_node;
+		if (flags & PORT_FLAG_RAW_F32) {
-		port->mix = &port->mix_node;
+			port->mix_node = schedule_mix_node;
+			port->mix = &port->mix_node;
+		}
 	} else {
 		n->out_ports[id] = p;
 		n->n_out_ports++;
@@ -790,7 +790,7 @@ static struct pw_node *make_node(struct impl *impl, const struct pw_properties *
 
 	snprintf(node_name, sizeof(node_name), "system_%s", alias);
 	for (i = 0; node_name[i]; i++) {
-		if (node_name[i] == ':')
+		if (node_name[i] == ':' || node_name[i] == ',')
 			node_name[i] = '_';
 	}
 	if ((alias = pw_properties_get(props, "alsa.card")) == NULL)
@@ -804,9 +804,9 @@ static struct pw_node *make_node(struct impl *impl, const struct pw_properties *
 	n->node = node;
 	n->impl = impl;
 	n->node_impl = node_impl;
-	n->channels = 2;
+	n->channels = DEFAULT_CHANNELS;
-	n->sample_rate = 44100;
+	n->sample_rate = DEFAULT_SAMPLE_RATE;
-	n->buffer_size = 1024 / sizeof(float);
+	n->buffer_size = DEFAULT_BUFFER_SIZE;
 	pw_node_set_implementation(node, &n->node_impl);
 
 	p = make_port(n, direction, 0, 0, NULL);
@@ -964,6 +964,7 @@ static int module_init(struct pw_module *module, struct pw_properties *propertie
 	pw_core_add_listener(core, &impl->core_listener, &core_events, impl);
 	pw_module_add_listener(module, &impl->module_listener, &module_events, impl);
 
+
 	return 0;
 }
 

src/pipewire/core.c

@@ -33,7 +33,16 @@
 #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;
+};
+
+
 struct resource_data {
 	struct spa_hook resource_listener;
 };
@@ -358,13 +367,16 @@ static const struct pw_global_events global_events = {
  */
 struct pw_core *pw_core_new(struct pw_loop *main_loop, struct pw_properties *properties)
 {
+	struct impl *impl;
 	struct pw_core *this;
 	const char *name;
 
-	this = calloc(1, sizeof(struct pw_core));
+	impl = calloc(1, sizeof(struct impl));
-	if (this == NULL)
+	if (impl == NULL)
 		return NULL;
 
+	this = &impl->this;
+
 	pw_log_debug("core %p: new", this);
 
 	if (properties == NULL)
@@ -384,6 +396,10 @@ 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);
@@ -686,6 +702,7 @@ struct pw_port *pw_core_find_port(struct pw_core *core,
 				continue;
 			}
 			best = p;
+			break;
 		}
 	}
 	if (best == NULL) {
@@ -747,6 +764,9 @@ int pw_core_find_format(struct pw_core *core,
 			asprintf(error, "error get output format: %s", spa_strerror(res));
 			goto error;
 		}
+		pw_log_debug("Got output format:");
+		if (pw_log_level_enabled(SPA_LOG_LEVEL_DEBUG))
+			spa_debug_pod(*format, SPA_DEBUG_FLAG_FORMAT);
 	} else if (out_state == PW_PORT_STATE_CONFIGURE && in_state > PW_PORT_STATE_CONFIGURE) {
 		/* only output needs format */
 		if ((res = spa_node_port_enum_params(input->node->node,
@@ -758,6 +778,9 @@ int pw_core_find_format(struct pw_core *core,
 			asprintf(error, "error get input format: %s", spa_strerror(res));
 			goto error;
 		}
+		pw_log_debug("Got input format:");
+		if (pw_log_level_enabled(SPA_LOG_LEVEL_DEBUG))
+			spa_debug_pod(*format, SPA_DEBUG_FLAG_FORMAT);
 	} else if (in_state == PW_PORT_STATE_CONFIGURE && out_state == PW_PORT_STATE_CONFIGURE) {
 		struct spa_pod_builder fb = { 0 };
 		uint8_t fbuf[4096];

src/pipewire/link.c

@@ -785,15 +785,9 @@ do_activate_link(struct spa_loop *loop,
         struct pw_link *this = user_data;
 
 	pw_log_trace("link %p: activate", this);
-	SPA_FLAG_UNSET(this->rt.mix[0].port.flags, SPA_GRAPH_PORT_FLAG_DISABLED);
-	SPA_FLAG_UNSET(this->rt.mix[1].port.flags, SPA_GRAPH_PORT_FLAG_DISABLED);
 
-	spa_list_append(&this->output->node->rt.links[SPA_DIRECTION_OUTPUT],
+	spa_graph_port_link(&this->rt.mix[SPA_DIRECTION_OUTPUT].port,
-			&this->rt.out_node_link);
+			    &this->rt.mix[SPA_DIRECTION_INPUT].port);
-	spa_list_append(&this->input->node->rt.links[SPA_DIRECTION_INPUT],
-			&this->rt.in_node_link);
-
-	__atomic_add_fetch(&this->input->node->rt.activation->required, 1, __ATOMIC_SEQ_CST);
 
 	return 0;
 }
@@ -1047,15 +1041,11 @@ do_deactivate_link(struct spa_loop *loop,
 		   bool async, uint32_t seq, const void *data, size_t size, void *user_data)
 {
         struct pw_link *this = user_data;
-	pw_log_debug("link %p: disable %p and %p", this, &this->rt.mix[0], &this->rt.mix[1]);
 
-	__atomic_sub_fetch(&this->input->node->rt.activation->required, 1, __ATOMIC_SEQ_CST);
+	pw_log_trace("link %p: disable %p and %p %d", this, &this->rt.mix[0], &this->rt.mix[1],
+			this->io->status);
 
-	SPA_FLAG_SET(this->rt.mix[0].port.flags, SPA_GRAPH_PORT_FLAG_DISABLED);
+	spa_graph_port_unlink(&this->rt.mix[SPA_DIRECTION_OUTPUT].port);
-	SPA_FLAG_SET(this->rt.mix[1].port.flags, SPA_GRAPH_PORT_FLAG_DISABLED);
-
-	spa_list_remove(&this->rt.out_node_link);
-	spa_list_remove(&this->rt.in_node_link);
 
 	return 0;
 }
@@ -1264,6 +1254,7 @@ struct pw_link *pw_link_new(struct pw_core *core,
 	this->info.props = this->properties ? &this->properties->dict : NULL;
 
 	impl->io = SPA_IO_BUFFERS_INIT;
+	impl->io.status = SPA_STATUS_NEED_BUFFER;
 
 	pw_port_init_mix(output, &this->rt.mix[SPA_DIRECTION_OUTPUT]);
 	pw_port_init_mix(input, &this->rt.mix[SPA_DIRECTION_INPUT]);
@@ -1272,14 +1263,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);
 
-	spa_graph_port_link(&this->rt.mix[SPA_DIRECTION_OUTPUT].port,
-			    &this->rt.mix[SPA_DIRECTION_INPUT].port);
-
 	/* nodes can be in different data loops so we do this twice */
 	pw_loop_invoke(output_node->data_loop, do_add_link,
 		       SPA_ID_INVALID, &output, sizeof(struct pw_port *), false, this);
 	pw_loop_invoke(input_node->data_loop, do_add_link,
-		       SPA_ID_INVALID, &input, sizeof(struct pw_port *), false, this);
+		       SPA_ID_INVALID, &input, sizeof(struct pw_port *), true, this);
 
 	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);

src/pipewire/node.c

@@ -317,6 +317,17 @@ static const struct pw_global_events global_events = {
 	.bind = global_bind,
 };
 
+static int
+do_node_add(struct spa_loop *loop,
+	    bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct pw_node *this = user_data;
+
+	spa_graph_node_add(this->rt.graph, &this->rt.node);
+
+	return 0;
+}
+
 int pw_node_register(struct pw_node *this,
 		     struct pw_client *owner,
 		     struct pw_global *parent,
@@ -338,6 +349,8 @@ 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);
@@ -379,7 +392,7 @@ static int impl_node_process(struct pw_node *node)
 	struct spa_graph_port *p;
 	int res = 0, old;
 
-	old = node->rt.activation->status;
+	old = node->rt.activation->state.status;
 
 	pw_log_trace("node %p: process %d", node, old);
 
@@ -407,7 +420,7 @@ static int impl_node_process(struct pw_node *node)
 	}
 	pw_log_trace("node %p: process %d", node, res);
 
-	node->rt.activation->status = res;
+	node->rt.activation->state.status = res;
 
 	return res;
 }
@@ -457,6 +470,8 @@ 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);
@@ -469,11 +484,12 @@ struct pw_node *pw_node_new(struct pw_core *core,
 	spa_list_init(&this->output_ports);
 	pw_map_init(&this->output_port_map, 64, 64);
 
-	spa_graph_node_init(&this->rt.node);
+	this->rt.activation = &impl->activation;
+
+	spa_graph_node_init(&this->rt.node, &this->rt.activation->state);
 	spa_list_init(&this->rt.links[SPA_DIRECTION_INPUT]);
 	spa_list_init(&this->rt.links[SPA_DIRECTION_OUTPUT]);
-	this->rt.activation = &impl->activation;
+	impl->activation.state.status = SPA_STATUS_NEED_BUFFER;
-	impl->activation.status = SPA_STATUS_NEED_BUFFER;
 
 	this->process = impl_node_process;
 
@@ -554,116 +570,24 @@ static void node_event(void *data, struct spa_event *event)
 	spa_hook_list_call(&node->listener_list, struct pw_node_events, event, event);
 }
 
-
-static void node_process(struct pw_node *node)
-{
-	pw_log_trace("node %p: pending %d required %d %d", node,
-			node->rt.activation->pending, node->rt.activation->required,
-			node->rt.activation->status);
-
-	if (--node->rt.activation->pending == 0) {
-		if (node->process)
-			node->rt.activation->status = node->process(node);
-		else
-			node->rt.activation->status = SPA_STATUS_OK;
-
-		pw_log_trace("node %p: %d", node, node->rt.activation->status);
-	}
-}
-
 static void node_need_input(void *data)
 {
-	struct pw_node *node = data, *n, *pn;
+	struct pw_node *node = data;
-	struct spa_list queue, pending;
-	struct spa_graph_port *p;
 
-	pw_log_trace("node %p: need input", node);
+	pw_log_trace("node %p: need input %d", node, node->rt.activation->state.status);
-
-	spa_list_for_each(p, &node->rt.node.ports[SPA_DIRECTION_INPUT], link)
-		spa_node_process_output(p->peer->node->implementation);
 
 	spa_hook_list_call(&node->listener_list, struct pw_node_events, need_input);
-
+	spa_graph_need_input(node->rt.graph, &node->rt.node);
-	spa_list_init(&queue);
-	spa_list_init(&pending);
-
-	node->rt.activation->status = SPA_STATUS_NEED_BUFFER;
-
-	if (node->rt.sched_link.next == NULL)
-		spa_list_append(&queue, &node->rt.sched_link);
-
-	while (!spa_list_is_empty(&queue)) {
-		struct pw_link *l;
-
-		n = spa_list_first(&queue, struct pw_node, rt.sched_link);
-		spa_list_remove(&n->rt.sched_link);
-		n->rt.sched_link.next = NULL;
-
-		n->rt.activation->pending = n->rt.activation->required + 1;
-		pw_log_trace("node %p: add %d %d status %d", n,
-				n->rt.activation->pending, n->rt.activation->required,
-				n->rt.activation->status);
-
-		spa_list_prepend(&pending, &n->rt.sched_link);
-
-		if (n->rt.activation->status == SPA_STATUS_HAVE_BUFFER)
-			continue;
-
-		spa_list_for_each(l, &n->rt.links[SPA_DIRECTION_INPUT], rt.in_node_link) {
-			pn = l->output->node;
-
-	                pw_log_trace("node %p: %p in %p %d %d", n, pn, l, l->io->status,
-					pn->rt.activation->status);
-
-			if (pn->rt.sched_link.next != NULL)
-				continue;
-
-			if (l->io->status == SPA_STATUS_NEED_BUFFER) {
-				spa_list_for_each(p, &pn->rt.node.ports[SPA_DIRECTION_OUTPUT], link)
-					spa_node_process_output(p->peer->node->implementation);
-
-				if (pn->node->process_output)
-					pn->rt.activation->status = spa_node_process_output(pn->node);
-
-				if (pn->rt.activation->status == SPA_STATUS_NEED_BUFFER) {
-					spa_list_for_each(p, &pn->rt.node.ports[SPA_DIRECTION_INPUT], link)
-						spa_node_process_output(p->peer->node->implementation);
-				}
-
-			} else {
-				n->rt.activation->pending--;
-			}
-
-			spa_list_append(&queue, &pn->rt.sched_link);
-		}
-	}
-	while (!spa_list_is_empty(&pending)) {
-		n = spa_list_first(&pending, struct pw_node, rt.sched_link);
-		spa_list_remove(&n->rt.sched_link);
-		n->rt.sched_link.next = NULL;
-
-                pw_log_trace("schedule node %p: %d", n, n->rt.activation->status);
-		node_process(n);
-	}
 }
 
 static void node_have_output(void *data)
 {
-	struct pw_node *node = data, *pn;
+	struct pw_node *node = data;
-	struct pw_link *l;
-	struct spa_graph_port *p;
 
 	pw_log_trace("node %p: have output", node);
-	spa_list_for_each(p, &node->rt.node.ports[SPA_DIRECTION_OUTPUT], link)
-		spa_node_process_input(p->peer->node->implementation);
 
 	spa_hook_list_call(&node->listener_list, struct pw_node_events, have_output);
-
+	spa_graph_have_output(node->rt.graph, &node->rt.node);
-	spa_list_for_each(l, &node->rt.links[SPA_DIRECTION_OUTPUT], rt.out_node_link) {
-		pn = l->input->node;
-                pw_log_trace("node %p: %p out %p %d", node, pn, l, l->io->status);
-		node_process(pn);
-	}
 }
 
 static void node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
@@ -681,7 +605,6 @@ static void node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
 	}
 }
 
-
 static const struct spa_node_callbacks node_callbacks = {
 	SPA_VERSION_NODE_CALLBACKS,
 	.done = node_done,
@@ -715,6 +638,19 @@ void pw_node_add_listener(struct pw_node *node,
 	spa_hook_list_append(&node->listener_list, listener, events, data);
 }
 
+static int
+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;
+}
+
 /** Destroy a node
  * \param node a node to destroy
  *
@@ -735,6 +671,7 @@ void pw_node_destroy(struct pw_node *node)
 	pause_node(node);
 
 	if (node->registered) {
+		pw_loop_invoke(node->data_loop, do_node_remove, 1, NULL, 0, true, node);
 		spa_list_remove(&node->link);
 	}
 
@@ -910,6 +847,7 @@ struct pw_port *pw_node_get_free_port(struct pw_node *node, enum pw_direction di
 	} else {
 		port = mixport;
 	}
+	pw_log_debug("node %p: return port %p", node, port);
 	return port;
 
       no_mem:

src/pipewire/port.c

@@ -61,8 +61,6 @@ static int schedule_tee_input(struct spa_node *data)
 		spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link) {
 			pw_log_trace("port %p: port %d %d %p->%p", this,
 					p->port_id, p->flags, io, p->io);
-			if (p->flags & SPA_GRAPH_PORT_FLAG_DISABLED)
-				continue;
 			*p->io = *io;
 		}
 	}
@@ -79,8 +77,6 @@ static int schedule_tee_output(struct spa_node *data)
 		pw_log_trace("port %p: port %d %d %p->%p %d %d",
 				this, p->port_id, p->flags, p->io, io,
 				p->io->status, p->io->buffer_id);
-		if (p->flags & SPA_GRAPH_PORT_FLAG_DISABLED)
-			continue;
 		*io = *p->io;
 	}
 	pw_log_trace("port %p: tee output %d %d", this, io->status, io->buffer_id);
@@ -115,8 +111,6 @@ static int schedule_mix_input(struct spa_node *data)
 	struct spa_io_buffers *io = this->rt.mix_port.io;
 
 	spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
-		if (p->flags & SPA_GRAPH_PORT_FLAG_DISABLED)
-			continue;
 		pw_log_trace("port %p: mix input %d %p->%p %d %d", this,
 				p->port_id, p->io, io, p->io->status, p->io->buffer_id);
 		*io = *p->io;
@@ -136,15 +130,9 @@ static int schedule_mix_output(struct spa_node *data)
 		spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
 			pw_log_trace("port %p: port %d %d %p->%p", this,
 					p->port_id, p->flags, io, p->io);
-			if (p->flags & SPA_GRAPH_PORT_FLAG_DISABLED)
-				continue;
 			*p->io = *io;
 		}
 	}
-	else {
-		io->status = SPA_STATUS_HAVE_BUFFER;
-		io->buffer_id = SPA_ID_INVALID;
-	}
 	pw_log_trace("port %p: output %d %d", this, io->status, io->buffer_id);
 	return io->status;
 }
@@ -182,7 +170,7 @@ int pw_port_init_mix(struct pw_port *port, struct pw_port_mix *mix)
 
 	spa_graph_port_init(&mix->port,
 			    port->direction, id,
-			    SPA_GRAPH_PORT_FLAG_DISABLED,
+			    0,
 			    NULL);
 
 	mix->port.scheduler_data = port;
@@ -256,7 +244,8 @@ struct pw_port *pw_port_new(enum pw_direction direction,
 			    this->port_id,
 			    0,
 			    &this->rt.io);
-	spa_graph_node_init(&this->rt.mix_node);
+	spa_graph_node_init(&this->rt.mix_node, &this->rt.mix_state);
+	this->rt.mix_state.status = SPA_STATUS_NEED_BUFFER;
 
 	this->mix_node = this->direction == PW_DIRECTION_INPUT ?
 				schedule_mix_node :
@@ -269,6 +258,7 @@ struct pw_port *pw_port_new(enum pw_direction direction,
 			    0,
 			    0,
 			    &this->rt.io);
+	this->rt.io.status = SPA_STATUS_NEED_BUFFER;
 
 	this->rt.mix_port.scheduler_data = this;
 	this->rt.port.scheduler_data = this;
@@ -342,6 +332,7 @@ 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);
 
@@ -517,6 +508,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;
 	pw_loop_invoke(node->data_loop, do_add_port, SPA_ID_INVALID, NULL, 0, false, port);
 
 	if (port->state <= PW_PORT_STATE_INIT)
@@ -554,6 +546,7 @@ 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);
 
 	return 0;
 }

src/pipewire/private.h

@@ -163,6 +163,10 @@ 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 {
@@ -228,15 +232,13 @@ struct pw_node_activation {
 #define TRIGGERED	1
 #define AWAKE		2
 #define FINISHED	3
-	int state;
+	int status;
 
 	uint64_t signal_time;
 	uint64_t awake_time;
 	uint64_t finish_time;
 
-	int status;
+	struct spa_graph_state state;
-	uint32_t required;
-	uint32_t pending;
 };
 
 struct pw_node {
@@ -275,6 +277,7 @@ struct pw_node {
 	int (*process) (struct pw_node *node);
 
 	struct {
+		struct spa_graph *graph;
 		struct spa_graph_node node;
 		struct spa_list links[2];
 		struct pw_node_activation *activation;
@@ -334,10 +337,12 @@ 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 */
 		struct spa_graph_node mix_node;	/**< mixer node */
+		struct spa_graph_state mix_state;	/**< mixer state */
 	} rt;					/**< data only accessed from the data thread */
 
         void *owner_data;		/**< extra owner data */

src/pipewire/remote.c

@@ -726,7 +726,6 @@ do_activate_mix(struct spa_loop *loop,
                 bool async, uint32_t seq, const void *data, size_t size, void *user_data)
 {
 	struct mix *mix = user_data;
-        SPA_FLAG_UNSET(mix->mix.port.flags, SPA_GRAPH_PORT_FLAG_DISABLED);
 	spa_graph_port_add(&mix->port->rt.mix_node, &mix->mix.port);
         return 0;
 }
@@ -908,6 +907,7 @@ static void do_start(struct node_data *data)
 		mix->mix.port.io->buffer_id = SPA_ID_INVALID;
 	}
 	if (!spa_list_is_empty(&data->mix[SPA_DIRECTION_INPUT])) {
+		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);
@@ -1332,7 +1332,6 @@ do_deactivate_mix(struct spa_loop *loop,
                 bool async, uint32_t seq, const void *data, size_t size, void *user_data)
 {
 	struct mix *mix = user_data;
-        SPA_FLAG_SET(mix->mix.port.flags, SPA_GRAPH_PORT_FLAG_DISABLED);
 	spa_graph_port_remove(&mix->mix.port);
         return 0;
 }