Wait for internal buffers to be drained

When ALSA client requests PCM drain, wait until our internal buffer
is consumed. Note, that samples might still be dropped (not played by
the Bluetooth device), because there is not possible to drain BlueZ
internal buffers (there is no API for that). One possible solution is
to releasing A2DP transport asynchronously with some arbitrary delay.

Fixes #67

src/asound/bluealsa-pcm.c

@@ -342,8 +342,9 @@ static int bluealsa_prepare(snd_pcm_ioplug_t *io) {
 }
 
 static int bluealsa_drain(snd_pcm_ioplug_t *io) {
-	(void)io;
-	debug("Draining");
+	struct bluealsa_pcm *pcm = io->private_data;
+	if (bluealsa_drain_transport(pcm->fd, pcm->transport) == -1)
+		return -errno;
 	return 0;
 }
 

src/ctl.c

@@ -455,6 +455,9 @@ static void ctl_thread_cmd_pcm_control(const struct request *req, int fd) {
 	case COMMAND_PCM_RESUME:
 		transport_set_state(t, TRANSPORT_ACTIVE);
 		break;
+	case COMMAND_PCM_DRAIN:
+		transport_drain_pcm(t);
+		break;
 	case COMMAND_PCM_READY:
 		break;
 	default:
@@ -483,6 +486,7 @@ static void *ctl_thread(void *arg) {
 		[COMMAND_PCM_CLOSE] = ctl_thread_cmd_pcm_close,
 		[COMMAND_PCM_PAUSE] = ctl_thread_cmd_pcm_control,
 		[COMMAND_PCM_RESUME] = ctl_thread_cmd_pcm_control,
+		[COMMAND_PCM_DRAIN] = ctl_thread_cmd_pcm_control,
 		[COMMAND_PCM_READY] = ctl_thread_cmd_pcm_control,
 	};
 

src/io.c

@@ -415,7 +415,12 @@ void *io_thread_a2dp_source_sbc(void *arg) {
 		/* add PCM socket to the poll if transport is active */
 		pfds[1].fd = t->state == TRANSPORT_ACTIVE ? t->a2dp.pcm.fd : -1;
 
-		if (poll(pfds, sizeof(pfds) / sizeof(*pfds), -1) == -1) {
+		switch (poll(pfds, sizeof(pfds) / sizeof(*pfds), 100)) {
+		case 0:
+			if (t->a2dp.pcm.sync)
+				pthread_cond_signal(&t->a2dp.pcm.drained);
+			continue;
+		case -1:
 			error("Transport poll error: %s", strerror(errno));
 			goto fail;
 		}
@@ -852,7 +857,12 @@ void *io_thread_a2dp_source_aac(void *arg) {
 		/* add PCM socket to the poll if transport is active */
 		pfds[1].fd = t->state == TRANSPORT_ACTIVE ? t->a2dp.pcm.fd : -1;
 
-		if (poll(pfds, sizeof(pfds) / sizeof(*pfds), -1) == -1) {
+		switch (poll(pfds, sizeof(pfds) / sizeof(*pfds), 100)) {
+		case 0:
+			if (t->a2dp.pcm.sync)
+				pthread_cond_signal(&t->a2dp.pcm.drained);
+			continue;
+		case -1:
 			error("Transport poll error: %s", strerror(errno));
 			goto fail;
 		}
@@ -1031,7 +1041,12 @@ void *io_thread_a2dp_source_aptx(void *arg) {
 		/* add PCM socket to the poll if transport is active */
 		pfds[1].fd = t->state == TRANSPORT_ACTIVE ? t->a2dp.pcm.fd : -1;
 
-		if (poll(pfds, sizeof(pfds) / sizeof(*pfds), -1) == -1) {
+		switch (poll(pfds, sizeof(pfds) / sizeof(*pfds), 100)) {
+		case 0:
+			if (t->a2dp.pcm.sync)
+				pthread_cond_signal(&t->a2dp.pcm.drained);
+			continue;
+		case -1:
 			error("Transport poll error: %s", strerror(errno));
 			goto fail;
 		}
@@ -1190,7 +1205,12 @@ void *io_thread_sco(void *arg) {
 		if (t->sco.mic_pcm.fd == -1)
 			pfds[1].fd = -1;
 
-		if (poll(pfds, sizeof(pfds) / sizeof(*pfds), -1) == -1) {
+		switch (poll(pfds, sizeof(pfds) / sizeof(*pfds), 100)) {
+		case 0:
+			if (t->sco.spk_pcm.sync)
+				pthread_cond_signal(&t->sco.spk_pcm.drained);
+			continue;
+		case -1:
 			error("Transport poll error: %s", strerror(errno));
 			goto fail;
 		}

src/shared/ctl-client.c

@@ -392,3 +392,28 @@ int bluealsa_pause_transport(int fd, const struct msg_transport *transport, bool
 
 	return bluealsa_send_request(fd, &req);
 }
+
+/**
+ * Drain PCM transport.
+ *
+ * @param fd Opened socket file descriptor.
+ * @param transport Address to the transport structure with the addr, type
+ *   and stream fields set - other fields are not used by this function.
+ * @return Upon success this function returns 0. Otherwise, -1 is returned. */
+int bluealsa_drain_transport(int fd, const struct msg_transport *transport) {
+
+	struct request req = {
+		.command = COMMAND_PCM_DRAIN,
+		.addr = transport->addr,
+		.type = transport->type,
+		.stream = transport->stream,
+	};
+
+#if DEBUG
+	char addr_[18];
+	ba2str_(&req.addr, addr_);
+	debug("Requesting PCM drain for %s", addr_);
+#endif
+
+	return bluealsa_send_request(fd, &req);
+}

src/shared/ctl-client.h

@@ -29,5 +29,6 @@ int bluealsa_get_transport_delay(int fd, const struct msg_transport *transport);
 int bluealsa_open_transport(int fd, const struct msg_transport *transport);
 int bluealsa_close_transport(int fd, const struct msg_transport *transport);
 int bluealsa_pause_transport(int fd, const struct msg_transport *transport, bool pause);
+int bluealsa_drain_transport(int fd, const struct msg_transport *transport);
 
 #endif

src/shared/ctl-proto.h

@@ -33,6 +33,7 @@ enum command {
 	COMMAND_PCM_CLOSE,
 	COMMAND_PCM_PAUSE,
 	COMMAND_PCM_RESUME,
+	COMMAND_PCM_DRAIN,
 	COMMAND_PCM_READY,
 	__COMMAND_MAX
 };

src/transport.c

@@ -285,6 +285,8 @@ struct ba_transport *transport_new_a2dp(
 
 	t->a2dp.pcm.fd = -1;
 	t->a2dp.pcm.client = -1;
+	pthread_cond_init(&t->a2dp.pcm.drained, NULL);
+	pthread_mutex_init(&t->a2dp.pcm.drained_mn, NULL);
 
 	bluealsa_event();
 	return t;
@@ -316,8 +318,13 @@ struct ba_transport *transport_new_rfcomm(
 
 	t_sco->sco.spk_pcm.fd = -1;
 	t_sco->sco.spk_pcm.client = -1;
+	pthread_cond_init(&t_sco->sco.spk_pcm.drained, NULL);
+	pthread_mutex_init(&t_sco->sco.spk_pcm.drained_mn, NULL);
+
 	t_sco->sco.mic_pcm.fd = -1;
 	t_sco->sco.mic_pcm.client = -1;
+	pthread_cond_init(&t_sco->sco.mic_pcm.drained, NULL);
+	pthread_mutex_init(&t_sco->sco.mic_pcm.drained_mn, NULL);
 
 	transport_set_state(t_sco, TRANSPORT_ACTIVE);
 
@@ -362,6 +369,8 @@ void transport_free(struct ba_transport *t) {
 	switch (t->type) {
 	case TRANSPORT_TYPE_A2DP:
 		transport_release_pcm(&t->a2dp.pcm);
+		pthread_cond_destroy(&t->a2dp.pcm.drained);
+		pthread_mutex_destroy(&t->a2dp.pcm.drained_mn);
 		free(t->a2dp.cconfig);
 		break;
 	case TRANSPORT_TYPE_RFCOMM:
@@ -371,7 +380,11 @@ void transport_free(struct ba_transport *t) {
 		break;
 	case TRANSPORT_TYPE_SCO:
 		transport_release_pcm(&t->sco.spk_pcm);
+		pthread_cond_destroy(&t->sco.spk_pcm.drained);
+		pthread_mutex_destroy(&t->sco.spk_pcm.drained_mn);
 		transport_release_pcm(&t->sco.mic_pcm);
+		pthread_cond_destroy(&t->sco.mic_pcm.drained);
+		pthread_mutex_destroy(&t->sco.mic_pcm.drained_mn);
 		t->sco.rfcomm->rfcomm.sco = NULL;
 		break;
 	}
@@ -711,21 +724,66 @@ int transport_set_state_from_string(struct ba_transport *t, const char *state) {
 	return 0;
 }
 
+int transport_drain_pcm(struct ba_transport *t) {
+
+	struct ba_pcm *pcm = NULL;
+
+	switch (t->profile) {
+	case BLUETOOTH_PROFILE_NULL:
+	case BLUETOOTH_PROFILE_A2DP_SINK:
+		break;
+	case BLUETOOTH_PROFILE_A2DP_SOURCE:
+		pcm = &t->a2dp.pcm;
+		break;
+	case BLUETOOTH_PROFILE_HSP_AG:
+	case BLUETOOTH_PROFILE_HFP_AG:
+		pcm = &t->sco.spk_pcm;
+		break;
+	case BLUETOOTH_PROFILE_HSP_HS:
+	case BLUETOOTH_PROFILE_HFP_HF:
+		break;
+	}
+
+	if (pcm == NULL || t->state != TRANSPORT_ACTIVE)
+		return 0;
+
+	pthread_mutex_lock(&pcm->drained_mn);
+
+	pcm->sync = true;
+	pthread_cond_wait(&pcm->drained, &pcm->drained_mn);
+	pcm->sync = false;
+
+	pthread_mutex_unlock(&pcm->drained_mn);
+
+	/* TODO: Asynchronous transport release.
+	 *
+	 * Unfortunately, BlueZ does not provide API for internal buffer drain.
+	 * Also, there is no specification for Bluetooth playback drain. In order
+	 * to make sure, that all samples are played out, we have to wait some
+	 * arbitrary time before releasing transport. In order to make it right,
+	 * there is a requirement for an asynchronous release mechanism, which
+	 * is not implemented - it requires a little bit of refactoring. */
+	usleep(200000);
+
+	debug("PCM drained");
+	return 0;
+}
+
 int transport_acquire_bt_a2dp(struct ba_transport *t) {
 
 	GDBusMessage *msg, *rep;
 	GUnixFDList *fd_list;
 	GError *err = NULL;
 
-	msg = g_dbus_message_new_method_call(t->dbus_owner, t->dbus_path, "org.bluez.MediaTransport1",
-			t->state == TRANSPORT_PENDING ? "TryAcquire" : "Acquire");
-
 	if (t->bt_fd != -1) {
 		warn("Closing dangling BT socket: %d", t->bt_fd);
 		close(t->bt_fd);
 		t->bt_fd = -1;
 	}
 
+	msg = g_dbus_message_new_method_call(t->dbus_owner, t->dbus_path, "org.bluez.MediaTransport1",
+			t->state == TRANSPORT_PENDING ? "TryAcquire" : "Acquire");
+
 	if ((rep = g_dbus_connection_send_message_with_reply_sync(config.dbus, msg,
 					G_DBUS_SEND_MESSAGE_FLAGS_NONE, -1, NULL, NULL, &err)) == NULL)
 		goto fail;

src/transport.h

@@ -81,6 +81,11 @@ struct ba_pcm {
 	 * by the PCM client lookup function - transport_lookup_pcm_client() */
 	int client;
 
+	/* variables used for PCM synchronization */
+	bool sync;
+	pthread_cond_t drained;
+	pthread_mutex_t drained_mn;
+
 };
 
 struct ba_transport {
@@ -229,6 +234,8 @@ int transport_set_volume(struct ba_transport *t, uint8_t ch1_muted, uint8_t ch2_
 int transport_set_state(struct ba_transport *t, enum ba_transport_state state);
 int transport_set_state_from_string(struct ba_transport *t, const char *state);
 
+int transport_drain_pcm(struct ba_transport *t);
+
 int transport_acquire_bt_a2dp(struct ba_transport *t);
 int transport_release_bt_a2dp(struct ba_transport *t);