[PATCH v2] bluez5-device: Rewrite of thread function, reduce send buffer size for a2dp sink

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

 



On 26.04.2018 15:13, Tanu Kaskinen wrote:
> On Thu, 2018-03-29 at 15:33 +0200, Georg Chini 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
>> ---
>> Changes in v2:
>>   - fix issues pointed out by Tanu
>>   - set writable to false for HSP only if a block really needs to be written
>>
>>   src/modules/bluetooth/module-bluez5-device.c | 289 ++++++++++++++++++---------
>>   1 file changed, 191 insertions(+), 98 deletions(-)
>>
>> diff --git a/src/modules/bluetooth/module-bluez5-device.c b/src/modules/bluetooth/module-bluez5-device.c
>> index c3acc1dc..dac1eb2a 100644
>> --- a/src/modules/bluetooth/module-bluez5-device.c
>> +++ b/src/modules/bluetooth/module-bluez5-device.c
>> @@ -56,9 +56,8 @@ 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_PLAYBACK_SCO  (25 * PA_USEC_PER_MSEC)
> Why is this changed? The commit message mentions nothing about this.

Ups, sorry, I forgot to mention. I changed it because it delivers much
better A/V sync for SCO. When testing the A2DP delay issue, I also
tested HSP. In fact, as HSP is a synchronous connection, the delay
for SCO playback/record should be smaller than for A2DP, but I
decided to use the same value for all four constants to keep it
simple.

>
>>   #define FIXED_LATENCY_RECORD_A2DP   (25 * PA_USEC_PER_MSEC)
>>   #define FIXED_LATENCY_RECORD_SCO    (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. */
>> +        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. */
>> +        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);
>> @@ -1068,12 +1110,12 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
>>       switch (code) {
>>   
>>           case PA_SINK_MESSAGE_GET_LATENCY: {
>> -            int64_t wi, ri;
>> +            int64_t wi = 0, ri = 0;
>>   
>>               if (u->read_smoother) {
>>                   ri = pa_smoother_get(u->read_smoother, pa_rtclock_now());
>>                   wi = pa_bytes_to_usec(u->write_index + u->write_block_size, &u->sample_spec);
>> -            } else {
>> +            } else if (u->started_at) {
>>                   ri = pa_rtclock_now() - u->started_at;
>>                   wi = pa_bytes_to_usec(u->write_index, &u->sample_spec);
>>               }
>> @@ -1415,12 +1457,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_debug("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);
>> @@ -1440,9 +1502,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 " :"",
>> @@ -1453,147 +1519,174 @@ static void thread_func(void *userdata) {
>>               if (pollfd->revents & POLLHUP) {
>>                   pollfd = NULL;
>>                   teardown_stream(u);
>> -                do_write = 0;
>> -                pending_read_bytes = 0;
>> -                writable = false;
>> +                blocks_to_write = 0;
>> +                bytes_to_write = 0;
>>                   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) {
>> -                    pa_usec_t time_passed;
>> -                    pa_usec_t audio_sent;
>> +                /* If we have a source, we let the source determine the timing
>> +                 * for the sink */
>> +                if (have_source) {
>>   
>> -                    /* Hmm, there is no input stream we could synchronize
>> -                     * to. So let's do things by time */
>> +                    if (writable && blocks_to_write > 0) {
>> +                        int result;
>>   
>> -                    time_passed = pa_rtclock_now() - u->started_at;
>> -                    audio_sent = pa_bytes_to_usec(u->write_index, &u->sample_spec);
>> +                        if ((result = write_block(u)) < 0)
>> +                            goto fail;
>>   
>> +                        blocks_to_write -= result;
>> +                        if (blocks_to_write > 0)
>> +                            writable = false;
> This "blocks_to_write > 0" check is new. In the previous version
> writable was set to false unconditionally, and that's how I believe it
> should be done. I guess this is an optimization: we won't write
> anything before we get POLLIN, so to you it seemed unnecessary to wake
> up on POLLOUT. Getting the POLLOUT notification is indeed unnecessary
> in itself, but we still need to set POLLOUT when polling, because
> otherwise when we wake up due to POLLIN, pollfd->revents will not have
> POLLOUT set even if the socket is writable. As a result we won't write
> when we're supposed to.
>
> Or actually... if we wake up on POLLIN, and POLLOUT isn't in revents
> even if the socket is writable, we'll set POLLOUT on the subsequent
> poll, and that will return immediately, so there's not much delay with
> the write. So maybe the "blocks_to_write > 0" is fine after all. Some
> kind of a comment would probably be good in any case. For example:
>
> "writable controls whether we set POLLOUT when polling - we set it to
> false to enable POLLOUT. If there are more blocks to write, we want to
> be woken up immediately when the socket becomes writable. If there
> aren't currently any more blocks to write, then we'll have to wait
> until we've received more data, so in that case we only want to set
> POLLIN. Note that when we are woken up the next time, POLLOUT won't be
> set in revents even if the socket has meanwhile become writable, which
> may seem bad, but in that case we'll set POLLOUT in the subsequent
> poll, and the poll will return immediately, so our writes won't be
> delayed."
>
Yes it is an optimization and it works fine and significantly reduces
CPU load on slow systems (which was the reason to implement it).
I will add your comment to clarify.



[Index of Archives]     [Linux Audio Users]     [AMD Graphics]     [Linux USB Devel]     [Linux Audio Users]     [Yosemite News]     [Linux Kernel]     [Linux SCSI]

  Powered by Linux