mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-19 07:00:10 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

Reorganise SPA tree

Browse source 
Reorganise the SPA includes to make it more extensible later
Simplify the naming of the buffer and meta params
This commit is contained in:
Wim Taymans 2017-11-10 13:36:14 +01:00
parent 58451d626c
commit caaeaff223
151 changed files with 1353 additions and 964 deletions
Download patch file Download diff file Expand all files Collapse all files

174
spa/include/spa/graph/graph-scheduler1.h Normal file
View file

@ -0,0 +1,174 @@
/* Simple Plugin API
* Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __SPA_GRAPH_SCHEDULER_H__
#define __SPA_GRAPH_SCHEDULER_H__
#ifdef __cplusplus
extern "C" {
#endif
#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)
{
struct spa_graph_node *node = port->node;
uint32_t required = node->required[SPA_DIRECTION_INPUT];
if (port->io->status == SPA_RESULT_HAVE_BUFFER)
node->ready[SPA_DIRECTION_INPUT]++;
spa_debug("port %p node %p check %d %d %d", port, node,
port->io->status, node->ready[SPA_DIRECTION_INPUT], required);
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)
{
bool res;
int state;
struct spa_graph_port *p;
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_RESULT_NEED_BUFFER)
n->state = SPA_GRAPH_STATE_CHECK_IN;
else if (state == SPA_RESULT_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_RESULT_NEED_BUFFER)
n->state = SPA_GRAPH_STATE_CHECK_IN;
else if (state == SPA_RESULT_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_RESULT_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_RESULT_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);
}
return res;
}
static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
{
struct spa_graph_data *d = data;
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(data));
return SPA_RESULT_OK;
}
static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node)
{
struct spa_graph_data *d = data;
spa_debug("node %p start push", node);
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(data));
return SPA_RESULT_OK;
}
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,
};
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif /* __SPA_GRAPH_SCHEDULER_H__ */

185
spa/include/spa/graph/graph-scheduler2.h Normal file
View file

