Keep transport thread signaling in PCM structure

src/a2dp-aac.c

@@ -593,10 +593,10 @@ void *a2dp_aac_dec_thread(struct ba_transport_pcm *t_pcm) {
 int a2dp_aac_transport_start(struct ba_transport *t) {
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aac_enc_thread, "ba-a2dp-aac", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aac_enc_thread, "ba-a2dp-aac");
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aac_dec_thread, "ba-a2dp-aac", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aac_dec_thread, "ba-a2dp-aac");
 
 	g_assert_not_reached();
 	return -1;

src/a2dp-aptx-hd.c

@@ -341,11 +341,11 @@ void *a2dp_aptx_hd_dec_thread(struct ba_transport_pcm *t_pcm) {
 int a2dp_aptx_hd_transport_start(struct ba_transport *t) {
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_hd_enc_thread, "ba-a2dp-aptx-hd", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_hd_enc_thread, "ba-a2dp-aptx-hd");
 
 #if HAVE_APTX_HD_DECODE
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_hd_dec_thread, "ba-a2dp-aptx-hd", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_hd_dec_thread, "ba-a2dp-aptx-hd");
 #endif
 
 	g_assert_not_reached();

src/a2dp-aptx.c

@@ -304,11 +304,11 @@ void *a2dp_aptx_dec_thread(struct ba_transport_pcm *t_pcm) {
 int a2dp_aptx_transport_start(struct ba_transport *t) {
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_enc_thread, "ba-a2dp-aptx", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_enc_thread, "ba-a2dp-aptx");
 
 #if HAVE_APTX_DECODE
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_dec_thread, "ba-a2dp-aptx", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_aptx_dec_thread, "ba-a2dp-aptx");
 #endif
 
 	g_assert_not_reached();

src/a2dp-faststream.c

@@ -323,17 +323,17 @@ int a2dp_faststream_transport_start(struct ba_transport *t) {
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE) {
 		if (t->a2dp.configuration.faststream.direction & FASTSTREAM_DIRECTION_MUSIC)
-			rv |= ba_transport_pcm_start(pcm, a2dp_faststream_enc_thread, "ba-a2dp-fs-m", true);
+			rv |= ba_transport_pcm_start(pcm, a2dp_faststream_enc_thread, "ba-a2dp-fs-m");
 		if (t->a2dp.configuration.faststream.direction & FASTSTREAM_DIRECTION_VOICE)
-			rv |= ba_transport_pcm_start(pcm_bc, a2dp_faststream_dec_thread, "ba-a2dp-fs-v", false);
+			rv |= ba_transport_pcm_start(pcm_bc, a2dp_faststream_dec_thread, "ba-a2dp-fs-v");
 		return rv;
 	}
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK) {
 		if (t->a2dp.configuration.faststream.direction & FASTSTREAM_DIRECTION_MUSIC)
-			rv |= ba_transport_pcm_start(pcm, a2dp_faststream_dec_thread, "ba-a2dp-fs-m", true);
+			rv |= ba_transport_pcm_start(pcm, a2dp_faststream_dec_thread, "ba-a2dp-fs-m");
 		if (t->a2dp.configuration.faststream.direction & FASTSTREAM_DIRECTION_VOICE)
-			rv |= ba_transport_pcm_start(pcm_bc, a2dp_faststream_enc_thread, "ba-a2dp-fs-v", false);
+			rv |= ba_transport_pcm_start(pcm_bc, a2dp_faststream_enc_thread, "ba-a2dp-fs-v");
 		return rv;
 	}
 

src/a2dp-lc3plus.c

@@ -609,10 +609,10 @@ void *a2dp_lc3plus_dec_thread(struct ba_transport_pcm *t_pcm) {
 int a2dp_lc3plus_transport_start(struct ba_transport *t) {
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_lc3plus_enc_thread, "ba-a2dp-lc3p", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_lc3plus_enc_thread, "ba-a2dp-lc3p");
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_lc3plus_dec_thread, "ba-a2dp-lc3p", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_lc3plus_dec_thread, "ba-a2dp-lc3p");
 
 	g_assert_not_reached();
 	return -1;

src/a2dp-ldac.c

