tests: Bluetooth: Tester: Refactor audio TX

Instead of relying on the bap_send_cmd and a ring buffer,
the tester will now automatically start transmitting on
streams can that transmit once they enter the streaming state,
and stop again once once they leave the streaming state.

This ensures that we are always sending, which help pass
the PTS tests that require us to send, without having the
autopts client constant telling the tester to send.

This also ensures that we actually send SDUs of the right size
by not relying on a ring buffer, but using a separate thread
that sends data from a predefined array of ISO mock data.

This is easily expandable to multiple streams (including a mix
of unicast and broadcast) using different SDU sizes and easy to
expand to also use one or more software codecs.

The design is based on the TX thread for the BAP Unicast
Client sample and the audio babblesim tests.

Signed-off-by: Emil Gydesen <[email protected]>

tests/bluetooth/tester/src/audio/btp_bap.c

@@ -2,11 +2,12 @@
 
 /*
  * Copyright (c) 2023 Codecoup
+ * Copyright (c) 2024 Nordic Semiconductor ASA
  *
  * SPDX-License-Identifier: Apache-2.0
  */
 
-
+#include <stdint.h>
 #include <stddef.h>
 #include <errno.h>
 
@@ -337,6 +338,20 @@ static uint8_t btp_bap_supported_commands(const void *cmd, uint16_t cmd_len,
 	return BTP_STATUS_SUCCESS;
 }
 