@ -0,0 +1,185 @@
/* Simple Plugin API
* Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __SPA_GRAPH_SCHEDULER_H__
#define __SPA_GRAPH_SCHEDULER_H__
#ifdef __cplusplus
extern "C" {
#endif
#include <spa/graph/graph.h>
static inline int spa_graph_scheduler_default(struct spa_graph_node *node)
{
int res;
struct spa_node *n = node->user_data;
if (node->action == SPA_GRAPH_ACTION_IN)
res = spa_node_process_input(n);
else if (node->action == SPA_GRAPH_ACTION_OUT)
res = spa_node_process_output(n);
else
res = SPA_RESULT_ERROR;
return res;
}
static inline void spa_graph_port_check(struct spa_graph *graph, struct spa_graph_port *port)
{
struct spa_graph_node *node = port->node;
if (port->io->status == SPA_RESULT_HAVE_BUFFER)
node->ready++;
spa_debug("port %p node %p check %d %d %d", port, node, port->io->status, node->ready, node->required);
if (node->required > 0 && node->ready == node->required) {
node->action = SPA_GRAPH_ACTION_IN;
if (node->ready_link.next == NULL)
spa_list_append(&graph->ready, &node->ready_link);
} else if (node->ready_link.next) {
spa_list_remove(&node->ready_link);
node->ready_link.next = NULL;
}
}
static inline void spa_graph_node_update(struct spa_graph *graph, struct spa_graph_node *node) {
struct spa_graph_port *p;
node->ready = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
if (p->io->status == SPA_RESULT_OK && !(node->flags & SPA_GRAPH_NODE_FLAG_ASYNC))
node->ready++;
}
spa_debug("node %p update %d ready", node, node->ready);
}
static inline bool spa_graph_scheduler_iterate(struct spa_graph *graph)
{
bool empty;
struct spa_graph_port *p;
struct spa_graph_node *n;
int iter = 1;
uint32_t state;
next:
empty = spa_list_is_empty(&graph->ready);
if (empty && !spa_list_is_empty(&graph->pending)) {
spa_debug("copy pending");
spa_list_insert_list(&graph->ready, &graph->pending);
spa_list_init(&graph->pending);
empty = false;
}
if (iter-- == 0 || empty)
return !empty;
n = spa_list_first(&graph->ready, struct spa_graph_node, ready_link);
spa_list_remove(&n->ready_link);
n->ready_link.next = NULL;
spa_debug("node %p state %d", n, n->state);
switch (n->state) {
case SPA_GRAPH_STATE_IN:
case SPA_GRAPH_STATE_OUT:
case SPA_GRAPH_STATE_END:
if (n->state == SPA_GRAPH_STATE_END)
n->state = SPA_GRAPH_STATE_OUT;
state = n->schedule(n);
spa_debug("node %p schedule %d res %d", n, action, state);
if (n->state == SPA_GRAPH_STATE_IN && n == graph->node)
break;
if (n->state != SPA_GRAPH_STATE_END) {
spa_debug("node %p add ready for CHECK", n);
if (state == SPA_RESULT_NEED_BUFFER)
n->state = SPA_GRAPH_STATE_CHECK_IN;
else if (state == SPA_RESULT_HAVE_BUFFER)
n->state = SPA_GRAPH_STATE_CHECK_OUT;
else if (state == SPA_RESULT_OK)
n->state = SPA_GRAPH_STATE_CHECK_OK;
spa_list_append(&graph->ready, &n->ready_link);
}
else {
spa_graph_node_update(graph, n);
}
break;
case SPA_GRAPH_STATE_CHECK_IN:
n->ready = 0;
spa_list_for_each(p, &n->ports[SPA_DIRECTION_INPUT], link) {
struct spa_graph_node *pn = p->peer->node;
if (p->io->status == SPA_RESULT_NEED_BUFFER) {
if (pn != graph->node
|| pn->flags & SPA_GRAPH_NODE_FLAG_ASYNC) {
pn->state = SPA_GRAPH_STATE_OUT;
spa_debug("node %p add ready OUT", n);
spa_list_append(&graph->ready, &pn->ready_link);
}
} else if (p->io->status == SPA_RESULT_OK)
n->ready++;
}
break;
case SPA_GRAPH_STATE_CHECK_OUT:
spa_list_for_each(p, &n->ports[SPA_DIRECTION_OUTPUT], link)
spa_graph_port_check(graph, p->peer);
spa_debug("node %p add pending", n);
n->state = SPA_GRAPH_STATE_END;
spa_list_insert(&graph->pending, &n->ready_link);
break;
case SPA_GRAPH_STATE_CHECK_OK:
spa_graph_node_update(graph, n);
break;
default:
break;
}
goto next;
}
static inline void spa_graph_scheduler_pull(struct spa_graph *graph, struct spa_graph_node *node)
{
node->action = SPA_GRAPH_ACTION_CHECK;
node->state = SPA_RESULT_NEED_BUFFER;
graph->node = node;
spa_debug("node %p start pull", node);
if (node->ready_link.next == NULL)
spa_list_append(&graph->ready, &node->ready_link);
}
static inline void spa_graph_scheduler_push(struct spa_graph *graph, struct spa_graph_node *node)
{
node->action = SPA_GRAPH_ACTION_OUT;
graph->node = node;
spa_debug("node %p start push", node);
if (node->ready_link.next == NULL)
spa_list_append(&graph->ready, &node->ready_link);
}
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif /* __SPA_GRAPH_SCHEDULER_H__ */

163
spa/include/spa/graph/graph-scheduler3.h Normal file
View file

