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

pulse: improve handling of buffer attributes

Browse source 
Repect minreq and fragsize for playback and capture.
Try to configure tlength at 2 seconds, minreq at around 25ms. Should
greatly improve compatibility with audacious and mpv.
Count since_underrun correctly.

Fixes #278
This commit is contained in:
Wim Taymans 2020-09-11 12:50:44 +02:00
parent 6eb4b552ad
commit 8f75056689
2 changed files with 74 additions and 84 deletions
Download patch file Download diff file Expand all files Collapse all files

3
pipewire-pulseaudio/src/internal.h
View file

@ -401,9 +401,6 @@ struct pa_mem {
void *user_data; void *user_data;
}; };
#define MAX_BUFFERS 64u
#define MASK_BUFFERS (MAX_BUFFERS-1)
struct pa_stream { struct pa_stream {
struct spa_list link; struct spa_list link;
int refcount; int refcount;

149
pipewire-pulseaudio/src/stream.c
View file

@ -33,14 +33,17 @@
#include "core-format.h" #include "core-format.h"
#include "internal.h" #include "internal.h"
#define MIN_QUEUED 1 #define MIN_BUFFERS 8u
#define MAX_BUFFERS 64u
#define MAX_SIZE (4*1024*1024)
#define BLOCK_SIZE (64*1024)
#define MAX_BUFFER_SAMPLES (8*1024u)
#define MAX_SIZE (4*1024*1024u)
static void dump_buffer_attr(pa_stream *s, pa_buffer_attr *attr) static void dump_buffer_attr(pa_stream *s, pa_buffer_attr *attr)
{ {
char b[1024];
pw_log_debug("stream %p: sample: %s", s, pa_sample_spec_snprint(b, sizeof(b), &s->sample_spec));
pw_log_debug("stream %p: stride: %zu", s, pa_frame_size(&s->sample_spec));
pw_log_debug("stream %p: maxlength: %u", s, attr->maxlength); pw_log_debug("stream %p: maxlength: %u", s, attr->maxlength);
pw_log_debug("stream %p: tlength: %u", s, attr->tlength); pw_log_debug("stream %p: tlength: %u", s, attr->tlength);
pw_log_debug("stream %p: minreq: %u", s, attr->minreq); pw_log_debug("stream %p: minreq: %u", s, attr->minreq);
@ -48,15 +51,6 @@ static void dump_buffer_attr(pa_stream *s, pa_buffer_attr *attr)
pw_log_debug("stream %p: fragsize: %u", s, attr->fragsize); pw_log_debug("stream %p: fragsize: %u", s, attr->fragsize);
} }
static void configure_buffers(pa_stream *s)
{
s->buffer_attr.maxlength = MAX_SIZE;
if (s->buffer_attr.prebuf == (uint32_t)-1)
s->buffer_attr.prebuf = s->buffer_attr.minreq;
s->buffer_attr.fragsize = s->buffer_attr.minreq;
dump_buffer_attr(s, &s->buffer_attr);
}
static void configure_device(pa_stream *s) static void configure_device(pa_stream *s)
{ {
struct global *g; struct global *g;
@ -131,7 +125,6 @@ static void stream_state_changed(void *data, enum pw_stream_state old,
break; break;
case PW_STREAM_STATE_STREAMING: case PW_STREAM_STATE_STREAMING:
configure_device(s); configure_device(s);
configure_buffers(s);
pa_stream_set_state(s, PA_STREAM_READY); pa_stream_set_state(s, PA_STREAM_READY);
if (s->suspended) { if (s->suspended) {
s->suspended = false; s->suspended = false;
@ -150,32 +143,19 @@ static const struct spa_pod *get_buffers_param(pa_stream *s, pa_buffer_attr *att
blocks = 1; blocks = 1;
stride = pa_frame_size(&s->sample_spec); stride = pa_frame_size(&s->sample_spec);
if (attr->tlength == (uint32_t)-1 || attr->tlength == 0) maxsize = attr->tlength;
maxsize = 1024; size = attr->minreq;
else buffers = SPA_CLAMP(maxsize / size, MIN_BUFFERS, MAX_BUFFERS);
maxsize = (attr->tlength / stride);
if (attr->minreq == (uint32_t)-1 || attr->minreq == 0)
size = maxsize;
else
size = SPA_MIN(attr->minreq / stride, maxsize);
if (attr->maxlength == (uint32_t)-1)
buffers = 3;
else
buffers = SPA_CLAMP(attr->maxlength / (size * stride), 3u, MAX_BUFFERS);
pw_log_debug("stream %p: stride %d maxsize %d size %u buffers %d", s, stride, maxsize, pw_log_debug("stream %p: stride %d maxsize %d size %u buffers %d", s, stride, maxsize,
size, buffers); size, buffers);
param = spa_pod_builder_add_object(b, param = spa_pod_builder_add_object(b,
SPA_TYPE_OBJECT_ParamBuffers, SPA_PARAM_Buffers, SPA_TYPE_OBJECT_ParamBuffers, SPA_PARAM_Buffers,
SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(buffers, buffers, MAX_BUFFERS), SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(buffers, MIN_BUFFERS, MAX_BUFFERS),
SPA_PARAM_BUFFERS_blocks, SPA_POD_Int(blocks), SPA_PARAM_BUFFERS_blocks, SPA_POD_Int(blocks),
SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int( SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int(
size * stride, size, size, maxsize),
size * stride,
maxsize * stride),
SPA_PARAM_BUFFERS_stride, SPA_POD_Int(stride), SPA_PARAM_BUFFERS_stride, SPA_POD_Int(stride),
SPA_PARAM_BUFFERS_align, SPA_POD_Int(16)); SPA_PARAM_BUFFERS_align, SPA_POD_Int(16));
return param; return param;
@ -184,6 +164,7 @@ static const struct spa_pod *get_buffers_param(pa_stream *s, pa_buffer_attr *att
static void patch_buffer_attr(pa_stream *s, pa_buffer_attr *attr, pa_stream_flags_t *flags) { static void patch_buffer_attr(pa_stream *s, pa_buffer_attr *attr, pa_stream_flags_t *flags) {
const char *e, *str; const char *e, *str;
char buf[100]; char buf[100];
uint32_t stride;
pa_assert(s); pa_assert(s);
pa_assert(attr); pa_assert(attr);
@ -228,21 +209,30 @@ static void patch_buffer_attr(pa_stream *s, pa_buffer_attr *attr, pa_stream_flag
*flags |= PA_STREAM_ADJUST_LATENCY; *flags |= PA_STREAM_ADJUST_LATENCY;
} }
} }
dump_buffer_attr(s, attr);
if (attr->maxlength == (uint32_t) -1) stride = pa_frame_size(&s->sample_spec);
if (attr->maxlength == (uint32_t) -1 || attr->maxlength == 0)
attr->maxlength = MAX_SIZE; /* 4MB is the maximum queue length PulseAudio <= 0.9.9 supported. */ attr->maxlength = MAX_SIZE; /* 4MB is the maximum queue length PulseAudio <= 0.9.9 supported. */
if (attr->tlength == (uint32_t) -1) if (attr->tlength == (uint32_t) -1 || attr->tlength == 0)
attr->tlength = (uint32_t) pa_usec_to_bytes(250*PA_USEC_PER_MSEC, &s->sample_spec); /* 250ms of buffering */ attr->tlength = (uint32_t) pa_usec_to_bytes(2*PA_USEC_PER_SEC, &s->sample_spec);
attr->tlength = SPA_MIN(attr->tlength, attr->maxlength);
if (attr->minreq == (uint32_t) -1) if (attr->minreq == (uint32_t) -1 || attr->minreq == 0)
attr->minreq = attr->tlength; /* Ask for more data when there are only 200ms left in the playback buffer */ attr->minreq = pa_usec_to_bytes(25*PA_USEC_PER_MSEC, &s->sample_spec);
attr->minreq = SPA_MIN(attr->minreq, attr->tlength / MIN_BUFFERS);
attr->minreq = SPA_MAX(attr->minreq, stride);
if (attr->prebuf == (uint32_t) -1) if (attr->fragsize == (uint32_t) -1 || attr->fragsize == 0)
attr->prebuf = attr->tlength; /* Start to play only when the playback is fully filled up once */ attr->fragsize = pa_usec_to_bytes(25*PA_USEC_PER_MSEC, &s->sample_spec);
attr->fragsize = SPA_MIN(attr->fragsize, attr->tlength / MIN_BUFFERS);
attr->fragsize = SPA_MAX(attr->fragsize, stride);
if (attr->fragsize == (uint32_t) -1) if (attr->prebuf == (uint32_t) -1 || attr->prebuf == 0)
attr->fragsize = attr->tlength; /* Pass data to the app only when the buffer is filled up once */ attr->prebuf = attr->tlength - attr->minreq;
attr->prebuf = SPA_MIN(attr->prebuf, attr->tlength - attr->minreq);
attr->prebuf = SPA_MAX(attr->prebuf, stride);
dump_buffer_attr(s, attr); dump_buffer_attr(s, attr);
} }
@ -355,7 +345,6 @@ static void update_timing_info(pa_stream *s)
ti->configured_source_usec = delay; ti->configured_source_usec = delay;
ti->write_index = pos; ti->write_index = pos;
} }
ti->since_underrun = 0;
s->timing_info_valid = true; s->timing_info_valid = true;
s->queued_bytes = pwt.queued; s->queued_bytes = pwt.queued;
@ -449,20 +438,25 @@ static void stream_process(void *data)
if (s->direction == PA_STREAM_PLAYBACK) { if (s->direction == PA_STREAM_PLAYBACK) {
pa_timing_info *i = &s->timing_info; pa_timing_info *i = &s->timing_info;
uint64_t queued, writable; uint64_t queued, writable, required;
queue_output(s); queue_output(s);
queued = i->write_index - SPA_MIN(i->read_index, i->write_index); queued = i->write_index - SPA_MIN(i->read_index, i->write_index);
writable = s->maxblock - SPA_MIN(queued, s->maxblock); writable = s->maxblock - SPA_MIN(queued, s->maxblock);
required = SPA_MIN(s->maxblock, s->buffer_attr.minreq);
if (s->write_callback && s->state == PA_STREAM_READY && writable > 0) if (s->write_callback && s->state == PA_STREAM_READY && writable >= required)
s->write_callback(s, writable, s->write_userdata); s->write_callback(s, writable, s->write_userdata);
} }
else { else {
uint64_t required;
pull_input(s); pull_input(s);
if (s->read_callback && s->ready_bytes > 0 && s->state == PA_STREAM_READY) required = SPA_MIN(s->maxblock, s->buffer_attr.fragsize);
if (s->read_callback && s->ready_bytes > required && s->state == PA_STREAM_READY)
s->read_callback(s, s->ready_bytes, s->read_userdata); s->read_callback(s, s->ready_bytes, s->read_userdata);
} }
} }
@ -554,23 +548,6 @@ static pa_stream* stream_new(pa_context *c, const char *name,
s->direct_on_input = PA_INVALID_INDEX; s->direct_on_input = PA_INVALID_INDEX;
s->stream_index = PA_INVALID_INDEX; s->stream_index = PA_INVALID_INDEX;
s->buffer_attr.maxlength = (uint32_t) -1;
if (ss)
s->buffer_attr.tlength = (uint32_t) pa_usec_to_bytes(250*PA_USEC_PER_MSEC, ss); /* 250ms of buffering */
else {
/* FIXME: We assume a worst-case compressed format corresponding to
* 48000 Hz, 2 ch, S16 PCM, but this can very well be incorrect */
pa_sample_spec tmp_ss = {
.format = PA_SAMPLE_S16NE,
.rate = 48000,
.channels = 2,
};
s->buffer_attr.tlength = (uint32_t) pa_usec_to_bytes(250*PA_USEC_PER_MSEC, &tmp_ss); /* 250ms of buffering */
}
s->buffer_attr.minreq = (uint32_t) -1;
s->buffer_attr.prebuf = (uint32_t) -1;
s->buffer_attr.fragsize = (uint32_t) -1;
s->maxblock = INT_MAX; s->maxblock = INT_MAX;
s->device_index = PA_INVALID_INDEX; s->device_index = PA_INVALID_INDEX;
@ -802,10 +779,9 @@ static int create_stream(pa_stream_direction_t direction,
int res; int res;
enum pw_stream_flags fl; enum pw_stream_flags fl;
const struct spa_pod *params[16]; const struct spa_pod *params[16];
uint32_t i, n_params = 0; uint32_t i, n_params = 0, stride;
uint8_t buffer[4096]; uint8_t buffer[4096];
struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer)); struct spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
uint32_t sample_rate = 0, stride = 0, latency_num;
const char *str; const char *str;
uint32_t devid, n_items; uint32_t devid, n_items;
struct global *g; struct global *g;
@ -886,16 +862,18 @@ static int create_stream(pa_stream_direction_t direction,
monitor = (flags & PA_STREAM_PEAK_DETECT); monitor = (flags & PA_STREAM_PEAK_DETECT);
no_remix = (flags & PA_STREAM_NO_REMIX_CHANNELS); no_remix = (flags & PA_STREAM_NO_REMIX_CHANNELS);
if (attr)
s->buffer_attr = *attr;
if (pa_sample_spec_valid(&s->sample_spec)) { if (pa_sample_spec_valid(&s->sample_spec)) {
params[n_params++] = pa_format_build_param(&b, SPA_PARAM_EnumFormat, params[n_params++] = pa_format_build_param(&b, SPA_PARAM_EnumFormat,
&s->sample_spec, &s->channel_map); &s->sample_spec, &s->channel_map);
sample_rate = s->sample_spec.rate;
stride = pa_frame_size(&s->sample_spec);
} }
else { else {
pa_sample_spec ss; pa_sample_spec ss;
pa_channel_map chmap; pa_channel_map chmap;
int i; int i;
uint32_t sample_rate = 0;
for (i = 0; i < s->n_formats; i++) { for (i = 0; i < s->n_formats; i++) {
if ((res = pa_format_info_to_sample_spec(s->req_formats[i], &ss, NULL)) < 0) { if ((res = pa_format_info_to_sample_spec(s->req_formats[i], &ss, NULL)) < 0) {
@ -911,17 +889,18 @@ static int create_stream(pa_stream_direction_t direction,
&ss, &chmap); &ss, &chmap);
if (ss.rate > sample_rate) { if (ss.rate > sample_rate) {
sample_rate = ss.rate; sample_rate = ss.rate;
stride = pa_frame_size(&ss); s->sample_spec = ss;
}
} }
} }
if (sample_rate == 0) { if (sample_rate == 0) {
sample_rate = 48000; s->sample_spec.format = PA_SAMPLE_S16NE;
stride = sizeof(int16_t) * 2; s->sample_spec.rate = 48000;
s->sample_spec.channels = 2;
} }
}
if (!pa_sample_spec_valid(&s->sample_spec))
return -EINVAL;
if (attr)
s->buffer_attr = *attr;
patch_buffer_attr(s, &s->buffer_attr, &flags); patch_buffer_attr(s, &s->buffer_attr, &flags);
if (direction == PA_STREAM_RECORD) if (direction == PA_STREAM_RECORD)
@ -972,8 +951,8 @@ static int create_stream(pa_stream_direction_t direction,
else else
str = "Music"; str = "Music";
latency_num = s->buffer_attr.minreq / stride; stride = pa_frame_size(&s->sample_spec);
sprintf(latency, "%u/%u", SPA_MAX(latency_num, 1u), sample_rate); sprintf(latency, "%u/%u", s->buffer_attr.minreq / stride, s->sample_spec.rate);
n_items = 0; n_items = 0;
items[n_items++] = SPA_DICT_ITEM_INIT(PW_KEY_NODE_LATENCY, latency); items[n_items++] = SPA_DICT_ITEM_INIT(PW_KEY_NODE_LATENCY, latency);
items[n_items++] = SPA_DICT_ITEM_INIT(PW_KEY_MEDIA_TYPE, "Audio"); items[n_items++] = SPA_DICT_ITEM_INIT(PW_KEY_MEDIA_TYPE, "Audio");
@ -1180,6 +1159,7 @@ int pa_stream_write_ext_free(pa_stream *s,
free_cb(free_cb_data); free_cb(free_cb_data);
s->timing_info.write_index += nbytes; s->timing_info.write_index += nbytes;
s->timing_info.since_underrun += nbytes;
pw_log_trace("stream %p: written %zd bytes", s, nbytes); pw_log_trace("stream %p: written %zd bytes", s, nbytes);
return 0; return 0;
@ -1253,7 +1233,7 @@ SPA_EXPORT
size_t pa_stream_writable_size(PA_CONST pa_stream *s) size_t pa_stream_writable_size(PA_CONST pa_stream *s)
{ {
const pa_timing_info *i; const pa_timing_info *i;
uint64_t now, then, queued, writable, elapsed; uint64_t now, then, queued, writable, elapsed, required;
struct timespec ts; struct timespec ts;
spa_assert(s); spa_assert(s);
@ -1279,13 +1259,21 @@ size_t pa_stream_writable_size(PA_CONST pa_stream *s)
queued -= SPA_MIN(queued, elapsed); queued -= SPA_MIN(queued, elapsed);
writable = s->maxblock - SPA_MIN(queued, s->maxblock); writable = s->maxblock - SPA_MIN(queued, s->maxblock);
pw_log_trace("stream %p: %"PRIu64, s, writable); required = SPA_MIN(s->maxblock, s->buffer_attr.minreq);
pw_log_debug("stream %p: %"PRIu64" minreq:%u maxblock:%zu", s,
writable, s->buffer_attr.minreq, s->maxblock);
if (writable < required)
writable = 0;
return writable; return writable;
} }
SPA_EXPORT SPA_EXPORT
size_t pa_stream_readable_size(PA_CONST pa_stream *s) size_t pa_stream_readable_size(PA_CONST pa_stream *s)
{ {
uint64_t readable, required;
spa_assert(s); spa_assert(s);
spa_assert(s->refcount >= 1); spa_assert(s->refcount >= 1);
@ -1294,8 +1282,13 @@ size_t pa_stream_readable_size(PA_CONST pa_stream *s)
PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD,
PA_ERR_BADSTATE, (size_t) -1); PA_ERR_BADSTATE, (size_t) -1);
pw_log_trace("stream %p: %zd", s, s->ready_bytes); readable = s->ready_bytes;
return s->ready_bytes; required = SPA_MIN(s->maxblock, s->buffer_attr.fragsize);
pw_log_trace("stream %p: %zd %zd", s, readable, required);
if (readable < required)
readable = 0;
return readable;
} }
struct success_ack { struct success_ack {