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

graph: new scheduling model

Browse source 
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
This commit is contained in:
Wim Taymans 2018-03-20 11:37:11 +01:00
parent 9b0a880afb
commit 33a322b96e
36 changed files with 401 additions and 750 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/extensions/client-node.h
View file

@ -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 */
};

55
src/modules/module-client-node/client-node.c
View file

@ -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);
}
}

51
src/pipewire/link.c
View file

@ -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;

90
src/pipewire/node.c
View file

@ -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);
}

6
src/pipewire/node.h
View file

@ -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);
};

24
src/pipewire/port.c
View file

@ -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)

11
src/pipewire/private.h
View file

@ -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;

139
src/pipewire/remote.c
View file

@ -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

101
src/pipewire/stream.c
View file

@ -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);