From 33a322b96e61ca5c42967f906ba361084a3f6646 Mon Sep 17 00:00:00 2001 From: Wim Taymans Date: Tue, 20 Mar 2018 11:37:11 +0100 Subject: [PATCH] graph: new scheduling model Make explicit links between elements that are used to activate the next element in the graph. Make subgraphs a special regular node. Make a link from the subgraph children to the parent so that the subgraph completes when all the children completed. Implement a single process function in plugins Remove many messages in the client node --- spa/include/spa/graph/graph-scheduler1.h | 3 +- spa/include/spa/graph/graph-scheduler2.h | 144 +++---------- spa/include/spa/graph/graph.h | 203 ++++++++++++------- spa/include/spa/node/node.h | 16 +- spa/include/spa/utils/defs.h | 13 ++ spa/plugins/alsa/alsa-source.c | 10 +- spa/plugins/alsa/alsa-utils.c | 4 +- spa/plugins/audiomixer/audiomixer.c | 7 +- spa/plugins/audiotestsrc/audiotestsrc.c | 12 +- spa/plugins/bluez5/a2dp-sink.c | 12 +- spa/plugins/ffmpeg/ffmpeg-dec.c | 10 +- spa/plugins/ffmpeg/ffmpeg-enc.c | 10 +- spa/plugins/support/loop.c | 15 +- spa/plugins/test/fakesink.c | 22 +- spa/plugins/test/fakesrc.c | 22 +- spa/plugins/v4l2/v4l2-source.c | 4 + spa/plugins/v4l2/v4l2-utils.c | 2 +- spa/plugins/videotestsrc/videotestsrc.c | 12 +- spa/plugins/volume/volume.c | 74 +++---- spa/tests/test-bluez5.c | 5 +- spa/tests/test-control.c | 11 +- spa/tests/test-graph.c | 12 +- spa/tests/test-graph2.c | 7 +- spa/tests/test-mixer.c | 11 +- spa/tests/test-perf.c | 19 +- spa/tests/test-ringbuffer.c | 5 +- spa/tests/test-v4l2.c | 4 +- src/extensions/client-node.h | 5 +- src/modules/module-client-node/client-node.c | 55 +---- src/pipewire/link.c | 51 ++--- src/pipewire/node.c | 90 ++++---- src/pipewire/node.h | 6 +- src/pipewire/port.c | 24 ++- src/pipewire/private.h | 11 +- src/pipewire/remote.c | 139 +------------ src/pipewire/stream.c | 101 +++------ 36 files changed, 401 insertions(+), 750 deletions(-) diff --git a/spa/include/spa/graph/graph-scheduler1.h b/spa/include/spa/graph/graph-scheduler1.h index 97d538885..ab28d542f 100644 --- a/spa/include/spa/graph/graph-scheduler1.h +++ b/spa/include/spa/graph/graph-scheduler1.h @@ -140,8 +140,7 @@ static inline int spa_graph_impl_run(void *data) 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, + .trigger = spa_graph_impl_have_output, }; #ifdef __cplusplus diff --git a/spa/include/spa/graph/graph-scheduler2.h b/spa/include/spa/graph/graph-scheduler2.h index 8d67664e0..d8813bdce 100644 --- a/spa/include/spa/graph/graph-scheduler2.h +++ b/spa/include/spa/graph/graph-scheduler2.h @@ -36,121 +36,6 @@ static inline void spa_graph_data_init(struct spa_graph_data *data, data->graph = graph; } -static inline int spa_graph_trigger(struct spa_graph *g, struct spa_graph_node *node) -{ - uint32_t val; - - spa_debug("node %p: pending %d required %d", node, - node->state->pending, node->state->required); - - if (node->state->pending == 0) { - spa_debug("node %p: nothing pending", node); - return node->state->status; - } - - 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 int spa_graph_impl_need_input(void *data, struct spa_graph_node *node) -{ -#if 0 - 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); - - 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 int spa_graph_impl_have_output(void *data, struct spa_graph_node *node) -{ - struct spa_graph_data *d = (struct spa_graph_data *) data; - struct spa_graph_port *p; - - spa_debug("node %p start push", node); - - 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 void spa_graph_impl_add_graph(struct spa_graph *g, struct spa_list *pending) -{ - struct spa_graph_node *n; - struct spa_graph *sg; - - spa_list_for_each(n, &g->nodes, link) { - n->state->pending = n->state->required + 1; - spa_debug("graph %p node %p: add %d %d status %d", g, n, - n->state->pending, n->state->required, n->state->status); - spa_list_append(pending, &n->sched_link); - } - spa_list_for_each(sg, &g->subgraphs, link) - spa_graph_impl_add_graph(sg, pending); -} - static inline int spa_graph_impl_run(void *data) { struct spa_graph_data *d = (struct spa_graph_data *) data; @@ -159,22 +44,43 @@ static inline int spa_graph_impl_run(void *data) struct spa_list pending; spa_debug("graph %p run", d->graph); + spa_graph_state_reset(g->state); spa_list_init(&pending); - spa_graph_impl_add_graph(g, &pending); + spa_list_for_each(n, &g->nodes, link) { + struct spa_graph_state *s = n->state; + spa_graph_state_reset(s); + + spa_debug("graph %p node %p: add %d status %d", g, n, s->pending, s->status); + + if (s->pending == 0) + spa_list_append(&pending, &n->sched_link); + } spa_list_for_each_safe(n, tmp, &pending, sched_link) - spa_graph_trigger(d->graph, n); + spa_graph_node_process(n); + + return 0; +} + +static inline int spa_graph_impl_finish(void *data) +{ + struct spa_graph_data *d = (struct spa_graph_data *) data; + struct spa_graph *g = d->graph; + + spa_debug("graph %p finish", d->graph); + + if (g->parent) + spa_graph_node_trigger(g->parent); 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, + .finish = spa_graph_impl_finish, }; #ifdef __cplusplus diff --git a/spa/include/spa/graph/graph.h b/spa/include/spa/graph/graph.h index 9f2710acf..54367d97c 100644 --- a/spa/include/spa/graph/graph.h +++ b/spa/include/spa/graph/graph.h @@ -35,62 +35,91 @@ extern "C" { struct spa_graph; struct spa_graph_node; +struct spa_graph_link; struct spa_graph_port; +struct spa_graph_state { + int status; /**< current status */ + uint32_t required; /**< required number of signals */ + uint32_t pending; /**< number of pending signals */ +}; + +static inline void spa_graph_state_reset(struct spa_graph_state *state) +{ + state->pending = state->required; +} + +struct spa_graph_link { + struct spa_list link; + struct spa_graph_state *state; + int (*signal) (void *data, void *target); + void *signal_data; +}; + +#define spa_graph_link_signal(l,t) ((l)->signal((l)->signal_data,(t))) + +static inline int spa_graph_link_trigger(struct spa_graph_link *link, void *target) +{ + struct spa_graph_state *state = link->state; + + if (state->pending == 0) { + spa_debug("link %p: nothing pending", link); + } else { + spa_debug("link %p: pending %d required %d", link, + state->pending, state->required); + if (__atomic_sub_fetch(&state->pending, 1, __ATOMIC_SEQ_CST) == 0) + spa_graph_link_signal(link, target); + } + return state->status; +} + struct spa_graph_callbacks { #define SPA_VERSION_GRAPH_CALLBACKS 0 uint32_t version; - int (*need_input) (void *data, struct spa_graph_node *node); - int (*have_output) (void *data, struct spa_graph_node *node); int (*run) (void *data); + int (*finish) (void *data); }; +#define spa_graph_run(g) ((g)->callbacks->run((g)->callbacks_data)) +#define spa_graph_finish(g) ((g)->callbacks->finish((g)->callbacks_data)) struct spa_graph { - struct spa_list link; /* link for subgraph */ #define SPA_GRAPH_FLAG_DRIVER (1 << 0) uint32_t flags; /* flags */ - struct spa_graph *parent; /* parent graph or NULL when driver */ + struct spa_graph_node *parent; /* parent node or NULL when driver */ + struct spa_graph_state *state; /* state of graph */ struct spa_list nodes; /* list of nodes of this graph */ - struct spa_list subgraphs; /* list of subgraphs */ const struct spa_graph_callbacks *callbacks; void *callbacks_data; }; -#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_run(g) ((g)->callbacks->run((g)->callbacks_data)) - -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_callbacks { #define SPA_VERSION_GRAPH_NODE_CALLBACKS 0 uint32_t version; + int (*trigger) (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); }; +#define spa_graph_node_trigger(n) ((n)->callbacks->trigger((n)->callbacks_data,(n))) +#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_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) + struct spa_list links; /**< list of links */ uint32_t flags; /**< node flags */ struct spa_graph_state *state; /**< state of the node */ + struct spa_graph_link graph_link; /**< link in graph */ + struct spa_graph *subgraph; /**< subgraph or NULL */ 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 */ @@ -100,14 +129,27 @@ struct spa_graph_port { uint32_t flags; /**< port flags */ struct spa_io_buffers *io; /**< io area of the port */ struct spa_graph_port *peer; /**< peer */ - void *scheduler_data; /**< scheduler private data */ }; -static inline void spa_graph_init(struct spa_graph *graph) +static inline int spa_graph_link_signal_node(void *data, void *arg) +{ + struct spa_graph_node *node = data; + return spa_graph_node_process(node); +} + +static inline int spa_graph_link_signal_graph(void *data, void *arg) +{ + struct spa_graph_node *node = data; + if (node->graph) + spa_graph_finish(node->graph); + return 0; +} + +static inline void spa_graph_init(struct spa_graph *graph, struct spa_graph_state *state) { spa_list_init(&graph->nodes); - spa_list_init(&graph->subgraphs); graph->flags = 0; + graph->state = state; spa_debug("graph %p init", graph); } @@ -120,25 +162,22 @@ spa_graph_set_callbacks(struct spa_graph *graph, graph->callbacks_data = data; } -static inline struct spa_graph *spa_graph_find_root(struct spa_graph *graph) +static inline void +spa_graph_link_add(struct spa_graph_node *out, + struct spa_graph_state *state, + struct spa_graph_link *link) { - while (graph->parent) - graph = graph->parent; - return graph; + spa_debug("node %p add link %p to node %p", out, link, state); + link->state = state; + state->required++; + spa_list_append(&out->links, &link->link); } -static inline void spa_graph_add_subgraph(struct spa_graph *graph, struct spa_graph *subgraph) +static inline void spa_graph_link_remove(struct spa_graph_link *link) { - subgraph->parent = graph; - spa_list_append(&graph->subgraphs, &subgraph->link); - spa_debug("graph %p add subgraph %p", graph, subgraph); -} - -static inline void spa_graph_remove_subgraph(struct spa_graph *subgraph) -{ - subgraph->parent = NULL; - spa_list_remove(&subgraph->link); - spa_debug("graph %p remove subgraph", subgraph); + spa_debug("link %p remove", link); + link->state->required--; + spa_list_remove(&link->link); } static inline void @@ -146,13 +185,47 @@ 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]); + spa_list_init(&node->links); node->flags = 0; + node->subgraph = NULL; node->state = state; node->state->required = node->state->pending = 0; node->state->status = SPA_STATUS_OK; + node->graph_link.signal = spa_graph_link_signal_graph; + node->graph_link.signal_data = node; spa_debug("node %p init", node); } +static inline int spa_graph_node_impl_trigger(void *data, struct spa_graph_node *node) +{ + struct spa_graph_link *l, *t; + spa_debug("node %p trigger", node); + spa_list_for_each_safe(l, t, &node->links, link) + spa_graph_link_trigger(l, node); + return 0; +} + +static inline int spa_graph_node_impl_sub_process(void *data, struct spa_graph_node *node) +{ + struct spa_graph *graph = node->subgraph; + spa_debug("node %p: sub process %p", node, graph); + return spa_graph_run(graph); +} + +static const struct spa_graph_node_callbacks spa_graph_node_sub_impl_default = { + SPA_VERSION_GRAPH_NODE_CALLBACKS, + .trigger = spa_graph_node_impl_trigger, + .process = spa_graph_node_impl_sub_process, +}; + +static inline void spa_graph_node_set_subgraph(struct spa_graph_node *node, + struct spa_graph *subgraph) +{ + node->subgraph = subgraph; + subgraph->parent = node; + spa_debug("node %p set subgraph %p", node, subgraph); +} + static inline void spa_graph_node_set_callbacks(struct spa_graph_node *node, const struct spa_graph_node_callbacks *callbacks, @@ -167,11 +240,19 @@ spa_graph_node_add(struct spa_graph *graph, struct spa_graph_node *node) { node->graph = graph; - node->sched_link.next = NULL; spa_list_append(&graph->nodes, &node->link); + spa_graph_link_add(node, graph->state, &node->graph_link); spa_debug("node %p add to graph %p", node, graph); } +static inline void spa_graph_node_remove(struct spa_graph_node *node) +{ + spa_debug("node %p remove", node); + spa_graph_link_remove(&node->graph_link); + spa_list_remove(&node->link); +} + + static inline void spa_graph_port_init(struct spa_graph_port *port, enum spa_direction direction, @@ -195,19 +276,10 @@ spa_graph_port_add(struct spa_graph_node *node, spa_list_append(&node->ports[port->direction], &port->link); } -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->sched_link.next) - 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); - port->node = NULL; } static inline void @@ -216,49 +288,27 @@ spa_graph_port_link(struct spa_graph_port *out, struct spa_graph_port *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->direction == SPA_DIRECTION_INPUT) { - in = port; - out = port->peer; - } else { - out = port; - in = port->peer; - } - - if (out && in) { - in->node->state->required--; - out->peer = NULL; - in->peer = NULL; + if (port->peer) { + port->peer->peer = NULL; + port->peer = NULL; } } 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 res = 0; - res = spa_node_process(n); + spa_debug("node %p: process %d", node, node->state->status); + if ((node->state->status = spa_node_process(n)) == SPA_STATUS_HAVE_BUFFER) + spa_graph_node_trigger(node); - spa_debug("node %p: process %d", node, res); - - node->state->status = res; - - if (res == SPA_STATUS_HAVE_BUFFER) - spa_graph_have_output(g, node); - - return res; + return node->state->status; } static inline int spa_graph_node_impl_reuse_buffer(void *data, struct spa_graph_node *node, @@ -270,6 +320,7 @@ static inline int spa_graph_node_impl_reuse_buffer(void *data, struct spa_graph_ static const struct spa_graph_node_callbacks spa_graph_node_impl_default = { SPA_VERSION_GRAPH_NODE_CALLBACKS, + .trigger = spa_graph_node_impl_trigger, .process = spa_graph_node_impl_process, .reuse_buffer = spa_graph_node_impl_reuse_buffer, }; diff --git a/spa/include/spa/node/node.h b/spa/include/spa/node/node.h index c08e49324..a36eb6ea4 100644 --- a/spa/include/spa/node/node.h +++ b/spa/include/spa/node/node.h @@ -83,23 +83,13 @@ struct spa_node_callbacks { /** * \param node a spa_node * - * The node needs more input. This callback is called from the + * The node is ready for processing. This callback is called from the * data thread. * * When this function is NULL, synchronous operation is requested - * on the input ports. + * on the ports. */ - void (*need_input) (void *data); - /** - * \param node a spa_node - * - * The node has output input. This callback is called from the - * data thread. - * - * When this function is NULL, synchronous operation is requested - * on the output ports. - */ - void (*have_output) (void *data); + void (*process) (void *data, int state); /** * \param node a spa_node diff --git a/spa/include/spa/utils/defs.h b/spa/include/spa/utils/defs.h index 0486c0251..81d8e0f68 100644 --- a/spa/include/spa/utils/defs.h +++ b/spa/include/spa/utils/defs.h @@ -29,6 +29,7 @@ extern "C" { #include #include #include +#include #define SPA_ASYNC_BIT (1 << 30) @@ -155,9 +156,21 @@ struct spa_fraction { #define spa_assert_se(expr) \ do { \ if (SPA_UNLIKELY(!(expr))) \ + fprintf(stderr, "'%s' failed at %s:%u %s()", \ + #expr , __FILE__, __LINE__, __func__); \ abort(); \ } while (false) +#define spa_assert(expr) \ + do { \ + if (SPA_UNLIKELY(!(expr))) { \ + fprintf(stderr, "'%s' failed at %s:%u %s()", \ + #expr , __FILE__, __LINE__, __func__); \ + abort(); \ + } \ + } while (false) + + /* Does exactly nothing */ #define spa_nop() do {} while (false) diff --git a/spa/plugins/alsa/alsa-source.c b/spa/plugins/alsa/alsa-source.c index f2c3dc031..808dd6100 100644 --- a/spa/plugins/alsa/alsa-source.c +++ b/spa/plugins/alsa/alsa-source.c @@ -593,12 +593,7 @@ impl_node_port_send_command(struct spa_node *node, return -ENOTSUP; } -static int impl_node_process_input(struct spa_node *node) -{ - return -ENOTSUP; -} - -static int impl_node_process_output(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct state *this; struct spa_io_buffers *io; @@ -647,8 +642,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_clock_enum_params(struct spa_clock *clock, uint32_t id, uint32_t *index, diff --git a/spa/plugins/alsa/alsa-utils.c b/spa/plugins/alsa/alsa-utils.c index 19f597292..bde448f1c 100644 --- a/spa/plugins/alsa/alsa-utils.c +++ b/spa/plugins/alsa/alsa-utils.c @@ -349,7 +349,7 @@ static inline void try_pull(struct state *state, snd_pcm_uframes_t frames, state->range->min_size = state->threshold * state->frame_size; state->range->max_size = frames * state->frame_size; } - state->callbacks->need_input(state->callbacks_data); + state->callbacks->process(state->callbacks_data, SPA_STATUS_NEED_BUFFER); } } @@ -480,7 +480,7 @@ push_frames(struct state *state, SPA_FLAG_SET(b->flags, BUFFER_FLAG_OUT); io->buffer_id = b->outbuf->id; io->status = SPA_STATUS_HAVE_BUFFER; - state->callbacks->have_output(state->callbacks_data); + state->callbacks->process(state->callbacks_data, SPA_STATUS_HAVE_BUFFER); } return total_frames; } diff --git a/spa/plugins/audiomixer/audiomixer.c b/spa/plugins/audiomixer/audiomixer.c index 89fbd3955..f8bc5c2fd 100644 --- a/spa/plugins/audiomixer/audiomixer.c +++ b/spa/plugins/audiomixer/audiomixer.c @@ -943,6 +943,7 @@ static int mix_output(struct impl *this, size_t n_bytes) return SPA_STATUS_HAVE_BUFFER; } +#if 0 static int impl_node_process_input(struct spa_node *node) { struct impl *this; @@ -1005,8 +1006,9 @@ static int impl_node_process_input(struct spa_node *node) } return outio->status; } +#endif -static int impl_node_process_output(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct impl *this; struct port *outport; @@ -1087,8 +1089,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_get_interface(struct spa_handle *handle, uint32_t interface_id, void **interface) diff --git a/spa/plugins/audiotestsrc/audiotestsrc.c b/spa/plugins/audiotestsrc/audiotestsrc.c index bf3d4b88f..50e43f64f 100644 --- a/spa/plugins/audiotestsrc/audiotestsrc.c +++ b/spa/plugins/audiotestsrc/audiotestsrc.c @@ -433,7 +433,7 @@ static void on_output(struct spa_source *source) res = make_buffer(this); if (res == SPA_STATUS_HAVE_BUFFER) - this->callbacks->have_output(this->callbacks_data); + this->callbacks->process(this->callbacks_data, res); } static int impl_node_send_command(struct spa_node *node, const struct spa_command *command) @@ -1010,12 +1010,7 @@ impl_node_port_send_command(struct spa_node *node, return -ENOTSUP; } -static int impl_node_process_input(struct spa_node *node) -{ - return -ENOTSUP; -} - -static int impl_node_process_output(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct impl *this; struct spa_io_buffers *io; @@ -1068,8 +1063,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_clock_enum_params(struct spa_clock *clock, uint32_t id, uint32_t *index, diff --git a/spa/plugins/bluez5/a2dp-sink.c b/spa/plugins/bluez5/a2dp-sink.c index 1ff759797..65df300bd 100644 --- a/spa/plugins/bluez5/a2dp-sink.c +++ b/spa/plugins/bluez5/a2dp-sink.c @@ -312,7 +312,7 @@ static inline void try_pull(struct impl *this, uint32_t frames, bool do_pull) this->range->min_size = this->threshold * this->frame_size; this->range->max_size = frames * this->frame_size; } - this->callbacks->need_input(this->callbacks_data); + this->callbacks->process(this->callbacks_data, SPA_STATUS_NEED_BUFFER); } } @@ -1257,7 +1257,7 @@ impl_node_port_send_command(struct spa_node *node, return -ENOTSUP; } -static int impl_node_process_input(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct impl *this; struct spa_io_buffers *input; @@ -1287,11 +1287,6 @@ static int impl_node_process_input(struct spa_node *node) return SPA_STATUS_OK; } -static int impl_node_process_output(struct spa_node *node) -{ - return -ENOTSUP; -} - static const struct spa_dict_item node_info_items[] = { { "media.class", "Audio/Sink" }, }; @@ -1320,8 +1315,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_get_interface(struct spa_handle *handle, uint32_t interface_id, void **interface) diff --git a/spa/plugins/ffmpeg/ffmpeg-dec.c b/spa/plugins/ffmpeg/ffmpeg-dec.c index 2b0a17585..46dea6115 100644 --- a/spa/plugins/ffmpeg/ffmpeg-dec.c +++ b/spa/plugins/ffmpeg/ffmpeg-dec.c @@ -437,12 +437,7 @@ spa_ffmpeg_dec_node_port_set_io(struct spa_node *node, return 0; } -static int spa_ffmpeg_dec_node_process_input(struct spa_node *node) -{ - return -EINVAL; -} - -static int spa_ffmpeg_dec_node_process_output(struct spa_node *node) +static int spa_ffmpeg_dec_node_process(struct spa_node *node) { struct impl *this; struct port *port; @@ -508,8 +503,7 @@ static const struct spa_node ffmpeg_dec_node = { spa_ffmpeg_dec_node_port_set_io, spa_ffmpeg_dec_node_port_reuse_buffer, spa_ffmpeg_dec_node_port_send_command, - spa_ffmpeg_dec_node_process_input, - spa_ffmpeg_dec_node_process_output, + spa_ffmpeg_dec_node_process, }; static int diff --git a/spa/plugins/ffmpeg/ffmpeg-enc.c b/spa/plugins/ffmpeg/ffmpeg-enc.c index 62fa31cd5..ed6f4ef60 100644 --- a/spa/plugins/ffmpeg/ffmpeg-enc.c +++ b/spa/plugins/ffmpeg/ffmpeg-enc.c @@ -442,12 +442,7 @@ spa_ffmpeg_enc_node_port_send_command(struct spa_node *node, return -ENOTSUP; } -static int spa_ffmpeg_enc_node_process_input(struct spa_node *node) -{ - return -EINVAL; -} - -static int spa_ffmpeg_enc_node_process_output(struct spa_node *node) +static int spa_ffmpeg_enc_node_process(struct spa_node *node) { struct impl *this; struct port *port; @@ -491,8 +486,7 @@ static const struct spa_node ffmpeg_enc_node = { spa_ffmpeg_enc_node_port_set_io, spa_ffmpeg_enc_node_port_reuse_buffer, spa_ffmpeg_enc_node_port_send_command, - spa_ffmpeg_enc_node_process_input, - spa_ffmpeg_enc_node_process_output, + spa_ffmpeg_enc_node_process, }; static int diff --git a/spa/plugins/support/loop.c b/spa/plugins/support/loop.c index 62d5b0cfe..c73cdfbca 100644 --- a/spa/plugins/support/loop.c +++ b/spa/plugins/support/loop.c @@ -269,15 +269,20 @@ static void wakeup_func(void *data, uint64_t count) struct impl *impl = data; uint32_t index; while (spa_ringbuffer_get_read_index(&impl->buffer, &index) > 0) { - struct invoke_item *item = - SPA_MEMBER(impl->buffer_data, index & (DATAS_SIZE - 1), struct invoke_item); + struct invoke_item *item; + bool block; + + item = SPA_MEMBER(impl->buffer_data, index & (DATAS_SIZE - 1), struct invoke_item); + block = item->block; + item->res = item->func(&impl->loop, true, item->seq, item->data, item->size, item->user_data); + spa_ringbuffer_read_update(&impl->buffer, index + item->item_size); - if (item->block) { - uint64_t count = 1; - if (write(impl->ack_fd, &count, sizeof(uint64_t)) != sizeof(uint64_t)) + if (block) { + uint64_t c = 1; + if (write(impl->ack_fd, &c, sizeof(uint64_t)) != sizeof(uint64_t)) spa_log_warn(impl->log, NAME " %p: failed to write event fd: %s", impl, strerror(errno)); } diff --git a/spa/plugins/test/fakesink.c b/spa/plugins/test/fakesink.c index b25b739cc..fecfd7003 100644 --- a/spa/plugins/test/fakesink.c +++ b/spa/plugins/test/fakesink.c @@ -211,7 +211,7 @@ static int impl_node_set_param(struct spa_node *node, uint32_t id, uint32_t flag static void set_timer(struct impl *this, bool enabled) { - if ((this->callbacks && this->callbacks->need_input) || this->props.live) { + if ((this->callbacks && this->callbacks->process) || this->props.live) { if (enabled) { if (this->props.live) { uint64_t next_time = this->start_time + this->elapsed_time; @@ -233,7 +233,7 @@ static inline void read_timer(struct impl *this) { uint64_t expirations; - if ((this->callbacks && this->callbacks->need_input) || this->props.live) { + if ((this->callbacks && this->callbacks->process) || this->props.live) { if (read(this->timer_source.fd, &expirations, sizeof(uint64_t)) != sizeof(uint64_t)) perror("read timerfd"); } @@ -253,8 +253,8 @@ static int consume_buffer(struct impl *this) if (spa_list_is_empty(&this->ready)) { io->status = SPA_STATUS_NEED_BUFFER; - if (this->callbacks->need_input) - this->callbacks->need_input(this->callbacks_data); + if (this->callbacks->process) + this->callbacks->process(this->callbacks_data, SPA_STATUS_NEED_BUFFER); } if (spa_list_is_empty(&this->ready)) { spa_log_error(this->log, NAME " %p: no buffers", this); @@ -356,7 +356,7 @@ impl_node_set_callbacks(struct spa_node *node, this = SPA_CONTAINER_OF(node, struct impl, node); - if (this->data_loop == NULL && callbacks != NULL && callbacks->need_input != NULL) { + if (this->data_loop == NULL && callbacks != NULL && callbacks->process != NULL) { spa_log_error(this->log, "a data_loop is needed for async operation"); return -EINVAL; } @@ -703,7 +703,7 @@ impl_node_port_send_command(struct spa_node *node, return -ENOTSUP; } -static int impl_node_process_input(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct impl *this; struct spa_io_buffers *input; @@ -732,17 +732,12 @@ static int impl_node_process_input(struct spa_node *node) input->buffer_id = SPA_ID_INVALID; input->status = SPA_STATUS_OK; } - if (this->callbacks == NULL || this->callbacks->need_input == NULL) + if (this->callbacks == NULL || this->callbacks->process == NULL) return consume_buffer(this); else return SPA_STATUS_OK; } -static int impl_node_process_output(struct spa_node *node) -{ - return -ENOTSUP; -} - static const struct spa_node impl_node = { SPA_VERSION_NODE, NULL, @@ -762,8 +757,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_clock_enum_params(struct spa_clock *clock, uint32_t id, uint32_t *index, diff --git a/spa/plugins/test/fakesrc.c b/spa/plugins/test/fakesrc.c index dd8748c1c..8540fa1d6 100644 --- a/spa/plugins/test/fakesrc.c +++ b/spa/plugins/test/fakesrc.c @@ -230,7 +230,7 @@ static int fill_buffer(struct impl *this, struct buffer *b) static void set_timer(struct impl *this, bool enabled) { - if ((this->callbacks && this->callbacks->have_output) || this->props.live) { + if ((this->callbacks && this->callbacks->process) || this->props.live) { if (enabled) { if (this->props.live) { uint64_t next_time = this->start_time + this->elapsed_time; @@ -252,7 +252,7 @@ static inline void read_timer(struct impl *this) { uint64_t expirations; - if ((this->callbacks && this->callbacks->have_output) || this->props.live) { + if ((this->callbacks && this->callbacks->process) || this->props.live) { if (read(this->timer_source.fd, &expirations, sizeof(uint64_t)) != sizeof(uint64_t)) perror("read timerfd"); } @@ -309,8 +309,8 @@ static void on_output(struct spa_source *source) res = make_buffer(this); - if (res == SPA_STATUS_HAVE_BUFFER && this->callbacks && this->callbacks->have_output) - this->callbacks->have_output(this->callbacks_data); + if (res == SPA_STATUS_HAVE_BUFFER && this->callbacks && this->callbacks->process) + this->callbacks->process(this->callbacks_data, res); } static int impl_node_send_command(struct spa_node *node, const struct spa_command *command) @@ -372,7 +372,7 @@ impl_node_set_callbacks(struct spa_node *node, this = SPA_CONTAINER_OF(node, struct impl, node); - if (this->data_loop == NULL && (callbacks != NULL && callbacks->have_output != NULL)) { + if (this->data_loop == NULL && (callbacks != NULL && callbacks->process != NULL)) { spa_log_error(this->log, "a data_loop is needed for async operation"); return -EINVAL; } @@ -747,12 +747,7 @@ impl_node_port_send_command(struct spa_node *node, return -ENOTSUP; } -static int impl_node_process_input(struct spa_node *node) -{ - return -ENOTSUP; -} - -static int impl_node_process_output(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct impl *this; struct spa_io_buffers *io; @@ -771,7 +766,7 @@ static int impl_node_process_output(struct spa_node *node) this->io->buffer_id = SPA_ID_INVALID; } - if ((this->callbacks == NULL || this->callbacks->have_output == NULL) && + if ((this->callbacks == NULL || this->callbacks->process == NULL) && (io->status == SPA_STATUS_NEED_BUFFER)) return make_buffer(this); else @@ -797,8 +792,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_clock_enum_params(struct spa_clock *clock, uint32_t id, uint32_t *index, diff --git a/spa/plugins/v4l2/v4l2-source.c b/spa/plugins/v4l2/v4l2-source.c index 112ea7ae3..720a43322 100644 --- a/spa/plugins/v4l2/v4l2-source.c +++ b/spa/plugins/v4l2/v4l2-source.c @@ -828,6 +828,8 @@ static int impl_node_process(struct spa_node *node) io = port->io; spa_return_val_if_fail(io != NULL, -EIO); + spa_log_trace(port->log, NAME " %p; status %d", node, io->status); + if (io->status == SPA_STATUS_HAVE_BUFFER) return SPA_STATUS_HAVE_BUFFER; @@ -862,6 +864,8 @@ static int impl_node_process(struct spa_node *node) b = spa_list_first(&port->queue, struct buffer, link); spa_list_remove(&b->link); + spa_log_trace(port->log, NAME " %p: dequeue buffer %d", node, b->outbuf->id); + io->buffer_id = b->outbuf->id; io->status = SPA_STATUS_HAVE_BUFFER; diff --git a/spa/plugins/v4l2/v4l2-utils.c b/spa/plugins/v4l2/v4l2-utils.c index 7a0803920..5cdad71f6 100644 --- a/spa/plugins/v4l2/v4l2-utils.c +++ b/spa/plugins/v4l2/v4l2-utils.c @@ -1181,7 +1181,7 @@ static int mmap_read(struct impl *this) spa_list_append(&port->queue, &b->link); spa_log_trace(port->log, "v4l2 %p: have output %d", this, buf.index); - this->callbacks->have_output(this->callbacks_data); + this->callbacks->process(this->callbacks_data, SPA_STATUS_HAVE_BUFFER); return 0; } diff --git a/spa/plugins/videotestsrc/videotestsrc.c b/spa/plugins/videotestsrc/videotestsrc.c index fb5478958..db9066076 100644 --- a/spa/plugins/videotestsrc/videotestsrc.c +++ b/spa/plugins/videotestsrc/videotestsrc.c @@ -358,7 +358,7 @@ static void on_output(struct spa_source *source) res = make_buffer(this); if (res == SPA_STATUS_HAVE_BUFFER) - this->callbacks->have_output(this->callbacks_data); + this->callbacks->process(this->callbacks_data, res); } static int impl_node_send_command(struct spa_node *node, const struct spa_command *command) @@ -849,12 +849,7 @@ impl_node_port_send_command(struct spa_node *node, return -ENOTSUP; } -static int impl_node_process_input(struct spa_node *node) -{ - return -ENOTSUP; -} - -static int impl_node_process_output(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct impl *this; struct spa_io_buffers *io; @@ -907,8 +902,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_clock_enum_params(struct spa_clock *clock, uint32_t id, uint32_t *index, diff --git a/spa/plugins/volume/volume.c b/spa/plugins/volume/volume.c index 7451c3ce3..f772e5d34 100644 --- a/spa/plugins/volume/volume.c +++ b/spa/plugins/volume/volume.c @@ -799,11 +799,11 @@ static void do_volume(struct impl *this, struct spa_buffer *dbuf, struct spa_buf dd[0].chunk->stride = 0; } -static int impl_node_process_input(struct spa_node *node) +static int impl_node_process(struct spa_node *node) { struct impl *this; - struct spa_io_buffers *input, *output; struct port *in_port, *out_port; + struct spa_io_buffers *input, *output; struct spa_buffer *dbuf, *sbuf; spa_return_val_if_fail(node != NULL, -EINVAL); @@ -814,50 +814,6 @@ static int impl_node_process_input(struct spa_node *node) output = out_port->io; spa_return_val_if_fail(output != NULL, -EIO); - if (output->status == SPA_STATUS_HAVE_BUFFER) - return SPA_STATUS_HAVE_BUFFER; - - in_port = GET_IN_PORT(this, 0); - input = in_port->io; - spa_return_val_if_fail(input != NULL, -EIO); - - if (input->buffer_id >= in_port->n_buffers) { - input->status = -EINVAL; - return -EINVAL; - } - - if ((dbuf = find_free_buffer(this, out_port)) == NULL) { - spa_log_error(this->log, NAME " %p: out of buffers", this); - return -EPIPE; - } - - sbuf = in_port->buffers[input->buffer_id].outbuf; - - input->status = SPA_STATUS_OK; - - spa_log_trace(this->log, NAME " %p: do volume %d -> %d", this, sbuf->id, dbuf->id); - do_volume(this, dbuf, sbuf); - - output->buffer_id = dbuf->id; - output->status = SPA_STATUS_HAVE_BUFFER; - - return SPA_STATUS_HAVE_BUFFER; -} - -static int impl_node_process_output(struct spa_node *node) -{ - struct impl *this; - struct port *in_port, *out_port; - struct spa_io_buffers *input, *output; - - spa_return_val_if_fail(node != NULL, -EINVAL); - - this = SPA_CONTAINER_OF(node, struct impl, node); - - out_port = GET_OUT_PORT(this, 0); - output = out_port->io; - spa_return_val_if_fail(output != NULL, -EIO); - if (output->status == SPA_STATUS_HAVE_BUFFER) return SPA_STATUS_HAVE_BUFFER; @@ -871,11 +827,32 @@ static int impl_node_process_output(struct spa_node *node) input = in_port->io; spa_return_val_if_fail(input != NULL, -EIO); + if (input->status != SPA_STATUS_HAVE_BUFFER) + return SPA_STATUS_NEED_BUFFER; + + if (input->buffer_id >= in_port->n_buffers) { + input->status = -EINVAL; + return -EINVAL; + } + + if ((dbuf = find_free_buffer(this, out_port)) == NULL) { + spa_log_error(this->log, NAME " %p: out of buffers", this); + return -EPIPE; + } + + sbuf = in_port->buffers[input->buffer_id].outbuf; + + spa_log_trace(this->log, NAME " %p: do volume %d -> %d", this, sbuf->id, dbuf->id); + do_volume(this, dbuf, sbuf); + + output->buffer_id = dbuf->id; + output->status = SPA_STATUS_HAVE_BUFFER; + if (in_port->range && out_port->range) *in_port->range = *out_port->range; input->status = SPA_STATUS_NEED_BUFFER; - return SPA_STATUS_NEED_BUFFER; + return SPA_STATUS_HAVE_BUFFER; } static const struct spa_node impl_node = { @@ -897,8 +874,7 @@ static const struct spa_node impl_node = { impl_node_port_set_io, impl_node_port_reuse_buffer, impl_node_port_send_command, - impl_node_process_input, - impl_node_process_output, + impl_node_process, }; static int impl_get_interface(struct spa_handle *handle, uint32_t interface_id, void **interface) diff --git a/spa/tests/test-bluez5.c b/spa/tests/test-bluez5.c index d60a2d4c0..abea5fe9d 100644 --- a/spa/tests/test-bluez5.c +++ b/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 -#include +#include #include @@ -117,6 +117,7 @@ struct data { struct spa_monitor *monitor; struct spa_graph graph; + struct spa_graph_state graph_state; struct spa_graph_data graph_data; struct spa_graph_node source_node; struct spa_graph_port source_out; @@ -305,7 +306,7 @@ int main(int argc, char *argv[]) data.monitor = iface; - spa_graph_init(&data.graph); + spa_graph_init(&data.graph, &data.graph_state); spa_graph_data_init(&data.graph_data, &data.graph); spa_graph_set_callbacks(&data.graph, &spa_graph_impl_default, &data.graph_data); diff --git a/spa/tests/test-control.c b/spa/tests/test-control.c index d01b295cd..6450d92a5 100644 --- a/spa/tests/test-control.c +++ b/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 -#include +#include #include @@ -114,6 +114,7 @@ struct data { uint32_t n_support; struct spa_graph graph; + struct spa_graph_state graph_state; struct spa_graph_data graph_data; struct spa_graph_node source_node; struct spa_graph_state source_state; @@ -255,13 +256,13 @@ static void update_props(struct data *data) data->volume_accum -= M_PI_M2; } -static void on_sink_need_input(void *_data) +static void on_sink_process(void *_data, int status) { struct data *data = _data; update_props(data); - spa_graph_need_input(&data->graph, &data->sink_node); + spa_graph_node_process(&data->sink_node); } static void @@ -276,7 +277,7 @@ static const struct spa_node_callbacks sink_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = on_sink_done, .event = on_sink_event, - .need_input = on_sink_need_input, + .process = on_sink_process, .reuse_buffer = on_sink_reuse_buffer }; @@ -564,7 +565,7 @@ int main(int argc, char *argv[]) int res; const char *str; - spa_graph_init(&data.graph); + spa_graph_init(&data.graph, &data.graph_state); spa_graph_data_init(&data.graph_data, &data.graph); spa_graph_set_callbacks(&data.graph, &spa_graph_impl_default, &data.graph_data); diff --git a/spa/tests/test-graph.c b/spa/tests/test-graph.c index fd2a6ba41..f65e3cb71 100644 --- a/spa/tests/test-graph.c +++ b/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 -#include +#include #include @@ -107,6 +107,7 @@ struct data { uint32_t n_support; struct spa_graph graph; + struct spa_graph_state graph_state; struct spa_graph_data graph_data; struct spa_graph_node source_node; struct spa_graph_state source_state; @@ -239,17 +240,16 @@ static void on_sink_event(void *data, struct spa_event *event) printf("got event %d\n", SPA_EVENT_TYPE(event)); } -static void on_sink_need_input(void *_data) +static void on_sink_process(void *_data, int status) { struct data *data = _data; - spa_graph_need_input(&data->graph, &data->sink_node); + spa_graph_node_process(&data->sink_node); } static void on_sink_reuse_buffer(void *_data, uint32_t port_id, uint32_t buffer_id) { struct data *data = _data; - data->volume_sink_io[0].buffer_id = buffer_id; } @@ -257,7 +257,7 @@ static const struct spa_node_callbacks sink_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = on_sink_done, .event = on_sink_event, - .need_input = on_sink_need_input, + .process = on_sink_process, .reuse_buffer = on_sink_reuse_buffer }; @@ -557,7 +557,7 @@ int main(int argc, char *argv[]) int res; const char *str; - spa_graph_init(&data.graph); + spa_graph_init(&data.graph, &data.graph_state); spa_graph_data_init(&data.graph_data, &data.graph); spa_graph_set_callbacks(&data.graph, &spa_graph_impl_default, &data.graph_data); diff --git a/spa/tests/test-graph2.c b/spa/tests/test-graph2.c index e71f36eca..288049ab5 100644 --- a/spa/tests/test-graph2.c +++ b/spa/tests/test-graph2.c @@ -35,7 +35,7 @@ #include #include #include -#include +#include static SPA_TYPE_MAP_IMPL(default_map, 4096); static SPA_LOG_IMPL(default_log); @@ -63,6 +63,7 @@ struct data { int writers; struct version version; struct spa_graph graph[2]; + struct spa_graph_state graph_state[2]; struct spa_graph_node source_node[2]; struct spa_graph_port source_out[2]; @@ -229,8 +230,8 @@ int main(int argc, char *argv[]) struct data data = { NULL }; const char *str; - spa_graph_init(&data.graph[0]); - spa_graph_init(&data.graph[1]); + spa_graph_init(&data.graph[0], &data.graph_state[0]); + spa_graph_init(&data.graph[1], &data.graph_state[1]); data.map = &default_map.map; data.log = &default_log.log; diff --git a/spa/tests/test-mixer.c b/spa/tests/test-mixer.c index 34bd7b784..01784eb2a 100644 --- a/spa/tests/test-mixer.c +++ b/spa/tests/test-mixer.c @@ -50,7 +50,7 @@ static SPA_LOG_IMPL(default_log); #define spa_debug(...) spa_log_trace(&default_log.log,__VA_ARGS__) #include -#include +#include struct type { uint32_t node; @@ -115,6 +115,7 @@ struct data { uint32_t n_support; struct spa_graph graph; + struct spa_graph_state graph_state; struct spa_graph_data graph_data; struct spa_graph_node source1_node; struct spa_graph_state source1_state; @@ -269,12 +270,12 @@ static void update_props(struct data *data) data->ctrl_volume[1].value = 1.0 - data->ctrl_volume[0].value; } -static void on_sink_need_input(void *_data) +static void on_sink_process(void *_data, int status) { struct data *data = _data; #ifdef USE_GRAPH - spa_graph_need_input(&data->graph, &data->sink_node); + spa_graph_node_process(&data->sink_node); #else int res; @@ -322,7 +323,7 @@ static const struct spa_node_callbacks sink_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = on_sink_done, .event = on_sink_event, - .need_input = &on_sink_need_input, + .process = &on_sink_process, .reuse_buffer = on_sink_reuse_buffer }; @@ -717,7 +718,7 @@ int main(int argc, char *argv[]) data.data_loop.remove_source = do_remove_source; data.data_loop.invoke = do_invoke; - spa_graph_init(&data.graph); + spa_graph_init(&data.graph, &data.graph_state); spa_graph_data_init(&data.graph_data, &data.graph); spa_graph_set_callbacks(&data.graph, &spa_graph_impl_default, &data.graph_data); diff --git a/spa/tests/test-perf.c b/spa/tests/test-perf.c index fa0174aa4..adc59f17a 100644 --- a/spa/tests/test-perf.c +++ b/spa/tests/test-perf.c @@ -36,7 +36,7 @@ #include #include #include -#include +#include #define MODE_SYNC_PUSH (1<<0) #define MODE_SYNC_PULL (1<<1) @@ -109,6 +109,7 @@ struct data { int iterations; struct spa_graph graph; + struct spa_graph_state graph_state; struct spa_graph_data graph_data; struct spa_graph_node source_node; struct spa_graph_state source_state; @@ -232,7 +233,7 @@ static void on_sink_pull(struct data *data) spa_node_process(data->source); spa_node_process(data->sink); } else { - spa_graph_need_input(&data->graph, &data->sink_node); + spa_graph_node_trigger(&data->sink_node); } } @@ -243,7 +244,7 @@ static void on_source_push(struct data *data) spa_node_process(data->source); spa_node_process(data->sink); } else { - spa_graph_have_output(&data->graph, &data->source_node); + spa_graph_node_trigger(&data->source_node); } } @@ -259,7 +260,7 @@ static void on_sink_event(void *_data, struct spa_event *event) spa_log_trace(data->log, "got sink event %d", SPA_EVENT_TYPE(event)); } -static void on_sink_need_input(void *_data) +static void on_sink_process(void *_data, int status) { struct data *data = _data; spa_log_trace(data->log, "need input"); @@ -280,7 +281,7 @@ static const struct spa_node_callbacks sink_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = on_sink_done, .event = on_sink_event, - .need_input = on_sink_need_input, + .process = on_sink_process, .reuse_buffer = on_sink_reuse_buffer }; @@ -296,7 +297,7 @@ static void on_source_event(void *_data, struct spa_event *event) spa_log_trace(data->log, "got source event %d", SPA_EVENT_TYPE(event)); } -static void on_source_have_output(void *_data) +static void on_source_process(void *_data, int status) { struct data *data = _data; spa_log_trace(data->log, "have_output"); @@ -309,7 +310,7 @@ static const struct spa_node_callbacks source_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = on_source_done, .event = on_source_event, - .have_output = on_source_have_output + .process = on_source_process }; @@ -378,7 +379,6 @@ static int make_nodes(struct data *data) 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; 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); @@ -386,7 +386,6 @@ static int make_nodes(struct data *data) 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; 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); @@ -545,7 +544,7 @@ int main(int argc, char *argv[]) int res; const char *str; - spa_graph_init(&data.graph); + spa_graph_init(&data.graph, &data.graph_state); spa_graph_data_init(&data.graph_data, &data.graph); spa_graph_set_callbacks(&data.graph, &spa_graph_impl_default, &data.graph_data); diff --git a/spa/tests/test-ringbuffer.c b/spa/tests/test-ringbuffer.c index 30d911588..5816aa805 100644 --- a/spa/tests/test-ringbuffer.c +++ b/spa/tests/test-ringbuffer.c @@ -214,7 +214,7 @@ static void on_sink_event(void *data, struct spa_event *event) printf("got event %d\n", SPA_EVENT_TYPE(event)); } -static void on_sink_need_input(void *_data) +static void on_sink_process(void *_data, int status) { struct data *data = _data; int res; @@ -222,7 +222,6 @@ static void on_sink_need_input(void *_data) res = spa_node_process(data->source); if (res != SPA_STATUS_HAVE_BUFFER) printf("got process error from source %d\n", res); - if ((res = spa_node_process(data->sink)) < 0) printf("got process error from sink %d\n", res); } @@ -240,7 +239,7 @@ static const struct spa_node_callbacks sink_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = on_sink_done, .event = on_sink_event, - .need_input = on_sink_need_input, + .process = on_sink_process, .reuse_buffer = on_sink_reuse_buffer }; diff --git a/spa/tests/test-v4l2.c b/spa/tests/test-v4l2.c index 5cba59be7..0e67f1bf1 100644 --- a/spa/tests/test-v4l2.c +++ b/spa/tests/test-v4l2.c @@ -198,7 +198,7 @@ static void on_source_event(void *_data, struct spa_event *event) printf("got event %d\n", SPA_EVENT_TYPE(event)); } -static void on_source_have_output(void *_data) +static void on_source_process(void *_data, int status) { struct data *data = _data; int res; @@ -284,7 +284,7 @@ static const struct spa_node_callbacks source_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = on_source_done, .event = on_source_event, - .have_output = on_source_have_output + .process = on_source_process }; static int do_add_source(struct spa_loop *loop, struct spa_source *source) diff --git a/src/extensions/client-node.h b/src/extensions/client-node.h index 56a55ebdf..cf505a204 100644 --- a/src/extensions/client-node.h +++ b/src/extensions/client-node.h @@ -97,10 +97,7 @@ struct pw_client_node_transport { #define pw_client_node_transport_parse_message(t,m) ((t)->parse_message((t), (m))) enum pw_client_node_message_type { - PW_CLIENT_NODE_MESSAGE_HAVE_OUTPUT, /*< signal that the node has output */ - PW_CLIENT_NODE_MESSAGE_NEED_INPUT, /*< signal that the node needs input */ - PW_CLIENT_NODE_MESSAGE_PROCESS_INPUT, /*< instruct the node to process input */ - PW_CLIENT_NODE_MESSAGE_PROCESS_OUTPUT, /*< instruct the node output is processed */ + PW_CLIENT_NODE_MESSAGE_PROCESS, /*< instruct the node to process */ PW_CLIENT_NODE_MESSAGE_PORT_REUSE_BUFFER, /*< reuse a buffer */ }; diff --git a/src/modules/module-client-node/client-node.c b/src/modules/module-client-node/client-node.c index b9fdfdfdf..255cbbca6 100644 --- a/src/modules/module-client-node/client-node.c +++ b/src/modules/module-client-node/client-node.c @@ -148,9 +148,6 @@ struct impl { int fds[2]; int other_fds[2]; - - uint32_t input_ready; - bool out_pending; }; /** \endcond */ @@ -845,48 +842,14 @@ impl_node_port_send_command(struct spa_node *node, static int impl_node_process(struct spa_node *node) { struct node *this = SPA_CONTAINER_OF(node, struct node, node); - struct impl *impl = this->impl; - int res; + uint64_t cmd = 1; - pw_log_trace("client-node %p: send process input", this); - pw_client_node_transport_add_message(impl->transport, - &PW_CLIENT_NODE_MESSAGE_INIT(PW_CLIENT_NODE_MESSAGE_PROCESS_INPUT)); - do_flush(this); - res = SPA_STATUS_OK; - return res; -} + pw_log_trace("client-node %p: send process", this); -static int handle_node_message(struct node *this, struct pw_client_node_message *message) -{ - struct impl *impl = SPA_CONTAINER_OF(this, struct impl, node); + if (write(this->writefd, &cmd, 8) != 8) + spa_log_warn(this->log, "node %p: error flushing : %s", this, strerror(errno)); - switch (PW_CLIENT_NODE_MESSAGE_TYPE(message)) { - case PW_CLIENT_NODE_MESSAGE_HAVE_OUTPUT: - impl->out_pending = false; - pw_log_trace("have output"); - this->callbacks->have_output(this->callbacks_data); - break; - - case PW_CLIENT_NODE_MESSAGE_NEED_INPUT: - pw_log_trace("need input"); - impl->input_ready++; - this->callbacks->need_input(this->callbacks_data); - break; - - case PW_CLIENT_NODE_MESSAGE_PORT_REUSE_BUFFER: - if (impl->client_reuse) { - struct pw_client_node_message_port_reuse_buffer *p = - (struct pw_client_node_message_port_reuse_buffer *) message; - this->callbacks->reuse_buffer(this->callbacks_data, p->body.port_id.value, - p->body.buffer_id.value); - } - break; - - default: - pw_log_warn("unhandled message %d", PW_CLIENT_NODE_MESSAGE_TYPE(message)); - return -ENOTSUP; - } - return 0; + return SPA_STATUS_OK; } static void @@ -995,7 +958,6 @@ static struct pw_client_node_proxy_methods client_node_methods = { static void node_on_data_fd_events(struct spa_source *source) { struct node *this = source->data; - struct impl *impl = this->impl; if (source->rmask & (SPA_IO_ERR | SPA_IO_HUP)) { spa_log_warn(this->log, "node %p: got error", this); @@ -1003,18 +965,13 @@ static void node_on_data_fd_events(struct spa_source *source) } if (source->rmask & SPA_IO_IN) { - struct pw_client_node_message message; uint64_t cmd; if (read(this->data_source.fd, &cmd, sizeof(uint64_t)) != sizeof(uint64_t)) spa_log_warn(this->log, "node %p: error reading message: %s", this, strerror(errno)); - while (pw_client_node_transport_next_message(impl->transport, &message) == 1) { - struct pw_client_node_message *msg = alloca(SPA_POD_SIZE(&message)); - pw_client_node_transport_parse_message(impl->transport, msg); - handle_node_message(this, msg); - } + this->callbacks->process(this->callbacks_data, SPA_STATUS_HAVE_BUFFER); } } diff --git a/src/pipewire/link.c b/src/pipewire/link.c index d2508e298..7692e803e 100644 --- a/src/pipewire/link.c +++ b/src/pipewire/link.c @@ -790,7 +790,9 @@ do_activate_link(struct spa_loop *loop, spa_graph_port_add(&this->input->rt.mix_node, &this->rt.mix[SPA_DIRECTION_INPUT].port); spa_graph_port_link(&this->rt.mix[SPA_DIRECTION_OUTPUT].port, &this->rt.mix[SPA_DIRECTION_INPUT].port); - + spa_graph_link_add(&this->output->node->rt.root, + this->input->node->rt.root.state, + &this->rt.link); return 0; } @@ -1031,6 +1033,7 @@ do_deactivate_link(struct spa_loop *loop, spa_graph_port_unlink(&this->rt.mix[SPA_DIRECTION_OUTPUT].port); spa_graph_port_remove(&this->rt.mix[SPA_DIRECTION_OUTPUT].port); spa_graph_port_remove(&this->rt.mix[SPA_DIRECTION_INPUT].port); + spa_graph_link_remove(&this->rt.link); return 0; } @@ -1160,21 +1163,23 @@ do_join_graphs(struct spa_loop *loop, { struct pw_link *this = user_data; struct spa_graph *in_graph, *out_graph; - struct spa_graph *in_root, *out_root; - in_graph = this->input->node->rt.node.graph; - out_graph = this->output->node->rt.node.graph; + in_graph = this->input->node->rt.root.graph; + out_graph = this->output->node->rt.root.graph; - in_root = spa_graph_find_root(in_graph); - out_root = spa_graph_find_root(out_graph); + if (in_graph != out_graph) { + if (SPA_FLAG_CHECK(in_graph->flags, SPA_GRAPH_FLAG_DRIVER)) { + spa_graph_node_remove(&this->output->node->rt.root); + spa_graph_node_add(in_graph, &this->output->node->rt.root); + } + else { + spa_graph_node_remove(&this->input->node->rt.root); + spa_graph_node_add(out_graph, &this->input->node->rt.root); + } + } + this->rt.link.signal = spa_graph_link_signal_node; + this->rt.link.signal_data = &this->input->node->rt.root; - if (out_root == in_root) - return 0; - - if (SPA_FLAG_CHECK(in_root->flags, SPA_GRAPH_FLAG_DRIVER)) - spa_graph_add_subgraph(in_root, out_root); - else - spa_graph_add_subgraph(out_root, in_root); return 0; } @@ -1189,8 +1194,6 @@ 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; - struct spa_graph *in_root, *out_root; if (output == input) goto same_ports; @@ -1201,21 +1204,6 @@ struct pw_link *pw_link_new(struct pw_core *core, input_node = input->node; output_node = output->node; - in_graph = input_node->rt.node.graph; - out_graph = output_node->rt.node.graph; - - pw_log_debug("link new %p %p", in_graph, out_graph); - - in_root = spa_graph_find_root(in_graph); - out_root = spa_graph_find_root(out_graph); - - pw_log_debug("link new %p %p", in_root, out_root); - - if (SPA_FLAG_CHECK(in_root->flags, SPA_GRAPH_FLAG_DRIVER) && - SPA_FLAG_CHECK(out_root->flags, SPA_GRAPH_FLAG_DRIVER) && - in_root != out_root) - goto link_not_supported; - impl = calloc(1, sizeof(struct impl) + user_data_size); if (impl == NULL) goto no_mem; @@ -1291,9 +1279,6 @@ 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; diff --git a/src/pipewire/node.c b/src/pipewire/node.c index f71b009ce..5d043c7c4 100644 --- a/src/pipewire/node.c +++ b/src/pipewire/node.c @@ -46,10 +46,16 @@ struct impl { struct pw_work_queue *work; bool pause_on_idle; + struct spa_graph driver_graph; + struct spa_graph_state driver_state; + struct spa_graph_data driver_data; + struct spa_graph graph; + struct spa_graph_state graph_state; struct spa_graph_data graph_data; - struct pw_node_activation activation; + struct pw_node_activation root_activation; + struct pw_node_activation node_activation; }; struct resource_data { @@ -394,19 +400,9 @@ static void check_properties(struct pw_node *node) node->driver = false; if (node->driver) - SPA_FLAG_SET(impl->graph.flags, SPA_GRAPH_FLAG_DRIVER); + SPA_FLAG_SET(impl->driver_graph.flags, SPA_GRAPH_FLAG_DRIVER); else - SPA_FLAG_UNSET(impl->graph.flags, SPA_GRAPH_FLAG_DRIVER); -} - -static int -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 impl *impl = SPA_CONTAINER_OF(this, struct impl, this); - spa_graph_node_add(&impl->graph, &this->rt.node); - return 0; + SPA_FLAG_UNSET(impl->driver_graph.flags, SPA_GRAPH_FLAG_DRIVER); } struct pw_node *pw_node_new(struct pw_core *core, @@ -455,19 +451,27 @@ 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_init(&impl->graph); + spa_graph_init(&impl->driver_graph, &impl->driver_state); + spa_graph_data_init(&impl->driver_data, &impl->driver_graph); + spa_graph_set_callbacks(&impl->driver_graph, + &spa_graph_impl_default, &impl->driver_data); + + this->rt.activation = &impl->root_activation; + spa_graph_node_init(&this->rt.root, &this->rt.activation->state); + spa_graph_node_add(&impl->driver_graph, &this->rt.root); + + spa_graph_init(&impl->graph, &impl->graph_state); spa_graph_data_init(&impl->graph_data, &impl->graph); spa_graph_set_callbacks(&impl->graph, &spa_graph_impl_default, &impl->graph_data); - this->rt.activation = &impl->activation; - spa_graph_node_init(&this->rt.node, &this->rt.activation->state); + spa_graph_node_set_subgraph(&this->rt.root, &impl->graph); + spa_graph_node_set_callbacks(&this->rt.root, + &spa_graph_node_sub_impl_default, this); - pw_loop_invoke(this->data_loop, do_node_join, 1, NULL, 0, true, this); - - spa_list_init(&this->rt.links[SPA_DIRECTION_INPUT]); - spa_list_init(&this->rt.links[SPA_DIRECTION_OUTPUT]); - impl->activation.state.status = SPA_STATUS_NEED_BUFFER; + impl->node_activation.state.status = SPA_STATUS_NEED_BUFFER; + spa_graph_node_init(&this->rt.node, &impl->node_activation.state); + spa_graph_node_add(&impl->graph, &this->rt.node); return this; @@ -546,38 +550,19 @@ 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_need_input(void *data) +static void node_process(void *data, int status) { 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); + pw_log_trace("node %p: process %d", node, node->driver); - spa_hook_list_call(&node->listener_list, struct pw_node_events, need_input); + spa_hook_list_call(&node->listener_list, struct pw_node_events, process); if (node->driver) - spa_graph_run(node->rt.node.graph); - else if (node->rt.node.graph) - spa_graph_need_input(node->rt.node.graph, &node->rt.node); + spa_graph_run(&impl->driver_graph); else - pw_log_error("node %p: not added in graph", node); -} - -static void node_have_output(void *data) -{ - struct pw_node *node = data; - - pw_log_trace("node %p: have output %d", node, node->driver); - - spa_hook_list_call(&node->listener_list, struct pw_node_events, have_output); - - if (node->driver) - spa_graph_run(node->rt.node.graph); - - 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); - + spa_graph_node_trigger(&node->rt.node); } static void node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id) @@ -599,8 +584,7 @@ static const struct spa_node_callbacks node_callbacks = { SPA_VERSION_NODE_CALLBACKS, .done = node_done, .event = node_event, - .need_input = node_need_input, - .have_output = node_have_output, + .process = node_process, .reuse_buffer = node_reuse_buffer, }; @@ -633,12 +617,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; - struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this); - - if (impl->graph.parent) - spa_graph_remove_subgraph(&impl->graph); - spa_graph_node_remove(&this->rt.node); - + spa_graph_node_remove(&this->rt.root); return 0; } @@ -661,8 +640,9 @@ void pw_node_destroy(struct pw_node *node) pause_node(node); + pw_loop_invoke(node->data_loop, do_node_remove, 1, NULL, 0, true, node); + if (node->registered) { - pw_loop_invoke(node->data_loop, do_node_remove, 1, NULL, 0, true, node); spa_list_remove(&node->link); } diff --git a/src/pipewire/node.h b/src/pipewire/node.h index dfa585b57..9e51c88e7 100644 --- a/src/pipewire/node.h +++ b/src/pipewire/node.h @@ -83,10 +83,8 @@ struct pw_node_events { /** an event is emited */ void (*event) (void *data, const struct spa_event *event); - /** the node wants input */ - void (*need_input) (void *data); - /** the node has output */ - void (*have_output) (void *data); + /** the node wants to process the graph */ + void (*process) (void *data); /** the node has a buffer to reuse */ void (*reuse_buffer) (void *data, uint32_t port_id, uint32_t buffer_id); }; diff --git a/src/pipewire/port.c b/src/pipewire/port.c index 451922f6e..e45052ba8 100644 --- a/src/pipewire/port.c +++ b/src/pipewire/port.c @@ -138,8 +138,6 @@ int pw_port_init_mix(struct pw_port *port, struct pw_port_mix *mix) 0, NULL); - mix->port.scheduler_data = port; - if (pi && pi->init_mix) res = pi->init_mix(port->implementation_data, mix); @@ -226,8 +224,6 @@ struct pw_port *pw_port_new(enum pw_direction direction, &this->rt.io); this->rt.io.status = SPA_STATUS_NEED_BUFFER; - this->rt.mix_port.scheduler_data = this; - this->rt.port.scheduler_data = this; return this; @@ -295,6 +291,7 @@ static int do_add_port(struct spa_loop *loop, bool async, uint32_t seq, const void *data, size_t size, void *user_data) { struct pw_port *this = user_data; + struct spa_graph_node *out, *in; this->rt.port.flags = this->spa_info->flags; spa_graph_port_add(&this->node->rt.node, &this->rt.port); @@ -302,6 +299,17 @@ static int do_add_port(struct spa_loop *loop, spa_graph_port_link(&this->rt.port, &this->rt.mix_port); spa_graph_node_add(this->node->rt.node.graph, &this->rt.mix_node); + if (this->direction == PW_DIRECTION_INPUT) { + out = &this->rt.mix_node; + in = &this->node->rt.node; + } else { + out = &this->node->rt.node; + in = &this->rt.mix_node; + } + spa_graph_link_add(out, in->state, &this->rt.mix_link); + this->rt.mix_link.signal = spa_graph_link_signal_node; + this->rt.mix_link.signal_data = in; + return 0; } @@ -474,7 +482,6 @@ 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.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) @@ -503,14 +510,10 @@ static int do_remove_port(struct spa_loop *loop, bool async, uint32_t seq, const void *data, size_t size, void *user_data) { struct pw_port *this = user_data; - struct spa_graph_port *p; + spa_graph_link_remove(&this->rt.mix_link); spa_graph_port_unlink(&this->rt.port); spa_graph_port_remove(&this->rt.port); - - spa_list_for_each(p, &this->rt.mix_node.ports[this->direction], link) - spa_graph_port_remove(p); - spa_graph_port_remove(&this->rt.mix_port); spa_graph_node_remove(&this->rt.mix_node); @@ -552,6 +555,7 @@ void pw_port_destroy(struct pw_port *port) pw_port_remove(port); + pw_log_debug("port %p: control destroy", port); spa_list_for_each_safe(control, ctemp, &port->control_list[0], port_link) pw_control_destroy(control); spa_list_for_each_safe(control, ctemp, &port->control_list[1], port_link) diff --git a/src/pipewire/private.h b/src/pipewire/private.h index b5e160248..e84ad4aff 100644 --- a/src/pipewire/private.h +++ b/src/pipewire/private.h @@ -272,10 +272,11 @@ struct pw_node { struct pw_loop *data_loop; /**< the data loop for this node */ struct { - struct spa_graph_node node; - struct spa_list links[2]; + struct spa_graph_node root; struct pw_node_activation *activation; - struct spa_list sched_link; + struct spa_graph_node node; + struct spa_graph_node subnode; + struct spa_graph_link sublink; } rt; void *user_data; /**< extra user data */ @@ -336,6 +337,7 @@ struct pw_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_link mix_link; /**< mixer link */ struct spa_graph_state mix_state; /**< mixer state */ } rt; /**< data only accessed from the data thread */ @@ -369,8 +371,7 @@ struct pw_link { struct { struct pw_port_mix mix[2]; - struct spa_list in_node_link; - struct spa_list out_node_link; + struct spa_graph_link link; /**< nodes link */ } rt; void *user_data; diff --git a/src/pipewire/remote.c b/src/pipewire/remote.c index b52751f28..d2b469f5c 100644 --- a/src/pipewire/remote.c +++ b/src/pipewire/remote.c @@ -479,128 +479,20 @@ static void unhandle_socket(struct node_data *data) do_remove_source, 1, NULL, 0, true, data); } -static void do_push(struct node_data *data, enum spa_direction direction) -{ - struct spa_graph_node *node = &data->node->rt.node; - struct spa_graph_port *p; - - spa_list_for_each(p, &node->ports[direction], link) { - if (p->peer) - spa_graph_node_process(p->peer->node); - } -} - -static void do_pull(struct node_data *data, enum spa_direction direction) -{ - struct spa_graph_node *node = &data->node->rt.node; - struct spa_graph_port *p; - - spa_list_for_each(p, &node->ports[direction], link) { - if (p->peer) - spa_graph_node_process(p->peer->node); - } -} - -static void node_need_input(void *data) +static void node_process(void *data) { struct node_data *d = data; uint64_t cmd = 1; - do_pull(data, SPA_DIRECTION_INPUT); - pw_log_trace("remote %p: send need input", data); - pw_client_node_transport_add_message(d->trans, - &PW_CLIENT_NODE_MESSAGE_INIT(PW_CLIENT_NODE_MESSAGE_NEED_INPUT)); + pw_log_trace("remote %p: process", data); write(d->rtwritefd, &cmd, 8); } -static void node_have_output(void *data) -{ - struct node_data *d = data; - uint64_t cmd = 1; - - do_push(data, SPA_DIRECTION_OUTPUT); - pw_log_trace("remote %p: send have output", data); - pw_client_node_transport_add_message(d->trans, - &PW_CLIENT_NODE_MESSAGE_INIT(PW_CLIENT_NODE_MESSAGE_HAVE_OUTPUT)); - write(d->rtwritefd, &cmd, 8); -} - -static int process_input(struct node_data *data) -{ - struct spa_graph_node *node = &data->node->rt.node; - int res; - - pw_log_trace("remote %p: process input", data->remote); - do_push(data, SPA_DIRECTION_INPUT); - - res = spa_graph_node_process(node); - - switch (res) { - case SPA_STATUS_HAVE_BUFFER: - node_have_output(data); - break; - case SPA_STATUS_NEED_BUFFER: -// node_need_input(data); - break; - } - return res; -} - -static int process_output(struct node_data *data) -{ - struct spa_graph_node *node = &data->node->rt.node; - int res; - - pw_log_trace("remote %p: process output", data->remote); - do_pull(data, SPA_DIRECTION_OUTPUT); - - res = spa_graph_node_process(node); - - switch (res) { - case SPA_STATUS_HAVE_BUFFER: - node_have_output(data); - break; - case SPA_STATUS_NEED_BUFFER: -// node_need_input(data); - break; - } - return res; -} - -static void handle_rtnode_message(struct pw_proxy *proxy, struct pw_client_node_message *message) -{ - struct node_data *data = proxy->user_data; - - switch (PW_CLIENT_NODE_MESSAGE_TYPE(message)) { - case PW_CLIENT_NODE_MESSAGE_PROCESS_INPUT: - process_input(data); - break; - - case PW_CLIENT_NODE_MESSAGE_PROCESS_OUTPUT: - process_output(data); - break; - - case PW_CLIENT_NODE_MESSAGE_PORT_REUSE_BUFFER: - { - struct pw_client_node_message_port_reuse_buffer *rb = - (struct pw_client_node_message_port_reuse_buffer *) message; - uint32_t port_id = rb->body.port_id.value; - uint32_t buffer_id = rb->body.buffer_id.value; - struct spa_graph_node *node = &data->node->rt.node; - - spa_graph_node_reuse_buffer(node, port_id, buffer_id); - break; - } - default: - pw_log_warn("unexpected node message %d", PW_CLIENT_NODE_MESSAGE_TYPE(message)); - break; - } -} - static void on_rtsocket_condition(void *user_data, int fd, enum spa_io mask) { struct pw_proxy *proxy = user_data; struct node_data *data = proxy->user_data; + struct spa_graph_node *node = &data->node->rt.node; if (mask & (SPA_IO_ERR | SPA_IO_HUP)) { pw_log_warn("got error"); @@ -609,21 +501,14 @@ on_rtsocket_condition(void *user_data, int fd, enum spa_io mask) } if (mask & SPA_IO_IN) { - struct pw_client_node_message message; uint64_t cmd; if (read(fd, &cmd, sizeof(uint64_t)) != sizeof(uint64_t)) pw_log_warn("proxy %p: read failed %m", proxy); - if (cmd > 1) - pw_log_warn("proxy %p: %ld messages", proxy, cmd); - - - while (pw_client_node_transport_next_message(data->trans, &message) == 1) { - struct pw_client_node_message *msg = alloca(SPA_POD_SIZE(&message)); - pw_client_node_transport_parse_message(data->trans, msg); - handle_rtnode_message(proxy, msg); - } + pw_log_trace("remote %p: process", data->remote); + spa_graph_run(node->graph); + node_process(data); } } @@ -899,15 +784,6 @@ 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) @@ -1315,8 +1191,7 @@ static const struct pw_node_events node_events = { PW_VERSION_NODE_EVENTS, .destroy = node_destroy, .active_changed = node_active_changed, - .need_input = node_need_input, - .have_output = node_have_output, + .process = node_process, }; static int diff --git a/src/pipewire/stream.c b/src/pipewire/stream.c index ac04d5689..769d2b9e4 100644 --- a/src/pipewire/stream.c +++ b/src/pipewire/stream.c @@ -495,23 +495,10 @@ static void add_port_update(struct pw_stream *stream, uint32_t change_mask) &impl->port_info); } -static inline void send_need_input(struct pw_stream *stream) -{ - struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this); - uint64_t cmd = 1; - - pw_client_node_transport_add_message(impl->trans, - &PW_CLIENT_NODE_MESSAGE_INIT(PW_CLIENT_NODE_MESSAGE_NEED_INPUT)); - write(impl->rtwritefd, &cmd, 8); -} - static inline void send_have_output(struct pw_stream *stream) { struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this); uint64_t cmd = 1; - - pw_client_node_transport_add_message(impl->trans, - &PW_CLIENT_NODE_MESSAGE_INIT(PW_CLIENT_NODE_MESSAGE_HAVE_OUTPUT)); write(impl->rtwritefd, &cmd, 8); } @@ -519,9 +506,6 @@ static inline void send_reuse_buffer(struct pw_stream *stream, uint32_t id) { struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this); uint64_t cmd = 1; - - pw_client_node_transport_add_message(impl->trans, (struct pw_client_node_message*) - &PW_CLIENT_NODE_MESSAGE_PORT_REUSE_BUFFER_INIT(impl->port_id, id)); write(impl->rtwritefd, &cmd, 8); } @@ -601,66 +585,43 @@ static inline void reuse_buffer(struct pw_stream *stream, uint32_t id) } } -static void handle_rtnode_message(struct pw_stream *stream, struct pw_client_node_message *message) +static void do_process(struct pw_stream *stream) { struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this); + struct spa_io_buffers *io = impl->io; + struct buffer *b; + uint32_t buffer_id; - switch (PW_CLIENT_NODE_MESSAGE_TYPE(message)) { - case PW_CLIENT_NODE_MESSAGE_PROCESS_INPUT: - { - struct spa_io_buffers *io = impl->io; - struct buffer *b; - uint32_t buffer_id; + if (impl->direction == SPA_DIRECTION_INPUT) { + buffer_id = io->buffer_id; - if (impl->direction == SPA_DIRECTION_INPUT) { - buffer_id = io->buffer_id; + pw_log_trace("stream %p: process input %d %d", stream, io->status, + buffer_id); - pw_log_trace("stream %p: process input %d %d", stream, io->status, - buffer_id); + if ((b = find_buffer(stream, buffer_id)) == NULL) + return; - if ((b = find_buffer(stream, buffer_id)) == NULL) - return; - - if (impl->client_reuse) - io->buffer_id = SPA_ID_INVALID; - - if (io->status == SPA_STATUS_HAVE_BUFFER) { - SPA_FLAG_SET(b->flags, BUFFER_FLAG_OUT); - - impl->in_new_buffer = true; - spa_hook_list_call(&stream->listener_list, struct pw_stream_events, - new_buffer, buffer_id); - impl->in_new_buffer = false; - } - io->status = SPA_STATUS_NEED_BUFFER; - } else { - reuse_buffer(stream, io->buffer_id); + if (impl->client_reuse) io->buffer_id = SPA_ID_INVALID; - pw_log_trace("stream %p: process output", stream); - impl->in_need_buffer = true; + if (io->status == SPA_STATUS_HAVE_BUFFER) { + SPA_FLAG_SET(b->flags, BUFFER_FLAG_OUT); + + impl->in_new_buffer = true; spa_hook_list_call(&stream->listener_list, struct pw_stream_events, - need_buffer); - impl->in_need_buffer = false; + new_buffer, buffer_id); + impl->in_new_buffer = false; } - break; - } - case PW_CLIENT_NODE_MESSAGE_PORT_REUSE_BUFFER: - { - struct pw_client_node_message_port_reuse_buffer *p = - (struct pw_client_node_message_port_reuse_buffer *) message; + io->status = SPA_STATUS_NEED_BUFFER; + } else { + reuse_buffer(stream, io->buffer_id); + io->buffer_id = SPA_ID_INVALID; - if (p->body.port_id.value != impl->port_id) - return; - if (impl->direction != SPA_DIRECTION_OUTPUT) - return; - - reuse_buffer(stream, p->body.buffer_id.value); - break; - } - default: - pw_log_warn("unexpected node message %d", PW_CLIENT_NODE_MESSAGE_TYPE(message)); - break; + pw_log_trace("stream %p: process output", stream); + impl->in_need_buffer = true; + spa_hook_list_call(&stream->listener_list, struct pw_stream_events, + need_buffer); + impl->in_need_buffer = false; } } @@ -677,17 +638,12 @@ on_rtsocket_condition(void *data, int fd, enum spa_io mask) } if (mask & SPA_IO_IN) { - struct pw_client_node_message message; uint64_t cmd; if (read(fd, &cmd, sizeof(uint64_t)) != sizeof(uint64_t)) pw_log_warn("stream %p: read failed %m", impl); - while (pw_client_node_transport_next_message(impl->trans, &message) == 1) { - struct pw_client_node_message *msg = alloca(SPA_POD_SIZE(&message)); - pw_client_node_transport_parse_message(impl->trans, msg); - handle_rtnode_message(stream, msg); - } + do_process(stream); } } @@ -752,7 +708,6 @@ static void client_node_command(void *data, uint32_t seq, const struct spa_comma if (impl->direction == SPA_DIRECTION_INPUT) { impl->io->status = SPA_STATUS_NEED_BUFFER; - send_need_input(stream); } else { impl->in_need_buffer = true; @@ -1035,7 +990,6 @@ static void clean_transport(struct pw_stream *stream) return; unhandle_socket(stream); - clear_buffers(stream); pw_client_node_transport_destroy(impl->trans); @@ -1057,6 +1011,7 @@ static void client_node_transport(void *data, uint32_t node_id, pw_log_info("stream %p: create client transport %p with fds %d %d for node %u", stream, impl->trans, readfd, writefd, node_id); + handle_socket(stream, readfd, writefd); stream_set_state(stream, PW_STREAM_STATE_CONFIGURE, NULL);