@ -0,0 +1,163 @@
/* Simple Plugin API
* Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __SPA_GRAPH_SCHEDULER_H__
#define __SPA_GRAPH_SCHEDULER_H__
#ifdef __cplusplus
extern "C" {
#endif
#include <spa/graph/graph.h>
static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
struct spa_graph_node *n, *t;
struct spa_list ready;
spa_debug("node %p start pull", node);
spa_list_init(&ready);
node->ready[SPA_DIRECTION_INPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
if ((pport = p->peer) == NULL)
continue;
pnode = pport->node;
spa_debug("node %p peer %p io %d %d", node, pnode, pport->io->status, pport->io->buffer_id);
if (pport->io->status == SPA_RESULT_NEED_BUFFER) {
if (pnode->ready_link.next == NULL)
spa_list_append(&ready, &pnode->ready_link);
}
else if (pport->io->status == SPA_RESULT_OK &&
!(pnode->flags & SPA_GRAPH_NODE_FLAG_ASYNC))
node->ready[SPA_DIRECTION_INPUT]++;
}
spa_list_for_each_safe(n, t, &ready, ready_link) {
n->state = spa_node_process_output(n->implementation);
spa_debug("peer %p processed out %d", n, n->state);
if (n->state == SPA_RESULT_NEED_BUFFER)
spa_graph_need_input(n->graph, n);
else {
spa_list_for_each(p, &n->ports[SPA_DIRECTION_OUTPUT], link) {
if (p->io->status == SPA_RESULT_HAVE_BUFFER)
node->ready[SPA_DIRECTION_INPUT]++;
}
}
spa_list_remove(&n->ready_link);
n->ready_link.next = NULL;
}
spa_debug("node %p ready:%d required:%d", node, node->ready[SPA_DIRECTION_INPUT], node->required[SPA_DIRECTION_INPUT]);
if (node->required[SPA_DIRECTION_INPUT] > 0 &&
node->ready[SPA_DIRECTION_INPUT] == node->required[SPA_DIRECTION_INPUT]) {
node->state = spa_node_process_input(node->implementation);
spa_debug("node %p processed in %d", node, node->state);
if (node->state == SPA_RESULT_HAVE_BUFFER) {
spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link) {
if (p->io->status == SPA_RESULT_HAVE_BUFFER)
if (p->peer)
p->peer->node->ready[SPA_DIRECTION_INPUT]++;
}
}
}
return SPA_RESULT_OK;
}
static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
struct spa_list ready;
struct spa_graph_node *n, *t;
spa_debug("node %p start push", node);
spa_list_init(&ready);
spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
uint32_t prequired, pready;
if ((pport = p->peer) == NULL) {
spa_debug("node %p port %p has no peer", node, p);
continue;
}
pnode = pport->node;
if (pport->io->status == SPA_RESULT_HAVE_BUFFER)
pnode->ready[SPA_DIRECTION_INPUT]++;
pready = pnode->ready[SPA_DIRECTION_INPUT];
prequired = pnode->required[SPA_DIRECTION_INPUT];
spa_debug("node %p peer %p io %d %d %d", node, pnode, pport->io->status,
pready, prequired);
if (prequired > 0 && pready == prequired)
if (pnode->ready_link.next == NULL)
spa_list_append(&ready, &pnode->ready_link);
}
spa_list_for_each_safe(n, t, &ready, ready_link) {
n->state = spa_node_process_input(n->implementation);
spa_debug("node %p chain processed in %d", n, n->state);
if (n->state == SPA_RESULT_HAVE_BUFFER)
spa_graph_have_output(n->graph, n);
else {
n->ready[SPA_DIRECTION_INPUT] = 0;
spa_list_for_each(p, &n->ports[SPA_DIRECTION_INPUT], link) {
if (p->io->status == SPA_RESULT_OK &&
!(n->flags & SPA_GRAPH_NODE_FLAG_ASYNC))
n->ready[SPA_DIRECTION_INPUT]++;
}
}
spa_list_remove(&n->ready_link);
n->ready_link.next = NULL;
}
node->state = spa_node_process_output(node->implementation);
spa_debug("node %p processed out %d", node, node->state);
if (node->state == SPA_RESULT_NEED_BUFFER) {
node->ready[SPA_DIRECTION_INPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
if (p->io->status == SPA_RESULT_OK && !(node->flags & SPA_GRAPH_NODE_FLAG_ASYNC))
node->ready[SPA_DIRECTION_INPUT]++;
}
}
return SPA_RESULT_OK;
}
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,
};
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif /* __SPA_GRAPH_SCHEDULER_H__ */

