libobs: Fix pulseaudio monitoring, once and for all

This makes multiple changes to resolve issues with monitoring. We have
added a number of things over time to resolve various issues due to our
misunderstanding of the pulse api and the explicitly wrong pulse api
documentation.

First is removing the prior stream write tracking. Write tracking
was incorrect as pulse sends you the total number of bytes it wants
written at that time and not a incremental, but also this write tracking
is unhelpful since we can just let begin_write signal when we should
stop writing.

Second removing underflow tracking, this sort of worked to detect high
latency sources like vlc which buffer ~1s of data before writing audio
data. It worked because it would grow the buffer in PA until we could
write enough data into it, but it didn't allow us to leverage that
larger buffer because we never paused the PA stream.

To solve the buffering/high latency issue instead start the PA stream
corked and only uncork it once we have enough data to reach the targeted
latency. We also check how much data we recieve from the source and if
it appears to be high latency providing us with much more data than our
target latency we increase the latency to match the source. Fixing the
VLC source issue and resulting in a smooth start to audio instead of
underruns while we play and increase latency.

Finally add some handling for pulses incorrectly documented API. Notably
that begin_write may request more or less data from you so handle both
cases. And stop attempting to use attr.maxlength which is not updated
after set_buffer_attr calls and was always -1, and tlength is just more
appropriate.

old mantis resulting in some changes, still fixed. https://obsproject.com/mantis/view.php?id=1076
fixes #7574

libobs/audio-monitoring/pulse/pulseaudio-output.c

@@ -20,7 +20,6 @@ struct audio_monitor {
 
 	struct circlebuf new_data;
 	audio_resampler_t *resampler;
-	size_t bytesRemaining;
 
 	bool ignore;
 	pthread_mutex_t playback_mutex;
@@ -190,26 +189,51 @@ static void do_stream_write(void *param)
 	PULSE_DATA(param);
 	uint8_t *buffer = NULL;
 
-	while (data->new_data.size > 0 && data->bytesRemaining > 0) {
-		size_t bytesToFill = data->new_data.size;
-		if (data->bytesRemaining < bytesToFill)
-			bytesToFill = data->bytesRemaining;
+	pulseaudio_lock();
+	pthread_mutex_lock(&data->playback_mutex);
 
-		pulseaudio_lock();
+	// If we have grown a large buffer internally, grow the pulse buffer to match so we can write our data out.
+	if (data->new_data.size > data->attr.tlength * 2) {
+		data->attr.fragsize = (uint32_t)-1;
+		data->attr.maxlength = (uint32_t)-1;
+		data->attr.prebuf = (uint32_t)-1;
+		data->attr.minreq = (uint32_t)-1;
+		data->attr.tlength = data->new_data.size;
+		pa_stream_set_buffer_attr(data->stream, &data->attr, NULL,
+					  NULL);
+	}
+
+	// Buffer up enough data before we start playing.
+	if (pa_stream_is_corked(data->stream)) {
+		if (data->new_data.size >= data->attr.tlength) {
+			pa_stream_cork(data->stream, 0, NULL, NULL);
+		} else {
+			goto finish;
+		}
+	}
+
+	while (data->new_data.size > 0) {
+		size_t bytesToFill = data->new_data.size;
 		if (pa_stream_begin_write(data->stream, (void **)&buffer,
-					  &bytesToFill)) {
-			pulseaudio_unlock();
-			return;
+					  &bytesToFill))
+			goto finish;
+
+		// PA may request we submit more or less data than we have.
+		// Wait for more data if we cannot perform a full write.
+		if (bytesToFill > data->new_data.size) {
+			pa_stream_cancel_write(data->stream);
+			goto finish;
 		}
 
 		circlebuf_pop_front(&data->new_data, buffer, bytesToFill);
 
 		pa_stream_write(data->stream, buffer, bytesToFill, NULL, 0LL,
 				PA_SEEK_RELATIVE);
-		pulseaudio_unlock();
-
-		data->bytesRemaining -= bytesToFill;
 	}
+
+finish:
+	pthread_mutex_unlock(&data->playback_mutex);
+	pulseaudio_unlock();
 }
 
 static void on_audio_playback(void *param, obs_source_t *source,
@@ -258,50 +282,6 @@ unlock:
 	do_stream_write(param);
 }
 