+uint8_t btp_bap_audio_stream_send(const void *cmd, uint16_t cmd_len, void *rsp, uint16_t *rsp_len)
+{
+	struct btp_bap_send_rp *rp = rsp;
+	const struct btp_bap_send_cmd *cp = cmd;
+
+	/* Always send dummy success for now until the command has be deprecated
+	 * https://github.com/auto-pts/auto-pts/issues/1317
+	 */
+	rp->data_len = cp->data_len;
+	*rsp_len = sizeof(*rp);
+
+	return BTP_STATUS_SUCCESS;
+}
+
 static const struct btp_handler bap_handlers[] = {
 	{
 		.opcode = BTP_BAP_READ_SUPPORTED_COMMANDS,
@@ -534,7 +549,7 @@ uint8_t tester_init_bap(void)
 		return BTP_STATUS_FAILED;
 	}
 
-	btp_bap_audio_stream_init_send_worker();
+	btp_bap_audio_stream_tx_init();
 
 	tester_register_command_handlers(BTP_SERVICE_ID_BAP, bap_handlers,
 					 ARRAY_SIZE(bap_handlers));

tests/bluetooth/tester/src/audio/btp_bap_audio_stream.c

@@ -2,188 +2,233 @@
 
 /*
  * Copyright (c) 2023 Codecoup
+ * Copyright (c) 2024 Nordic Semiconductor ASA
  *
  * SPDX-License-Identifier: Apache-2.0
  */
 
-
-#include <stddef.h>
 #include <errno.h>
-
-#include <zephyr/types.h>
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <string.h>
+
+#include <zephyr/autoconf.h>
+#include <zephyr/bluetooth/audio/cap.h>
+#include <zephyr/bluetooth/audio/bap.h>
+#include <zephyr/bluetooth/bluetooth.h>
+#include <zephyr/bluetooth/hci_types.h>
+#include <zephyr/bluetooth/iso.h>
 #include <zephyr/kernel.h>
-#include <zephyr/sys/ring_buffer.h>
-
+#include <zephyr/kernel/thread_stack.h>
 #include <zephyr/logging/log.h>
+#include <zephyr/logging/log_core.h>
+#include <zephyr/net_buf.h>
+#include <zephyr/sys/__assert.h>
+#include <zephyr/sys/atomic.h>
+#include <zephyr/sys/atomic_types.h>
 #include <zephyr/sys/byteorder.h>
-#define LOG_MODULE_NAME bttester_bap_audio_stream
-LOG_MODULE_REGISTER(LOG_MODULE_NAME, CONFIG_BTTESTER_LOG_LEVEL);
+#include <zephyr/sys/util.h>
+#include <zephyr/sys/util_macro.h>
+#include <zephyr/types.h>
+
 #include "btp/btp.h"
 #include "btp_bap_audio_stream.h"
 
-NET_BUF_POOL_FIXED_DEFINE(tx_pool, MAX(CONFIG_BT_ASCS_MAX_ASE_SRC_COUNT,
-			  CONFIG_BT_BAP_UNICAST_CLIENT_ASE_SRC_COUNT),
-			  BT_ISO_SDU_BUF_SIZE(CONFIG_BT_ISO_TX_MTU),
-			  CONFIG_BT_CONN_TX_USER_DATA_SIZE, NULL);
+LOG_MODULE_REGISTER(bttester_bap_audio_stream, CONFIG_BTTESTER_LOG_LEVEL);
 
-RING_BUF_DECLARE(audio_ring_buf, CONFIG_BT_ISO_TX_MTU);
-
-#define ISO_DATA_THREAD_STACK_SIZE 512
-#define ISO_DATA_THREAD_PRIORITY -7
-K_THREAD_STACK_DEFINE(iso_data_thread_stack_area, ISO_DATA_THREAD_STACK_SIZE);
-static struct k_work_q iso_data_work_q;
-static bool send_worker_inited;
+/** Enqueue at least 2 buffers per stream, but otherwise equal distribution based on the buf count*/
+#define MAX_ENQUEUE_CNT MAX(2, (CONFIG_BT_ISO_TX_BUF_COUNT / CONFIG_BT_ISO_MAX_CHAN))
 
 static inline struct bt_bap_stream *audio_stream_to_bap_stream(struct btp_bap_audio_stream *stream)
 {
 	return &stream->cap_stream.bap_stream;
 }
+struct tx_stream {
+	struct bt_bap_stream *bap_stream;
+	uint16_t seq_num;
+	size_t tx_completed;
+	atomic_t enqueued;
+} tx_streams[CONFIG_BT_ISO_MAX_CHAN];
 
-static void audio_clock_timeout(struct k_work *work)
+static bool stream_is_streaming(const struct bt_bap_stream *bap_stream)
 {
-	struct btp_bap_audio_stream *stream;
-	struct k_work_delayable *dwork;
+	struct bt_bap_ep_info ep_info;
+	int err;
+
+	if (bap_stream == NULL) {
+		return false;
+	}
+
+	/* No-op if stream is not configured */
+	if (bap_stream->ep == NULL) {
+		return false;
+	}
 
-	dwork = k_work_delayable_from_work(work);
-	stream = CONTAINER_OF(dwork, struct btp_bap_audio_stream, audio_clock_work);
-	atomic_inc(&stream->seq_num);
+	err = bt_bap_ep_get_info(bap_stream->ep, &ep_info);
+	__ASSERT_NO_MSG(err == 0);
 
-	k_work_schedule(dwork, K_USEC(audio_stream_to_bap_stream(stream)->qos->interval));
+	if (ep_info.iso_chan == NULL || ep_info.iso_chan->state != BT_ISO_STATE_CONNECTED) {
+		return false;
+	}
+
+	return ep_info.state == BT_BAP_EP_STATE_STREAMING;
 }
 
-static void audio_send_timeout(struct k_work *work)
+static void tx_thread_func(void *arg1, void *arg2, void *arg3)
 {
-	struct bt_iso_tx_info info;
-	struct btp_bap_audio_stream *stream;
-	struct bt_bap_stream *bap_stream;
-	struct k_work_delayable *dwork;
-	struct net_buf *buf;
-	uint32_t size;
-	uint8_t *data;
-	int err;
-
-	dwork = k_work_delayable_from_work(work);
-	stream = CONTAINER_OF(dwork, struct btp_bap_audio_stream, audio_send_work);
-	bap_stream = audio_stream_to_bap_stream(stream);
+	NET_BUF_POOL_FIXED_DEFINE(tx_pool, CONFIG_BT_ISO_TX_BUF_COUNT,
+				  BT_ISO_SDU_BUF_SIZE(CONFIG_BT_ISO_TX_MTU),
+				  CONFIG_BT_CONN_TX_USER_DATA_SIZE, NULL);
+
+	/* This loop will attempt to send on all streams in the streaming state in a round robin
+	 * fashion.
+	 * The TX is controlled by the number of buffers configured, and increasing
+	 * CONFIG_BT_ISO_TX_BUF_COUNT will allow for more streams in parallel, or to submit more
+	 * buffers per stream.
+	 * Once a buffer has been freed by the stack, it triggers the next TX.
+	 */
+	while (true) {
+		int err = -ENOEXEC;
+
+		for (size_t i = 0U; i < ARRAY_SIZE(tx_streams); i++) {
+			struct bt_bap_stream *bap_stream = tx_streams[i].bap_stream;
+			struct net_buf *buf;
+
+			if (!stream_is_streaming(bap_stream) ||
+			    atomic_get(&tx_streams[i].enqueued) >= MAX_ENQUEUE_CNT) {
+				continue;
+			}
 
-	if (stream->last_req_seq_num % 201 == 200) {
-		err = bt_bap_stream_get_tx_sync(bap_stream, &info);
-		if (err != 0) {
-			LOG_DBG("Failed to get last seq num: err %d", err);
-		} else if (stream->last_req_seq_num > info.seq_num) {
-			LOG_DBG("Previous TX request rejected by the controller: requested seq %u,"
-				" last accepted seq %u", stream->last_req_seq_num, info.seq_num);
-			stream->last_sent_seq_num = info.seq_num;
-		} else {
-			LOG_DBG("Host and Controller sequence number is in sync.");
-			stream->last_sent_seq_num = info.seq_num;
-		}
-		/* TODO: Synchronize the Host clock with the Controller clock */
-	}
+			buf = net_buf_alloc(&tx_pool, K_SECONDS(1));
+			__ASSERT(buf != NULL, "Failed to get a TX buffer");
 
-	/* Get buffer within a ring buffer memory */
-	size = ring_buf_get_claim(&audio_ring_buf, &data, bap_stream->qos->sdu);
-	if (size > 0) {
-		buf = net_buf_alloc(&tx_pool, K_NO_WAIT);
-		if (!buf) {
-			LOG_ERR("Cannot allocate net_buf. Dropping data.");
-		} else {
 			net_buf_reserve(buf, BT_ISO_CHAN_SEND_RESERVE);
-			net_buf_add_mem(buf, data, size);
-
-			/* Because the seq_num field of the audio_stream struct is atomic_val_t
-			 * (4 bytes), let's allow an overflow and just cast it to uint16_t.
-			 */
-			stream->last_req_seq_num = (uint16_t)atomic_get(&stream->seq_num);
 
-			LOG_DBG("Sending data to stream %p len %d seq %d", bap_stream, size,
-				stream->last_req_seq_num);
+			net_buf_add_mem(buf, btp_bap_audio_stream_mock_data, bap_stream->qos->sdu);
+
+			err = bt_bap_stream_send(bap_stream, buf, tx_streams[i].seq_num);
+			if (err == 0) {
+				tx_streams[i].seq_num++;
+				atomic_inc(&tx_streams[i].enqueued);
+			} else {
+				if (!stream_is_streaming(bap_stream)) {
+					/* Can happen if we disconnected while waiting for a
+					 * buffer - Ignore
+					 */
+				} else {
+					LOG_ERR("Unable to send: %d", err);
+				}
 
-			err = bt_bap_stream_send(bap_stream, buf, 0);
-			if (err != 0) {
-				LOG_ERR("Failed to send audio data to stream %p, err %d",
-					bap_stream, err);
 				net_buf_unref(buf);
 			}
 		}
 
-		/* Free ring buffer memory */
-		err = ring_buf_get_finish(&audio_ring_buf, size);
 		if (err != 0) {
-			LOG_ERR("Error freeing ring buffer memory: %d", err);
+			/* In case of any errors, retry with a delay */
+			k_sleep(K_MSEC(10));
 		}
 	}
-
-	k_work_schedule_for_queue(&iso_data_work_q, dwork,
-				  K_USEC(bap_stream->qos->interval));
 }
 
-void btp_bap_audio_stream_started(struct btp_bap_audio_stream *a_stream)
+int btp_bap_audio_stream_tx_register(struct btp_bap_audio_stream *stream)
 {
-	struct bt_bap_ep_info info;
-	struct bt_bap_stream *bap_stream = audio_stream_to_bap_stream(a_stream);
-
-	/* Callback called on transition to Streaming state */
+	if (stream == NULL) {
+		return -EINVAL;
+	}
 
-	LOG_DBG("Started stream %p", bap_stream);
+	if (!btp_bap_audio_stream_can_send(stream)) {
+		return -EINVAL;
+	}
 
-	(void)bt_bap_ep_get_info(bap_stream->ep, &info);
-	if (info.can_send == true) {
-		/* Schedule first TX ISO data at seq_num 1 instead of 0 to ensure
-		 * we are in sync with the controller at start of streaming.
-		 */
-		a_stream->seq_num = 1;
+	for (size_t i = 0U; i < ARRAY_SIZE(tx_streams); i++) {
+		if (tx_streams[i].bap_stream == NULL) {
+			tx_streams[i].bap_stream = audio_stream_to_bap_stream(stream);
 
-		/* Run audio clock work in system work queue */
-		k_work_init_delayable(&a_stream->audio_clock_work, audio_clock_timeout);
-		k_work_schedule(&a_stream->audio_clock_work, K_NO_WAIT);
+			LOG_INF("Registered %p (%p) for TX", stream, tx_streams[i].bap_stream);
 
-		/* Run audio send work in user defined work queue */
-		k_work_init_delayable(&a_stream->audio_send_work, audio_send_timeout);
-		k_work_schedule_for_queue(&iso_data_work_q, &a_stream->audio_send_work,
-					  K_USEC(bap_stream->qos->interval));
+			return 0;
+		}
 	}
+
+	return -ENOMEM;
 }
 
-void btp_bap_audio_stream_stopped(struct btp_bap_audio_stream *a_stream)
+int btp_bap_audio_stream_tx_unregister(struct btp_bap_audio_stream *stream)
 {
-	/* Stop send timer */
-	k_work_cancel_delayable(&a_stream->audio_clock_work);
-	k_work_cancel_delayable(&a_stream->audio_send_work);
+	const struct bt_bap_stream *bap_stream;
+
+	if (stream == NULL) {
+		return -EINVAL;
+	}
+
+	bap_stream = audio_stream_to_bap_stream(stream);
+
+	for (size_t i = 0U; i < ARRAY_SIZE(tx_streams); i++) {
+		if (tx_streams[i].bap_stream == bap_stream) {
+			memset(&tx_streams[i], 0, sizeof(tx_streams[i]));
+
+			LOG_INF("Unregistered %p for TX", bap_stream);
+
+			return 0;
+		}
+	}
+
+	return -ENODATA;
 }
 
-uint8_t btp_bap_audio_stream_send_data(const uint8_t *data, uint8_t data_len)
+void btp_bap_audio_stream_tx_init(void)
 {
-	return ring_buf_put(&audio_ring_buf, data, data_len);
+	static bool thread_started;
+
+	if (!thread_started) {
+		static K_KERNEL_STACK_DEFINE(tx_thread_stack, 1024U);
+		const int tx_thread_prio = K_PRIO_PREEMPT(5);
+		static struct k_thread tx_thread;
+
+		k_thread_create(&tx_thread, tx_thread_stack, K_KERNEL_STACK_SIZEOF(tx_thread_stack),
+				tx_thread_func, NULL, NULL, NULL, tx_thread_prio, 0, K_NO_WAIT);
+		k_thread_name_set(&tx_thread, "TX thread");
+		thread_started = true;
+	}
 }
 
-uint8_t btp_bap_audio_stream_send(const void *cmd, uint16_t cmd_len,
-				  void *rsp, uint16_t *rsp_len)
+bool btp_bap_audio_stream_can_send(struct btp_bap_audio_stream *stream)
 {
-	struct btp_bap_send_rp *rp = rsp;
-	const struct btp_bap_send_cmd *cp = cmd;
-	uint32_t ret;
+	struct bt_bap_stream *bap_stream;
+	struct bt_bap_ep_info info;
+	int err;
+
+	if (stream == NULL) {
+		return false;
+	}
 
-	ret = btp_bap_audio_stream_send_data(cp->data, cp->data_len);
+	bap_stream = audio_stream_to_bap_stream(stream);
+	if (bap_stream->ep == NULL) {
+		return false;
+	}
 
-	rp->data_len = ret;
-	*rsp_len = sizeof(*rp);
+	err = bt_bap_ep_get_info(bap_stream->ep, &info);
+	__ASSERT_NO_MSG(err == 0);
 
-	return BTP_STATUS_SUCCESS;
+	return info.can_send;
 }
 
-int btp_bap_audio_stream_init_send_worker(void)
+void btp_bap_audio_stream_sent_cb(struct bt_bap_stream *stream)
 {
-	if (send_worker_inited) {
-		return 0;
-	}
+	for (size_t i = 0U; i < ARRAY_SIZE(tx_streams); i++) {
+		if (tx_streams[i].bap_stream == stream) {
+			const atomic_val_t old = atomic_dec(&tx_streams[i].enqueued);
 
-	k_work_queue_init(&iso_data_work_q);
-	k_work_queue_start(&iso_data_work_q, iso_data_thread_stack_area,
-			   K_THREAD_STACK_SIZEOF(iso_data_thread_stack_area),
-			   ISO_DATA_THREAD_PRIORITY, NULL);
+			__ASSERT_NO_MSG(old != 0);
 
-	send_worker_inited = true;
+			tx_streams[i].tx_completed++;
+			if ((tx_streams[i].tx_completed % 100U) == 0U) {
+				LOG_INF("Stream %p sent %zu SDUs of size %u", stream,
+					tx_streams[i].tx_completed, stream->qos->sdu);
+			}
 
-	return 0;
+			break;
+		}
+	}
 }