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

audiomixer: rework the port logic

Browse source 
Use port lists for faster and safer updates of the mixer ports.
This commit is contained in:
Wim Taymans 2025-07-30 15:57:47 +02:00
parent 77494086c1
commit 8f45cfcbc9
1 changed files with 78 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

143
spa/plugins/audiomixer/audiomixer.c
View file

@ -59,6 +59,8 @@ struct buffer {
}; };
struct port { struct port {
struct spa_list link;
uint32_t direction; uint32_t direction;
uint32_t id; uint32_t id;
@ -70,7 +72,6 @@ struct port {
struct spa_port_info info; struct spa_port_info info;
struct spa_param_info params[8]; struct spa_param_info params[8];
unsigned int valid:1;
unsigned int have_format:1; unsigned int have_format:1;
struct buffer buffers[MAX_BUFFERS]; struct buffer buffers[MAX_BUFFERS];
@ -78,6 +79,9 @@ struct port {
struct spa_list queue; struct spa_list queue;
size_t queued_bytes; size_t queued_bytes;
struct spa_list mix_link;
bool active;
}; };
struct impl { struct impl {
@ -103,10 +107,10 @@ struct impl {
struct spa_hook_list hooks; struct spa_hook_list hooks;
uint32_t port_count;
uint32_t last_port;
struct port *in_ports[MAX_PORTS]; struct port *in_ports[MAX_PORTS];
struct port out_ports[1]; struct port out_ports[1];
struct spa_list port_list;
struct spa_list free_list;
struct buffer *mix_buffers[MAX_PORTS]; struct buffer *mix_buffers[MAX_PORTS];
const void *mix_datas[MAX_PORTS]; const void *mix_datas[MAX_PORTS];
@ -118,12 +122,13 @@ struct impl {
unsigned int started:1; unsigned int started:1;
uint32_t stride; uint32_t stride;
uint32_t blocks; uint32_t blocks;
struct spa_list mix_list;
}; };
#define PORT_VALID(p) ((p) != NULL && (p)->valid)
#define CHECK_ANY_IN(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) == SPA_ID_INVALID) #define CHECK_ANY_IN(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) == SPA_ID_INVALID)
#define CHECK_FREE_IN_PORT(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) < MAX_PORTS && !PORT_VALID(this->in_ports[(p)])) #define CHECK_FREE_IN_PORT(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) < MAX_PORTS && this->in_ports[(p)] == NULL)
#define CHECK_IN_PORT(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) < MAX_PORTS && PORT_VALID(this->in_ports[(p)])) #define CHECK_IN_PORT(this,d,p) ((d) == SPA_DIRECTION_INPUT && (p) < MAX_PORTS && this->in_ports[(p)] != NULL)
#define CHECK_OUT_PORT(this,d,p) ((d) == SPA_DIRECTION_OUTPUT && (p) == 0) #define CHECK_OUT_PORT(this,d,p) ((d) == SPA_DIRECTION_OUTPUT && (p) == 0)
#define CHECK_PORT(this,d,p) (CHECK_OUT_PORT(this,d,p) || CHECK_IN_PORT (this,d,p)) #define CHECK_PORT(this,d,p) (CHECK_OUT_PORT(this,d,p) || CHECK_IN_PORT (this,d,p))
#define CHECK_PORT_ANY(this,d,p) (CHECK_ANY_IN(this,d,p) || CHECK_PORT(this,d,p)) #define CHECK_PORT_ANY(this,d,p) (CHECK_ANY_IN(this,d,p) || CHECK_PORT(this,d,p))
@ -209,7 +214,7 @@ static int impl_node_add_listener(void *object,
{ {
struct impl *this = object; struct impl *this = object;
struct spa_hook_list save; struct spa_hook_list save;
uint32_t i; struct port *port;
spa_return_val_if_fail(this != NULL, -EINVAL); spa_return_val_if_fail(this != NULL, -EINVAL);
@ -217,10 +222,8 @@ static int impl_node_add_listener(void *object,
emit_node_info(this, true); emit_node_info(this, true);
emit_port_info(this, GET_OUT_PORT(this, 0), true); emit_port_info(this, GET_OUT_PORT(this, 0), true);
for (i = 0; i < this->last_port; i++) { spa_list_for_each(port, &this->port_list, link)
if (PORT_VALID(this->in_ports[i])) emit_port_info(this, port, true);
emit_port_info(this, GET_IN_PORT(this, i), true);
}
spa_hook_list_join(&this->hooks, &save); spa_hook_list_join(&this->hooks, &save);
@ -235,6 +238,18 @@ impl_node_set_callbacks(void *object,
return 0; return 0;
} }
static struct port *get_free_port(struct impl *this)
{
struct port *port;
if (!spa_list_is_empty(&this->free_list)) {
port = spa_list_first(&this->free_list, struct port, link);
spa_list_remove(&port->link);
} else {
port = calloc(1, sizeof(struct port));
}
return port;
}
static int impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id, static int impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id,
const struct spa_dict *props) const struct spa_dict *props)
{ {
@ -244,13 +259,9 @@ static int impl_node_add_port(void *object, enum spa_direction direction, uint32
spa_return_val_if_fail(this != NULL, -EINVAL); spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(CHECK_FREE_IN_PORT(this, direction, port_id), -EINVAL); spa_return_val_if_fail(CHECK_FREE_IN_PORT(this, direction, port_id), -EINVAL);
port = GET_IN_PORT(this, port_id); if ((port = get_free_port(this)) == NULL)
if (port == NULL) { return -errno;
port = calloc(1, sizeof(struct port));
if (port == NULL)
return -errno;
this->in_ports[port_id] = port;
}
port->direction = SPA_DIRECTION_INPUT; port->direction = SPA_DIRECTION_INPUT;
port->id = port_id; port->id = port_id;
@ -272,13 +283,10 @@ static int impl_node_add_port(void *object, enum spa_direction direction, uint32
port->info.params = port->params; port->info.params = port->params;
port->info.n_params = 5; port->info.n_params = 5;
this->port_count++; this->in_ports[port_id] = port;
if (this->last_port <= port_id) spa_list_append(&this->port_list, &port->link);
this->last_port = port_id + 1;
port->valid = true;
spa_log_debug(this->log, "%p: add port %d:%d %d", this, spa_log_debug(this->log, "%p: add port %d:%d", this, direction, port_id);
direction, port_id, this->last_port);
emit_port_info(this, port, true); emit_port_info(this, port, true);
return 0; return 0;
@ -294,26 +302,18 @@ impl_node_remove_port(void *object, enum spa_direction direction, uint32_t port_
spa_return_val_if_fail(CHECK_IN_PORT(this, direction, port_id), -EINVAL); spa_return_val_if_fail(CHECK_IN_PORT(this, direction, port_id), -EINVAL);
port = GET_IN_PORT(this, port_id); port = GET_IN_PORT(this, port_id);
this->in_ports[port_id] = NULL;
spa_list_remove(&port->link);
port->valid = false;
this->port_count--;
if (port->have_format && this->have_format) { if (port->have_format && this->have_format) {
if (--this->n_formats == 0) if (--this->n_formats == 0)
this->have_format = false; this->have_format = false;
} }
spa_memzero(port, sizeof(struct port)); spa_memzero(port, sizeof(struct port));
spa_list_append(&this->free_list, &port->link);
if (port_id + 1 == this->last_port) { spa_log_debug(this->log, "%p: remove port %d:%d", this,
int i; direction, port_id);
for (i = this->last_port - 1; i >= 0; i--)
if (PORT_VALID(GET_IN_PORT(this, i)))
break;
this->last_port = i + 1;
}
spa_log_debug(this->log, "%p: remove port %d:%d %d", this,
direction, port_id, this->last_port);
spa_node_emit_port_info(&this->hooks, direction, port_id, NULL); spa_node_emit_port_info(&this->hooks, direction, port_id, NULL);
@ -697,6 +697,7 @@ impl_node_port_use_buffers(void *object,
} }
struct io_info { struct io_info {
struct impl *impl;
struct port *port; struct port *port;
void *data; void *data;
size_t size; size_t size;
@ -706,17 +707,29 @@ static int do_port_set_io(struct spa_loop *loop, bool async, uint32_t seq,
const void *data, size_t size, void *user_data) const void *data, size_t size, void *user_data)
{ {
struct io_info *info = user_data; struct io_info *info = user_data;
if (info->size >= sizeof(struct spa_io_async_buffers)) { struct port *port = info->port;
struct spa_io_async_buffers *ab = info->data;
info->port->io[0] = &ab->buffers[info->port->direction]; if (info->data == NULL || info->size < sizeof(struct spa_io_buffers)) {
info->port->io[1] = &ab->buffers[info->port->direction^1]; port->io[0] = NULL;
} else if (info->size >= sizeof(struct spa_io_buffers)) { port->io[1] = NULL;
info->port->io[0] = info->data; if (port->active) {
info->port->io[1] = info->data; spa_list_remove(&port->mix_link);
} else { port->active = false;
info->port->io[0] = NULL; }
info->port->io[1] = NULL; } else {
} if (info->size >= sizeof(struct spa_io_async_buffers)) {
struct spa_io_async_buffers *ab = info->data;
port->io[0] = &ab->buffers[port->direction];
port->io[1] = &ab->buffers[port->direction^1];
} else {
port->io[0] = info->data;
port->io[1] = info->data;
}
if (!port->active) {
spa_list_append(&info->impl->mix_list, &port->mix_link);
port->active = true;
}
}
return 0; return 0;
} }
@ -737,6 +750,7 @@ impl_node_port_set_io(void *object,
spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL); spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
port = GET_PORT(this, direction, port_id); port = GET_PORT(this, direction, port_id);
info.impl = this;
info.port = port; info.port = port;
info.data = data; info.data = data;
info.size = size; info.size = size;
@ -771,9 +785,9 @@ static int impl_node_port_reuse_buffer(void *object, uint32_t port_id, uint32_t
static int impl_node_process(void *object) static int impl_node_process(void *object)
{ {
struct impl *this = object; struct impl *this = object;
struct port *outport; struct port *outport, *inport;
struct spa_io_buffers *outio; struct spa_io_buffers *outio;
uint32_t n_buffers, i, maxsize; uint32_t n_buffers, maxsize;
struct buffer **buffers; struct buffer **buffers;
struct buffer *outb; struct buffer *outb;
const void **datas; const void **datas;
@ -803,24 +817,17 @@ static int impl_node_process(void *object)
maxsize = UINT32_MAX; maxsize = UINT32_MAX;
for (i = 0; i < this->last_port; i++) { spa_list_for_each(inport, &this->mix_list, mix_link) {
struct port *inport = GET_IN_PORT(this, i); struct spa_io_buffers *inio = inport->io[cycle];
struct spa_io_buffers *inio = NULL;
struct buffer *inb; struct buffer *inb;
struct spa_data *bd; struct spa_data *bd;
uint32_t size, offs; uint32_t size, offs;
if (SPA_UNLIKELY(!PORT_VALID(inport) || (inio = inport->io[cycle]) == NULL)) {
spa_log_trace_fp(this->log, "%p: skip input idx:%d valid:%d io:%p/%p/%d",
this, i, PORT_VALID(inport),
inport->io[0], inport->io[1], cycle);
continue;
}
if (inio->buffer_id >= inport->n_buffers || if (inio->buffer_id >= inport->n_buffers ||
inio->status != SPA_STATUS_HAVE_DATA) { inio->status != SPA_STATUS_HAVE_DATA) {
spa_log_trace_fp(this->log, "%p: skip input idx:%d " spa_log_trace_fp(this->log, "%p: skip input idx:%d "
"io:%p status:%d buf_id:%d n_buffers:%d", this, "io:%p status:%d buf_id:%d n_buffers:%d", this,
i, inio, inio->status, inio->buffer_id, inport->n_buffers); inport->id, inio, inio->status, inio->buffer_id, inport->n_buffers);
continue; continue;
} }
@ -832,7 +839,7 @@ static int impl_node_process(void *object)
maxsize = SPA_MIN(size, maxsize); maxsize = SPA_MIN(size, maxsize);
spa_log_trace_fp(this->log, "%p: mix input %d %p->%p %d %d %d:%d/%d", this, spa_log_trace_fp(this->log, "%p: mix input %d %p->%p %d %d %d:%d/%d", this,
i, inio, outio, inio->status, inio->buffer_id, inport->id, inio, outio, inio->status, inio->buffer_id,
offs, size, this->stride); offs, size, this->stride);
if (!SPA_FLAG_IS_SET(bd->chunk->flags, SPA_CHUNK_FLAG_EMPTY)) { if (!SPA_FLAG_IS_SET(bd->chunk->flags, SPA_CHUNK_FLAG_EMPTY)) {
@ -912,14 +919,18 @@ static int impl_get_interface(struct spa_handle *handle, const char *type, void
static int impl_clear(struct spa_handle *handle) static int impl_clear(struct spa_handle *handle)
{ {
struct impl *this; struct impl *this;
uint32_t i; struct port *port;
spa_return_val_if_fail(handle != NULL, -EINVAL); spa_return_val_if_fail(handle != NULL, -EINVAL);
this = (struct impl *) handle; this = (struct impl *) handle;
for (i = 0; i < MAX_PORTS; i++) spa_list_insert_list(&this->free_list, &this->port_list);
free(this->in_ports[i]); spa_list_insert_list(&this->free_list, &this->mix_list);
spa_list_consume(port, &this->free_list, link) {
spa_list_remove(&port->link);
free(port);
}
mix_ops_free(&this->ops); mix_ops_free(&this->ops);
return 0; return 0;
} }
@ -973,6 +984,9 @@ impl_init(const struct spa_handle_factory *factory,
} }
spa_hook_list_init(&this->hooks); spa_hook_list_init(&this->hooks);
spa_list_init(&this->port_list);
spa_list_init(&this->free_list);
spa_list_init(&this->mix_list);
this->node.iface = SPA_INTERFACE_INIT( this->node.iface = SPA_INTERFACE_INIT(
SPA_TYPE_INTERFACE_Node, SPA_TYPE_INTERFACE_Node,
@ -985,7 +999,6 @@ impl_init(const struct spa_handle_factory *factory,
this->info.flags = SPA_NODE_FLAG_RT | SPA_NODE_FLAG_IN_DYNAMIC_PORTS; this->info.flags = SPA_NODE_FLAG_RT | SPA_NODE_FLAG_IN_DYNAMIC_PORTS;
port = GET_OUT_PORT(this, 0); port = GET_OUT_PORT(this, 0);
port->valid = true;
port->direction = SPA_DIRECTION_OUTPUT; port->direction = SPA_DIRECTION_OUTPUT;
port->id = 0; port->id = 0;
port->info = SPA_PORT_INFO_INIT(); port->info = SPA_PORT_INFO_INIT();