@@ -407,11 +407,11 @@ void *a2dp_ldac_dec_thread(struct ba_transport_pcm *t_pcm) {
 int a2dp_ldac_transport_start(struct ba_transport *t) {
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_ldac_enc_thread, "ba-a2dp-ldac", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_ldac_enc_thread, "ba-a2dp-ldac");
 
 #if HAVE_LDAC_DECODE
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_ldac_dec_thread, "ba-a2dp-ldac", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_ldac_dec_thread, "ba-a2dp-ldac");
 #endif
 
 	g_assert_not_reached();

src/a2dp-mpeg.c

@@ -607,16 +607,16 @@ int a2dp_mpeg_transport_start(struct ba_transport *t) {
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE) {
 #if ENABLE_MP3LAME
 		if (t->a2dp.configuration.mpeg.layer == MPEG_LAYER_MP3)
-			return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_mp3_enc_thread, "ba-a2dp-mp3", true);
+			return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_mp3_enc_thread, "ba-a2dp-mp3");
 #endif
 	}
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK) {
 #if ENABLE_MPG123
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_mpeg_dec_thread, "ba-a2dp-mpeg", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_mpeg_dec_thread, "ba-a2dp-mpeg");
 #elif ENABLE_MP3LAME
 		if (t->a2dp.configuration.mpeg.layer == MPEG_LAYER_MP3)
-			return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_mpeg_dec_thread, "ba-a2dp-mp3", true);
+			return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_mpeg_dec_thread, "ba-a2dp-mp3");
 #endif
 	}
 

src/a2dp-sbc.c

@@ -440,10 +440,10 @@ void *a2dp_sbc_dec_thread(struct ba_transport_pcm *t_pcm) {
 int a2dp_sbc_transport_start(struct ba_transport *t) {
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SOURCE)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_sbc_enc_thread, "ba-a2dp-sbc", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_sbc_enc_thread, "ba-a2dp-sbc");
 
 	if (t->profile & BA_TRANSPORT_PROFILE_A2DP_SINK)
-		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_sbc_dec_thread, "ba-a2dp-sbc", true);
+		return ba_transport_pcm_start(&t->a2dp.pcm, a2dp_sbc_dec_thread, "ba-a2dp-sbc");
 
 	g_assert_not_reached();
 	return -1;

src/ba-transport-pcm.c

