CRAS: audio_thread - Register callbacks for arbitrary events

Bluetooth audio relies on polling A2DP or SCO socket to write or read
packets. However polling for POLLOUT or POLLIN is not sufficient for
CRAS to handle the special case when connection terminates by remote.
Audio thread could keep waiting for next writable or readable event
until main thread stops the stream or close device.

BUG=b:155282782
TEST=Run PTS HFP/AG/ACC/BV-11-I to verify when remote side terminates
SCO connection, the file descriptor can be closed and reopen without
issue when next input stream connects.

Change-Id: If44534ac608c5703fd0575f6dc29ead673d9549f
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/third_party/adhd/+/2172847
Reviewed-by: En-Shuo Hsu <enshuo@chromium.org>
Tested-by: Hsinyu Chao <hychao@chromium.org>
Commit-Queue: Hsinyu Chao <hychao@chromium.org>

9 files changed, 71 insertions, 69 deletions

diff --git a/cras/src/server/audio_thread.c b/cras/src/server/audio_thread.c
index e103226a..5482514b 100644
--- a/cras/src/server/audio_thread.c
+++ b/cras/src/server/audio_thread.c
@@ -117,7 +117,7 @@ static struct iodev_callback_list *iodev_callbacks;
 
 struct iodev_callback_list {
 	int fd;
-	int is_write;
+	int events;
 	int enabled;
 	thread_callback cb;
 	void *cb_data;
@@ -125,8 +125,8 @@ struct iodev_callback_list {
 	struct iodev_callback_list *prev, *next;
 };
 
-static void _audio_thread_add_callback(int fd, thread_callback cb, void *data,
-				       int is_write)
+void audio_thread_add_events_callback(int fd, thread_callback cb, void *data,
+				      int events)
 {
 	struct iodev_callback_list *iodev_cb;
 
@@ -140,21 +140,11 @@ static void _audio_thread_add_callback(int fd, thread_callback cb, void *data,
 	iodev_cb->cb = cb;
 	iodev_cb->cb_data = data;
 	iodev_cb->enabled = 1;
-	iodev_cb->is_write = is_write;
+	iodev_cb->events = events;
 
 	DL_APPEND(iodev_callbacks, iodev_cb);
 }
 
-void audio_thread_add_callback(int fd, thread_callback cb, void *data)
-{
-	_audio_thread_add_callback(fd, cb, data, 0);
-}
-
-void audio_thread_add_write_callback(int fd, thread_callback cb, void *data)
-{
-	_audio_thread_add_callback(fd, cb, data, 1);
-}
-
 void audio_thread_rm_callback(int fd)
 {
 	struct iodev_callback_list *iodev_cb;
@@ -742,13 +732,10 @@ static int fill_next_sleep_interval(struct audio_thread *thread,
 }
 
 static struct pollfd *add_pollfd(struct audio_thread *thread, int fd,
-				 int is_write)
+				 int events)
 {
 	thread->pollfds[thread->num_pollfds].fd = fd;
-	if (is_write)
-		thread->pollfds[thread->num_pollfds].events = POLLOUT;
-	else
-		thread->pollfds[thread->num_pollfds].events = POLLIN;
+	thread->pollfds[thread->num_pollfds].events = events;
 	thread->num_pollfds++;
 	if (thread->num_pollfds >= thread->pollfds_size) {
 		thread->pollfds_size *= 2;
@@ -850,7 +837,7 @@ static void *audio_io_thread(void *arg)
 				continue;
 			}
 			iodev_cb->pollfd = add_pollfd(thread, iodev_cb->fd,
-						      iodev_cb->is_write);
+						      iodev_cb->events);
 			if (!iodev_cb->pollfd)
 				goto restart_poll_loop;
 		}
@@ -861,7 +848,7 @@ static void *audio_io_thread(void *arg)
 				int fd = dev_stream_poll_stream_fd(curr);
 				if (fd < 0)
 					continue;
-				if (!add_pollfd(thread, fd, 0))
+				if (!add_pollfd(thread, fd, POLLIN))
 					goto restart_poll_loop;
 			}
 		}
