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

bluez5: iso-io: track and apply corrections to tx latency

Browse source 
Use TX timestamps to get accurate reading of queue length and latency on
kernel + controller side.

This is new kernel BT feature, so requires kernel with the necessary
patches, available currently only in bluetooth-next/master branch.
Enabling Poll Errqueue kernel experimental Bluetooth feature is also
required for this.

Use the latency information to mitigate controller issues where ISO
streams are desynchronized due to tx problems or spontaneously when some
packets that should have been sent are left sitting in the queue, and
transmission is off by a multiple of the ISO interval.  This state is
visible in the latency information, so if we see streams in a group have
persistently different latencies, drop packets to resynchronize them.

Also make corrections if the kernel/controller queues get too long, so
that we don't have too big latency there.

Since BlueZ watches the same socket for errors, and TX timestamps arrive
via the socket error queue, we need to set BT_POLL_ERRQUEUE in addition
to SO_TIMESTAMPING so that BlueZ doesn't think TX timestamps are errors.

Link: https://github.com/bluez/bluez/issues/515
Link: https://lore.kernel.org/linux-bluetooth/cover.1710440392.git.pav@iki.fi/
Link: https://lore.kernel.org/linux-bluetooth/f57e065bb571d633f811610d273711c7047af335.1712499936.git.pav@iki.fi/
This commit is contained in:
Pauli Virtanen 2024-02-20 23:01:07 +02:00
parent 9165291c43
commit a6dcdfae0c
7 changed files with 298 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

84
spa/plugins/bluez5/iso-io.c
View file

