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

Rework how targets are managed

Browse source 
Manage them like we do on the client and reuse logic. Make a node
function to safely add and remove a target.

Activate the targets from the process loop when we can be sure that we
can resume them. This avoids incrementing the pending state when we are
not going to be able to resume the nodes (like when the cycle is ongoing
and we have already been scheduled) and avoids glitches and xruns.

When a node is added to the poll loop, it can activate its own targets.
This is mostly for driver so that they have something to schedule and
can then activate the other targets.

Try to resume the target when it is removed and we are supposed to be
scheduled.

Also add targets to the target_list when the node is remote to make sure
the profiler can see the targets as well.

Keep the node in the INACTIVE state as long as the eventfd of the node
is not added to the loop. Skip nodes in the INACTIVE state from going to
the NOT_TRIGGERED status, which avoids scheduling the node.

Make sure we remove any local targets we have in a node when we export
it, we will receive new targets from the server.

This should eliminate any glitches when adding and removing nodes from
the graph.

See #4026, #2468
This commit is contained in:
Wim Taymans 2024-05-28 18:19:51 +02:00
parent ce985def73
commit 46f71b1cd2
6 changed files with 296 additions and 345 deletions
Download patch file Download diff file Expand all files Collapse all files

96
pipewire-jack/src/pipewire-jack.c
View file