@@ -65,15 +65,18 @@ static const char *transport_get_dbus_path_type(
 int transport_pcm_init(
 		struct ba_transport_pcm *pcm,
 		enum ba_transport_pcm_mode mode,
-		struct ba_transport_thread *th) {
+		struct ba_transport_thread *th,
+		bool master) {
 
 	struct ba_transport *t = th->t;
 
 	pcm->t = t;
 	pcm->mode = mode;
+	pcm->master = master;
 	pcm->fd = -1;
 	pcm->fd_bt = -1;
-	pcm->active = true;
+	pcm->pipe[0] = -1;
+	pcm->pipe[1] = -1;
 
 	/* link PCM and transport thread */
 	pcm->th = th;
@@ -89,6 +92,9 @@ int transport_pcm_init(
 	pthread_mutex_init(&pcm->client_mtx, NULL);
 	pthread_cond_init(&pcm->cond, NULL);
 
+	if (pipe(pcm->pipe) == -1)
+		return -1;
+
 	pcm->delay_adjustments = g_hash_table_new(NULL, NULL);
 
 	pcm->ba_dbus_path = g_strdup_printf("%s/%s/%s",
@@ -110,6 +116,11 @@ void transport_pcm_free(
 	pthread_mutex_destroy(&pcm->client_mtx);
 	pthread_cond_destroy(&pcm->cond);
 
+	if (pcm->pipe[0] != -1)
+		close(pcm->pipe[0]);
+	if (pcm->pipe[1] != -1)
+		close(pcm->pipe[1]);
+
 	g_hash_table_unref(pcm->delay_adjustments);
 	g_free(pcm->ba_dbus_path);
 
@@ -142,8 +153,8 @@ void ba_transport_pcm_thread_cleanup(struct ba_transport_pcm *pcm) {
 	 * ba_transport_pcm_start() function or in the IO thread itself. */
 	ba_transport_pcm_bt_release(pcm);
 
-	/* If we are closing master thread, release underlying BT transport. */
-	if (th->master)
+	/* If we are closing master PCM, release underlying BT transport. */
+	if (pcm->master)
 		ba_transport_release(t);
 
 #if DEBUG
@@ -216,8 +227,7 @@ int ba_transport_pcm_bt_release(struct ba_transport_pcm *pcm) {
 int ba_transport_pcm_start(
 		struct ba_transport_pcm *pcm,
 		ba_transport_pcm_thread_func th_func,
-		const char *name,
-		bool master) {
+		const char *name) {
 
 	struct ba_transport *t = pcm->t;
 	struct ba_transport_thread *th = pcm->th;
@@ -226,7 +236,6 @@ int ba_transport_pcm_start(
 
 	pthread_mutex_lock(&th->mutex);
 
-	th->master = master;
 	th->state = BA_TRANSPORT_THREAD_STATE_STARTING;
 
 	/* Please note, this call here does not guarantee that the BT socket
@@ -367,20 +376,20 @@ int ba_transport_pcm_pause(struct ba_transport_pcm *pcm) {
 
 	pthread_mutex_lock(&pcm->mutex);
 	debug("PCM pause: %d", pcm->fd);
-	pcm->active = false;
+	pcm->paused = true;
 	pthread_mutex_unlock(&pcm->mutex);
 
-	return ba_transport_thread_signal_send(pcm->th, BA_TRANSPORT_THREAD_SIGNAL_PCM_PAUSE);
+	return ba_transport_pcm_signal_send(pcm, BA_TRANSPORT_PCM_SIGNAL_PAUSE);
 }
 
 int ba_transport_pcm_resume(struct ba_transport_pcm *pcm) {
 
 	pthread_mutex_lock(&pcm->mutex);
 	debug("PCM resume: %d", pcm->fd);
-	pcm->active = true;
+	pcm->paused = false;
 	pthread_mutex_unlock(&pcm->mutex);
 
-	return ba_transport_thread_signal_send(pcm->th, BA_TRANSPORT_THREAD_SIGNAL_PCM_RESUME);
+	return ba_transport_pcm_signal_send(pcm, BA_TRANSPORT_PCM_SIGNAL_RESUME);
 }
 
 int ba_transport_pcm_drain(struct ba_transport_pcm *pcm) {
@@ -395,7 +404,7 @@ int ba_transport_pcm_drain(struct ba_transport_pcm *pcm) {
 	debug("PCM drain: %d", pcm->fd);
 
 	pcm->synced = false;
-	ba_transport_thread_signal_send(pcm->th, BA_TRANSPORT_THREAD_SIGNAL_PCM_SYNC);
+	ba_transport_pcm_signal_send(pcm, BA_TRANSPORT_PCM_SIGNAL_SYNC);
 
 	while (!pcm->synced)
 		pthread_cond_wait(&pcm->cond, &pcm->mutex);
@@ -427,16 +436,64 @@ int ba_transport_pcm_drop(struct ba_transport_pcm *pcm) {
 	if (io_pcm_flush(pcm) == -1)
 		return -1;
 
-	int rv = ba_transport_thread_signal_send(pcm->th, BA_TRANSPORT_THREAD_SIGNAL_PCM_DROP);
+	int rv = ba_transport_pcm_signal_send(pcm, BA_TRANSPORT_PCM_SIGNAL_DROP);
 	if (rv == -1 && errno == ESRCH)
 		rv = 0;
 
 	return rv;
 }
 
+int ba_transport_pcm_signal_send(
+		struct ba_transport_pcm *pcm,
+		enum ba_transport_pcm_signal signal) {
+
+	struct ba_transport_thread *th = pcm->th;
+	int ret = -1;
+
+	pthread_mutex_lock(&th->mutex);
+
+	if (th->state != BA_TRANSPORT_THREAD_STATE_RUNNING) {
+		errno = ESRCH;
+		goto fail;
+	}
+
+	if (write(pcm->pipe[1], &signal, sizeof(signal)) != sizeof(signal)) {
+		warn("Couldn't write transport PCM signal: %s", strerror(errno));
+		goto fail;
+	}
+
+	ret = 0;
+
+fail:
+	pthread_mutex_unlock(&th->mutex);
+	return ret;
+}
+
+/**
+ * Receive signal sent by ba_transport_pcm_signal_send().
+ *
+ * @note
+ * In case of error, this function will return -1 instead of signal value. */
+enum ba_transport_pcm_signal ba_transport_pcm_signal_recv(
+		struct ba_transport_pcm *pcm) {
+
+	enum ba_transport_pcm_signal signal;
+	ssize_t ret;
+
+	while ((ret = read(pcm->pipe[0], &signal, sizeof(signal))) == -1 &&
+			errno == EINTR)
+		continue;
+
+	if (ret == sizeof(signal))
+		return signal;
+
+	warn("Couldn't read transport PCM signal: %s", strerror(errno));
+	return -1;
+}
+
 bool ba_transport_pcm_is_active(const struct ba_transport_pcm *pcm) {
 	pthread_mutex_lock(MUTABLE(&pcm->mutex));
-	bool active = pcm->fd != -1 && pcm->active;
+	bool active = pcm->fd != -1 && !pcm->paused;
 	pthread_mutex_unlock(MUTABLE(&pcm->mutex));
 	return active;
 }

src/ba-transport-pcm.h

@@ -56,6 +56,15 @@ struct ba_transport_pcm_volume {
 	double scale;
 };
 
+enum ba_transport_pcm_signal {
+	BA_TRANSPORT_PCM_SIGNAL_OPEN,
+	BA_TRANSPORT_PCM_SIGNAL_CLOSE,
+	BA_TRANSPORT_PCM_SIGNAL_PAUSE,
+	BA_TRANSPORT_PCM_SIGNAL_RESUME,
+	BA_TRANSPORT_PCM_SIGNAL_SYNC,
+	BA_TRANSPORT_PCM_SIGNAL_DROP,
+};
+
 struct ba_transport_thread;
 
 struct ba_transport_pcm {
@@ -67,6 +76,8 @@ struct ba_transport_pcm {
 
 	/* PCM stream operation mode */
 	enum ba_transport_pcm_mode mode;
+	/* indicates a master PCM */
+	bool master;
 
 	/* guard PCM data updates */
 	pthread_mutex_t mutex;
@@ -78,8 +89,8 @@ struct ba_transport_pcm {
 	/* clone of BT socket */
 	int fd_bt;
 
-	/* indicates whether PCM shall be active */
-	bool active;
+	/* indicates whether PCM is running */
+	bool paused;
 
 	/* 16-bit stream format identifier */
 	uint16_t format;
@@ -111,6 +122,9 @@ struct ba_transport_pcm {
 	/* new PCM client mutex */
 	pthread_mutex_t client_mtx;
 
+	/* notification PIPE */
+	int pipe[2];
+
 	/* exported PCM D-Bus API */
 	char *ba_dbus_path;
 	bool ba_dbus_exported;
@@ -120,7 +134,8 @@ struct ba_transport_pcm {
 int transport_pcm_init(
 		struct ba_transport_pcm *pcm,
 		enum ba_transport_pcm_mode mode,
-		struct ba_transport_thread *th);
+		struct ba_transport_thread *th,
+		bool master);
 void transport_pcm_free(
 		struct ba_transport_pcm *pcm);
 
@@ -142,8 +157,7 @@ int ba_transport_pcm_bt_release(struct ba_transport_pcm *pcm);
 int ba_transport_pcm_start(
 		struct ba_transport_pcm *pcm,
 		ba_transport_pcm_thread_func th_func,
-		const char *name,
-		bool master);
+		const char *name);
 void ba_transport_pcm_stop(
 		struct ba_transport_pcm *pcm);
 
@@ -154,6 +168,12 @@ int ba_transport_pcm_resume(struct ba_transport_pcm *pcm);
 int ba_transport_pcm_drain(struct ba_transport_pcm *pcm);
 int ba_transport_pcm_drop(struct ba_transport_pcm *pcm);
 
+int ba_transport_pcm_signal_send(
+		struct ba_transport_pcm *pcm,
+		enum ba_transport_pcm_signal signal);
+enum ba_transport_pcm_signal ba_transport_pcm_signal_recv(
+		struct ba_transport_pcm *pcm);
+
 bool ba_transport_pcm_is_active(const struct ba_transport_pcm *pcm);
 
 int ba_transport_pcm_volume_level_to_range(int value, int max);