@ -7,7 +7,6 @@
#include <stdio.h>
#include <errno.h>
#include <limits.h>
#include <sys/socket.h>
#include <spa/support/loop.h>
#include <spa/support/log.h>
@ -26,9 +25,14 @@ SPA_LOG_TOPIC_DEFINE_STATIC(log_topic, "spa.bluez5.iso");
#undef SPA_LOG_TOPIC_DEFAULT
#define SPA_LOG_TOPIC_DEFAULT &log_topic
#include "bt-latency.h"
#define IDLE_TIME (500 * SPA_NSEC_PER_MSEC)
#define EMPTY_BUF_SIZE 65536
#define LATENCY_PERIOD (200 * SPA_NSEC_PER_MSEC)
#define MAX_PACKET_QUEUE 3
struct group {
struct spa_log *log;
struct spa_loop *data_loop;
@ -39,7 +43,6 @@ struct group {
uint8_t id;
uint64_t next;
uint64_t duration;
uint32_t paused;
bool started;
};
@ -55,6 +58,8 @@ struct stream {
const struct media_codec *codec;
uint32_t block_size;
struct spa_bt_latency tx_latency;
};
struct modify_info
@ -131,15 +136,20 @@ static int set_timeout(struct group *group, uint64_t time)
group->timerfd, SPA_FD_TIMER_ABSTIME, &ts, NULL);
}
static int set_timers(struct group *group)
static uint64_t get_time_ns(struct spa_system *system, clockid_t clockid)
{
struct timespec now;
spa_system_clock_gettime(system, clockid, &now);
return SPA_TIMESPEC_TO_NSEC(&now);
}
static int set_timers(struct group *group)
{
if (group->duration == 0)
return -EINVAL;
spa_system_clock_gettime(group->data_system, CLOCK_MONOTONIC, &now);
group->next = SPA_ROUND_UP(SPA_TIMESPEC_TO_NSEC(&now) + group->duration,
group->next = SPA_ROUND_UP(get_time_ns(group->data_system, CLOCK_MONOTONIC) + group->duration,
group->duration);
return set_timeout(group, group->next);
@ -170,10 +180,6 @@ static void group_on_timeout(struct spa_source *source)
return;
}
/*
* If a stream failed, pause output of all streams for a while to avoid
* desynchronization.
*/
spa_list_for_each(stream, &group->streams, link) {
if (!stream->sink) {
if (!stream->pull) {
@ -183,6 +189,8 @@ static void group_on_timeout(struct spa_source *source)
continue;
}
spa_bt_latency_recv_errqueue(&stream->tx_latency, stream->fd, group->log);
if (stream->this.need_resync) {
resync = true;
stream->this.need_resync = false;
@ -192,18 +200,16 @@ static void group_on_timeout(struct spa_source *source)
group->started = true;
}
if (group->paused) {
--group->paused;
spa_log_debug(group->log, "%p: ISO group:%u paused:%u", group, group->id, group->paused);
}
/* Produce output */
spa_list_for_each(stream, &group->streams, link) {
int res;
uint64_t now;
int32_t min_latency = INT32_MAX, max_latency = INT32_MIN;
struct stream *other;
if (!stream->sink)
continue;
if (group->paused || !group->started) {
if (!group->started) {
stream->this.resync = true;
stream->this.size = 0;
continue;
@ -217,21 +223,50 @@ static void group_on_timeout(struct spa_source *source)
}
}
spa_list_for_each(other, &group->streams, link) {
if (!other->sink || stream == other || !other->tx_latency.valid)
continue;
min_latency = SPA_MIN(min_latency, other->tx_latency.ptp.min);
max_latency = SPA_MAX(max_latency, other->tx_latency.ptp.max);
}
if (stream->tx_latency.valid && min_latency <= max_latency &&
stream->tx_latency.ptp.min > min_latency + (int64_t)group->duration/2 &&
stream->tx_latency.ptp.max > max_latency + (int64_t)group->duration/2) {
spa_log_debug(group->log, "%p: ISO group:%u latency skip align fd:%d", group, group->id, stream->fd);
spa_bt_latency_reset(&stream->tx_latency);
goto stream_done;
}
/* TODO: this should use rate match */
if (stream->tx_latency.valid &&
stream->tx_latency.ptp.min > MAX_PACKET_QUEUE * (int64_t)group->duration) {
spa_log_debug(group->log, "%p: ISO group:%u latency skip fd:%d", group, group->id, stream->fd);
spa_bt_latency_reset(&stream->tx_latency);
goto stream_done;
}
now = get_time_ns(group->data_system, CLOCK_REALTIME);
res = send(stream->fd, stream->this.buf, stream->this.size, MSG_DONTWAIT | MSG_NOSIGNAL);
if (res < 0) {
res = -errno;
fail = true;
} else {
spa_bt_latency_sent(&stream->tx_latency, now);
}
spa_log_trace(group->log, "%p: ISO group:%u sent fd:%d size:%u ts:%u idle:%d res:%d",
stream_done:
spa_log_trace(group->log, "%p: ISO group:%u sent fd:%d size:%u ts:%u idle:%d res:%d latency:%d..%d us",
group, group->id, stream->fd, (unsigned)stream->this.size,
(unsigned)stream->this.timestamp, stream->idle, res);
(unsigned)stream->this.timestamp, stream->idle, res,
stream->tx_latency.valid ? stream->tx_latency.ptp.min/1000 : -1,
stream->tx_latency.valid ? stream->tx_latency.ptp.max/1000 : -1);
stream->this.size = 0;
}
if (fail)
group->paused = 1u + IDLE_TIME / group->duration;
spa_log_debug(group->log, "%p: ISO group:%d send failure", group, group->id);
/* Pull data for the next interval */
group->next += exp * group->duration;
@ -399,6 +434,8 @@ static struct stream *stream_create(struct spa_bt_transport *t, struct group *gr
stream->this.format = format;
stream->block_size = block_size;
spa_bt_latency_init(&stream->tx_latency, stream->fd, LATENCY_PERIOD, group->log);
if (sink)
stream_silence(stream);
@ -453,6 +490,8 @@ void spa_bt_iso_io_destroy(struct spa_bt_iso_io *this)
stream_unlink(stream);
spa_bt_latency_flush(&stream->tx_latency, stream->fd, stream->group->log);
if (spa_list_is_empty(&stream->group->streams))
group_destroy(stream->group);
@ -508,3 +547,12 @@ void spa_bt_iso_io_set_cb(struct spa_bt_iso_io *this, spa_bt_iso_io_pull_t pull,
return;
}
}
/** Must be called from data thread */
int spa_bt_iso_io_recv_errqueue(struct spa_bt_iso_io *this)
{
struct stream *stream = SPA_CONTAINER_OF(this, struct stream, this);
struct group *group = stream->group;
return spa_bt_latency_recv_errqueue(&stream->tx_latency, stream->fd, group->log);
}