From 623b6df8a20d2dc27996ea02f655a817d22ec19e Mon Sep 17 00:00:00 2001
From: Pauli Virtanen <pav@iki.fi>
Date: Thu, 19 Aug 2021 22:48:30 +0300
Subject: [PATCH] bluez5: work around obscure issue with a2dp-source with
 duplex

For unknown reason the BT socket when working with A2DP "duplex" stream,
sometimes stops waking up poll when data arrives. Regardless, recv() can
read data packets from it.

To work around this, when A2DP source is in duplex mode, instead of
polling on data, we poll on a timer.
---
 spa/plugins/bluez5/a2dp-source.c | 83 +++++++++++++++++++++++++++++---
 1 file changed, 75 insertions(+), 8 deletions(-)

diff --git a/spa/plugins/bluez5/a2dp-source.c b/spa/plugins/bluez5/a2dp-source.c
index ad0e07b6d..cc56a68a8 100644
--- a/spa/plugins/bluez5/a2dp-source.c
+++ b/spa/plugins/bluez5/a2dp-source.c
@@ -129,6 +129,10 @@ struct impl {
 	unsigned int transport_acquired:1;
 	unsigned int following:1;
 
+	unsigned int is_input:1;
+	unsigned int is_duplex:1;
+
+	int fd;
 	struct spa_source source;
 
 	struct spa_io_clock *clock;
@@ -145,8 +149,8 @@ struct impl {
 	uint64_t sample_count;
 	uint64_t skip_count;
 
-	bool is_input;
-	bool is_duplex;
+	int duplex_timerfd;
+	uint64_t duplex_timeout;
 };
 
 #define NAME "a2dp-source"
@@ -379,7 +383,7 @@ static int32_t read_data(struct impl *this) {
 
 again:
 	/* read data from socket */
-	size_read = read(this->source.fd, this->buffer_read, b_size);
+	size_read = recv(this->fd, this->buffer_read, b_size, MSG_DONTWAIT);
 
 	if (size_read == 0)
 		return 0;
@@ -599,6 +603,30 @@ stop:
 		spa_loop_remove_source(this->data_loop, &this->source);
 }
 
+static int set_duplex_timeout(struct impl *this, uint64_t timeout)
+{
+	struct itimerspec ts;
+	ts.it_value.tv_sec = timeout / SPA_NSEC_PER_SEC;
+	ts.it_value.tv_nsec = timeout % SPA_NSEC_PER_SEC;
+	ts.it_interval.tv_sec = 0;
+	ts.it_interval.tv_nsec = 0;
+	return spa_system_timerfd_settime(this->data_system,
+			this->duplex_timerfd, 0, &ts, NULL);
+}
+
+static void a2dp_on_duplex_timeout(struct spa_source *source)
+{
+	struct impl *this = source->data;
+	uint64_t exp;
+
+	if (spa_system_timerfd_read(this->data_system, this->duplex_timerfd, &exp) < 0)
+		spa_log_warn(this->log, "error reading timerfd: %s", strerror(errno));
+
+	set_duplex_timeout(this, this->duplex_timeout);
+
+	a2dp_on_ready_read(source);
+}
+
 static int transport_start(struct impl *this)
 {
 	int res, val;
@@ -643,12 +671,36 @@ static int transport_start(struct impl *this)
 
 	reset_buffers(&this->port);
 
+	this->fd = this->transport->fd;
+
 	this->source.data = this;
-	this->source.fd = this->transport->fd;
-	this->source.func = a2dp_on_ready_read;
-	this->source.mask = SPA_IO_IN;
-	this->source.rmask = 0;
-	spa_loop_add_source(this->data_loop, &this->source);
+
+	if (!this->is_duplex) {
+		this->source.fd = this->transport->fd;
+		this->source.func = a2dp_on_ready_read;
+		this->source.mask = SPA_IO_IN;
+		this->source.rmask = 0;
+		spa_loop_add_source(this->data_loop, &this->source);
+	} else {
+		/*
+		 * XXX: For an unknown reason (on Linux 5.13.10), the socket when working with
+		 * XXX: "duplex" stream sometimes stops waking up from the poll, even though
+		 * XXX: you can recv() from the socket with no problem.
+		 * XXX:
+		 * XXX: The reason for this should be found and fixed.
+		 * XXX: To work around this, for now we just do the stupid thing and poll
+		 * XXX: on a timer, chosen so that it's fast enough for the aptX-LL codec
+		 * XXX: we currently support (which sends mSBC data).
+		 */
+		this->source.fd = this->duplex_timerfd;
+		this->source.func = a2dp_on_duplex_timeout;
+		this->source.mask = SPA_IO_IN;
+		this->source.rmask = 0;
+		spa_loop_add_source(this->data_loop, &this->source);
+
+		this->duplex_timeout = SPA_NSEC_PER_MSEC * 75/10;
+		set_duplex_timeout(this, this->duplex_timeout);
+	}
 
 	this->sample_count = 0;
 	this->skip_count = 0;
@@ -688,6 +740,10 @@ static int do_remove_source(struct spa_loop *loop,
 {
 	struct impl *this = user_data;
 
+	spa_log_debug(this->log, NAME" %p: remove source", this);
+
+	set_duplex_timeout(this, 0);
+
 	if (this->source.loop)
 		spa_loop_remove_source(this->data_loop, &this->source);
 
@@ -1287,6 +1343,10 @@ static int impl_clear(struct spa_handle *handle)
 		this->codec->clear_props(this->codec_props);
 	if (this->transport)
 		spa_hook_remove(&this->transport_listener);
+	if (this->duplex_timerfd >= 0) {
+		spa_system_close(this->data_system, this->duplex_timerfd);
+		this->duplex_timerfd = -1;
+	}
 	return 0;
 }
 
@@ -1411,6 +1471,13 @@ impl_init(const struct spa_handle_factory *factory,
 	spa_bt_transport_add_listener(this->transport,
 			&this->transport_listener, &transport_events, this);
 
+	if (this->is_duplex) {
+		this->duplex_timerfd = spa_system_timerfd_create(this->data_system,
+				CLOCK_MONOTONIC, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
+	} else {
+		this->duplex_timerfd = -1;
+	}
+
 	return 0;
 }