212
spa/include/spa/graph/graph-scheduler4.h Normal file
View file

@ -0,0 +1,212 @@
/* Simple Plugin API
* Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __SPA_GRAPH_SCHEDULER_H__
#define __SPA_GRAPH_SCHEDULER_H__
#ifdef __cplusplus
extern "C" {
#endif
#include <spa/graph/graph.h>
struct spa_graph_data {
struct spa_graph *graph;
};
static inline void spa_graph_data_init(struct spa_graph_data *data,
struct spa_graph *graph)
{
data->graph = graph;
}
static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node);
static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node);
static inline void check_input(struct spa_graph_node *node)
{
struct spa_graph_port *p;
node->ready[SPA_DIRECTION_INPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
if ((pport = p->peer) == NULL)
continue;
pnode = pport->node;
debug("node %p input peer %p io %d %d\n", node, pnode, pport->io->status, pport->io->buffer_id);
pnode->ready[SPA_DIRECTION_OUTPUT]++;
if (pport->io->status == SPA_RESULT_OK)
node->ready[SPA_DIRECTION_INPUT]++;
debug("node %p input peer %p out %d %d\n", node, pnode,
pnode->required[SPA_DIRECTION_OUTPUT],
pnode->ready[SPA_DIRECTION_OUTPUT]);
}
}
static inline void check_output(struct spa_graph_node *node)
{
struct spa_graph_port *p;
node->ready[SPA_DIRECTION_OUTPUT] = 0;
node->required[SPA_DIRECTION_OUTPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
if ((pport = p->peer) == NULL)
continue;
pnode = pport->node;
debug("node %p output peer %p io %d %d\n", node, pnode, pport->io->status, pport->io->buffer_id);
if (pport->io->status == SPA_RESULT_HAVE_BUFFER) {
pnode->ready[SPA_DIRECTION_INPUT]++;
node->required[SPA_DIRECTION_OUTPUT]++;
}
debug("node %p output peer %p out %d %d\n", node, pnode,
pnode->required[SPA_DIRECTION_INPUT],
pnode->ready[SPA_DIRECTION_INPUT]);
}
}
static inline void spa_graph_impl_activate(void *data, struct spa_graph_node *node)
{
int res;
debug("node %p activate %d\n", node, node->state);
if (node->state == SPA_RESULT_NEED_BUFFER) {
res = spa_node_process_input(node->implementation);
debug("node %p process in %d\n", node, res);
}
else if (node->state == SPA_RESULT_HAVE_BUFFER) {
res = spa_node_process_output(node->implementation);
debug("node %p process out %d\n", node, res);
}
else
return;
if (res == SPA_RESULT_NEED_BUFFER || (res == SPA_RESULT_OK && node->state == SPA_RESULT_NEED_BUFFER)) {
check_input(node);
}
else if (res == SPA_RESULT_HAVE_BUFFER) {
check_output(node);
}
node->state = res;
debug("node %p activate end %d\n", node, res);
}
static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
debug("node %p start pull\n", node);
node->state = SPA_RESULT_NEED_BUFFER;
node->ready[SPA_DIRECTION_INPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
uint32_t prequired;
if ((pport = p->peer) == NULL)
continue;
pnode = pport->node;
prequired = pnode->required[SPA_DIRECTION_OUTPUT];
debug("node %p pull peer %p io %d %d\n", node, pnode, pport->io->status, pport->io->buffer_id);
pnode->ready[SPA_DIRECTION_OUTPUT]++;
if (pport->io->status == SPA_RESULT_OK)
node->ready[SPA_DIRECTION_INPUT]++;
debug("node %p pull peer %p out %d %d\n", node, pnode, prequired, pnode->ready[SPA_DIRECTION_OUTPUT]);
if (prequired > 0 && pnode->ready[SPA_DIRECTION_OUTPUT] >= prequired) {
pnode->state = SPA_RESULT_HAVE_BUFFER;
spa_graph_impl_activate(data, pnode);
}
}
debug("node %p end pull\n", node);
return SPA_RESULT_OK;
}
static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
uint32_t required;
debug("node %p start push\n", node);
node->state = SPA_RESULT_HAVE_BUFFER;
node->ready[SPA_DIRECTION_OUTPUT] = 0;
node->required[SPA_DIRECTION_OUTPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
uint32_t prequired;
if ((pport = p->peer) == NULL)
continue;
pnode = pport->node;
prequired = pnode->required[SPA_DIRECTION_INPUT];
debug("node %p push peer %p io %d %d\n", node, pnode, pport->io->status, pport->io->buffer_id);
if (pport->io->status == SPA_RESULT_HAVE_BUFFER) {
pnode->ready[SPA_DIRECTION_INPUT]++;
node->required[SPA_DIRECTION_OUTPUT]++;
}
debug("node %p push peer %p in %d %d\n", node, pnode, prequired, pnode->ready[SPA_DIRECTION_INPUT]);
if (prequired > 0 && pnode->ready[SPA_DIRECTION_INPUT] >= prequired) {
pnode->state = SPA_RESULT_NEED_BUFFER;
spa_graph_impl_activate(data, pnode);
}
}
required = node->required[SPA_DIRECTION_OUTPUT];
if (required > 0 && node->ready[SPA_DIRECTION_OUTPUT] >= required) {
}
debug("node %p end push\n", node);
return SPA_RESULT_OK;
}
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,
};
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif /* __SPA_GRAPH_SCHEDULER_H__ */

