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

jack: work on mixing

Browse source
This commit is contained in:
Wim Taymans 2017-08-25 10:01:04 +02:00
parent 68f5c05b9f
commit b7b80683bc
7 changed files with 195 additions and 95 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/daemon/pipewire.conf.in
View file

@ -1,11 +1,11 @@
#load-module libpipewire-module-protocol-dbus
load-module libpipewire-module-protocol-native
load-module libpipewire-module-suspend-on-idle
#load-module libpipewire-module-spa-monitor alsa/libspa-alsa alsa-monitor alsa
load-module libpipewire-module-spa-monitor alsa/libspa-alsa alsa-monitor alsa
load-module libpipewire-module-spa-monitor v4l2/libspa-v4l2 v4l2-monitor v4l2
#load-module libpipewire-module-spa-node videotestsrc/libspa-videotestsrc videotestsrc videotestsrc media.class=Video/Source Spa:POD:Object:Props:patternType=Spa:POD:Object:Props:patternType:snow
load-module libpipewire-module-autolink
#load-module libpipewire-module-mixer
load-module libpipewire-module-client-node
load-module libpipewire-module-flatpak
#load-module libpipewire-module-jack
load-module libpipewire-module-jack

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

@ -165,10 +165,10 @@ struct pw_client_node_buffer {
#define PW_CLIENT_NODE_PROXY_METHOD_DONE 0
#define PW_CLIENT_NODE_PROXY_METHOD_UPDATE 1
#define PW_CLIENT_NODE_PROXY_METHOD_PORT_UPDATE 2
#define PW_CLIENT_NODE_PROXY_METHOD_PORT_UPDATE 2
#define PW_CLIENT_NODE_PROXY_METHOD_EVENT 3
#define PW_CLIENT_NODE_PROXY_METHOD_DESTROY 4
#define PW_CLIENT_NODE_PROXY_METHOD_NUM 5
#define PW_CLIENT_NODE_PROXY_METHOD_NUM 5
/** \ref pw_client_node methods */
struct pw_client_node_proxy_methods {

173
src/modules/module-jack/jack-node.c
View file

@ -77,7 +77,7 @@ struct node_data {
struct spa_hook_list listener_list;
int port_count;
int port_count[2];
int status;
};
@ -102,10 +102,16 @@ struct port_data {
struct spa_port_io *io;
bool have_buffers;
struct buffer buffers[64];
uint32_t n_buffers;
struct spa_list empty;
struct spa_buffer *bufs[1];
struct spa_buffer buf;
struct spa_data data[1];
struct spa_chunk chunk[1];
uint8_t buffer[1024];
};
@ -156,33 +162,7 @@ static void recycle_buffer(struct pw_jack_node *this, struct port_data *pd, uint
static int driver_process_input(void *data)
{
struct pw_jack_node *this = data;
struct spa_graph_node *node = &this->node->rt.node;
struct spa_graph_port *p;
struct buffer *out;
struct port_data *opd = SPA_CONTAINER_OF(this->driverport, struct port_data, port);
struct spa_port_io *out_io = opd->io;
pw_log_trace("process input");
if (out_io->status == SPA_RESULT_HAVE_BUFFER)
return SPA_RESULT_HAVE_BUFFER;
out = buffer_dequeue(this, opd);
if (out == NULL)
return SPA_RESULT_OUT_OF_BUFFERS;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
struct pw_port *port = p->callbacks_data;
struct port_data *ipd = pw_port_get_user_data(port);
struct spa_port_io *in_io = ipd->io;
in_io->status = SPA_RESULT_NEED_BUFFER;
}
out_io->buffer_id = out->outbuf->id;
out_io->status = SPA_RESULT_HAVE_BUFFER;
return SPA_RESULT_HAVE_BUFFER;
return SPA_RESULT_NOT_IMPLEMENTED;
}
static void conv_f32_s16(int16_t *out, float *in, int n_samples, int stride)
@ -206,6 +186,12 @@ static void fill_s16(int16_t *out, int n_samples, int stride)
out += stride;
}
}
static void add_f32(float *out, float *in, int n_samples)
{
int i;
for (i = 0; i < n_samples; i++)
out[i] += in[i];
}
static int driver_process_output(void *data)
{
@ -213,7 +199,7 @@ static int driver_process_output(void *data)
struct pw_jack_node *this = &nd->node;
struct spa_graph_node *node = &this->node->rt.node;
struct spa_graph_port *p;
struct port_data *opd = SPA_CONTAINER_OF(this->driverport, struct port_data, port);
struct port_data *opd = SPA_CONTAINER_OF(this->driver_out, struct port_data, port);
struct spa_port_io *out_io = opd->io;
struct jack_engine_control *ctrl = this->server->engine_control;
struct buffer *out;
@ -224,7 +210,7 @@ static int driver_process_output(void *data)
if (out_io->status == SPA_RESULT_HAVE_BUFFER)
return SPA_RESULT_HAVE_BUFFER;
if (out_io->buffer_id != SPA_ID_INVALID) {
if (out_io->buffer_id < opd->n_buffers) {
recycle_buffer(this, opd, out_io->buffer_id);
out_io->buffer_id = SPA_ID_INVALID;
}
@ -247,7 +233,7 @@ static int driver_process_output(void *data)
struct buffer *in;
int stride = 2;
if (in_io->buffer_id != SPA_ID_INVALID && in_io->status == SPA_RESULT_HAVE_BUFFER) {
if (in_io->buffer_id < ipd->n_buffers && in_io->status == SPA_RESULT_HAVE_BUFFER) {
in = &ipd->buffers[in_io->buffer_id];
conv_f32_s16(op, in->ptr, ctrl->buffer_size, stride);
}
@ -437,6 +423,10 @@ static int port_use_buffers(void *data, struct spa_buffer **buffers, uint32_t n_
struct type *t = &pd->node->type;
int i;
if (pd->have_buffers)
return SPA_RESULT_OK;
pw_log_debug("use_buffers %d", n_buffers);
for (i = 0; i < n_buffers; i++) {
struct buffer *b;
struct spa_data *d = buffers[i]->datas;
@ -466,6 +456,7 @@ static int port_alloc_buffers(void *data,
struct type *t = &pd->node->type;
int i;
pw_log_debug("alloc %d", *n_buffers);
for (i = 0; i < *n_buffers; i++) {
struct buffer *b;
struct spa_data *d = buffers[i]->datas;
@ -473,6 +464,7 @@ static int port_alloc_buffers(void *data,
b = &pd->buffers[i];
b->outbuf = buffers[i];
d[0].type = t->data.MemPtr;
d[0].maxsize = pd->node->node.server->engine_control->buffer_size;
b->ptr = d[0].data = pd->port.ptr;
spa_list_append(&pd->empty, &b->link);
}
@ -506,6 +498,64 @@ static const struct pw_port_implementation port_impl = {
.send_command = port_send_command,
};
static int schedule_mix_input(void *data)
{
struct pw_jack_port *this = data;
struct spa_graph_node *node = &this->port->rt.mix_node;
struct spa_graph_port *p;
struct spa_port_io *io = this->port->rt.mix_port.io;
struct port_data *pd = SPA_CONTAINER_OF(this, struct port_data, port);
size_t buffer_size = pd->node->node.server->engine_control->buffer_size;
int layer = 0;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link) {
struct pw_link *link = p->callbacks_data;
struct spa_buffer *inbuf;
pw_log_trace("mix %p: input %d %d", node, p->io->buffer_id, link->output->n_buffers);
if (!(p->io->buffer_id < link->output->n_buffers && p->io->status == SPA_RESULT_HAVE_BUFFER))
continue;
inbuf = link->output->buffers[p->io->buffer_id];
if (layer++ == 0)
memcpy(pd->buffers[0].ptr, inbuf->datas[0].data, buffer_size * sizeof(float));
else
add_f32(pd->buffers[0].ptr, inbuf->datas[0].data, buffer_size);
pw_log_trace("mix %p: input %p %p->%p %d %d", node,
p, p->io, io, p->io->status, p->io->buffer_id);
*io = *p->io;
io->buffer_id = 0;
p->io->status = SPA_RESULT_OK;
p->io->buffer_id = SPA_ID_INVALID;
}
return SPA_RESULT_HAVE_BUFFER;
}
static int schedule_mix_output(void *data)
{
struct pw_jack_port *this = data;
struct spa_graph_node *node = &this->port->rt.mix_node;
struct spa_graph_port *p;
struct spa_port_io *io = this->port->rt.mix_port.io;
io->status = SPA_RESULT_NEED_BUFFER;
spa_list_for_each(p, &node->ports[SPA_DIRECTION_INPUT], link)
*p->io = *io;
io->buffer_id = SPA_ID_INVALID;
return SPA_RESULT_NEED_BUFFER;
}
static const struct spa_graph_node_callbacks schedule_mix_node = {
SPA_VERSION_GRAPH_NODE_CALLBACKS,
schedule_mix_input,
schedule_mix_output,
};
static void port_destroy(void *data)
{
struct port_data *pd = data;
@ -539,14 +589,25 @@ static void port_free(void *data)
spa_hook_list_call(&pd->listener_list, struct pw_jack_port_events, free);
}
static void port_link_added(void *data, struct pw_link *link)
{
}
static void port_link_removed(void *data, struct pw_link *link)
{
}
static const struct pw_port_events port_events = {
PW_VERSION_PORT_EVENTS,
.destroy = port_destroy,
.free = port_free,
.link_added = port_link_added,
.link_removed = port_link_removed,
};
struct pw_jack_port *
alloc_port(struct pw_jack_node *node, enum pw_direction direction, uint32_t port_id, size_t user_data_size)
alloc_port(struct pw_jack_node *node, enum pw_direction direction,
uint32_t port_id, size_t user_data_size)
{
struct node_data *nd = SPA_CONTAINER_OF(node, struct node_data, node);
struct pw_port *p;
@ -561,8 +622,8 @@ alloc_port(struct pw_jack_node *node, enum pw_direction direction, uint32_t port
pd->node = nd;
spa_hook_list_init(&pd->listener_list);
spa_list_init(&pd->empty);
port = &pd->port;
port = &pd->port;
port->node = node;
port->direction = direction;
port->port = p;
@ -573,7 +634,6 @@ alloc_port(struct pw_jack_node *node, enum pw_direction direction, uint32_t port
pw_port_add_listener(p, &pd->port_listener, &port_events, pd);
pw_port_set_implementation(p, &port_impl, pd);
pw_port_add(p, node->node);
return port;
}
@ -590,6 +650,7 @@ pw_jack_node_add_port(struct pw_jack_node *node,
struct jack_graph_manager *mgr = server->graph_manager;
struct jack_connection_manager *conn;
struct pw_jack_port *port;
struct port_data *pd;
jack_port_type_id_t type_id;
jack_port_id_t port_id;
enum pw_direction direction;
@ -611,7 +672,7 @@ pw_jack_node_add_port(struct pw_jack_node *node,
return NULL;
}
port = alloc_port(node, direction, nd->port_count++, user_data_size);
port = alloc_port(node, direction, nd->port_count[direction]++, user_data_size);
if (port == NULL)
return NULL;
@ -619,6 +680,8 @@ pw_jack_node_add_port(struct pw_jack_node *node,
port->jack_port = jack_graph_manager_get_port(mgr, port_id);
port->ptr = (float *)((uintptr_t)port->jack_port->buffer & ~31L) + 8;
pd = SPA_CONTAINER_OF(port, struct port_data, port);
conn = jack_graph_manager_next_start(mgr);
if (direction == PW_DIRECTION_INPUT)
jack_connection_manager_add_inport(conn, ref_num, port_id);
@ -626,6 +689,33 @@ pw_jack_node_add_port(struct pw_jack_node *node,
jack_connection_manager_add_outport(conn, ref_num, port_id);
jack_graph_manager_next_stop(mgr);
{
struct spa_buffer *b = &pd->buf;
struct type *t = &pd->node->type;
pd->bufs[0] = b;
b->id = 0;
b->n_metas = 0;
b->metas = NULL;
b->n_datas = 1;
b->datas = pd->data;
pd->data[0].data = pd->port.ptr;
pd->data[0].chunk = pd->chunk;
pd->data[0].type = t->data.MemPtr;
pd->data[0].maxsize = pd->node->node.server->engine_control->buffer_size;
port->port->state = PW_PORT_STATE_READY;
pw_port_use_buffers(port->port, pd->bufs, 1);
pd->have_buffers = true;
port->port->state = PW_PORT_STATE_PAUSED;
}
if (direction == PW_DIRECTION_INPUT) {
port->port->mix = port;
spa_graph_node_set_callbacks(&port->port->rt.mix_node, &schedule_mix_node, port);
}
pw_port_add(port->port, node->node);
return port;
}
@ -816,11 +906,14 @@ pw_jack_driver_new(struct pw_core *core,
}
jack_graph_manager_next_stop(mgr);
if (n_capture_channels > 0)
this->driverport = alloc_port(this, PW_DIRECTION_INPUT, 0, 0);
if (n_playback_channels > 0)
this->driverport = alloc_port(this, PW_DIRECTION_OUTPUT, 0, 0);
if (n_capture_channels > 0) {
this->driver_in = alloc_port(this, PW_DIRECTION_INPUT, 0, 0);
pw_port_add(this->driver_in->port, node);
}
if (n_playback_channels > 0) {
this->driver_out = alloc_port(this, PW_DIRECTION_OUTPUT, 0, 0);
pw_port_add(this->driver_out->port, node);
}
pw_node_register(node);
return this;

3
src/modules/module-jack/jack-node.h
View file

@ -38,7 +38,8 @@ struct pw_jack_node {
struct jack_client_control *control;
struct pw_jack_port *driverport;
struct pw_jack_port *driver_out;
struct pw_jack_port *driver_in;
struct spa_list graph_link;

91
src/pipewire/link.c
View file

@ -50,11 +50,6 @@ struct impl {
struct spa_hook input_node_listener;
struct spa_hook output_port_listener;
struct spa_hook output_node_listener;
void *buffer_owner;
struct pw_memblock buffer_mem;
struct spa_buffer **buffers;
uint32_t n_buffers;
};
struct resource_data {
@ -497,7 +492,7 @@ static int do_allocation(struct pw_link *this, uint32_t in_state, uint32_t out_s
spa_debug_port_info(iinfo);
}
if (impl->buffers == NULL) {
if (this->buffers == NULL) {
struct spa_param **params, *param;
uint8_t buffer[4096];
struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
@ -564,18 +559,18 @@ static int do_allocation(struct pw_link *this, uint32_t in_state, uint32_t out_s
if (this->output->n_buffers) {
out_flags = 0;
in_flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
impl->n_buffers = this->output->n_buffers;
impl->buffers = this->output->buffers;
impl->buffer_owner = this->output;
pw_log_debug("reusing %d output buffers %p", impl->n_buffers,
impl->buffers);
} else if (this->input->n_buffers) {
this->n_buffers = this->output->n_buffers;
this->buffers = this->output->buffers;
this->buffer_owner = this->output;
pw_log_debug("reusing %d output buffers %p", this->n_buffers,
this->buffers);
} else if (this->input->n_buffers && this->input->mix == NULL) {
out_flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
in_flags = 0;
impl->n_buffers = this->input->n_buffers;
impl->buffers = this->input->buffers;
impl->buffer_owner = this->input;
pw_log_debug("reusing %d input buffers %p", impl->n_buffers, impl->buffers);
this->n_buffers = this->input->n_buffers;
this->buffers = this->input->buffers;
this->buffer_owner = this->input;
pw_log_debug("reusing %d input buffers %p", this->n_buffers, this->buffers);
} else {
size_t data_sizes[1];
ssize_t data_strides[1];
@ -583,64 +578,64 @@ static int do_allocation(struct pw_link *this, uint32_t in_state, uint32_t out_s
data_sizes[0] = minsize;
data_strides[0] = stride;
impl->buffer_owner = this;
impl->n_buffers = max_buffers;
impl->buffers = alloc_buffers(this,
impl->n_buffers,
this->buffer_owner = this;
this->n_buffers = max_buffers;
this->buffers = alloc_buffers(this,
this->n_buffers,
n_params,
params,
1,
data_sizes, data_strides,
&impl->buffer_mem);
&this->buffer_mem);
pw_log_debug("allocating %d input buffers %p %zd %zd", impl->n_buffers,
impl->buffers, minsize, stride);
pw_log_debug("allocating %d input buffers %p %zd %zd", this->n_buffers,
this->buffers, minsize, stride);
}
if (out_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS) {
if ((res = pw_port_alloc_buffers(this->output,
params, n_params,
impl->buffers, &impl->n_buffers)) < 0) {
this->buffers, &this->n_buffers)) < 0) {
asprintf(&error, "error alloc output buffers: %d", res);
goto error;
}
if (SPA_RESULT_IS_ASYNC(res))
pw_work_queue_add(impl->work, this->output->node, res, complete_paused,
this->output);
this->output->buffer_mem = impl->buffer_mem;
impl->buffer_owner = this->output;
pw_log_debug("allocated %d buffers %p from output port", impl->n_buffers,
impl->buffers);
this->output->buffer_mem = this->buffer_mem;
this->buffer_owner = this->output;
pw_log_debug("allocated %d buffers %p from output port", this->n_buffers,
this->buffers);
} else if (in_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS) {
if ((res = pw_port_alloc_buffers(this->input,
params, n_params,
impl->buffers, &impl->n_buffers)) < 0) {
this->buffers, &this->n_buffers)) < 0) {
asprintf(&error, "error alloc input buffers: %d", res);
goto error;
}
if (SPA_RESULT_IS_ASYNC(res))
pw_work_queue_add(impl->work, this->input->node, res, complete_paused,
this->input);
this->input->buffer_mem = impl->buffer_mem;
impl->buffer_owner = this->input;
pw_log_debug("allocated %d buffers %p from input port", impl->n_buffers,
impl->buffers);
this->input->buffer_mem = this->buffer_mem;
this->buffer_owner = this->input;
pw_log_debug("allocated %d buffers %p from input port", this->n_buffers,
this->buffers);
}
}
if (in_flags & SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS) {
pw_log_debug("using %d buffers %p on input port", impl->n_buffers, impl->buffers);
pw_log_debug("using %d buffers %p on input port", this->n_buffers, this->buffers);
if ((res = pw_port_use_buffers(this->input,
impl->buffers, impl->n_buffers)) < 0) {
this->buffers, this->n_buffers)) < 0) {
asprintf(&error, "error use input buffers: %d", res);
goto error;
}
if (SPA_RESULT_IS_ASYNC(res))
pw_work_queue_add(impl->work, this->input->node, res, complete_paused, this->input);
} else if (out_flags & SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS) {
pw_log_debug("using %d buffers %p on output port", impl->n_buffers, impl->buffers);
pw_log_debug("using %d buffers %p on output port", this->n_buffers, this->buffers);
if ((res = pw_port_use_buffers(this->output,
impl->buffers, impl->n_buffers)) < 0) {
this->buffers, this->n_buffers)) < 0) {
asprintf(&error, "error use output buffers: %d", res);
goto error;
}
@ -784,9 +779,7 @@ output_node_async_complete(void *data, uint32_t seq, int res)
static void clear_port_buffers(struct pw_link *link, struct pw_port *port)
{
struct impl *impl = SPA_CONTAINER_OF(link, struct impl, this);
if (impl->buffer_owner != port)
if (link->buffer_owner != port)
pw_port_use_buffers(port, NULL, 0);
}
@ -838,7 +831,6 @@ static void output_remove(struct pw_link *this, struct pw_port *port)
static void on_port_destroy(struct pw_link *this, struct pw_port *port)
{
struct impl *impl = (struct impl *) this;
struct pw_port *other;
if (port == this->input) {
@ -850,13 +842,13 @@ static void on_port_destroy(struct pw_link *this, struct pw_port *port)
} else
return;
if (impl->buffer_owner == port) {
impl->buffers = NULL;
impl->n_buffers = 0;
if (this->buffer_owner == port) {
this->buffers = NULL;
this->n_buffers = 0;
pw_log_debug("link %p: clear allocated buffers on port %p", this, other);
pw_port_use_buffers(other, NULL, 0);
impl->buffer_owner = NULL;
this->buffer_owner = NULL;
}
spa_hook_list_call(&this->listener_list, struct pw_link_events, port_unlinked, port);
@ -1114,6 +1106,9 @@ struct pw_link *pw_link_new(struct pw_core *core,
0,
&this->io);
this->rt.in_port.callbacks_data = this;
this->rt.out_port.callbacks_data = this;
pw_loop_invoke(output_node->data_loop,
do_add_link,
SPA_ID_INVALID, sizeof(struct pw_port *), &output, false, this);
@ -1176,9 +1171,9 @@ void pw_link_destroy(struct pw_link *link)
if (link->info.format)
free(link->info.format);
if (impl->buffer_owner == link) {
free(impl->buffers);
pw_memblock_free(&impl->buffer_mem);
if (link->buffer_owner == link) {
free(link->buffers);
pw_memblock_free(&link->buffer_mem);
}
free(impl);
}

10
src/pipewire/port.c
View file

@ -106,6 +106,7 @@ static int schedule_mix_input(void *data)
*io = *p->io;
p->io->status = SPA_RESULT_OK;
p->io->buffer_id = SPA_ID_INVALID;
break;
}
return SPA_RESULT_HAVE_BUFFER;
}
@ -292,7 +293,8 @@ void pw_port_add(struct pw_port *port, struct pw_node *node)
port->rt.graph = node->rt.graph;
pw_loop_invoke(node->data_loop, do_add_port, SPA_ID_INVALID, 0, NULL, false, port);
port_update_state(port, PW_PORT_STATE_CONFIGURE);
if (port->state <= PW_PORT_STATE_INIT)
port_update_state(port, PW_PORT_STATE_CONFIGURE);
spa_hook_list_call(&node->listener_list, struct pw_node_events, port_added, port);
}
@ -480,7 +482,7 @@ int pw_port_use_buffers(struct pw_port *port, struct spa_buffer **buffers, uint3
port->n_buffers = n_buffers;
port->allocated = false;
if (port->n_buffers == 0)
if (n_buffers == 0)
port_update_state (port, PW_PORT_STATE_READY);
else if (!SPA_RESULT_IS_ASYNC(res))
port_update_state (port, PW_PORT_STATE_PAUSED);
@ -512,6 +514,10 @@ int pw_port_alloc_buffers(struct pw_port *port,
else
res = SPA_RESULT_NOT_IMPLEMENTED;
if (port->allocated) {
free(port->buffers);
pw_memblock_free(&port->buffer_mem);
}
port->buffers = buffers;
port->n_buffers = *n_buffers;
port->allocated = true;

5
src/pipewire/private.h
View file

@ -180,6 +180,11 @@ struct pw_link {
struct spa_hook_list listener_list;
void *buffer_owner;
struct pw_memblock buffer_mem;
struct spa_buffer **buffers;
uint32_t n_buffers;
struct {
struct spa_graph_port out_port;
struct spa_graph_port in_port;