Hi George,
On Mon, Mar 5, 2018 at 9:49 AM, Georg Chini <[email protected]> wrote:
> The rewrite of the thread function does not change functionality much,
> most of it is only cleanup, minor bug fixing and documentation work.
>
> This patch also changes the send buffer size for a2dp sink to avoid lags
> after temporary connection drops, following the proof-of-concept patch
> posted by Dmitry Kalyanov.
>
> Bug-Link: https://bugs.freedesktop.org/show_bug.cgi?id=58746
> ---
> src/modules/bluetooth/module-bluez5-device.c | 275
> ++++++++++++++++++---------
> 1 file changed, 182 insertions(+), 93 deletions(-)
>
> diff --git a/src/modules/bluetooth/module-bluez5-device.c
> b/src/modules/bluetooth/module-bluez5-device.c
> index 7970dda7..149d1708 100644
> --- a/src/modules/bluetooth/module-bluez5-device.c
> +++ b/src/modules/bluetooth/module-bluez5-device.c
> @@ -56,7 +56,6 @@ PA_MODULE_LOAD_ONCE(false);
> PA_MODULE_USAGE("path=<device object path>"
> "autodetect_mtu=<boolean>");
>
> -#define MAX_PLAYBACK_CATCH_UP_USEC (100 * PA_USEC_PER_MSEC)
> #define FIXED_LATENCY_PLAYBACK_A2DP (25 * PA_USEC_PER_MSEC)
> #define FIXED_LATENCY_PLAYBACK_SCO (125 * PA_USEC_PER_MSEC)
> #define FIXED_LATENCY_RECORD_A2DP (25 * PA_USEC_PER_MSEC)
> @@ -660,6 +659,38 @@ static int a2dp_process_push(struct userdata *u) {
> return ret;
> }
>
> +static void update_buffer_size(struct userdata *u) {
> + int old_bufsize;
> + socklen_t len = sizeof(int);
> + int ret;
> +
> + ret = getsockopt(u->stream_fd, SOL_SOCKET, SO_SNDBUF, &old_bufsize,
> &len);
> + if (ret == -1) {
> + pa_log_warn("Changing bluetooth buffer size: Failed to
> getsockopt(SO_SNDBUF): %s", pa_cstrerror(errno));
> + } else {
> + int new_bufsize;
> +
> + /* Set send buffer size as small as possible. The minimum value is
> 1024 according to the
> + * socket man page. The data is written to the socket in chunks of
> write_block_size, so
> + * there should at least be room for two chunks in the buffer.
> Generally, write_block_size
> + * is larger than 512. If not, use the next multiple of
> write_block_size which is larger
> + * than 1024. */
Btw, the man page actually says it is 2048 bytes but perhaps you are
considering the kernel doubles the valuse set for bookkeeping? Anyway
getsockopt shall always return 2048 at least.
> + new_bufsize = 2 * u->write_block_size;
> + if (new_bufsize < 1024)
> + new_bufsize = (1024 / u->write_block_size + 1) *
> u->write_block_size;
> +
> + /* The kernel internally doubles the buffer size that was set by
> setsockopt and getsockopt
> + * returns the doubled value. */
Btw, what happens if the buffer is reduced, for example when the
bitpool is reduced, and there are packets still on the buffer? Are
they discarded?
> + if (new_bufsize != old_bufsize / 2) {
> + ret = setsockopt(u->stream_fd, SOL_SOCKET, SO_SNDBUF,
> &new_bufsize, len);
> + if (ret == -1)
> + pa_log_warn("Changing bluetooth buffer size: Failed to
> change from %d to %d: %s", old_bufsize / 2, new_bufsize, pa_cstrerror(errno));
> + else
> + pa_log_info("Changing bluetooth buffer size: Changed from %d
> to %d", old_bufsize / 2, new_bufsize);
> + }
> + }
> +}
> +
> /* Run from I/O thread */
> static void a2dp_set_bitpool(struct userdata *u, uint8_t bitpool) {
> struct sbc_info *sbc_info;
> @@ -694,6 +725,15 @@ static void a2dp_set_bitpool(struct userdata *u, uint8_t
> bitpool) {
> pa_sink_set_max_request_within_thread(u->sink, u->write_block_size);
> pa_sink_set_fixed_latency_within_thread(u->sink,
> FIXED_LATENCY_PLAYBACK_A2DP +
> pa_bytes_to_usec(u->write_block_size, &u->sample_spec));
> +
> + /* If there is still data in the memchunk, we have to discard it
> + * because the write_block_size may have changed. */
> + if (u->write_memchunk.memblock) {
> + pa_memblock_unref(u->write_memchunk.memblock);
> + pa_memchunk_reset(&u->write_memchunk);
> + }
> +
> + update_buffer_size(u);
> }
>
> /* Run from I/O thread */
> @@ -852,8 +892,10 @@ static void setup_stream(struct userdata *u) {
>
> pa_log_debug("Stream properly set up, we're ready to roll!");
>
> - if (u->profile == PA_BLUETOOTH_PROFILE_A2DP_SINK)
> + if (u->profile == PA_BLUETOOTH_PROFILE_A2DP_SINK) {
> a2dp_set_bitpool(u, u->sbc_info.max_bitpool);
> + update_buffer_size(u);
> + }
>
> u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
> pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
> @@ -861,7 +903,7 @@ static void setup_stream(struct userdata *u) {
> pollfd->events = pollfd->revents = 0;
>
> u->read_index = u->write_index = 0;
> - u->started_at = 0;
> + u->started_at = pa_rtclock_now();
> u->stream_setup_done = true;
>
> if (u->source)
> @@ -1407,12 +1449,32 @@ static int init_profile(struct userdata *u) {
> return r;
> }
>
> +static int write_block(struct userdata *u) {
> + int n_written;
> +
> + if (u->write_index <= 0)
> + u->started_at = pa_rtclock_now();
> +
> + if (u->profile == PA_BLUETOOTH_PROFILE_A2DP_SINK) {
> + if ((n_written = a2dp_process_render(u)) < 0)
> + return -1;
> + } else {
> + if ((n_written = sco_process_render(u)) < 0)
> + return -1;
> + }
> +
> + if (n_written == 0)
> + pa_log("Got EAGAIN on write() after POLLOUT, probably there is a
> temporary connection loss.");
> +
> + return n_written;
> +}
> +
> +
> /* I/O thread function */
> static void thread_func(void *userdata) {
> struct userdata *u = userdata;
> - unsigned do_write = 0;
> - unsigned pending_read_bytes = 0;
> - bool writable = false;
> + unsigned blocks_to_write = 0;
> + unsigned bytes_to_write = 0;
>
> pa_assert(u);
> pa_assert(u->transport);
> @@ -1432,9 +1494,13 @@ static void thread_func(void *userdata) {
> struct pollfd *pollfd;
> int ret;
> bool disable_timer = true;
> + bool writable = false;
> + bool have_source = u->source ?
> PA_SOURCE_IS_LINKED(u->source->thread_info.state) : false;
> + bool have_sink = u->sink ?
> PA_SINK_IS_LINKED(u->sink->thread_info.state) : false;
>
> pollfd = u->rtpoll_item ? pa_rtpoll_item_get_pollfd(u->rtpoll_item,
> NULL) : NULL;
>
> + /* Check for stream error or close */
> if (pollfd && (pollfd->revents & ~(POLLOUT|POLLIN))) {
> pa_log_info("FD error: %s%s%s%s",
> pollfd->revents & POLLERR ? "POLLERR " :"",
> @@ -1445,147 +1511,170 @@ static void thread_func(void *userdata) {
> if (pollfd->revents & POLLHUP) {
> pollfd = NULL;
> teardown_stream(u);
> - do_write = 0;
> - pending_read_bytes = 0;
> + blocks_to_write = 0;
> + bytes_to_write = 0;
> writable = false;
> pa_asyncmsgq_post(pa_thread_mq_get()->outq,
> PA_MSGOBJECT(u->msg), BLUETOOTH_MESSAGE_STREAM_FD_HUP, NULL, 0, NULL, NULL);
> } else
> goto fail;
> }
>
> - if (u->source && PA_SOURCE_IS_LINKED(u->source->thread_info.state)) {
> + /* If there is a pollfd, the stream is set up and we need to do
> something */
> + if (pollfd) {
>
> - /* We should send two blocks to the device before we expect
> - * a response. */
> + /* Handle source if present */
> + if (have_source) {
>
> - if (u->write_index == 0 && u->read_index <= 0)
> - do_write = 2;
> + /* We should send two blocks to the device before we expect
> a response. */
> + if (u->write_index == 0 && u->read_index <= 0)
> + blocks_to_write = 2;
>
> - if (pollfd && (pollfd->revents & POLLIN)) {
> - int n_read;
> + /* If we got woken up by POLLIN let's do some reading */
> + if (pollfd->revents & POLLIN) {
> + int n_read;
>
> - if (u->profile == PA_BLUETOOTH_PROFILE_A2DP_SOURCE)
> - n_read = a2dp_process_push(u);
> - else
> - n_read = sco_process_push(u);
> + if (u->profile == PA_BLUETOOTH_PROFILE_A2DP_SOURCE)
> + n_read = a2dp_process_push(u);
> + else
> + n_read = sco_process_push(u);
>
> - if (n_read < 0)
> - goto fail;
> + if (n_read < 0)
> + goto fail;
>
> - if (n_read > 0) {
> - /* We just read something, so we are supposed to write
> something, too */
> - pending_read_bytes += n_read;
> - do_write += pending_read_bytes / u->write_block_size;
> - pending_read_bytes = pending_read_bytes %
> u->write_block_size;
> + if (n_read > 0) {
> + /* We just read something, so we are supposed to
> write something, too */
> + bytes_to_write += n_read;
> + blocks_to_write += bytes_to_write /
> u->write_block_size;
> + bytes_to_write = bytes_to_write %
> u->write_block_size;
> + }
> }
> }
> - }
>
> - if (u->sink && PA_SINK_IS_LINKED(u->sink->thread_info.state)) {
> + /* Handle sink if present */
> + if (have_sink) {
>
> - if (PA_UNLIKELY(u->sink->thread_info.rewind_requested))
> - pa_sink_process_rewind(u->sink, 0);
> + /* Process rewinds */
> + if (PA_UNLIKELY(u->sink->thread_info.rewind_requested))
> + pa_sink_process_rewind(u->sink, 0);
>
> - if (pollfd) {
> + /* Test if the stream is writable */
> if (pollfd->revents & POLLOUT)
> writable = true;
>
> - if ((!u->source ||
> !PA_SOURCE_IS_LINKED(u->source->thread_info.state)) && do_write <= 0 &&
> writable) {
> + /* If we have a source, we let the source determine the
> timing
> + * for the sink */
> + if (have_source) {
> +
> + if (writable && blocks_to_write > 0) {
> + int result;
> +
> + if ((result = write_block(u)) < 0)
> + goto fail;
> +
> + blocks_to_write -= result;
> + writable = false;
> + }
> +
> + /* There is no source, we have to use the system clock for
> timing */
> + } else {
> + bool have_written = false;
> pa_usec_t time_passed;
> pa_usec_t audio_sent;
>
> - /* Hmm, there is no input stream we could synchronize
> - * to. So let's do things by time */
> -
> time_passed = pa_rtclock_now() - u->started_at;
> audio_sent = pa_bytes_to_usec(u->write_index,
> &u->sample_spec);
>
> + /* A new block needs to be sent. */
> if (audio_sent <= time_passed) {
> - pa_usec_t audio_to_send = time_passed - audio_sent;
> + size_t bytes_to_send = pa_usec_to_bytes(time_passed
> - audio_sent, &u->sample_spec);
>
> - /* Never try to catch up for more than 100ms */
> - if (u->write_index > 0 && audio_to_send >
> MAX_PLAYBACK_CATCH_UP_USEC) {
> - pa_usec_t skip_usec;
> + /* There are more than two blocks that need to be
> written.
> + * We cannot catch up, therefore discard everything
> older
> + * than two block sizes. */
> + if (bytes_to_send > 2 * u->write_block_size) {
> uint64_t skip_bytes;
> + pa_memchunk tmp;
> + size_t mempool_max_block_size =
> pa_mempool_block_size_max(u->core->mempool);
> + pa_usec_t skip_usec;
>
> - skip_usec = audio_to_send -
> MAX_PLAYBACK_CATCH_UP_USEC;
> - skip_bytes = pa_usec_to_bytes(skip_usec,
> &u->sample_spec);
> + skip_bytes = bytes_to_send - 2 *
> u->write_block_size;
> + skip_usec = pa_bytes_to_usec(skip_bytes,
> &u->sample_spec);
>
> - if (skip_bytes > 0) {
> - pa_memchunk tmp;
> + pa_log_warn("Skipping %llu us (= %llu bytes) in
> audio stream",
> + (unsigned long long) skip_usec,
> + (unsigned long long) skip_bytes);
>
> - pa_log_warn("Skipping %llu us (= %llu bytes)
> in audio stream",
> - (unsigned long long) skip_usec,
> - (unsigned long long) skip_bytes);
> + while (skip_bytes > 0) {
> + size_t bytes_to_render;
>
> - pa_sink_render_full(u->sink, skip_bytes,
> &tmp);
> - pa_memblock_unref(tmp.memblock);
> - u->write_index += skip_bytes;
> + if (skip_bytes > mempool_max_block_size)
> + bytes_to_render = mempool_max_block_size;
> + else
> + bytes_to_render = skip_bytes;
>
> - if (u->profile ==
> PA_BLUETOOTH_PROFILE_A2DP_SINK)
> - a2dp_reduce_bitpool(u);
> + pa_sink_render_full(u->sink,
> bytes_to_render, &tmp);
> + pa_memblock_unref(tmp.memblock);
> + u->write_index += bytes_to_render;
> + skip_bytes -= bytes_to_render;
> }
> +
> + if (u->write_index > 0 && u->profile ==
> PA_BLUETOOTH_PROFILE_A2DP_SINK)
> + a2dp_reduce_bitpool(u);
> }
>
> - do_write = 1;
> - pending_read_bytes = 0;
> + blocks_to_write = 1;
> }
> - }
> -
> - if (writable && do_write > 0) {
> - int n_written;
>
> - if (u->write_index <= 0)
> - u->started_at = pa_rtclock_now();
> + /* If the stream is writable, send some data if
> necessary */
> + if (writable && blocks_to_write > 0) {
> + int result;
>
> - if (u->profile == PA_BLUETOOTH_PROFILE_A2DP_SINK) {
> - if ((n_written = a2dp_process_render(u)) < 0)
> - goto fail;
> - } else {
> - if ((n_written = sco_process_render(u)) < 0)
> + if ((result = write_block(u)) < 0)
> goto fail;
> - }
> -
> - if (n_written == 0)
> - pa_log("Broken kernel: we got EAGAIN on write()
> after POLLOUT!");
>
> - do_write -= n_written;
> - writable = false;
> - }
> + blocks_to_write -= result;
> + writable = false;
> + have_written = true;
> + }
>
> - if ((!u->source ||
> !PA_SOURCE_IS_LINKED(u->source->thread_info.state)) && do_write <= 0) {
> - pa_usec_t sleep_for;
> - pa_usec_t time_passed, next_write_at;
> -
> - if (writable) {
> - /* Hmm, there is no input stream we could synchronize
> - * to. So let's estimate when we need to wake up the
> latest */
> - time_passed = pa_rtclock_now() - u->started_at;
> - next_write_at = pa_bytes_to_usec(u->write_index,
> &u->sample_spec);
> - sleep_for = time_passed < next_write_at ?
> next_write_at - time_passed : 0;
> - /* pa_log("Sleeping for %lu; time passed %lu, next
> write at %lu", (unsigned long) sleep_for, (unsigned long) time_passed,
> (unsigned long)next_write_at); */
> - } else
> - /* drop stream every 500 ms */
> - sleep_for = PA_USEC_PER_MSEC * 500;
> -
> - pa_rtpoll_set_timer_relative(u->rtpoll, sleep_for);
> - disable_timer = false;
> + /* If nothing was written during this iteration, either
> the stream
> + * is not writable or there was no write pending. Set up
> a timer that
> + * will wake up the thread when the next data needs to
> be written. */
> + if (!have_written) {
> + pa_usec_t sleep_for;
> + pa_usec_t next_write_at;
> +
> + if (writable) {
> + /* There was no write pending on this iteration
> of the loop.
> + * Let's estimate when we need to wake up next */
> + next_write_at = pa_bytes_to_usec(u->write_index,
> &u->sample_spec);
> + sleep_for = time_passed < next_write_at ?
> next_write_at - time_passed : 0;
> + /* pa_log("Sleeping for %lu; time passed %lu,
> next write at %lu", (unsigned long) sleep_for, (unsigned long) time_passed,
> (unsigned long)next_write_at); */
> + } else
> + /* We could not write because the stream was not
> ready. Let's try
> + * again in 500 ms and drop audio if we still
> can't write. The
> + * thread will also be woken up when we can
> write again. */
> + sleep_for = PA_USEC_PER_MSEC * 500;
> +
> + pa_rtpoll_set_timer_relative(u->rtpoll, sleep_for);
> + disable_timer = false;
> + }
> }
> }
> +
> + /* Set events to wake up the thread */
> + pollfd->events = (short) (((have_sink && !writable) ? POLLOUT :
> 0) | (have_source ? POLLIN : 0));
> +
> }
>
> if (disable_timer)
> pa_rtpoll_set_timer_disabled(u->rtpoll);
>
> - /* Hmm, nothing to do. Let's sleep */
> - if (pollfd)
> - pollfd->events = (short) (((u->sink &&
> PA_SINK_IS_LINKED(u->sink->thread_info.state) && !writable) ? POLLOUT : 0) |
> - (u->source &&
> PA_SOURCE_IS_LINKED(u->source->thread_info.state) ? POLLIN : 0));
> -
> if ((ret = pa_rtpoll_run(u->rtpoll)) < 0) {
> pa_log_debug("pa_rtpoll_run failed with: %d", ret);
> goto fail;
> }
> +
> if (ret == 0) {
> pa_log_debug("IO thread shutdown requested, stopping cleanly");
> transport_release(u);
> --
> 2.14.1
>
> _______________________________________________
> pulseaudio-discuss mailing list
> [email protected]
> https://lists.freedesktop.org/mailman/listinfo/pulseaudio-discuss
--
Luiz Augusto von Dentz
_______________________________________________
pulseaudio-discuss mailing list
[email protected]
https://lists.freedesktop.org/mailman/listinfo/pulseaudio-discuss