158
spa/include/spa/graph/graph-scheduler5.h Normal file
View file

@ -0,0 +1,158 @@
/* Simple Plugin API
* Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __SPA_GRAPH_SCHEDULER_H__
#define __SPA_GRAPH_SCHEDULER_H__
#ifdef __cplusplus
extern "C" {
#endif
#include <spa/graph/graph.h>
struct spa_graph_data {
struct spa_graph *graph;
};
static inline void spa_graph_data_init(struct spa_graph_data *data,
struct spa_graph *graph)
{
data->graph = graph;
}
static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node);
static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node);
static inline void spa_graph_impl_activate(void *data, struct spa_graph_node *node, bool recurse)
{
int res = node->state;
debug("node %p activate %d\n", node, node->state);
if (node->state == SPA_RESULT_NEED_BUFFER) {
res = spa_node_process_input(node->implementation);
debug("node %p process in %d\n", node, res);
}
else if (node->state == SPA_RESULT_HAVE_BUFFER) {
res = spa_node_process_output(node->implementation);
debug("node %p process out %d\n", node, res);
}
if (recurse && (res == SPA_RESULT_NEED_BUFFER || res == SPA_RESULT_OK))
spa_graph_impl_need_input(data, node);
else if (recurse && (res == SPA_RESULT_HAVE_BUFFER))
spa_graph_impl_have_output(data, node);
else
node->state = res;
debug("node %p activate end %d\n", node, node->state);
}
static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
uint32_t required;
debug("node %p start pull\n", node);
node->state = SPA_RESULT_NEED_BUFFER;
node->ready[SPA_DIRECTION_INPUT] = 0;
required = node->required[SPA_DIRECTION_INPUT];
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
uint32_t prequired;
if ((pport = p->peer) == NULL)
continue;
pnode = pport->node;
prequired = pnode->required[SPA_DIRECTION_OUTPUT];
debug("node %p pull peer %p io %d %d\n", node, pnode, pport->io->status, pport->io->buffer_id);
if (pport->io->status == SPA_RESULT_NEED_BUFFER)
pnode->ready[SPA_DIRECTION_OUTPUT]++;
else if (pport->io->status == SPA_RESULT_OK)
node->ready[SPA_DIRECTION_INPUT]++;
debug("node %p pull peer %p out %d %d\n", node, pnode, prequired, pnode->ready[SPA_DIRECTION_OUTPUT]);
if (prequired > 0 && pnode->ready[SPA_DIRECTION_OUTPUT] >= prequired) {
if (pnode->state == SPA_RESULT_NEED_BUFFER)
pnode->state = SPA_RESULT_HAVE_BUFFER;
spa_graph_impl_activate(data, pnode, true);
}
}
if (required > 0 && node->ready[SPA_DIRECTION_INPUT] >= required)
spa_graph_impl_activate(data, node, false);
debug("node %p end pull\n", node);
return SPA_RESULT_OK;
}
static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
debug("node %p start push\n", node);
node->state = SPA_RESULT_HAVE_BUFFER;
node->ready[SPA_DIRECTION_OUTPUT] = 0;
node->required[SPA_DIRECTION_OUTPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
uint32_t prequired;
if ((pport = p->peer) == NULL)
continue;
pnode = pport->node;
prequired = pnode->required[SPA_DIRECTION_INPUT];
debug("node %p push peer %p io %d %d\n", node, pnode, pport->io->status, pport->io->buffer_id);
if (pport->io->status == SPA_RESULT_HAVE_BUFFER) {
pnode->ready[SPA_DIRECTION_INPUT]++;
node->required[SPA_DIRECTION_OUTPUT]++;
}
debug("node %p push peer %p in %d %d\n", node, pnode, prequired, pnode->ready[SPA_DIRECTION_INPUT]);
if (prequired > 0 && pnode->ready[SPA_DIRECTION_INPUT] >= prequired)
spa_graph_impl_activate(data, pnode, true);
}
required = node->required[SPA_DIRECTION_OUTPUT];
if (required > 0 && node->ready[SPA_DIRECTION_OUTPUT] >= required)
spa_graph_impl_activate(data, node, false);
debug("node %p end push\n", node);
return SPA_RESULT_OK;
}
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,
};
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif /* __SPA_GRAPH_SCHEDULER_H__ */