@@ -870,7 +857,7 @@ static void *audio_io_thread(void *arg)
 				int fd = dev_stream_poll_stream_fd(curr);
 				if (fd < 0)
 					continue;
-				if (!add_pollfd(thread, fd, 0))
+				if (!add_pollfd(thread, fd, POLLIN))
 					goto restart_poll_loop;
 			}
 		}
@@ -899,10 +886,12 @@ static void *audio_io_thread(void *arg)
 
 		DL_FOREACH (iodev_callbacks, iodev_cb) {
 			if (iodev_cb->pollfd &&
-			    iodev_cb->pollfd->revents & (POLLIN | POLLOUT)) {
+			    iodev_cb->pollfd->revents & iodev_cb->events) {
 				ATLOG(atlog, AUDIO_THREAD_IODEV_CB,
-				      iodev_cb->is_write, 0, 0);
-				iodev_cb->cb(iodev_cb->cb_data);
+				      iodev_cb->pollfd->revents,
+				      iodev_cb->events, 0);
+				iodev_cb->cb(iodev_cb->cb_data,
+					     iodev_cb->pollfd->revents);
 			}
 		}
 	}
diff --git a/cras/src/server/audio_thread.h b/cras/src/server/audio_thread.h
index 5e5e9956..fbb2dbab 100644
--- a/cras/src/server/audio_thread.h
+++ b/cras/src/server/audio_thread.h
@@ -48,8 +48,9 @@ struct audio_thread {
 /* Callback function to be handled in main loop in audio thread.
  * Args:
  *    data - The data for callback function.
+ *    revent - The returned event from ppoll().
  */
-typedef int (*thread_callback)(void *data);
+typedef int (*thread_callback)(void *data, int revent);
 
 /* Creates an audio thread.
  * Returns:
@@ -83,23 +84,16 @@ int audio_thread_rm_open_dev(struct audio_thread *thread,
 int audio_thread_is_dev_open(struct audio_thread *thread,
 			     struct cras_iodev *dev);
 
-/* Adds an thread_callback to audio thread.
+/* Adds a thread_callback to audio thread for requested events.
  * Args:
  *    fd - The file descriptor to be polled for the callback.
- *      The callback will be called when fd is readable.
+ *      The callback will be called when any of requested events matched.
  *    cb - The callback function.
  *    data - The data for the callback function.
+ *    events - The requested events to ppoll().
  */
-void audio_thread_add_callback(int fd, thread_callback cb, void *data);
-
-/* Adds an thread_callback to audio thread.
- * Args:
- *    fd - The file descriptor to be polled for the callback.
- *      The callback will be called when fd is writeable.
- *    cb - The callback function.
- *    data - The data for the callback function.
- */
-void audio_thread_add_write_callback(int fd, thread_callback cb, void *data);
+void audio_thread_add_events_callback(int fd, thread_callback cb, void *data,
+				      int events);
 
 /* Removes an thread_callback from audio thread.
  * Args:
diff --git a/cras/src/server/cras_a2dp_iodev.c b/cras/src/server/cras_a2dp_iodev.c
index de8bd928..20a949dc 100644
--- a/cras/src/server/cras_a2dp_iodev.c
+++ b/cras/src/server/cras_a2dp_iodev.c
@@ -230,7 +230,7 @@ static int encode_a2dp_packet(struct a2dp_io *a2dpio)
 /*
  * To be called when a2dp socket becomes writable.
  */
-static int a2dp_socket_write_cb(void *arg)
+static int a2dp_socket_write_cb(void *arg, int revent)
 {
 	struct cras_iodev *iodev = (struct cras_iodev *)arg;
 	return encode_and_flush(iodev);
@@ -300,8 +300,9 @@ static int configure_dev(struct cras_iodev *iodev)
 	 */
 	iodev->min_buffer_level = a2dpio->write_block;
 
-	audio_thread_add_write_callback(cras_bt_transport_fd(a2dpio->transport),
-					a2dp_socket_write_cb, iodev);
+	audio_thread_add_events_callback(
+		cras_bt_transport_fd(a2dpio->transport), a2dp_socket_write_cb,
+		iodev, POLLOUT | POLLERR | POLLHUP);
 	audio_thread_enable_callback(cras_bt_transport_fd(a2dpio->transport),
 				     0);
 	return 0;
diff --git a/cras/src/server/cras_alsa_io.c b/cras/src/server/cras_alsa_io.c
index eabbc2e9..aeb6739b 100644
--- a/cras/src/server/cras_alsa_io.c
+++ b/cras/src/server/cras_alsa_io.c
@@ -359,7 +359,7 @@ static int close_dev(struct cras_iodev *iodev)
 	return 0;
 }
 
-static int dummy_hotword_cb(void *arg)
+static int dummy_hotword_cb(void *arg, int revents)
 {
 	/* Only need this once. */
 	struct alsa_io *aio = (struct alsa_io *)arg;
@@ -474,8 +474,8 @@ static int configure_dev(struct cras_iodev *iodev)
 		free(ufds);
 
 		if (aio->poll_fd >= 0)
-			audio_thread_add_callback(aio->poll_fd,
-						  dummy_hotword_cb, aio);
+			audio_thread_add_events_callback(
+				aio->poll_fd, dummy_hotword_cb, aio, POLLIN);
 	}
 
 	/* Capture starts right away, playback will wait for samples. */
diff --git a/cras/src/server/cras_hfp_info.c b/cras/src/server/cras_hfp_info.c
index 8d05f089..9842fde6 100644
--- a/cras/src/server/cras_hfp_info.c
+++ b/cras/src/server/cras_hfp_info.c
@@ -541,7 +541,7 @@ recv_sample:
  * 2. When input device not attached, ignore the data just read.
  * 3. When output device attached, write one chunk of MTU bytes of data.
  */
-static int hfp_info_callback(void *arg)
+static int hfp_info_callback(void *arg, int revents)
 {
 	struct hfp_info *info = (struct hfp_info *)arg;
 	int err;
@@ -549,16 +549,23 @@ static int hfp_info_callback(void *arg)
 	if (!info->started)
 		return 0;
 
-	err = info->read_cb(info);
-	if (err < 0) {
-		syslog(LOG_ERR, "Read error");
-		goto read_write_error;
+	/* Allow last read before handling error or hang-up events. */
+	if (revents & POLLIN) {
+		err = info->read_cb(info);
+		if (err < 0) {
+			syslog(LOG_ERR, "Read error");
+			goto read_write_error;
+		}
 	}
-
 	/* Ignore the bytes just read if input dev not in present */
 	if (!info->input_format_bytes)
 		buf_increment_read(info->capture_buf, err);
 
+	if (revents & (POLLERR | POLLHUP)) {
+		syslog(LOG_ERR, "Error polling SCO socket, revent %d", revents);
+		goto read_write_error;
+	}
+
 	/* Without output stream's presence, we shall still send zero packets
 	 * to HF. This is required for some HF devices to start sending non-zero
 	 * data to AG.
@@ -642,7 +649,8 @@ int hfp_info_start(int fd, unsigned int mtu, struct hfp_info *info)
 	buf_reset(info->playback_buf);
 	buf_reset(info->capture_buf);
 
-	audio_thread_add_callback(info->fd, hfp_info_callback, info);
+	audio_thread_add_events_callback(info->fd, hfp_info_callback, info,
+					 POLLIN | POLLERR | POLLHUP);
 
 	info->started = 1;
 	info->msbc_num_out_frames = 0;
diff --git a/cras/src/tests/a2dp_iodev_unittest.cc b/cras/src/tests/a2dp_iodev_unittest.cc
index a3521eec..1f6d991f 100644
--- a/cras/src/tests/a2dp_iodev_unittest.cc
+++ b/cras/src/tests/a2dp_iodev_unittest.cc
@@ -297,7 +297,7 @@ TEST_F(A2dpIodev, FramesQueued) {
 
   /* Some time has passed, same amount of frames are queued. */
   time_now.tv_nsec = 15000000;
-  write_callback(write_callback_data);
+  write_callback(write_callback_data, POLLOUT);
   EXPECT_EQ(104, iodev->frames_queued(iodev, &tstamp));
 
   /* Put 900 more frames. next_flush_time not yet passed so expect
@@ -407,7 +407,7 @@ TEST_F(A2dpIodev, SleepTimeWithWriteThrottle) {
   /* Fake the event that socket becomes writable so data continues to flush.
    * next_flush_time fast forwards by another flush_period. */
   a2dp_write_return_val[3] = 0;
-  write_callback(write_callback_data);
+  write_callback(write_callback_data, POLLOUT);
   EXPECT_EQ(312, iodev->frames_queued(iodev, &tstamp)); /* 1208 - 896 */
 
   /* Expect to sleep the time between now and next_flush_time(~60.9ms). */
@@ -801,7 +801,10 @@ struct audio_thread* cras_iodev_list_get_audio_thread() {
 // From audio_thread
 struct audio_thread_event_log* atlog;
 
-void audio_thread_add_write_callback(int fd, thread_callback cb, void* data) {
+void audio_thread_add_events_callback(int fd,
+                                      thread_callback cb,
+                                      void* data,
+                                      int events) {
   write_callback = cb;
   write_callback_data = data;
 }
diff --git a/cras/src/tests/alsa_io_unittest.cc b/cras/src/tests/alsa_io_unittest.cc
index c6500213..7f4202d3 100644
--- a/cras/src/tests/alsa_io_unittest.cc
+++ b/cras/src/tests/alsa_io_unittest.cc
@@ -2142,7 +2142,7 @@ TEST(AlsaHotwordNode, HotwordTriggeredSendMessage) {
   ASSERT_EQ(0, rc);
 
   ASSERT_NE(reinterpret_cast<thread_callback>(NULL), audio_thread_cb);
-  audio_thread_cb(audio_thread_cb_data);
+  audio_thread_cb(audio_thread_cb_data, POLLIN);
   EXPECT_EQ(1, hotword_send_triggered_msg_called);
   alsa_iodev_destroy(iodev);
 }
@@ -2763,7 +2763,10 @@ void cras_audio_area_config_buf_pointers(struct cras_audio_area* area,
                                          const struct cras_audio_format* fmt,
                                          uint8_t* base_buffer) {}
 
-void audio_thread_add_callback(int fd, thread_callback cb, void* data) {
+void audio_thread_add_events_callback(int fd,
+                                      thread_callback cb,
+                                      void* data,
+                                      int events) {
   audio_thread_cb = cb;
   audio_thread_cb_data = data;
 }
diff --git a/cras/src/tests/hfp_info_unittest.cc b/cras/src/tests/hfp_info_unittest.cc
index 482c3a99..99daeff9 100644
--- a/cras/src/tests/hfp_info_unittest.cc
+++ b/cras/src/tests/hfp_info_unittest.cc
@@ -256,7 +256,7 @@ TEST(HfpInfo, StartHfpInfoAndRead) {
   send(sock[0], sample, 48, 0);
 
   /* Trigger thread callback */
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
 
   dev.direction = CRAS_STREAM_INPUT;
   ASSERT_EQ(0, hfp_info_add_iodev(info, dev.direction, dev.format));
@@ -268,7 +268,7 @@ TEST(HfpInfo, StartHfpInfoAndRead) {
   /* Trigger thread callback after idev added. */
   ts.tv_sec = 0;
   ts.tv_nsec = 5000000;
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
 
   rc = hfp_buf_queued(info, dev.direction);
   ASSERT_EQ(48 / 2, rc);
@@ -300,7 +300,7 @@ TEST(HfpInfo, StartHfpInfoAndWrite) {
   send(sock[0], sample, 48, 0);
 
   /* Trigger thread callback */
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
 
   /* Without odev in presence, zero packet should be sent. */
   rc = recv(sock[0], sample, 48, 0);
@@ -314,7 +314,7 @@ TEST(HfpInfo, StartHfpInfoAndWrite) {
 
   /* Put some fake data and trigger thread callback again */
   buf_increment_write(info->playback_buf, 1008);
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
 
   /* Assert some samples written */
   rc = recv(sock[0], sample, 48, 0);
@@ -373,7 +373,7 @@ TEST(HfpInfo, StartHfpInfoAndReadMsbc) {
   send_mSBC_packet(sock[0], pkt_count++, 0);
 
   /* Trigger thread callback */
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
 
   /* Expect one empty mSBC packet is send, because no odev in presence. */
   rc = recv(sock[0], sample, MSBC_PKT_SIZE, 0);
@@ -388,7 +388,7 @@ TEST(HfpInfo, StartHfpInfoAndReadMsbc) {
   send_mSBC_packet(sock[0], pkt_count, 0);
 
   /* Trigger thread callback after idev added. */
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
   rc = recv(sock[0], sample, MSBC_PKT_SIZE, 0);
   ASSERT_EQ(MSBC_PKT_SIZE, rc);
 
@@ -401,7 +401,7 @@ TEST(HfpInfo, StartHfpInfoAndReadMsbc) {
    */
   pkt_count++;
   send_mSBC_packet(sock[0], pkt_count, 0);
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
   rc = recv(sock[0], sample, MSBC_PKT_SIZE, 0);
   ASSERT_EQ(MSBC_PKT_SIZE, rc);
 
@@ -418,7 +418,7 @@ TEST(HfpInfo, StartHfpInfoAndReadMsbc) {
 
   set_sbc_codec_decoded_fail(1);
 
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
   rc = recv(sock[0], sample, MSBC_PKT_SIZE, 0);
   ASSERT_EQ(MSBC_PKT_SIZE, rc);
 
@@ -434,7 +434,7 @@ TEST(HfpInfo, StartHfpInfoAndReadMsbc) {
 
   set_sbc_codec_decoded_fail(1);
 
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
   rc = recv(sock[0], sample, MSBC_PKT_SIZE, 0);
   ASSERT_EQ(MSBC_PKT_SIZE, rc);
 
@@ -466,7 +466,7 @@ TEST(HfpInfo, StartHfpInfoAndWriteMsbc) {
   send(sock[0], sample, 63, 0);
 
   /* Trigger thread callback */
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
 
   dev.direction = CRAS_STREAM_OUTPUT;
   ASSERT_EQ(0, hfp_info_add_iodev(info, dev.direction, dev.format));
@@ -477,7 +477,7 @@ TEST(HfpInfo, StartHfpInfoAndWriteMsbc) {
   /* Put some fake data and trigger thread callback again */
   send(sock[0], sample, 63, 0);
   buf_increment_write(info->playback_buf, 240);
-  thread_cb((struct hfp_info*)cb_data);
+  thread_cb((struct hfp_info*)cb_data, POLLIN);
 
   /* Assert some samples written */
   rc = recv(sock[0], sample, 60, 0);
@@ -496,7 +496,10 @@ struct audio_thread* cras_iodev_list_get_audio_thread() {
   return NULL;
 }
 
-void audio_thread_add_callback(int fd, thread_callback cb, void* data) {
+void audio_thread_add_events_callback(int fd,
+                                      thread_callback cb,
+                                      void* data,
+                                      int events) {
   thread_cb = cb;
   cb_data = data;
   return;
diff --git a/cras/src/tools/cras_test_client/cras_test_client.c b/cras/src/tools/cras_test_client/cras_test_client.c
index 0b0a29cf..9c43f446 100644
--- a/cras/src/tools/cras_test_client/cras_test_client.c
+++ b/cras/src/tools/cras_test_client/cras_test_client.c
@@ -637,7 +637,8 @@ static void show_alog_tag(const struct audio_thread_event_log *log,
 		printf("%-30s dev:%u\n", "DEV_REMOVED", data1);
 		break;
 	case AUDIO_THREAD_IODEV_CB:
-		printf("%-30s is_write:%u\n", "IODEV_CB", data1);
+		printf("%-30s revents:%u events:%u\n", "IODEV_CB", data1,
+		       data2);
 		break;
 	case AUDIO_THREAD_PB_MSG:
 		printf("%-30s msg_id:%u\n", "PB_MSG", data1);