-static void pulseaudio_stream_write(pa_stream *p, size_t nbytes, void *userdata)
-{
-	UNUSED_PARAMETER(p);
-	PULSE_DATA(userdata);
-
-	pthread_mutex_lock(&data->playback_mutex);
-	data->bytesRemaining += nbytes;
-	pthread_mutex_unlock(&data->playback_mutex);
-
-	pulseaudio_signal(0);
-}
-
-static void pulseaudio_underflow(pa_stream *p, void *userdata)
-{
-	UNUSED_PARAMETER(p);
-	PULSE_DATA(userdata);
-
-	pa_sample_spec spec = {0};
-	spec.format = data->format;
-	spec.rate = (uint32_t)data->samples_per_sec;
-	spec.channels = data->channels;
-	uint64_t latency = pa_bytes_to_usec(data->attr.tlength, &spec);
-
-	pthread_mutex_lock(&data->playback_mutex);
-	if (obs_source_active(data->source) && latency < 1000000) {
-		data->attr.fragsize = (uint32_t)-1;
-		data->attr.maxlength = (uint32_t)-1;
-		data->attr.prebuf = (uint32_t)-1;
-		data->attr.minreq = (uint32_t)-1;
-		data->attr.tlength = (data->attr.tlength * 3) / 2;
-		pa_stream_set_buffer_attr(data->stream, &data->attr, NULL,
-					  NULL);
-		data->bytesRemaining = data->attr.maxlength;
-	}
-	pthread_mutex_unlock(&data->playback_mutex);
-
-	if (latency >= 1000000) {
-		blog(LOG_WARNING, "source monitor reached max latency %ldms",
-		     latency / 1000);
-	}
-
-	pulseaudio_signal(0);
-}
-
 static void pulseaudio_server_info(pa_context *c, const pa_server_info *i,
 				   void *userdata)
 {
@@ -371,7 +351,6 @@ static void pulseaudio_stop_playback(struct audio_monitor *monitor)
 		/* Remove the callbacks, to ensure we no longer try to do anything
 		 * with this stream object */
 		pulseaudio_write_callback(monitor->stream, NULL, NULL);
-		pulseaudio_set_underflow_callback(monitor->stream, NULL, NULL);
 
 		/* Unreference the stream and drop it. PA will free it when it can. */
 		pulseaudio_lock();
@@ -489,13 +468,8 @@ static bool audio_monitor_init(struct audio_monitor *monitor,
 	monitor->attr.tlength = pa_usec_to_bytes(25000, &spec);
 
 	pa_stream_flags_t flags = PA_STREAM_INTERPOLATE_TIMING |
-				  PA_STREAM_AUTO_TIMING_UPDATE;
-
-	if (pthread_mutex_init(&monitor->playback_mutex, NULL) != 0) {
-		blog(LOG_WARNING, "%s: %s", __FUNCTION__,
-		     "Failed to init mutex");
-		return false;
-	}
+				  PA_STREAM_AUTO_TIMING_UPDATE |
+				  PA_STREAM_START_CORKED;
 
 	int_fast32_t ret = pulseaudio_connect_playback(
 		monitor->stream, monitor->device, &monitor->attr, flags);
@@ -505,8 +479,6 @@ static bool audio_monitor_init(struct audio_monitor *monitor,
 		return false;
 	}
 
-	monitor->bytesRemaining = monitor->attr.maxlength;
-
 	blog(LOG_INFO, "Started Monitoring in '%s'", monitor->device);
 	return true;
 }
@@ -518,12 +490,6 @@ static void audio_monitor_init_final(struct audio_monitor *monitor)
 
 	obs_source_add_audio_capture_callback(monitor->source,
 					      on_audio_playback, monitor);
-
-	pulseaudio_write_callback(monitor->stream, pulseaudio_stream_write,
-				  (void *)monitor);
-
-	pulseaudio_set_underflow_callback(monitor->stream, pulseaudio_underflow,
-					  (void *)monitor);
 }
 
 static inline void audio_monitor_free(struct audio_monitor *monitor)