@ -279,6 +279,7 @@ struct link {
struct pw_memmap *mem;
struct pw_node_activation *activation;
int signalfd;
bool active;
};
struct context {
@ -1888,9 +1889,58 @@ static inline uint32_t cycle_wait(struct client *c)
return nframes;
}
static void trigger_link(struct link *l, uint64_t nsec)
{
struct client *c = l->client;
struct pw_node_activation *a = l->activation;
struct pw_node_activation_state *state = &a->state[0];
uint64_t cmd = 1;
pw_log_trace_fp("%p: link %p-%d %p %d/%d", c, l, l->node_id, state,
state->pending, state->required);
if (pw_node_activation_state_dec(state)) {
if (SPA_ATOMIC_CAS(a->status,
PW_NODE_ACTIVATION_NOT_TRIGGERED,
PW_NODE_ACTIVATION_TRIGGERED)) {
a->signal_time = nsec;
pw_log_trace_fp("%p: signal %p %p", c, l, state);
if (SPA_UNLIKELY(write(l->signalfd, &cmd, sizeof(cmd)) != sizeof(cmd)))
pw_log_warn("%p: write failed %m", c);
}
}
}
static inline void activate_link(struct client *c, struct link *l)
{
if (SPA_UNLIKELY(!l->active)) {
if (!c->async) {
struct pw_node_activation_state *state = &l->activation->state[0];
SPA_ATOMIC_INC(state->required);
SPA_ATOMIC_INC(state->pending);
}
l->active = true;
}
}
static inline void deactivate_link(struct client *c, struct link *l, uint64_t trigger)
{
if (SPA_UNLIKELY(l->active)) {
if (!c->async) {
struct pw_node_activation_state *state = &l->activation->state[0];
if (trigger != 0)
trigger_link(l, trigger);
SPA_ATOMIC_DEC(state->required);
}
l->active = false;
}
}
static inline void signal_sync(struct client *c)
{
uint64_t cmd, nsec;
uint64_t nsec;
struct link *l;
struct pw_node_activation *activation = c->activation;
int old_status;
@ -1904,30 +1954,12 @@ static inline void signal_sync(struct client *c)
if (c->async || old_status != PW_NODE_ACTIVATION_AWAKE)
return;
cmd = 1;
spa_list_for_each(l, &c->rt.target_links, target_link) {
struct pw_node_activation_state *state;
if (SPA_UNLIKELY(l->activation == NULL))
continue;
state = &l->activation->state[0];
pw_log_trace_fp("%p: link %p %p %d/%d", c, l, state,
state->pending, state->required);
if (pw_node_activation_state_dec(state)) {
if (SPA_ATOMIC_CAS(l->activation->status,
PW_NODE_ACTIVATION_NOT_TRIGGERED,
PW_NODE_ACTIVATION_TRIGGERED)) {
l->activation->signal_time = nsec;
pw_log_trace_fp("%p: signal %p %p", c, l, state);
if (SPA_UNLIKELY(write(l->signalfd, &cmd, sizeof(cmd)) != sizeof(cmd)))
pw_log_warn("%p: write failed %m", c);
}
}
activate_link(c, l);
trigger_link(l, nsec);
}
}
@ -2203,6 +2235,7 @@ static int do_prepare_client(struct spa_loop *loop, bool async, uint32_t seq,
if (c->rt.prepared)
return 0;
SPA_ATOMIC_STORE(c->activation->status, PW_NODE_ACTIVATION_FINISHED);
pw_loop_update_io(c->l,
c->socket_source,
SPA_IO_IN | SPA_IO_ERR | SPA_IO_HUP);
@ -2217,10 +2250,20 @@ static int do_unprepare_client(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
struct client *c = user_data;
int old_state;
uint64_t trigger = 0;
struct link *l;
if (!c->rt.prepared)
return 0;
old_state = SPA_ATOMIC_XCHG(c->activation->status, PW_NODE_ACTIVATION_INACTIVE);
if (old_state != PW_NODE_ACTIVATION_FINISHED)
trigger = get_time_ns(c->l->system);
spa_list_for_each(l, &c->rt.target_links, target_link)
deactivate_link(c, l, trigger);
pw_loop_update_io(c->l,
c->socket_source, SPA_IO_ERR | SPA_IO_HUP);
@ -2240,7 +2283,6 @@ static int client_node_command(void *data, const struct spa_command *command)
if (c->started) {
pw_data_loop_invoke(c->loop,
do_unprepare_client, SPA_ID_INVALID, NULL, 0, false, c);
c->started = false;
}
break;
@ -2944,8 +2986,18 @@ do_remove_link(struct spa_loop *loop,
bool async, uint32_t seq, const void *data, size_t size, void *user_data)
{
struct link *link = user_data;
struct client *c = link->client;
pw_log_trace("link %p activate", link);
spa_list_remove(&link->target_link);
if (c->rt.prepared) {
int old_state = SPA_ATOMIC_LOAD(c->activation->status);
uint64_t trigger = 0;
if (old_state != PW_NODE_ACTIVATION_FINISHED)
trigger = get_time_ns(c->l->system);
deactivate_link(c, link, trigger);
}
free_link(link);
return 0;
}

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

@ -1347,8 +1347,6 @@ void pw_impl_client_node_registered(struct pw_impl_client_node *this, struct pw_
0,
sizeof(struct pw_node_activation));
node_peer_added(impl, node);
if (impl->bind_node_id) {
pw_global_bind(global, client, PW_PERM_ALL,
impl->bind_node_version, impl->bind_node_id);
@ -1707,16 +1705,6 @@ static void node_port_removed(void *data, struct pw_impl_port *port)
clear_port(impl, p);
}
static void node_driver_changed(void *data, struct pw_impl_node *old, struct pw_impl_node *driver)
{
struct impl *impl = data;
pw_log_debug("%p: driver changed %p -> %p", impl, old, driver);
node_peer_removed(data, old);
node_peer_added(data, driver);
}
static const struct pw_impl_node_events node_events = {
PW_VERSION_IMPL_NODE_EVENTS,
.free = node_free,
@ -1726,7 +1714,6 @@ static const struct pw_impl_node_events node_events = {
.port_removed = node_port_removed,
.peer_added = node_peer_added,
.peer_removed = node_peer_removed,
.driver_changed = node_driver_changed,
};
static const struct pw_resource_events resource_events = {

34
src/modules/module-client-node/remote-node.c
View file

@ -94,21 +94,10 @@ static struct link *find_activation(struct spa_list *links, uint32_t node_id)
return NULL;
}
static int
do_deactivate_link(struct spa_loop *loop,
bool async, uint32_t seq, const void *data, size_t size, void *user_data)
{
struct link *link = user_data;
pw_log_trace("link %p deactivate", link);
spa_list_remove(&link->target.link);
return 0;
}
static void clear_link(struct node_data *data, struct link *link)
{
pw_log_debug("link %p", link);
pw_loop_invoke(data->data_loop,
do_deactivate_link, SPA_ID_INVALID, NULL, 0, true, link);
pw_impl_node_remove_target(data->node, &link->target);
pw_memmap_free(link->map);
spa_system_close(link->target.system, link->target.fd);
spa_list_remove(&link->link);
@ -802,17 +791,6 @@ exit:
return res;
}
static int
do_activate_link(struct spa_loop *loop,
bool async, uint32_t seq, const void *data, size_t size, void *user_data)
{
struct link *link = user_data;
struct node_data *d = link->data;
pw_log_trace("link %p activate", link);
spa_list_append(&d->node->rt.target_list, &link->target.link);
return 0;
}
static int
client_node_set_activation(void *_data,
uint32_t node_id,
@ -863,8 +841,7 @@ client_node_set_activation(void *_data,
link->target.fd = signalfd;
spa_list_append(&data->links, &link->link);
pw_loop_invoke(data->data_loop,
do_activate_link, SPA_ID_INVALID, NULL, 0, false, link);
pw_impl_node_add_target(node, &link->target);
pw_log_debug("node %p: add link %p: memid:%u fd:%d id:%u state:%p pending:%d/%d",
node, link, memid, signalfd, node_id,
@ -1206,6 +1183,8 @@ static struct pw_proxy *node_export(struct pw_core *core, void *object, bool do_
if (node->data_loop == NULL)
goto error;
pw_log_debug("%p: export node %p", core, object);
user_data_size = SPA_ROUND_UP_N(user_data_size, __alignof__(struct node_data));
client_node = pw_core_create_object(core,
@ -1228,6 +1207,11 @@ static struct pw_proxy *node_export(struct pw_core *core, void *object, bool do_
data->client_node = (struct pw_client_node *)client_node;
data->remote_id = SPA_ID_INVALID;
/* the node might have been registered and added to a driver. When we export,
* we will be assigned a new driver target from the server and we can forget our
* local ones. */
pw_impl_node_remove_target(node, &node->rt.driver_target);
pw_impl_node_remove_target(node->driver_node, &node->rt.target);
data->allow_mlock = pw_properties_get_bool(node->properties, "mem.allow-mlock",
data->context->settings.mem_allow_mlock);

71
src/pipewire/impl-link.c
View file

@ -69,13 +69,12 @@ static struct pw_node_peer *pw_node_peer_ref(struct pw_impl_node *onode, struct
peer->ref = 1;
peer->output = onode;
peer->active_count = 0;
copy_target(&peer->target, &inode->rt.target);
peer->target.flags = PW_NODE_TARGET_PEER;
spa_list_append(&onode->peer_list, &peer->link);
pw_log_debug("new peer %p from %p to %p", peer, onode, inode);
pw_impl_node_emit_peer_added(onode, inode);
pw_impl_node_add_target(onode, &peer->target);
return peer;
}
@ -86,48 +85,10 @@ static void pw_node_peer_unref(struct pw_node_peer *peer)
return;
spa_list_remove(&peer->link);
pw_log_debug("remove peer %p from %p to %p", peer, peer->output, peer->target.node);
pw_impl_node_emit_peer_removed(peer->output, peer->target.node);
pw_impl_node_remove_target(peer->output, &peer->target);
free(peer);
}
static void pw_node_peer_activate(struct pw_node_peer *peer)
{
struct pw_node_activation_state *state;
state = &peer->target.activation->state[0];
if (peer->active_count++ == 0) {
spa_list_append(&peer->output->rt.target_list, &peer->target.link);
if (!peer->target.active && peer->output->rt.driver_target.node != NULL) {
if (!peer->output->async)
SPA_ATOMIC_INC(state->required);
peer->target.active = true;
}
}
pw_log_trace("%p: node:%s state:%p pending:%d/%d", peer->output,
peer->target.name, state, state->pending, state->required);
}
static void pw_node_peer_deactivate(struct pw_node_peer *peer)
{
struct pw_node_activation_state *state;
state = &peer->target.activation->state[0];
if (--peer->active_count == 0) {
spa_list_remove(&peer->target.link);
if (peer->target.active) {
if (!peer->output->async) {
SPA_ATOMIC_DEC(state->required);
trigger_target(&peer->target, get_time_ns(peer->target.system));
}
peer->target.active = false;
}
}
pw_log_trace("%p: node:%s state:%p pending:%d/%d", peer->output,
peer->target.name, state, state->pending, state->required);
}
static void info_changed(struct pw_impl_link *link)
{
struct pw_resource *resource;
@ -694,17 +655,6 @@ error:
return res;
}
static int
do_activate_link(struct spa_loop *loop,
bool async, uint32_t seq, const void *data, size_t size, void *user_data)
{
struct pw_impl_link *this = user_data;
pw_log_trace("%p: activate", this);
if (this->peer)
pw_node_peer_activate(this->peer);
return 0;
}
int pw_impl_link_activate(struct pw_impl_link *this)
{
struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
@ -733,9 +683,6 @@ int pw_impl_link_activate(struct pw_impl_link *this)
io_size, &this->rt.out_mix)) < 0)
goto error_clean;
pw_loop_invoke(this->output->node->data_loop,
do_activate_link, SPA_ID_INVALID, NULL, 0, false, this);
impl->activated = true;
pw_log_info("(%s) activated", this->name);
link_update_state(this, PW_LINK_STATE_ACTIVE, 0, NULL);
@ -902,17 +849,6 @@ int pw_impl_link_prepare(struct pw_impl_link *this)
return 0;
}
static int
do_deactivate_link(struct spa_loop *loop,
bool async, uint32_t seq, const void *data, size_t size, void *user_data)
{
struct pw_impl_link *this = user_data;
pw_log_trace("%p: disable out %p", this, &this->rt.out_mix);
if (this->peer)
pw_node_peer_deactivate(this->peer);
return 0;
}
int pw_impl_link_deactivate(struct pw_impl_link *this)
{
struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
@ -922,9 +858,6 @@ int pw_impl_link_deactivate(struct pw_impl_link *this)
if (!impl->activated)
return 0;
pw_loop_invoke(this->output->node->data_loop,
do_deactivate_link, SPA_ID_INVALID, NULL, 0, true, this);
port_set_io(this, this->output, SPA_IO_Buffers, NULL, 0,
&this->rt.out_mix);
port_set_io(this, this->input, SPA_IO_Buffers, NULL, 0,

405
src/pipewire/impl-node.c
View file

@ -70,25 +70,86 @@ struct resource_data {
struct spa_hook listener;
};
static inline void activate_target(struct pw_impl_node *node, struct pw_node_target *t)
{
struct pw_node_activation_state *state = &t->activation->state[0];
if (!t->active) {
if (!node->async) {
SPA_ATOMIC_INC(state->required);
SPA_ATOMIC_INC(state->pending);
}
t->active = true;
pw_log_debug("%p: target state:%p id:%d pending:%d/%d",
node, state, t->id, state->pending, state->required);
}
}
static inline void deactivate_target(struct pw_impl_node *node, struct pw_node_target *t, uint64_t trigger)
{
if (t->active) {
struct pw_node_activation_state *state = &t->activation->state[0];
if (!node->async) {
/* the driver copies the required to the pending state
* so first try to resume the node and then decrement the
* required state. This way we either resume with the old value
* or we don't when the driver has not yet copied */
if (trigger != 0)
trigger_target(t, trigger);
SPA_ATOMIC_DEC(state->required);
}
t->active = false;
pw_log_debug("%p: target state:%p id:%d pending:%d/%d trigger:%"PRIu64,
node, state, t->id, state->pending, state->required, trigger);
}
}
/* called from data-loop when all the targets of a node need to be triggered */
static inline void trigger_targets(struct pw_impl_node *node, int status, uint64_t nsec)
{
struct pw_node_target *ta;
pw_log_trace_fp("%p: (%s-%u) trigger targets %"PRIu64,
node, node->name, node->info.id, nsec);
spa_list_for_each(ta, &node->rt.target_list, link)
trigger_target(ta, nsec);
}
static inline void trigger_activate_targets(struct pw_impl_node *node, int status, uint64_t nsec)
{
struct pw_node_target *ta;
pw_log_trace_fp("%p: (%s-%u) activate and trigger targets %"PRIu64,
node, node->name, node->info.id, nsec);
spa_list_for_each(ta, &node->rt.target_list, link) {
activate_target(node, ta);
trigger_target(ta, nsec);
}
node->rt.trigger_targets = trigger_targets;
}
/** \endcond */
/* Called from the node data loop when a node needs to be scheduled by
* the given driver.
*
* - the node adds the source to the data loop
* - the node needs to trigger the driver when it completes. This means
* the driver is added to the target list.
* - from all the targets of the node, we can now be scheduled ourselves
* so activate our target.
* - When we get scheduled, we will activate our peer targets
*/
static int
do_node_prepare(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
struct pw_impl_node *this = user_data;
struct pw_impl_node *driver = this->driver_node;
struct pw_node_target *t;
uint64_t dummy;
int res;
pw_log_trace("%p: prepare driver %p %d", this, driver, this->rt.prepared);
pw_log_trace("%p: prepare %d remote:%d", this, this->rt.prepared, this->remote);
if (this->rt.prepared)
return 0;
@ -98,17 +159,16 @@ do_node_prepare(struct spa_loop *loop, bool async, uint32_t seq,
if (SPA_UNLIKELY(res != -EAGAIN && res != 0))
pw_log_warn("%p: read failed %m", this);
/* remote nodes have their source added in client-node instead */
if (!this->remote) {
pw_log_trace("%p %p %p", this, loop, this->source.loop);
spa_loop_add_source(loop, &this->source);
SPA_ATOMIC_STORE(this->rt.target.activation->status, PW_NODE_ACTIVATION_FINISHED);
spa_loop_add_source(loop, &this->source);
spa_list_for_each(t, &this->rt.target_list, link) {
/* we can now trigger ourself */
if (t->id == this->info.id)
activate_target(this, t);
}
if (!this->exported) {
/* trigger the driver when we complete */
copy_target(&this->rt.driver_target, &driver->rt.target);
spa_list_append(&this->rt.target_list, &this->rt.driver_target.link);
}
this->rt.trigger_targets = trigger_activate_targets;
this->rt.prepared = true;
return 0;
@ -120,198 +180,46 @@ do_node_unprepare(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
struct pw_impl_node *this = user_data;
struct pw_node_target *t;
int old_state;
uint64_t trigger = 0;
pw_log_trace("%p: unprepare driver %s %d", this, this->rt.driver_target.name,
this->rt.prepared);
pw_log_trace("%p: unprepare %d remote:%d", this, this->rt.prepared, this->remote);
if (!this->rt.prepared)
return 0;
if (!this->remote) {
pw_log_trace("%p %p %p", this, loop, this->source.loop);
spa_loop_remove_source(loop, &this->source);
}
/* We mark ourself as finished now, this will avoid going further into the process loop
* in case our fd was ready (removing ourselfs from the loop should avoid that as well).
* If we were supposed to be scheduled make sure we continue the graph for the peers we
* were supposed to trigger */
old_state = SPA_ATOMIC_XCHG(this->rt.target.activation->status, PW_NODE_ACTIVATION_INACTIVE);
if (old_state != PW_NODE_ACTIVATION_FINISHED)
trigger = get_time_ns(this->rt.target.system);
spa_list_for_each(t, &this->rt.target_list, link)
deactivate_target(this, t, trigger);
spa_loop_remove_source(loop, &this->source);
if (!this->exported) {
spa_list_remove(&this->rt.driver_target.link);
spa_zero(this->rt.driver_target);
}
this->rt.prepared = false;
return 0;
}
/* Called from the driver data loop when a node needs to be scheduled by
* the driver. 2 things needs to happen:
*
* - the node is added to the driver target list and the required state
* is incremented. This makes sure the node is woken up when the driver
* starts a new cycle.
* - the node targets (including the driver we added in do_node_prepare above)
* have their required state incremented. We can safely read the target
* list because we always sync on updates.
*/
static int
do_node_add(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
struct pw_impl_node *this = user_data;
struct pw_impl_node *driver = this->driver_node;
struct pw_node_activation_state *dstate, *nstate;
struct pw_node_target *t;
pw_log_trace("%p: add to driver %p %p %p %d", this, driver,
driver->rt.target.activation, this->rt.target.activation,
this->rt.added);
if (this->rt.added)
return 0;
/* let the driver trigger us as part of the processing cycle */
spa_list_append(&driver->rt.target_list, &this->rt.target.link);
nstate = &this->rt.target.activation->state[0];
if (!this->rt.target.active) {
SPA_ATOMIC_INC(nstate->required);
this->rt.target.active = true;
}
/* now increment the required states of all this node targets, including
* the driver we added above */
spa_list_for_each(t, &this->rt.target_list, link) {
dstate = &t->activation->state[0];
if (!t->active) {
if (!this->async)
SPA_ATOMIC_INC(dstate->required);
t->active = true;
}
pw_log_trace("%p: driver state:%p pending:%d/%d, node state:%p pending:%d/%d",
this, dstate, dstate->pending, dstate->required,
nstate, nstate->pending, nstate->required);
}
this->rt.added = true;
return 0;
}
/* called from the driver data loop and undoes the changes done in do_node_add. */
static int
do_node_remove(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
struct pw_impl_node *this = user_data;
struct pw_node_activation_state *dstate, *nstate;
struct pw_node_target *t;
int old_state;
pw_log_trace("%p: remove from driver %s %p %p %d",
this, this->rt.driver_target.name,
this->rt.driver_target.activation, this->rt.target.activation,
this->rt.added);
if (!this->rt.added)
return 0;
old_state = SPA_ATOMIC_XCHG(this->rt.target.activation->status, PW_NODE_ACTIVATION_FINISHED);
/* remove from driver */
spa_list_remove(&this->rt.target.link);
nstate = &this->rt.target.activation->state[0];
if (this->rt.target.active) {
SPA_ATOMIC_DEC(nstate->required);
this->rt.target.active = false;
}
spa_list_for_each(t, &this->rt.target_list, link) {
dstate = &t->activation->state[0];
if (t->active) {
if (!this->async) {
SPA_ATOMIC_DEC(dstate->required);
if (old_state != PW_NODE_ACTIVATION_FINISHED)
trigger_target(t, get_time_ns(this->rt.target.system));
}
t->active = false;
}
pw_log_trace("%p: driver state:%p pending:%d/%d, node state:%p pending:%d/%d",
this, dstate, dstate->pending, dstate->required,
nstate, nstate->pending, nstate->required);
}
this->rt.added = false;
return 0;
}
static int
do_node_prepare_add(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
struct pw_impl_node *this = user_data;
do_node_prepare(loop, async, seq, data, size, user_data);
if (!this->exported)
do_node_add(loop, async, seq, data, size, user_data);
return 0;
}
static void add_node_to_graph(struct pw_impl_node *node)
{
struct pw_impl_node *driver = node->driver_node;
if (node->remote)
return;
pw_log_debug("%p: driver:%p", node, driver);
if (node->data_loop == driver->data_loop) {
pw_loop_invoke(node->data_loop, do_node_prepare_add, 1, NULL, 0, true, node);
} else {
pw_loop_invoke(node->data_loop, do_node_prepare, 1, NULL, 0, true, node);
if (!node->exported)
pw_loop_invoke(driver->data_loop, do_node_add, 1, NULL, 0, true, node);
}
}
static int
do_node_remove_unprepare(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
struct pw_impl_node *this = user_data;
if (!this->exported)
do_node_remove(loop, async, seq, data, size, user_data);
do_node_unprepare(loop, async, seq, data, size, user_data);
return 0;
pw_loop_invoke(node->data_loop, do_node_prepare, 1, NULL, 0, true, node);
}
static void remove_node_from_graph(struct pw_impl_node *node)
{
struct pw_impl_node *driver = node->driver_node;
pw_log_debug("%p: driver:%p", node, driver);
if (node->data_loop == driver->data_loop) {
pw_loop_invoke(node->data_loop, do_node_remove_unprepare, 1, NULL, 0, true, node);
} else {
if (!node->exported)
pw_loop_invoke(driver->data_loop, do_node_remove, 1, NULL, 0, true, node);
pw_loop_invoke(node->data_loop, do_node_unprepare, 1, NULL, 0, true, node);
}
}
static int
do_node_move(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data)
{
do_node_remove_unprepare(loop, async, seq, data, size, user_data);
do_node_prepare_add(loop, async, seq, data, size, user_data);
return 0;
}
static void move_node_to_graph(struct pw_impl_node *node)
{
struct pw_impl_node *driver = node->driver_node;
pw_log_debug("%p: driver:%p prepared:%d", node, driver, node->rt.prepared);
if (!node->rt.prepared)
if (node->remote)
return;
if (node->data_loop == driver->data_loop) {
pw_loop_invoke(node->data_loop, do_node_move, 1, NULL, 0, true, node);
} else {
remove_node_from_graph(node);
add_node_to_graph(node);
}
pw_loop_invoke(node->data_loop, do_node_unprepare, 1, NULL, 0, true, node);
}
static void node_deactivate(struct pw_impl_node *this)
@ -867,7 +775,7 @@ int pw_impl_node_set_io(struct pw_impl_node *this, uint32_t id, void *data, size
this->driving = this->driver && this->rt.clock && this->rt.position &&
this->rt.position->clock.id == this->rt.clock->id;
pw_log_debug("%p: driving:%d %d %d", this, this->driving,
pw_log_debug("%p: driving:%d clock-id:%d driver-id:%d", this, this->driving,
this->rt.clock ? this->rt.clock->id : SPA_ID_INVALID,
this->rt.position ? this->rt.position->clock.id : SPA_ID_INVALID);
@ -883,6 +791,71 @@ int pw_impl_node_set_io(struct pw_impl_node *this, uint32_t id, void *data, size
return res;
}
static int
do_add_target(struct spa_loop *loop,
bool async, uint32_t seq, const void *data, size_t size, void *user_data)
{
struct pw_node_target *t = user_data;
struct pw_impl_node *node = *(struct pw_impl_node**)data;
pw_log_debug("%p: target:%p id:%d added:%d prepared:%d", node, t, t->id, t->added, node->rt.prepared);
if (!t->added) {
spa_list_append(&node->rt.target_list, &t->link);
t->added = true;
if (node->rt.prepared)
node->rt.trigger_targets = trigger_activate_targets;
}
return 0;
}
SPA_EXPORT
int pw_impl_node_add_target(struct pw_impl_node *node, struct pw_node_target *t)
{
pw_loop_invoke(node->data_loop,
do_add_target, SPA_ID_INVALID, &node, sizeof(void *), true, t);
if (t->node)
pw_impl_node_emit_peer_added(node, t->node);
return 0;
}
static int
do_remove_target(struct spa_loop *loop,
bool async, uint32_t seq, const void *data, size_t size, void *user_data)
{
struct pw_node_target *t = user_data;
struct pw_impl_node *node = *(struct pw_impl_node**)data;
pw_log_debug("%p: target:%p id:%d added:%d prepared:%d", node, t, t->id, t->added, node->rt.prepared);
if (t->added) {
spa_list_remove(&t->link);
t->added = false;
if (node->rt.prepared) {
int old_state = SPA_ATOMIC_LOAD(node->rt.target.activation->status);
uint64_t trigger = 0;
if (old_state != PW_NODE_ACTIVATION_FINISHED)
trigger = get_time_ns(node->rt.target.system);
deactivate_target(node, t, trigger);
}
}
return 0;
}
SPA_EXPORT
int pw_impl_node_remove_target(struct pw_impl_node *node, struct pw_node_target *t)
{
/* we also update the target list for remote nodes so that the profiler
* can inspect the nodes as well */
pw_loop_invoke(node->data_loop,
do_remove_target, SPA_ID_INVALID, &node, sizeof(void *), true, t);
if (t->node)
pw_impl_node_emit_peer_removed(node, t->node);
return 0;
}
static void update_io(struct pw_impl_node *node)
{
struct pw_node_target *t = &node->rt.target;
@ -893,8 +866,6 @@ static void update_io(struct pw_impl_node *node)
sizeof(struct spa_io_clock));
pw_impl_node_set_io(node, SPA_IO_Position, &t->activation->position,
sizeof(struct spa_io_position));
pw_impl_node_emit_peer_added(node, node);
}
SPA_EXPORT
@ -925,7 +896,7 @@ int pw_impl_node_register(struct pw_impl_node *this,
struct pw_context *context = this->context;
struct pw_impl_port *port;
pw_log_debug("%p: register", this);
pw_log_debug("%p: register remote:%d exported:%d", this, this->remote, this->exported);
if (this->registered)
goto error_existed;
@ -949,6 +920,10 @@ int pw_impl_node_register(struct pw_impl_node *this,
this->rt.target.id = this->info.id;
this->rt.target.activation->position.clock.id = this->global->id;
copy_target(&this->rt.driver_target, &this->rt.target);
pw_impl_node_add_target(this, &this->rt.driver_target);
pw_impl_node_add_target(this, &this->rt.target);
pw_properties_setf(this->properties, PW_KEY_OBJECT_ID, "%d", this->global->id);
pw_properties_setf(this->properties, PW_KEY_OBJECT_SERIAL, "%"PRIu64,
pw_global_get_serial(this->global));
@ -1032,16 +1007,19 @@ int pw_impl_node_set_driver(struct pw_impl_node *node, struct pw_impl_node *driv
node->name, node->info.id,
old->name, old->info.id, driver->name, driver->info.id);
pw_impl_node_remove_target(old, &node->rt.target);
pw_impl_node_remove_target(node, &node->rt.driver_target);
spa_zero(node->rt.driver_target);
node->driver_node = driver;
node->moved = true;
move_node_to_graph(node);
copy_target(&node->rt.driver_target, &driver->rt.target);
pw_impl_node_add_target(node, &node->rt.driver_target);
pw_impl_node_add_target(driver, &node->rt.target);
pw_impl_node_emit_driver_changed(node, old, driver);
pw_impl_node_emit_peer_removed(old, node);
pw_impl_node_emit_peer_added(driver, node);
if (no_driver) {
/* We don't have a driver, so remove the property */
pw_properties_set(node->properties, PW_KEY_NODE_DRIVER_ID, NULL);
@ -1296,6 +1274,8 @@ static const char *str_status(uint32_t status)
return "awake";
case PW_NODE_ACTIVATION_FINISHED:
return "finished";
case PW_NODE_ACTIVATION_INACTIVE:
return "inactive";
}
return "unknown";
}
@ -1349,20 +1329,6 @@ static void check_states(struct pw_impl_node *driver, uint64_t nsec)
}
}
/* called from data-loop when all the targets of a node need to be triggered */
static inline int trigger_targets(struct pw_node_target *t, int status, uint64_t nsec)
{
struct pw_node_target *ta;
pw_log_trace_fp("%p: (%s-%u) trigger targets %"PRIu64,
t->node, t->name, t->id, nsec);
spa_list_for_each(ta, &t->node->rt.target_list, link)
trigger_target(ta, nsec);
return 0;
}
static inline void calculate_stats(struct pw_impl_node *this, struct pw_node_activation *a)
{
uint64_t signal_time = a->signal_time;
@ -1445,7 +1411,7 @@ static inline int process_node(void *data)
* graph because that means we finished the graph. */
if (SPA_LIKELY(!this->driving)) {
if (!this->async && old_status == PW_NODE_ACTIVATION_AWAKE)
trigger_targets(&this->rt.target, status, nsec);
this->rt.trigger_targets(this, status, nsec);
} else {
/* calculate CPU time when finished */
a->signal_time = this->driver_start;
@ -1621,7 +1587,7 @@ struct pw_impl_node *pw_context_create_node(struct pw_context *context,
reset_position(this, &this->rt.target.activation->position);
this->rt.target.activation->sync_timeout = DEFAULT_SYNC_TIMEOUT;
this->rt.target.activation->sync_left = 0;
this->rt.target.activation->status = PW_NODE_ACTIVATION_FINISHED;
this->rt.target.activation->status = PW_NODE_ACTIVATION_INACTIVE;
this->rt.rate_limit.interval = 2 * SPA_NSEC_PER_SEC;
this->rt.rate_limit.burst = 1;
@ -1986,7 +1952,7 @@ static int node_ready(void *data, int status)
" pending %d/%d cycle:%u", node->name, node->info.id,
state, a->position.clock.duration,
pending, state->required, a->position.clock.cycle);
SPA_ATOMIC_XCHG(a->status, PW_NODE_ACTIVATION_TRIGGERED);
SPA_ATOMIC_STORE(a->status, PW_NODE_ACTIVATION_TRIGGERED);
process_node(node);
check_states(node, nsec);
pw_impl_node_rt_emit_incomplete(node);
@ -2018,9 +1984,18 @@ again:
struct pw_node_activation *ta = t->activation;
uint32_t id = t->id;
ta->status = PW_NODE_ACTIVATION_NOT_TRIGGERED;
ta->driver_id = driver->info.id;
pw_node_activation_state_reset(&ta->state[0]);
/* we don't change the state of inactive nodes and don't use them
* for reposition. The pending will be at least 1 and they might
* get decremented to 0 but since the status is inactive, we don't
* do the atomic CAS from NOT_TRIGGERED to TRIGGERED and we don't
* write the eventfd. */
if (SPA_ATOMIC_LOAD(ta->status) == PW_NODE_ACTIVATION_INACTIVE)
continue;
SPA_ATOMIC_STORE(ta->status, PW_NODE_ACTIVATION_NOT_TRIGGERED);
/* this is the node with reposition info */
if (SPA_UNLIKELY(id == reposition_owner))
@ -2076,7 +2051,8 @@ again:
spa_node_process_fast(p->mix);
}
/* now signal all the nodes we drive */
return trigger_targets(&node->rt.target, status, nsec);
node->rt.trigger_targets(node, status, nsec);
return 0;
}
static int node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
@ -2283,7 +2259,8 @@ void pw_impl_node_destroy(struct pw_impl_node *node)
/* remove ourself as a follower from the driver node */
spa_list_remove(&node->follower_link);
pw_impl_node_emit_peer_removed(node->driver_node, node);
pw_impl_node_remove_target(node, &node->rt.driver_target);
pw_impl_node_remove_target(node->driver_node, &node->rt.target);
remove_segment_owner(node->driver_node, node->info.id);
spa_list_consume(follower, &node->follower_list, follower_link) {

22
src/pipewire/private.h
View file

@ -542,6 +542,7 @@ struct pw_node_target {
struct spa_system *system;
int fd;
unsigned int active:1;
unsigned int added:1;
};
static inline void copy_target(struct pw_node_target *dst, const struct pw_node_target *src)
@ -554,11 +555,26 @@ static inline void copy_target(struct pw_node_target *dst, const struct pw_node_
dst->fd = src->fd;
}
/* nodes start as INACTIVE, when they are ready to be scheduled, they add their
* fd to the loop and change status to FINISHED. When the node shuts down, the
* status is set back to INACTIVE.
*
* We have status changes (using compare-and-swap) from
*
* INACTIVE -> FINISHED (node is added to loop and can be scheduled)
* * -> INACTIVE (node can not be scheduled anymore)
*
* !INACTIVE -> NOT_TRIGGERED (node is prepared by the driver)
* NOT_TRIGGERED -> TRIGGERED (eventfd is written)
* TRIGGERED -> AWAKE (eventfd is read, node starts processing)
* AWAKE -> FINISHED (node completed processing and triggered the peers)
*/
struct pw_node_activation {
#define PW_NODE_ACTIVATION_NOT_TRIGGERED 0
#define PW_NODE_ACTIVATION_TRIGGERED 1
#define PW_NODE_ACTIVATION_AWAKE 2
#define PW_NODE_ACTIVATION_FINISHED 3
#define PW_NODE_ACTIVATION_INACTIVE 4
uint32_t status;
unsigned int version:1;
@ -774,7 +790,7 @@ struct pw_impl_node {
struct spa_ratelimit rate_limit;
bool prepared; /**< the node was added to loop */
bool added; /**< the node was added to driver */
void (*trigger_targets) (struct pw_impl_node *node, int status, uint64_t nsec);
} rt;
struct spa_fraction target_rate;
uint64_t target_quantum;
@ -923,7 +939,6 @@ struct pw_control_link {
struct pw_node_peer {
int ref;
int active_count;
struct spa_list link; /**< link in peer list */
struct pw_impl_node *output; /**< the output node */
struct pw_node_target target; /**< target of the input node */
@ -1297,6 +1312,9 @@ int pw_impl_node_trigger(struct pw_impl_node *node);
int pw_impl_node_set_io(struct pw_impl_node *node, uint32_t id, void *data, size_t size);
int pw_impl_node_add_target(struct pw_impl_node *node, struct pw_node_target *t);
int pw_impl_node_remove_target(struct pw_impl_node *node, struct pw_node_target *t);
/** Prepare a link
* Starts the negotiation of formats and buffers on \a link */
int pw_impl_link_prepare(struct pw_impl_link *link);