136
spa/include/spa/graph/graph-scheduler6.h Normal file
View file

@ -0,0 +1,136 @@
/* Simple Plugin API
* Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __SPA_GRAPH_SCHEDULER_H__
#define __SPA_GRAPH_SCHEDULER_H__
#ifdef __cplusplus
extern "C" {
#endif
#include <spa/graph/graph.h>
struct spa_graph_data {
struct spa_graph *graph;
};
static inline void spa_graph_data_init(struct spa_graph_data *data,
struct spa_graph *graph)
{
data->graph = graph;
}
static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
spa_debug("node %p start pull", node);
node->ready[SPA_DIRECTION_INPUT] = 0;
node->required[SPA_DIRECTION_INPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
uint32_t prequired, pready;
if ((pport = p->peer) == NULL) {
spa_debug("node %p port %p has no peer", node, p);
continue;
}
pnode = pport->node;
if (pport->io->status == SPA_RESULT_NEED_BUFFER) {
pnode->ready[SPA_DIRECTION_OUTPUT]++;
node->required[SPA_DIRECTION_INPUT]++;
}
pready = pnode->ready[SPA_DIRECTION_OUTPUT];
prequired = pnode->required[SPA_DIRECTION_OUTPUT];
spa_debug("node %p peer %p io %d %d %d %d", node, pnode, pport->io->status,
pport->io->buffer_id, pready, prequired);
if (prequired > 0 && pready >= prequired) {
pnode->state = spa_node_process_output(pnode->implementation);
spa_debug("peer %p processed out %d", pnode, pnode->state);
if (pnode->state == SPA_RESULT_NEED_BUFFER)
spa_graph_need_input(pnode->graph, pnode);
else if (pnode->state == SPA_RESULT_HAVE_BUFFER)
spa_graph_have_output(pnode->graph, pnode);
}
}
return SPA_RESULT_OK;
}
static inline int spa_graph_impl_have_output(void *data, struct spa_graph_node *node)
{
struct spa_graph_port *p;
spa_debug("node %p start push", node);
node->ready[SPA_DIRECTION_OUTPUT] = 0;
node->required[SPA_DIRECTION_OUTPUT] = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_OUTPUT], link) {
struct spa_graph_port *pport;
struct spa_graph_node *pnode;
uint32_t prequired, pready;
if ((pport = p->peer) == NULL) {
spa_debug("node %p port %p has no peer", node, p);
continue;
}
pnode = pport->node;
if (p->io->status == SPA_RESULT_HAVE_BUFFER) {
pnode->ready[SPA_DIRECTION_INPUT]++;
node->required[SPA_DIRECTION_OUTPUT]++;
}
pready = pnode->ready[SPA_DIRECTION_INPUT];
prequired = pnode->required[SPA_DIRECTION_INPUT];
spa_debug("node %p peer %p io %d %d %d", node, pnode, pport->io->status,
pready, prequired);
if (prequired > 0 && pready >= prequired) {
pnode->state = spa_node_process_input(pnode->implementation);
spa_debug("node %p chain processed in %d", pnode, pnode->state);
if (pnode->state == SPA_RESULT_HAVE_BUFFER)
spa_graph_have_output(pnode->graph, pnode);
else if (pnode->state == SPA_RESULT_NEED_BUFFER)
spa_graph_need_input(pnode->graph, pnode);
}
}
return SPA_RESULT_OK;
}
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,
};
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif /* __SPA_GRAPH_SCHEDULER_H__ */

