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

sco-source: clean up and support different buffer sizes

Browse source
This commit is contained in:
Julian Bouzas 2020-07-07 13:06:11 -04:00
parent f743fff694
commit ba96eecba0
1 changed files with 96 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

225
spa/plugins/bluez5/sco-source.c
View file

@ -55,7 +55,6 @@ struct props {
uint32_t max_latency; uint32_t max_latency;
}; };
#define FILL_FRAMES 2
#define MAX_BUFFERS 32 #define MAX_BUFFERS 32
struct buffer { struct buffer {
@ -83,7 +82,8 @@ struct port {
struct spa_list free; struct spa_list free;
struct spa_list ready; struct spa_list ready;
size_t ready_offset; struct buffer *current_buffer;
uint32_t ready_offset;
}; };
struct impl { struct impl {
@ -109,12 +109,11 @@ struct impl {
struct port port; struct port port;
unsigned int started:1; unsigned int started:1;
unsigned int following:1;
struct spa_source source; struct spa_source source;
struct spa_io_clock *clock; struct spa_io_clock *clock;
struct spa_io_position *position; struct spa_io_position *position;
struct timespec now; struct timespec now;
uint32_t sample_count; uint32_t sample_count;
@ -123,10 +122,10 @@ struct impl {
#define NAME "sco-source" #define NAME "sco-source"
#define CHECK_PORT(this,d,p) ((d) == SPA_DIRECTION_OUTPUT && (p) == 0) #define CHECK_PORT(this,d,p) ((d) == SPA_DIRECTION_OUTPUT && (p) == 0)
static const uint32_t default_min_latency = 64; static const uint32_t default_min_latency = 128;
static const uint32_t default_max_latency = 256; static const uint32_t default_max_latency = 512;
static void reset_props(struct props *props) static void reset_props(struct props *props)
{ {
@ -148,10 +147,10 @@ static int impl_node_enum_params(void *object, int seq,
spa_return_val_if_fail(this != NULL, -EINVAL); spa_return_val_if_fail(this != NULL, -EINVAL);
spa_return_val_if_fail(num != 0, -EINVAL); spa_return_val_if_fail(num != 0, -EINVAL);
result.id = id; result.id = id;
result.next = start; result.next = start;
next: next:
result.index = result.next++; result.index = result.next++;
spa_pod_builder_init(&b, buffer, sizeof(buffer)); spa_pod_builder_init(&b, buffer, sizeof(buffer));
@ -211,25 +210,9 @@ static int impl_node_enum_params(void *object, int seq,
return 0; return 0;
} }
static int do_reassign_follower(struct spa_loop *loop,
bool async,
uint32_t seq,
const void *data,
size_t size,
void *user_data)
{
return 0;
}
static inline bool is_following(struct impl *this)
{
return this->position && this->clock && this->position->clock.id != this->clock->id;
}
static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size) static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
{ {
struct impl *this = object; struct impl *this = object;
bool following;
spa_return_val_if_fail(this != NULL, -EINVAL); spa_return_val_if_fail(this != NULL, -EINVAL);
@ -244,12 +227,6 @@ static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
return -ENOENT; return -ENOENT;
} }
following = is_following(this);
if (this->started && following != this->following) {
spa_log_debug(this->log, "sco-source %p: reassign follower %d->%d", this, this->following, following);
this->following = following;
spa_loop_invoke(this->data_loop, do_reassign_follower, 0, NULL, 0, true, this);
}
return 0; return 0;
} }
@ -296,36 +273,26 @@ static void reset_buffers(struct port *port)
} }
} }
static bool read_data(struct impl *this, uint8_t *data, uint32_t size, uint32_t *total_read) static int read_data(struct impl *this, uint8_t *data, uint32_t data_size)
{ {
const uint32_t mtu_size = this->read_mtu; int res = 0;
uint32_t local_total_read = 0;
/* Read chunks of mtu_size */ again:
while (local_total_read <= (size - mtu_size)) { res = read(this->sock_fd, data, data_size);
const int bytes_read = read(this->sock_fd, data, mtu_size); if (res <= 0) {
if (bytes_read < 0) { /* retry if interrupted */
/* Retry */ if (errno == EINTR)
if (errno == EINTR) goto again;
continue;
/* Socked has no data */ /* return socked has no data */
if (errno == EAGAIN || errno == EWOULDBLOCK) if (errno == EAGAIN || errno == EWOULDBLOCK)
goto done; return res;
/* Error */ /* error */
spa_log_error(this->log, "read error: %s", strerror(errno)); return -errno;
return false;
}
data += bytes_read;
local_total_read += bytes_read;
} }
done: return res;
if (total_read)
*total_read = local_total_read;
return true;
} }
static void recycle_buffer(struct impl *this, struct port *port, uint32_t buffer_id) static void recycle_buffer(struct impl *this, struct port *port, uint32_t buffer_id)
@ -344,70 +311,81 @@ static void sco_on_ready_read(struct spa_source *source)
struct impl *this = source->data; struct impl *this = source->data;
struct port *port = &this->port; struct port *port = &this->port;
struct spa_io_buffers *io = port->io; struct spa_io_buffers *io = port->io;
int32_t io_done_status = io->status; int size_read;
struct buffer *buffer; struct spa_data *datas;
struct spa_data *buffer_data;
uint32_t total_read;
spa_return_if_fail(io != NULL); /* make sure the source has input data */
if ((source->rmask & SPA_IO_IN) == 0) {
spa_log_error(this->log, "source has no input data, rmask=%d", source->rmask);
goto stop;
}
if (this->transport == NULL) {
spa_log_debug(this->log, "no transport, stop reading");
goto stop;
}
/* Read a buffer if there is one free */ /* get buffer */
if (!spa_list_is_empty(&port->free)) { if (!port->current_buffer) {
/* Get the free buffer and remove it from the free list */ if (spa_list_is_empty(&port->free)) {
buffer = spa_list_first(&port->free, struct buffer, link); spa_log_warn(this->log, "buffer not available");
spa_list_remove(&buffer->link);
buffer_data = &buffer->buf->datas[0];
spa_assert(buffer_data->data);
/* Read sco data */
if (!read_data(this, buffer_data->data, buffer_data->maxsize, &total_read)) {
if (this->source.loop)
spa_loop_remove_source(this->data_loop, &this->source);
return; return;
} }
port->current_buffer = spa_list_first(&port->free, struct buffer, link);
spa_list_remove(&port->current_buffer->link);
port->ready_offset = 0;
}
datas = port->current_buffer->buf->datas;
/* Append a ready buffer if data could be read */ /* read */
if (total_read > 0) { size_read = read_data(this, (uint8_t *)datas[0].data + port->ready_offset, datas[0].maxsize);
/* Update the buffer offset, size and stride */ if (size_read < 0) {
buffer_data->chunk->offset = 0; spa_log_error(this->log, "failed to read data");
buffer_data->chunk->size = total_read; goto stop;
buffer_data->chunk->stride = port->frame_size; }
spa_log_debug(this->log, "read socket data %d", size_read);
/* Update the sample count */ /* send buffer if full */
this->sample_count += buffer_data->chunk->size / port->frame_size; port->ready_offset += size_read;
if ((this->read_mtu + port->ready_offset) > (this->props.max_latency * port->frame_size)) {
datas[0].chunk->offset = 0;
datas[0].chunk->size = port->ready_offset;
datas[0].chunk->stride = port->frame_size;
/* Add the buffer to the ready list */ this->sample_count += datas[0].chunk->size / port->frame_size;
buffer->outstanding = true; spa_list_append(&port->ready, &port->current_buffer->link);
spa_list_append(&port->ready, &buffer->link); port->current_buffer = NULL;
}
} }
/* Process a buffer if there is one ready and IO does not have one */ /* done if there are no buffers ready */
if (!spa_list_is_empty(&port->ready) && io->status != SPA_STATUS_HAVE_DATA) { if (spa_list_is_empty(&port->ready))
/* Get the ready buffer and remove it from the ready list */ return;
buffer = spa_list_first(&port->ready, struct buffer, link);
spa_list_remove(&buffer->link);
/* Mark the buffer to be processed */ /* process the buffer if IO does not have any */
io->buffer_id = buffer->id; if (io->status != SPA_STATUS_HAVE_DATA) {
struct buffer *b;
if (io->buffer_id < port->n_buffers)
recycle_buffer(this, port, io->buffer_id);
b = spa_list_first(&port->ready, struct buffer, link);
spa_list_remove(&b->link);
b->outstanding = true;
io->buffer_id = b->id;
io->status = SPA_STATUS_HAVE_DATA; io->status = SPA_STATUS_HAVE_DATA;
/* Add the buffer to the free list */
spa_list_append(&port->free, &buffer->link);
buffer->outstanding = false;
/* Set the done status as have buffer */
io_done_status = SPA_STATUS_HAVE_DATA;
} }
/* Notify the current status */ /* notify ready */
spa_node_call_ready(&this->callbacks, io_done_status); spa_node_call_ready(&this->callbacks, SPA_STATUS_HAVE_DATA);
return;
stop:
if (this->source.loop)
spa_loop_remove_source(this->data_loop, &this->source);
} }
static int do_start(struct impl *this) static int do_start(struct impl *this)
{ {
int val;
bool do_accept; bool do_accept;
/* Dont do anything if the node has already started */ /* Dont do anything if the node has already started */
@ -425,22 +403,6 @@ static int do_start(struct impl *this)
if (this->sock_fd < 0) if (this->sock_fd < 0)
return -1; return -1;
/* Set the write MTU */
val = FILL_FRAMES * this->transport->write_mtu;
if (setsockopt(this->sock_fd, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val)) < 0)
spa_log_warn(this->log, "sco-source %p: SO_SNDBUF %m", this);
/* Set the read MTU */
this->read_mtu = this->transport->read_mtu;
val = FILL_FRAMES * this->read_mtu;
if (setsockopt(this->sock_fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val)) < 0)
spa_log_warn(this->log, "sco-source %p: SO_RCVBUF %m", this);
/* Set the priority */
val = 6;
if (setsockopt(this->sock_fd, SOL_SOCKET, SO_PRIORITY, &val, sizeof(val)) < 0)
spa_log_warn(this->log, "SO_PRIORITY failed: %m");
/* Reset the buffers and sample count */ /* Reset the buffers and sample count */
reset_buffers(&this->port); reset_buffers(&this->port);
this->sample_count = 0; this->sample_count = 0;
@ -641,7 +603,7 @@ impl_node_port_enum_params(void *object, int seq,
result.id = id; result.id = id;
result.next = start; result.next = start;
next: next:
result.index = result.next++; result.index = result.next++;
spa_pod_builder_init(&b, buffer, sizeof(buffer)); spa_pod_builder_init(&b, buffer, sizeof(buffer));
@ -686,10 +648,12 @@ impl_node_port_enum_params(void *object, int seq,
param = spa_pod_builder_add_object(&b, param = spa_pod_builder_add_object(&b,
SPA_TYPE_OBJECT_ParamBuffers, id, SPA_TYPE_OBJECT_ParamBuffers, id,
/* 8 buffers are enough to make sure we always have one available when decoding */ SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(2, 1, MAX_BUFFERS),
SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(8, 8, MAX_BUFFERS),
SPA_PARAM_BUFFERS_blocks, SPA_POD_Int(1), SPA_PARAM_BUFFERS_blocks, SPA_POD_Int(1),
SPA_PARAM_BUFFERS_size, SPA_POD_Int(this->props.max_latency * port->frame_size), SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int(
this->props.max_latency * port->frame_size,
this->props.min_latency * port->frame_size,
INT32_MAX),
SPA_PARAM_BUFFERS_stride, SPA_POD_Int(port->frame_size), SPA_PARAM_BUFFERS_stride, SPA_POD_Int(port->frame_size),
SPA_PARAM_BUFFERS_align, SPA_POD_Int(16)); SPA_PARAM_BUFFERS_align, SPA_POD_Int(16));
break; break;
@ -932,22 +896,25 @@ static int impl_node_process(void *object)
if (io->status == SPA_STATUS_HAVE_DATA) if (io->status == SPA_STATUS_HAVE_DATA)
return SPA_STATUS_HAVE_DATA; return SPA_STATUS_HAVE_DATA;
/* Return if there is not buffers ready to be processed */ /* Recycle */
if (io->buffer_id < port->n_buffers) {
recycle_buffer(this, port, io->buffer_id);
io->buffer_id = SPA_ID_INVALID;
}
/* Return if there are no buffers ready to be processed */
if (spa_list_is_empty(&port->ready)) if (spa_list_is_empty(&port->ready))
return io->status; return SPA_STATUS_OK;
/* Get the new buffer from the ready list */ /* Get the new buffer from the ready list */
buffer = spa_list_first(&port->ready, struct buffer, link); buffer = spa_list_first(&port->ready, struct buffer, link);
spa_list_remove(&buffer->link); spa_list_remove(&buffer->link);
buffer->outstanding = false;
/* Set the new buffer in IO */ /* Set the new buffer in IO */
io->buffer_id = buffer->id; io->buffer_id = buffer->id;
io->status = SPA_STATUS_HAVE_DATA; io->status = SPA_STATUS_HAVE_DATA;
/* Add the buffer to the free list */
spa_list_append(&port->free, &buffer->link);
buffer->outstanding = false;
/* Notify we have a buffer ready to be processed */ /* Notify we have a buffer ready to be processed */
return SPA_STATUS_HAVE_DATA; return SPA_STATUS_HAVE_DATA;
} }
@ -980,7 +947,7 @@ static void transport_destroy(void *data)
static const struct spa_bt_transport_events transport_events = { static const struct spa_bt_transport_events transport_events = {
SPA_VERSION_BT_TRANSPORT_EVENTS, SPA_VERSION_BT_TRANSPORT_EVENTS,
.destroy = transport_destroy, .destroy = transport_destroy,
}; };
static int impl_get_interface(struct spa_handle *handle, const char *type, void **interface) static int impl_get_interface(struct spa_handle *handle, const char *type, void **interface)