188
spa/include/spa/graph/graph.h Normal file
View file

@ -0,0 +1,188 @@
/* Simple Plugin API
* Copyright (C) 2017 Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __SPA_GRAPH_H__
#define __SPA_GRAPH_H__
#ifdef __cplusplus
extern "C" {
#endif
#include <spa/utils/defs.h>
#include <spa/utils/list.h>
#include <spa/node/node.h>
#ifndef spa_debug
#define spa_debug(...)
#endif
struct spa_graph;
struct spa_graph_node;
struct spa_graph_port;
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);
};
struct spa_graph {
struct spa_list nodes;
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_reuse_buffer(g,n,p,i) ((g)->callbacks->reuse_buffer((g)->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 */
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 */
void *scheduler_data; /**< scheduler private data */
};
struct spa_graph_port {
struct spa_list link; /**< link in node port list */
struct spa_graph_node *node; /**< owner node */
enum spa_direction direction; /**< port direction */
uint32_t port_id; /**< port id */
uint32_t flags; /**< port flags */
struct spa_port_io *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)
{
spa_list_init(&graph->nodes);
}
static inline void
spa_graph_set_callbacks(struct spa_graph *graph,
const struct spa_graph_callbacks *callbacks,
void *data)
{
graph->callbacks = callbacks;
graph->callbacks_data = data;
}
static inline void
spa_graph_node_init(struct spa_graph_node *node)
{
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->required[SPA_DIRECTION_OUTPUT] = node->ready[SPA_DIRECTION_OUTPUT] = 0;
spa_debug("node %p init", node);
}
static inline void
spa_graph_node_set_implementation(struct spa_graph_node *node,
struct spa_node *implementation)
{
node->implementation = implementation;
}
static inline void
spa_graph_node_add(struct spa_graph *graph,
struct spa_graph_node *node)
{
node->graph = graph;
node->state = SPA_RESULT_OK;
node->ready_link.next = NULL;
spa_list_append(&graph->nodes, &node->link);
spa_debug("node %p add", node);
}
static inline void
spa_graph_port_init(struct spa_graph_port *port,
enum spa_direction direction,
uint32_t port_id,
uint32_t flags,
struct spa_port_io *io)
{
spa_debug("port %p init type %d id %d", port, direction, port_id);
port->direction = direction;
port->port_id = port_id;
port->flags = flags;
port->io = io;
}
static inline void
spa_graph_port_add(struct spa_graph_node *node,
struct spa_graph_port *port)
{
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)
spa_list_remove(&node->ready_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->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);
out->peer = in;
in->peer = out;
}
static inline void
spa_graph_port_unlink(struct spa_graph_port *port)
{
spa_debug("port %p unlink from %p", port, port->peer);
if (port->peer) {
port->peer->peer = NULL;
port->peer = NULL;
}
}
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif /* __SPA_GRAPH_H__ */